feat(blob): support reading map blob values

src/paimon/common/data/blob_utils.cpp

@@ -108,6 +108,26 @@ bool BlobUtils::IsBlobField(const std::shared_ptr<arrow::Field>& field) {
     return IsBlobMetadata(field->metadata());
 }
 
+bool BlobUtils::IsMapBlobField(const std::shared_ptr<arrow::Field>& field) {
+    if (field == nullptr || field->type()->id() != arrow::Type::MAP) {
+        return false;
+    }
+    const auto& map_type = checked_cast<const arrow::MapType&>(*field->type());
+    // Arrow's C schema bridge does not retain nested field metadata for MapType. Paimon's
+    // ordinary binary type is BINARY, so LARGE_BINARY uniquely identifies a BLOB value here.
+    return map_type.item_type()->id() == arrow::Type::LARGE_BINARY;
+}
+
+Status BlobUtils::ValidateMapBlobWriteSchema(const std::shared_ptr<arrow::Schema>& schema) {
+    for (const auto& field : schema->fields()) {
+        if (IsMapBlobField(field)) {
+            return Status::NotImplemented(
+                "Writing a table with MAP<..., BLOB> is not supported by the C++ writer.");
+        }
+    }
+    return Status::OK();
+}
+
 bool BlobUtils::IsBlobMetadata(const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) {
     if (!metadata) {
         return false;

src/paimon/common/data/blob_utils.h

@@ -73,6 +73,10 @@ class PAIMON_EXPORT BlobUtils {
         const std::set<std::string>& inline_fields);
 
     static bool IsBlobField(const std::shared_ptr<arrow::Field>& field);
+    /// Returns whether the field is a top-level MAP whose values are BLOBs.
+    static bool IsMapBlobField(const std::shared_ptr<arrow::Field>& field);
+    /// Rejects schemas that the C++ writer cannot safely mutate.
+    static Status ValidateMapBlobWriteSchema(const std::shared_ptr<arrow::Schema>& schema);
     static bool IsBlobMetadata(const std::shared_ptr<const arrow::KeyValueMetadata>& metadata);
     static bool IsBlobFile(const std::string& file_name);
 

src/paimon/common/reader/blob_fallback_batch_reader.cpp

@@ -56,7 +56,8 @@ Result<std::unique_ptr<BlobFallbackBatchReader>> BlobFallbackBatchReader::Create
     }
     int32_t blob_field_idx = -1;
     for (int32_t i = 0; i < read_schema->num_fields(); i++) {
-        if (BlobUtils::IsBlobField(read_schema->field(i))) {
+        if (BlobUtils::IsBlobField(read_schema->field(i)) ||
+            BlobUtils::IsMapBlobField(read_schema->field(i))) {
             if (blob_field_idx != -1) {
                 return Status::Invalid(
                     "Blob fallback read schema should contain exactly one blob field.");
@@ -193,10 +194,29 @@ Result<std::vector<bool>> BlobFallbackBatchReader::ComputePlaceholderFlags(
             std::fill(flags.begin() + pos, flags.begin() + pos + chunk.length, true);
         } else {
             std::shared_ptr<arrow::Array> blob_col = chunk.array->field(blob_field_idx_);
-            if (!blob_col || blob_col->type_id() != arrow::Type::LARGE_BINARY) {
-                return Status::Invalid(fmt::format(
-                    "Blob fallback expects the blob column to be large binary, but got {}",
-                    blob_col ? blob_col->type()->ToString() : "null"));
+            if (!blob_col) {
+                return Status::Invalid("Blob fallback got a null blob column.");
+            }
+            if (blob_col->type_id() == arrow::Type::MAP) {
+                auto map_col = checked_pointer_cast<arrow::MapArray>(blob_col);
+                const std::shared_ptr<arrow::Array>& keys = map_col->keys();
+                const std::shared_ptr<arrow::Array>& items = map_col->items();
+                for (int64_t k = 0; k < chunk.length; k++) {
+                    int64_t idx = chunk.offset + k;
+                    if (!map_col->IsNull(idx) && map_col->value_length(idx) == 2) {
+                        int64_t entry_idx = map_col->value_offset(idx);
+                        flags[pos + k] =
+                            items->IsNull(entry_idx) && items->IsNull(entry_idx + 1) &&
+                            keys->RangeEquals(entry_idx, entry_idx + 1, entry_idx + 1, *keys);
+                    }
+                }
+                pos += chunk.length;
+                continue;
+            }
+            if (blob_col->type_id() != arrow::Type::LARGE_BINARY) {
+                return Status::Invalid(
+                    fmt::format("Blob fallback expects a BLOB or MAP<..., BLOB> column, but got {}",
+                                blob_col->type()->ToString()));
             }
             auto binary_col = checked_pointer_cast<arrow::LargeBinaryArray>(blob_col);
             for (int64_t k = 0; k < chunk.length; k++) {

src/paimon/common/reader/blob_fallback_batch_reader.h

@@ -54,9 +54,8 @@ namespace paimon {
 ///    vector has to reach every group the same way, through the file segments' readers and
 ///    through the row ids the caller leaves in a gap segment's `gap_selected_ranges`.
 /// 3. Each output row takes the first group, in max-sequence order, whose row is not a
-///    placeholder. Placeholder rows are identified by exact equality with the
-///    BlobDefs::kPlaceholderSentinel bytes, emitted by the blob format reader when
-///    BlobDefs::kEmitPlaceholderSentinelKey is set.
+///    placeholder. The blob format reader emits placeholders as BlobDefs::kPlaceholderSentinel
+///    bytes for scalar BLOBs, or as a two-entry map with duplicate keys for MAP<..., BLOB>.
 /// 4. A row that is a placeholder in every group degrades to a null blob: it keeps its
 ///    _ROW_ID, reports -1 as its _SEQUENCE_NUMBER, and returns null for every other field.
 class BlobFallbackBatchReader : public BatchReader {

src/paimon/core/append/append_compact_coordinator.cpp

@@ -25,6 +25,7 @@
 #include <vector>
 
 #include "paimon/common/data/binary_row.h"
+#include "paimon/common/data/blob_utils.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/linked_hash_map.h"
 #include "paimon/core/append/append_compact_task.h"
@@ -198,7 +199,9 @@ Result<std::pair<std::shared_ptr<TableSchema>, CoreOptions>> LoadSchemaAndOption
 
 /// Validate that the table is an append-only unaware-bucket table without DV.
 Status ValidateTable(const std::shared_ptr<TableSchema>& table_schema,
+                     const std::shared_ptr<arrow::Schema>& arrow_schema,
                      const CoreOptions& core_options) {
+    PAIMON_RETURN_NOT_OK(BlobUtils::ValidateMapBlobWriteSchema(arrow_schema));
     if (!table_schema->PrimaryKeys().empty() || core_options.GetBucket() != -1) {
         return Status::Invalid(
             "AppendCompactCoordinator only supports append-only tables "
@@ -320,12 +323,12 @@ Result<std::vector<std::shared_ptr<CommitMessage>>> AppendCompactCoordinator::Ru
     PAIMON_ASSIGN_OR_RAISE(schema_and_options,
                            LoadSchemaAndOptions(table_path, options, file_system));
     const auto& [table_schema, core_options] = schema_and_options;
+    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
 
     // Validate table type
-    PAIMON_RETURN_NOT_OK(ValidateTable(table_schema, core_options));
+    PAIMON_RETURN_NOT_OK(ValidateTable(table_schema, arrow_schema, core_options));
 
     // Build shared objects
-    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
     PAIMON_ASSIGN_OR_RAISE(
         std::shared_ptr<arrow::Schema> partition_schema,
         FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys()));

src/paimon/core/append/append_compact_coordinator_test.cpp

@@ -511,6 +511,31 @@ TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnDvTable) {
                         "not support for dv in UNAWARE_BUCKET mode");
 }
 
+TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnLoadedMapBlobTable) {
+    auto fs = dir_->GetFileSystem();
+    SchemaManager schema_manager(fs, TablePath());
+    std::string schema_json = R"json({
+        "version" : 3,
+        "id" : 0,
+        "fields" : [ {
+            "id" : 0,
+            "name" : "blob_map",
+            "type" : {"type":"MAP", "key":"STRING", "value":"BLOB"}
+        } ],
+        "highestFieldId" : 0,
+        "partitionKeys" : [],
+        "primaryKeys" : [],
+        "options" : {"bucket":"-1"},
+        "timeMillis" : 1721614341162
+    })json";
+    ASSERT_OK(fs->AtomicStore(PathUtil::JoinPath(schema_manager.SchemaDirectory(), "schema-0"),
+                              schema_json));
+
+    ASSERT_NOK_WITH_MSG(
+        AppendCompactCoordinator::Run(TablePath(), /*options=*/{}, /*partitions=*/{}, fs, pool_),
+        "Writing a table with MAP<..., BLOB> is not supported by the C++ writer");
+}
+
 /// Test that compact output files are written to external path when configured.
 TEST_F(AppendCompactCoordinatorTest, TestCompactWithExternalPath) {
     auto external_dir = UniqueTestDirectory::Create("local");

src/paimon/core/operation/file_store_write.cpp

@@ -23,6 +23,7 @@
 #include <utility>
 
 #include "fmt/format.h"
+#include "paimon/common/data/blob_utils.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/fields_comparator.h"
 #include "paimon/core/core_options.h"
@@ -109,14 +110,15 @@ Result<std::unique_ptr<FileStoreWrite>> FileStoreWrite::Create(std::unique_ptr<W
         return Status::Invalid(fmt::format("cannot found latest schema in branch {}", branch));
     }
     const auto& schema = table_schema.value();
+    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(schema->Fields());
+    PAIMON_RETURN_NOT_OK(BlobUtils::ValidateMapBlobWriteSchema(arrow_schema));
     auto opts = schema->Options();
     for (const auto& [key, value] : ctx->GetOptions()) {
         opts[key] = value;
     }
     PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
                            CoreOptions::FromMap(opts, ctx->GetSpecificFileSystem(),
                                                 ctx->GetFileSystemSchemeToIdentifierMap()));
-    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(schema->Fields());
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> partition_schema,
                            FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys()));
 

src/paimon/core/operation/file_store_write_test.cpp

@@ -75,6 +75,34 @@ TEST(FileStoreWriteTest, TestCreateAppendTable) {
                          FileStoreWrite::Create(std::move(write_context)));
 }
 
+TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) {
+    auto dir = UniqueTestDirectory::Create();
+    std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
+    auto fs = std::make_shared<LocalFileSystem>();
+    SchemaManager schema_manager(fs, table_path);
+    std::string schema_json = R"json({
+        "version" : 3,
+        "id" : 0,
+        "fields" : [ {
+            "id" : 0,
+            "name" : "blob_map",
+            "type" : {"type":"MAP", "key":"STRING", "value":"BLOB"}
+        } ],
+        "highestFieldId" : 0,
+        "partitionKeys" : [],
+        "primaryKeys" : [],
+        "options" : {},
+        "timeMillis" : 1721614341162
+    })json";
+    ASSERT_OK(fs->AtomicStore(PathUtil::JoinPath(schema_manager.SchemaDirectory(), "schema-0"),
+                              schema_json));
+
+    WriteContextBuilder context_builder(table_path, "commit_user_1");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context, context_builder.Finish());
+    ASSERT_NOK_WITH_MSG(FileStoreWrite::Create(std::move(write_context)),
+                        "Writing a table with MAP<..., BLOB> is not supported by the C++ writer");
+}
+
 TEST(FileStoreWriteTest, TestCreateAppendTableWithInvalidBucket) {
     auto dir = UniqueTestDirectory::Create();
     arrow::FieldVector fields = {

src/paimon/core/schema/arrow_schema_validator.cpp

@@ -164,8 +164,10 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
             const auto& item_field = checked_cast<arrow::MapType*>(type.get())->item_field();
             PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
                 key_field->type(), key_field->metadata(), /*allow_blob=*/false, field_id_set));
