feat(blob): support writing MAP<K, BLOB> fields

Write each MAP<K, BLOB> row as the nested payload of Java's
MapBlobElementSerializer, except that a map of exactly two entries with
equal keys and null values is persisted as a placeholder entry, so that
a data-evolution partial update keeps the older value of the row. For
any other map, the keys are serialized before any byte of the entry is
written and checked to be unique, as in Java, and non-null, since Arrow
map keys cannot hold the one null key Java allows. A decimal key must
also fit its precision and a string key must be valid UTF-8, so that
the C++ reader can read the key back. Each value is copied like an
ARRAY<BLOB> element, from raw bytes or a serialized BlobDescriptor, and
the blob-write-null-on-* options write NULL for an unreachable value.

Allow creating and writing tables with MAP<K, BLOB> columns whose key
type is BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, DATE, DECIMAL, CHAR,
VARCHAR, BINARY or VARBINARY; TIME keys remain unsupported. Restore the
BLOB marker on map values when creating a table schema, as Arrow's C
schema bridge drops it.

Author: SteNicholas

docs/source/user_guide/data_types.rst

@@ -294,9 +294,13 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
    * - ``BLOB``
 
        ``ARRAY<BLOB>``
+
+       ``MAP<kt, BLOB>``
      - LargeBinary
 
        List<LargeBinary>
+
+       Map<kt, LargeBinary>
      - Data type of a binary large object, such as an image or a video, stored
        in dedicated ``.blob`` files instead of the normal data files. A BLOB
        column listed in ``blob-descriptor-field`` or ``blob-view-field`` instead
@@ -308,12 +312,17 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        field. An ``ARRAY<BLOB>`` field is a top-level ``List`` field whose
        element field is a BLOB field, which stores an ordered collection of
        objects (e.g. the frames of one sample) in a single blob entry. Both the
-       array and its elements may be null.
-
-       Tables with BLOB or ``ARRAY<BLOB>`` columns must enable
-       ``row-tracking.enabled`` and ``data-evolution.enabled``, and must have
-       at least one non-BLOB column. These columns cannot be partition keys.
-       ``ARRAY<BLOB>`` cannot be nested inside other types or listed in
-       ``blob-descriptor-field`` or ``blob-view-field``. Paimon C++ does not
-       support compacting these tables yet, and can read but not write
-       ``MAP<kt, BLOB>`` columns. See :doc:`write` for writing BLOB columns.
+       array and its elements may be null. A ``MAP<kt, BLOB>`` field is a
+       top-level ``Map`` field whose values are ``LargeBinary``, which stores
+       keyed objects (e.g. the views of one sample) in a single blob entry. Both
+       the map and its values may be null, while its keys must be unique and
+       non-null. ``kt`` must be BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, DATE,
+       DECIMAL, CHAR, VARCHAR, BINARY or VARBINARY.
+
+       Tables with BLOB, ``ARRAY<BLOB>`` or ``MAP<kt, BLOB>`` columns must
+       enable ``row-tracking.enabled`` and ``data-evolution.enabled``, and must
+       have at least one non-BLOB column. These columns cannot be partition
+       keys. ``ARRAY<BLOB>`` and ``MAP<kt, BLOB>`` cannot be nested inside other
+       types or listed in ``blob-descriptor-field`` or ``blob-view-field``.
+       Paimon C++ does not support compacting these tables yet. See :doc:`write`
+       for writing BLOB columns.

docs/source/user_guide/write.rst

@@ -75,10 +75,11 @@ Writing BLOB Columns
 ~~~~~~~~~~~~~~~~~~~~
 
 A ``BLOB`` column is a ``LargeBinary`` field carrying Paimon's BLOB field
-metadata, and an ``ARRAY<BLOB>`` column is a top-level ``List`` field whose
-element field carries it. Build the BLOB field with ``paimon::Blob::ArrowField``
-and import it into Arrow; the element field of an ``ARRAY<BLOB>`` column must
-keep that metadata:
+metadata, an ``ARRAY<BLOB>`` column is a top-level ``List`` field whose element
+field carries it, and a ``MAP<kt, BLOB>`` column is a top-level ``Map`` field
+whose values are ``LargeBinary``. Build the BLOB field with
+``paimon::Blob::ArrowField`` and import it into Arrow; the element field of an
+``ARRAY<BLOB>`` column must keep that metadata:
 
 .. code-block:: cpp
 