+            bool allow_direct_blob_item =
+                allow_blob && item_field->type()->id() == arrow::Type::LARGE_BINARY;
             PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
-                item_field->type(), item_field->metadata(), /*allow_blob=*/false, field_id_set));
+                item_field->type(), item_field->metadata(), allow_direct_blob_item, field_id_set));
             break;
         }
         case arrow::Type::type::LARGE_BINARY: {
@@ -248,7 +250,9 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr<arrow::Field>&
             const auto& item_field =
                 checked_cast<const arrow::MapType&>(*field->type()).item_field();
             PAIMON_RETURN_NOT_OK(ValidateField(key_field, /*allow_blob=*/false));
-            PAIMON_RETURN_NOT_OK(ValidateField(item_field, /*allow_blob=*/false));
+            bool allow_direct_blob_item =
+                allow_blob && item_field->type()->id() == arrow::Type::LARGE_BINARY;
+            PAIMON_RETURN_NOT_OK(ValidateField(item_field, allow_direct_blob_item));
             break;
         }
         case arrow::Type::type::LARGE_BINARY: {

src/paimon/core/schema/arrow_schema_validator_test.cpp

@@ -195,6 +195,16 @@ TEST(ArrowSchemaValidatorTest, TestBlobFieldMustBeTopLevel) {
         ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
                             "Blob field must be a top-level field.");
     }
+    {
+        auto map_blob_field = arrow::field(
+            "map_blob", arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", true)));
+        auto arrow_schema = arrow::schema(arrow::FieldVector({map_blob_field}));
+        ASSERT_OK(ArrowSchemaValidator::ValidateSchema(*arrow_schema));
+
+        std::vector<DataField> fields = {DataField(0, map_blob_field)};
+        arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+        ASSERT_OK(ArrowSchemaValidator::ValidateSchemaWithFieldId(*arrow_schema));
+    }
     {
         auto map_blob_field = arrow::field(
             "map_blob",
@@ -203,6 +213,15 @@ TEST(ArrowSchemaValidatorTest, TestBlobFieldMustBeTopLevel) {
         ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
                             "Blob field must be a top-level field.");
     }
+    {
+        auto nested_map_blob_field = arrow::field(
+            "nested_map_blob",
+            arrow::map(arrow::utf8(),
+                       arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", true))));
+        auto arrow_schema = arrow::schema(arrow::FieldVector({nested_map_blob_field}));
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
+                            "Blob field must be a top-level field.");
+    }
     {
         std::vector<DataField> nested_fields = {
             DataField(1, BlobUtils::ToArrowField("blob", true))};

src/paimon/core/schema/schema_validation_test.cpp

@@ -25,6 +25,7 @@
 #include "gtest/gtest.h"
 #include "paimon/common/data/blob_utils.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
+#include "paimon/common/utils/checked_cast.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/defs.h"
 #include "paimon/testing/utils/testharness.h"
@@ -1159,8 +1160,6 @@ TEST(SchemaValidationTest, TestMapSharedShreddingRequiresNonNullableKey) {
 }
 
 TEST(SchemaValidationTest, TestMapSharedShreddingRejectsBlobValue) {
-    auto direct_blob_map =
-        arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", /*nullable=*/true));
     auto nested_blob_map = arrow::map(
         arrow::utf8(), arrow::field("value", arrow::struct_({BlobUtils::ToArrowField("blob")})));
     std::map<std::string, std::string> options = {
@@ -1169,15 +1168,38 @@ TEST(SchemaValidationTest, TestMapSharedShreddingRejectsBlobValue) {
         {"fields.f1.map.storage-layout", "shared-shredding"},
     };
 
-    for (const auto& map_type : {direct_blob_map, nested_blob_map}) {
-        auto schema = arrow::schema({
-            arrow::field("f0", arrow::utf8()),
-            arrow::field("f1", map_type),
-        });
-        ASSERT_NOK_WITH_MSG(TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
-                                                /*primary_keys=*/{}, options),
-                            "Blob field must be a top-level field.");
-    }
+    const std::string loaded_schema = R"json({
+        "version": 3,
+        "id": 0,
+        "fields": [
+            {"id": 0, "name": "f0", "type": "STRING"},
+            {"id": 1, "name": "f1",
+             "type": {"type": "MAP", "key": "STRING", "value": "BLOB"}}
+        ],
+        "highestFieldId": 1,
+        "partitionKeys": [],
+        "primaryKeys": [],
+        "options": {
+            "bucket": "1",
+            "bucket-key": "f0",
+            "fields.f1.map.storage-layout": "shared-shredding"
+        },
+        "timeMillis": 0
+    })json";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                         TableSchema::CreateFromJson(loaded_schema));
+    auto loaded_map = checked_pointer_cast<arrow::MapType>(table_schema->Fields()[1].Type());
+    ASSERT_TRUE(BlobUtils::IsBlobField(loaded_map->item_field()));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "MAP shared-shredding currently cannot contain BLOB fields.");
+
+    auto nested_schema = arrow::schema({
+        arrow::field("f0", arrow::utf8()),
+        arrow::field("f1", nested_blob_map),
+    });
+    ASSERT_NOK_WITH_MSG(TableSchema::Create(/*schema_id=*/0, nested_schema,
+                                            /*partition_keys=*/{}, /*primary_keys=*/{}, options),
+                        "Blob field must be a top-level field.");
 }
 
 TEST(SchemaValidationTest, TestMapSharedShreddingCompression) {

src/paimon/core/schema/table_schema.cpp

@@ -27,6 +27,7 @@
 #include "arrow/api.h"
 #include "arrow/c/bridge.h"
 #include "fmt/format.h"
+#include "paimon/common/data/blob_utils.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -56,6 +57,7 @@ Result<std::unique_ptr<TableSchema>> TableSchema::Create(
     for (const auto& primary_key : primary_keys) {
         primary_key_set.insert(primary_key);
     }
+    PAIMON_RETURN_NOT_OK(BlobUtils::ValidateMapBlobWriteSchema(schema));
     for (const auto& field : schema->fields()) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Field> field_with_id,
                                AssignFieldIdsRecursively(field, /*set_field_id=*/true, &field_id));