@@ -87,26 +88,31 @@ keep that metadata:
    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Field> element,
                                      arrow::ImportField(c_element.get()));
    std::shared_ptr<arrow::Schema> schema = arrow::schema(
-       {arrow::field("id", arrow::int32()), arrow::field("frames", arrow::list(element))});
+       {arrow::field("id", arrow::int32()), arrow::field("frames", arrow::list(element)),
+        arrow::field("views", arrow::map(arrow::utf8(), element))});
 
-For a column stored in blob files, which includes every ``ARRAY<BLOB>`` column,
-each value or element holds either the raw bytes or a serialized
-``paimon::BlobDescriptor`` produced by ``paimon::Blob::ToDescriptor``; the writer
-copies the referenced data into the blob file. A BLOB column listed in
-``blob-descriptor-field`` or ``blob-view-field`` keeps the reference in the data
-file instead, so the referenced data must remain available.
+For a column stored in blob files, which includes every ``ARRAY<BLOB>`` and
+``MAP<kt, BLOB>`` column, each value, element or map value holds either the raw
+bytes or a serialized ``paimon::BlobDescriptor`` produced by
+``paimon::Blob::ToDescriptor``; the writer copies the referenced data into the
+blob file. A BLOB column listed in ``blob-descriptor-field`` or
+``blob-view-field`` keeps the reference in the data file instead, so the
+referenced data must remain available.
 
 If the referenced data cannot be reached, the write fails unless a write-null
 option covers the failure: ``blob-write-null-on-missing-file`` covers a
 referenced file that does not exist, and ``blob-write-null-on-fetch-failure``
 covers any other failure to resolve the descriptor or open the data, including
 a missing file when the former is disabled and an offset past the end of the
 file for a descriptor with a dynamic length (``-1``). A covered value is written
-as NULL; in an ``ARRAY<BLOB>`` only that element becomes NULL, not the array.
+as NULL; in an ``ARRAY<BLOB>`` or a ``MAP<kt, BLOB>`` only that element or map
+value becomes NULL, not the array or map.
 Any other failure fails the write, such as a failure to write the blob file or
 to close a referenced file. As in Paimon Java, this includes a file too short
 for the range of a descriptor with a known length, which is only detected while
 the data is copied.
+Except for the placeholder marker defined below, a ``MAP<kt, BLOB>`` value with
+a repeated or null key also fails the write, before any of its data is written.
 See :doc:`data_types` for the table requirements and restrictions.
 
 A data-evolution write can update columns of existing rows. The write does not
@@ -125,20 +131,40 @@ payload was serialized with.
 In such an update, a row whose BLOB or ``ARRAY<BLOB>`` value stays unchanged is
 marked with the reserved bytes ``_PAIMON_BLOB_PLACEHOLDER``: as the value itself
 for a BLOB column, or as the only element of the array for an ``ARRAY<BLOB>``
-column. As in Paimon Java, the marker works whatever other columns the update
-carries. Such a row keeps its value from the older files when read. Every write
-stores a value equal to the reserved bytes as such a marker, so that value is
-not supported.
+column. Every write stores a value equal to the reserved bytes as such a marker,
+so that value is not supported. A row whose ``MAP<kt, BLOB>`` value stays
+unchanged is marked with a map of exactly two entries with equal keys and null
+values, which no other map can hold since its keys must be unique. As in Paimon
+Java, a marker works whatever other columns the update carries, and its row
+keeps its value from the older files when read.
+
+Append the two entries of a ``MAP<kt, BLOB>`` marker one by one, for example
+with ``arrow::MapBuilder``, since a JSON object or a dictionary merges equal
+keys into one entry:
+
+.. code-block:: cpp
+
+   // map_builder builds the MAP<STRING, BLOB> column of the update.
+   auto* keys = static_cast<arrow::StringBuilder*>(map_builder->key_builder());
+   auto* values = static_cast<arrow::LargeBinaryBuilder*>(map_builder->item_builder());
+   ARROW_RETURN_NOT_OK(map_builder->Append());
+   for (int32_t i = 0; i < 2; ++i) {
+       ARROW_RETURN_NOT_OK(keys->Append("k"));
+       ARROW_RETURN_NOT_OK(values->AppendNull());
+   }
 
 .. note::
    The C++ writer differs from Paimon Java in these respects:
 
-   - A placeholder is identified by the reserved bytes; Java uses a dedicated
-     placeholder object, which cannot collide with a user value.
+   - A placeholder is identified by the reserved bytes, which can collide with a
+     user value, or by the two-entry map marker above; Java uses a dedicated
+     placeholder object.
+   - A ``MAP<kt, BLOB>`` key cannot be null, as Arrow map keys are not
+     nullable, and cannot be TIME; Java allows one null key and TIME keys.
    - A missing file is detected with ``FileSystem::Exists``. Java detects a
-     missing file for an ``ARRAY<BLOB>`` element only from an HTTP 404, and
-     does not write NULL for a 404 under ``blob-write-null-on-fetch-failure``
-     alone.
+     missing file for an ``ARRAY<BLOB>`` element or a ``MAP<kt, BLOB>`` value
+     only from an HTTP 404, and does not write NULL for a 404 under
+     ``blob-write-null-on-fetch-failure`` alone.
    - A descriptor with a dynamic length is copied up to the file length read
      when it is opened, so data appended to the file during the copy is left
      out, and a file truncated during the copy fails the write. Java reads it

include/paimon/defs.h

@@ -574,17 +574,19 @@ struct PAIMON_EXPORT Options {
     /// path and requires manual configuration by the user. No default value.
     static const char BLOB_VIEW_UPSTREAM_WAREHOUSE[];
     /// "blob-write-null-on-missing-file" - Whether to write NULL for a descriptor BLOB value,
-    /// including an ARRAY<BLOB> element, when the referenced file does not exist at write time.
+    /// including an ARRAY<BLOB> element or a MAP<..., BLOB> value, when the referenced file does
+    /// not exist at write time.
     /// When false, a missing file is treated like any other fetch failure, following
     /// "blob-write-null-on-fetch-failure". Default value is "false".
     static const char BLOB_WRITE_NULL_ON_MISSING_FILE[];
     /// "blob-write-null-on-fetch-failure" - Whether to write NULL for a descriptor BLOB value,
-    /// including an ARRAY<BLOB> element, when the referenced data cannot be reached at write time,
-    /// i.e. the descriptor cannot be deserialized or the data cannot be opened (e.g. the offset of
-    /// a descriptor with dynamic length -1 is past the end of the file). A missing file is handled
-    /// by "blob-write-null-on-missing-file" when that option is enabled. When false, such a
-    /// failure fails the write. A failure while copying the data, such as a read error or a file
-    /// too short for the range of a descriptor with a known length, always fails the write.
+    /// including an ARRAY<BLOB> element or a MAP<..., BLOB> value, when the referenced data cannot
+    /// be reached at write time, i.e. the descriptor cannot be deserialized or the data cannot be
+    /// opened (e.g. the offset of a descriptor with dynamic length -1 is past the end of the
+    /// file). A missing file is handled by "blob-write-null-on-missing-file" when that option is
+    /// enabled. When false, such a failure fails the write. A failure while copying the data, such
+    /// as a read error or a file too short for the range of a descriptor with a known length,
+    /// always fails the write.
     /// Default value is "false".
     static const char BLOB_WRITE_NULL_ON_FETCH_FAILURE[];
     /// "global-index.enabled" - Whether to enable global index for scan. Default value is "true".

src/paimon/common/data/blob_defs.h

@@ -76,6 +76,11 @@ class BlobDefs {
     /// merge resolves from an older layer, or degrades to a null blob when every layer holds a
     /// placeholder for the row. The marker is distinctive enough that these collisions are
     /// accepted as negligibly improbable.
+    ///
+    /// A MAP<..., BLOB> placeholder does not use these bytes: both channels mark it as a map of
+    /// exactly two entries with equal keys and null values (see BlobUtils::IsMapBlobPlaceholder).
+    /// The writer rejects duplicate keys in any other map, so that marker never collides with a
+    /// user value.
     static constexpr char kPlaceholderSentinel[] = "_PAIMON_BLOB_PLACEHOLDER";
     /// Byte length of kPlaceholderSentinel, excluding the literal's terminating NUL.
     static constexpr int32_t kPlaceholderSentinelLength = sizeof(kPlaceholderSentinel) - 1;
@@ -119,6 +124,10 @@ class BlobDefs {
     static constexpr int32_t kArrayBlobMagicNumber = 1094861634;
     /// ARRAY<BLOB> nested payload version.
     static constexpr int8_t kArrayBlobVersion = 1;
+    /// Magic number identifying the nested payload of a MAP<..., BLOB> entry.
+    static constexpr int32_t kMapBlobMagicNumber = 0x4D424342;
+    /// MAP<..., BLOB> nested payload version.
+    static constexpr int8_t kMapBlobVersion = 1;
 };
 
 }  // namespace paimon

src/paimon/common/data/blob_utils.cpp

@@ -166,11 +166,33 @@ bool BlobUtils::IsMapBlobPlaceholder(const arrow::MapArray& array, int64_t row)
     return keys->RangeEquals(entry_index, entry_index + 1, entry_index + 1, *keys);
 }
 
+bool BlobUtils::IsSupportedMapBlobKeyType(const arrow::DataType& key_type) {
+    switch (key_type.id()) {
+        case arrow::Type::BOOL:
+        case arrow::Type::INT8:
+        case arrow::Type::INT16:
+        case arrow::Type::INT32:
+        case arrow::Type::INT64:
+        case arrow::Type::DATE32:
+        case arrow::Type::DECIMAL128:
+        case arrow::Type::STRING:
+        case arrow::Type::BINARY:
+            return true;
+        default:
+            return false;
+    }
+}
+
 Status BlobUtils::ValidateContainerBlobWriteSchema(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.");
+        if (!IsMapBlobField(field)) {
+            continue;
+        }
+        const auto& map_type = checked_cast<const arrow::MapType&>(*field->type());
+        if (!IsSupportedMapBlobKeyType(*map_type.key_type())) {
+            return Status::NotImplemented(fmt::format(
+                "Writing a table with MAP<{}, BLOB> field {} is not supported by the C++ writer.",
+                map_type.key_type()->ToString(), field->name()));
         }
     }
     return Status::OK();

src/paimon/common/data/blob_utils.h

@@ -30,6 +30,7 @@
 #include "paimon/visibility.h"
 
 namespace arrow {
+class DataType;
 class Field;
 class KeyValueMetadata;
 class ListArray;
@@ -88,7 +89,10 @@ class PAIMON_EXPORT BlobUtils {
     static bool IsArrayBlobPlaceholder(const arrow::ListArray& array, int64_t row);
     /// Returns whether a MAP<..., BLOB> row is the internal fallback sentinel.
     static bool IsMapBlobPlaceholder(const arrow::MapArray& array, int64_t row);
-    /// Rejects MAP<..., BLOB>, which is currently supported by the C++ reader only.
+    /// Returns whether Paimon C++ supports `key_type` as the key type of MAP<..., BLOB>:
+    /// BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, DATE, DECIMAL, CHAR/VARCHAR and BINARY/VARBINARY.
+    static bool IsSupportedMapBlobKeyType(const arrow::DataType& key_type);
+    /// Rejects MAP<..., BLOB> fields whose key type the C++ blob writer does not support.
     static Status ValidateContainerBlobWriteSchema(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/data/blob_utils_test.cpp

@@ -100,9 +100,29 @@ TEST_F(BlobUtilsTest, ValidateContainerBlobWriteSchema) {
 
     auto map_blob_field =
         arrow::field("map_blob", arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", true)));
-    ASSERT_NOK_WITH_MSG(BlobUtils::ValidateContainerBlobWriteSchema(
-                            arrow::schema({array_blob_field, map_blob_field})),
-                        "Writing a table with MAP<..., BLOB> is not supported by the C++ writer");
+    ASSERT_OK(BlobUtils::ValidateContainerBlobWriteSchema(
+        arrow::schema({array_blob_field, map_blob_field})));
+
+    auto float_key_map_blob_field = arrow::field(
+        "float_map_blob", arrow::map(arrow::float32(), BlobUtils::ToArrowField("value", true)));
+    ASSERT_NOK_WITH_MSG(
+        BlobUtils::ValidateContainerBlobWriteSchema(
+            arrow::schema({map_blob_field, float_key_map_blob_field})),
+        "Writing a table with MAP<float, BLOB> field float_map_blob is not supported by the C++ "
+        "writer");
+}
+
+TEST_F(BlobUtilsTest, IsSupportedMapBlobKeyType) {
+    for (const auto& key_type : {arrow::boolean(), arrow::int8(), arrow::int16(), arrow::int32(),
+                                 arrow::int64(), arrow::date32(), arrow::decimal128(10, 2),
+                                 arrow::decimal128(38, 2), arrow::utf8(), arrow::binary()}) {
+        ASSERT_TRUE(BlobUtils::IsSupportedMapBlobKeyType(*key_type)) << key_type->ToString();
+    }
+    for (const auto& key_type :
+         {arrow::float32(), arrow::float64(), arrow::time32(arrow::TimeUnit::MILLI),
+          arrow::timestamp(arrow::TimeUnit::MICRO), arrow::large_binary(), arrow::large_utf8()}) {
+        ASSERT_FALSE(BlobUtils::IsSupportedMapBlobKeyType(*key_type)) << key_type->ToString();
+    }
 }
 
 TEST_F(BlobUtilsTest, IsArrayBlobPlaceholder) {

src/paimon/core/append/append_compact_coordinator_test.cpp

@@ -528,7 +528,7 @@ TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnDataEvolutionTable) {
                         "not support for data evolution in UNAWARE_BUCKET mode");
 }
 
-TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnArrayBlobTable) {
+TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnContainerBlobTable) {
     std::map<std::string, std::string> options = {{Options::FILE_FORMAT, "parquet"},
                                                   {Options::BUCKET, "-1"},
                                                   {Options::FILE_SYSTEM, "local"},
@@ -537,7 +537,9 @@ TEST_F(AppendCompactCoordinatorTest, TestValidateFailsOnArrayBlobTable) {
 
     arrow::FieldVector fields = {
         arrow::field("f0", arrow::int32()),
-        arrow::field("a0", arrow::list(BlobUtils::ToArrowField("item", /*nullable=*/true)))};
+        arrow::field("a0", arrow::list(BlobUtils::ToArrowField("item", /*nullable=*/true))),
+        arrow::field(
+            "m0", arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", /*nullable=*/true)))};
     CreateTable(fields, /*partition_keys=*/{}, options);
 
     ASSERT_NOK_WITH_MSG(AppendCompactCoordinator::Run(TablePath(), options, /*partitions=*/{},

src/paimon/core/operation/file_store_write_test.cpp

@@ -253,7 +253,7 @@ TEST(FileStoreWriteTest, TestCatalogWriteUsesPerTableFileSystem) {
     ASSERT_EQ(requests.back().GetFullName(), identifier.GetFullName());
 }
 
-TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) {
+TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTableWithUnsupportedKey) {
     auto dir = UniqueTestDirectory::Create();
     std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
     auto fs = std::make_shared<LocalFileSystem>();
@@ -264,7 +264,7 @@ TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) {
         "fields" : [ {
             "id" : 0,
             "name" : "blob_map",
-            "type" : {"type":"MAP", "key":"STRING", "value":"BLOB"}
+            "type" : {"type":"MAP", "key":"TIME(3)", "value":"BLOB"}
         } ],
         "highestFieldId" : 0,
         "partitionKeys" : [],
@@ -277,8 +277,10 @@ TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) {
 
     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");
+    ASSERT_NOK_WITH_MSG(
+        FileStoreWrite::Create(std::move(write_context)),
+        "Writing a table with MAP<time32[ms], BLOB> field blob_map is not supported by the C++ "
+        "writer");
 }
 
 TEST(FileStoreWriteTest, TestCreateAppendTableWithInvalidBucket) {