This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 3e9304a6 feat(blob): support writing MAP<K, BLOB> fields (#416)
3e9304a6 is described below
commit 3e9304a6236ff80738a8a930d88a3f563d474f95
Author: Nicholas Jiang <[email protected]>
AuthorDate: Thu Oct 8 14:09:11 2026 +0700
feat(blob): support writing MAP<K, BLOB> fields (#416)
---
docs/source/user_guide/data_types.rst | 27 +-
docs/source/user_guide/write.rst | 68 ++-
include/paimon/defs.h | 16 +-
src/paimon/common/data/blob_defs.h | 9 +
src/paimon/common/data/blob_utils.cpp | 21 +-
src/paimon/common/data/blob_utils.h | 6 +-
src/paimon/common/data/blob_utils_test.cpp | 22 +-
.../append/append_compact_coordinator_test.cpp | 6 +-
src/paimon/core/operation/file_store_write.cpp | 2 -
.../core/operation/file_store_write_test.cpp | 28 --
src/paimon/core/schema/schema_validation.cpp | 14 +
src/paimon/core/schema/schema_validation_test.cpp | 35 ++
src/paimon/core/schema/table_schema.cpp | 10 +-
src/paimon/core/schema/table_schema_test.cpp | 32 +-
src/paimon/format/blob/blob_file_batch_reader.cpp | 6 +-
src/paimon/format/blob/blob_format_writer.cpp | 223 +++++++++-
src/paimon/format/blob/blob_format_writer.h | 53 ++-
src/paimon/format/blob/blob_format_writer_test.cpp | 480 ++++++++++++++++++++-
test/inte/blob_table_inte_test.cpp | 127 ++++++
19 files changed, 1042 insertions(+), 143 deletions(-)
diff --git a/docs/source/user_guide/data_types.rst
b/docs/source/user_guide/data_types.rst
index e893d498..f864596d 100644
--- a/docs/source/user_guide/data_types.rst
+++ b/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.
diff --git a/docs/source/user_guide/write.rst b/docs/source/user_guide/write.rst
index 89f467db..4a3d27db 100644
--- a/docs/source/user_guide/write.rst
+++ b/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,14 +88,16 @@ 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
@@ -102,11 +105,14 @@ 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
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index bea13f61..d6abd047 100644
--- a/include/paimon/defs.h
+++ b/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".
diff --git a/src/paimon/common/data/blob_defs.h
b/src/paimon/common/data/blob_defs.h
index 68d396da..ce5932b4 100644
--- a/src/paimon/common/data/blob_defs.h
+++ b/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
diff --git a/src/paimon/common/data/blob_utils.cpp
b/src/paimon/common/data/blob_utils.cpp
index a601ee4b..fd12cbcd 100644
--- a/src/paimon/common/data/blob_utils.cpp
+++ b/src/paimon/common/data/blob_utils.cpp
@@ -166,14 +166,21 @@ bool BlobUtils::IsMapBlobPlaceholder(const
arrow::MapArray& array, int64_t row)
return keys->RangeEquals(entry_index, entry_index + 1, entry_index + 1,
*keys);
}
-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.");
- }
+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;
}
- return Status::OK();
}
bool BlobUtils::IsBlobMetadata(const std::shared_ptr<const
arrow::KeyValueMetadata>& metadata) {
diff --git a/src/paimon/common/data/blob_utils.h
b/src/paimon/common/data/blob_utils.h
index b7c6c6c2..8292c108 100644
--- a/src/paimon/common/data/blob_utils.h
+++ b/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,8 +89,9 @@ 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.
- static Status ValidateContainerBlobWriteSchema(const
std::shared_ptr<arrow::Schema>& schema);
+ /// 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);
static bool IsBlobMetadata(const std::shared_ptr<const
arrow::KeyValueMetadata>& metadata);
static bool IsBlobFile(const std::string& file_name);
diff --git a/src/paimon/common/data/blob_utils_test.cpp
b/src/paimon/common/data/blob_utils_test.cpp
index 9e989c86..b4d9a21b 100644
--- a/src/paimon/common/data/blob_utils_test.cpp
+++ b/src/paimon/common/data/blob_utils_test.cpp
@@ -92,17 +92,17 @@ TEST_F(BlobUtilsTest, IsArrayBlobField) {
arrow::field("nested",
arrow::list(arrow::list(BlobUtils::ToArrowField("item", true))))));
}
-TEST_F(BlobUtilsTest, ValidateContainerBlobWriteSchema) {
- auto array_blob_field =
- arrow::field("array_blob", arrow::list(BlobUtils::ToArrowField("item",
true)));
- ASSERT_OK(BlobUtils::ValidateContainerBlobWriteSchema(
- arrow::schema({arrow::field("id", arrow::int32()),
array_blob_field})));
-
- 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");
+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) {
diff --git a/src/paimon/core/append/append_compact_coordinator_test.cpp
b/src/paimon/core/append/append_compact_coordinator_test.cpp
index 57758ace..acdf59cb 100644
--- a/src/paimon/core/append/append_compact_coordinator_test.cpp
+++ b/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=*/{},
diff --git a/src/paimon/core/operation/file_store_write.cpp
b/src/paimon/core/operation/file_store_write.cpp
index 6056ba1e..2ae664f8 100644
--- a/src/paimon/core/operation/file_store_write.cpp
+++ b/src/paimon/core/operation/file_store_write.cpp
@@ -26,7 +26,6 @@
#include "fmt/format.h"
#include "paimon/catalog/catalog.h"
-#include "paimon/common/data/blob_utils.h"
#include "paimon/common/global_index/btree/btree_defs.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/fields_comparator.h"
@@ -239,7 +238,6 @@ Result<std::unique_ptr<FileStoreWrite>>
FileStoreWrite::Create(std::unique_ptr<W
}
const std::shared_ptr<TableSchema>& schema = latest_schema;
auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(schema->Fields());
-
PAIMON_RETURN_NOT_OK(BlobUtils::ValidateContainerBlobWriteSchema(arrow_schema));
auto opts = schema->Options();
for (const auto& [key, value] : ctx->GetOptions()) {
opts[key] = value;
diff --git a/src/paimon/core/operation/file_store_write_test.cpp
b/src/paimon/core/operation/file_store_write_test.cpp
index 3893d857..520e850e 100644
--- a/src/paimon/core/operation/file_store_write_test.cpp
+++ b/src/paimon/core/operation/file_store_write_test.cpp
@@ -253,34 +253,6 @@ TEST(FileStoreWriteTest,
TestCatalogWriteUsesPerTableFileSystem) {
ASSERT_EQ(requests.back().GetFullName(), identifier.GetFullName());
}
-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 = {
diff --git a/src/paimon/core/schema/schema_validation.cpp
b/src/paimon/core/schema/schema_validation.cpp
index 9a5b2983..a5992455 100644
--- a/src/paimon/core/schema/schema_validation.cpp
+++ b/src/paimon/core/schema/schema_validation.cpp
@@ -801,6 +801,20 @@ Status SchemaValidation::ValidateRowTracking(const
TableSchema& table_schema,
}
Status SchemaValidation::ValidateBlobFields(const TableSchema& schema, const
CoreOptions& options) {
+ // A MAP<..., BLOB> field goes to blob files by its type alone, so its key
type is checked
+ // whether or not a blob option names it.
+ for (const auto& field : schema.Fields()) {
+ if (!BlobUtils::IsMapBlobField(field.ArrowField())) {
+ continue;
+ }
+ const auto& map_type = checked_cast<const
arrow::MapType&>(*field.Type());
+ if (!BlobUtils::IsSupportedMapBlobKeyType(*map_type.key_type())) {
+ return Status::NotImplemented(
+ fmt::format("Field '{}' has unsupported MAP<..., BLOB> key
type: {}.", field.Name(),
+ map_type.key_type()->ToString()));
+ }
+ }
+
const auto& configured_blob_names = options.GetBlobFields();
const auto& blob_descriptor_names = options.GetBlobDescriptorFields();
const auto& blob_view_names = options.GetBlobViewFields();
diff --git a/src/paimon/core/schema/schema_validation_test.cpp
b/src/paimon/core/schema/schema_validation_test.cpp
index bccbad8e..f0823157 100644
--- a/src/paimon/core/schema/schema_validation_test.cpp
+++ b/src/paimon/core/schema/schema_validation_test.cpp
@@ -459,6 +459,41 @@ TEST(SchemaValidationTest, TestContainerBlobRowTracking) {
}
}
+TEST(SchemaValidationTest, TestMapBlobKeyType) {
+ const std::string options_json = R"({
+ "bucket": "-1",
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true"
+ })";
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TableSchema> table_schema,
+
MakeContainerBlobSchema(R"json({"type":"MAP","key":"STRING","value":"BLOB"})json",
"[]",
+ options_json));
+ ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+
+ // Checked without any blob option naming the field.
+ ASSERT_OK_AND_ASSIGN(
+ table_schema,
+
MakeContainerBlobSchema(R"json({"type":"MAP","key":"TIME(3)","value":"BLOB"})json",
"[]",
+ options_json));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "Field 'blob' has unsupported MAP<..., BLOB> key type:
time32[ms].");
+
+ std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_FIELD,
"blob_map"}};
+ auto schema = arrow::schema(
+ {arrow::field("id", arrow::int32()),
+ arrow::field("blob_map", arrow::map(arrow::float32(),
BlobUtils::ToArrowField(
+ "value",
/*nullable=*/true)))});
+ ASSERT_OK_AND_ASSIGN(table_schema,
+ TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{},
+ /*primary_keys=*/{}, options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "Field 'blob_map' has unsupported MAP<..., BLOB> key
type: float.");
+}
+
TEST(SchemaValidationTest, TestWithBlobField) {
auto f0 = arrow::field("f0", arrow::utf8());
auto f1 = arrow::field("f1", arrow::int32());
diff --git a/src/paimon/core/schema/table_schema.cpp
b/src/paimon/core/schema/table_schema.cpp
index 5e70ff5a..1d69c033 100644
--- a/src/paimon/core/schema/table_schema.cpp
+++ b/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_defs.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"
@@ -57,7 +58,6 @@ 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::ValidateContainerBlobWriteSchema(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));
@@ -132,6 +132,14 @@ Result<std::shared_ptr<arrow::Field>>
TableSchema::AssignFieldIdsRecursively(
auto map_type = checked_pointer_cast<arrow::MapType>(field->type());
std::shared_ptr<arrow::Field> key_field = map_type->key_field();
std::shared_ptr<arrow::Field> value_field = map_type->item_field();
+ // Arrow's C schema bridge drops the metadata of a map value field,
which marks a BLOB.
+ // Paimon BYTES is BINARY, so a LARGE_BINARY map value can only be a
BLOB (see
+ // BlobUtils::IsMapBlobField); restore the marker.
+ if (value_field->type()->id() == arrow::Type::LARGE_BINARY &&
+ !BlobUtils::IsBlobMetadata(value_field->metadata())) {
+ value_field =
value_field->WithMergedMetadata(arrow::KeyValueMetadata::Make(
+ {BlobDefs::kExtensionTypeKey},
{BlobDefs::kExtensionTypeValue}));
+ }
PAIMON_ASSIGN_OR_RAISE(
key_field, AssignFieldIdsRecursively(key_field,
/*set_field_id=*/false, field_id));
PAIMON_ASSIGN_OR_RAISE(
diff --git a/src/paimon/core/schema/table_schema_test.cpp
b/src/paimon/core/schema/table_schema_test.cpp
index e0f75ca4..73909d41 100644
--- a/src/paimon/core/schema/table_schema_test.cpp
+++ b/src/paimon/core/schema/table_schema_test.cpp
@@ -22,6 +22,7 @@
#include <utility>
#include "arrow/api.h"
+#include "arrow/c/bridge.h"
#include "gtest/gtest.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/data/variant/variant_type_utils.h"
@@ -1358,12 +1359,33 @@ TEST_F(TableSchemaTest, MapBlobSchemaLoadsFromJson) {
ASSERT_EQ(restored_json, serialized);
}
-TEST_F(TableSchemaTest, CreatingMapBlobSchemaIsRejected) {
+TEST_F(TableSchemaTest, CreatingMapBlobSchemaIsAllowed) {
auto map_type = arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value",
/*nullable=*/true));
- ASSERT_NOK_WITH_MSG(
- TableSchema::Create(/*schema_id=*/0,
arrow::schema({arrow::field("blob_map", map_type)}),
- /*partition_keys=*/{}, /*primary_keys=*/{},
/*options=*/{}),
- "not supported by the C++ writer");
+ auto schema =
+ arrow::schema({arrow::field("id", arrow::int32()),
arrow::field("blob_map", map_type)});
+ ::ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+ std::shared_ptr<arrow::Schema> imported =
arrow::ImportSchema(&c_schema).ValueOrDie();
+ auto missing_metadata_value = arrow::field("value", arrow::large_binary(),
/*nullable=*/true);
+ ASSERT_FALSE(BlobUtils::IsBlobField(missing_metadata_value));
+ auto missing_metadata = arrow::schema(
+ {arrow::field("id", arrow::int32()),
+ arrow::field("blob_map", arrow::map(arrow::utf8(),
missing_metadata_value))});
+
+ for (const auto& input : {schema, imported, missing_metadata}) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> table_schema,
+ TableSchema::Create(/*schema_id=*/0, input,
/*partition_keys=*/{},
+ /*primary_keys=*/{},
/*options=*/{}));
+ ASSERT_OK_AND_ASSIGN(std::string serialized,
table_schema->ToJsonString());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> restored,
+ TableSchema::CreateFromJson(serialized));
+ for (const auto& fields : {table_schema->Fields(),
restored->Fields()}) {
+ auto field = DataField::ConvertDataFieldToArrowField(fields[1]);
+ ASSERT_TRUE(BlobUtils::IsMapBlobField(field));
+ ASSERT_TRUE(BlobUtils::IsBlobField(
+ checked_cast<const
arrow::MapType&>(*field->type()).item_field()));
+ }
+ }
}
TEST_F(TableSchemaTest, CreatingArrayBlobSchemaIsAllowed) {
diff --git a/src/paimon/format/blob/blob_file_batch_reader.cpp
b/src/paimon/format/blob/blob_file_batch_reader.cpp
index 86ff1f1c..5856e88d 100644
--- a/src/paimon/format/blob/blob_file_batch_reader.cpp
+++ b/src/paimon/format/blob/blob_file_batch_reader.cpp
@@ -53,8 +53,6 @@ constexpr int32_t kArrayBlobHeaderLength = 9;
constexpr int32_t kArrayBlobIndexLengthSize = 4;
constexpr int32_t kArrayBlobMinPayloadLength = kArrayBlobHeaderLength +
kArrayBlobIndexLengthSize;
-constexpr int32_t kMapBlobMagicNumber = 0x4D424342;
-constexpr int8_t kMapBlobVersion = 1;
constexpr int32_t kMapBlobHeaderLength = 9;
constexpr int32_t kMapBlobIndexLengthsSize = 8;
constexpr int32_t kMapBlobMinPayloadLength = kMapBlobHeaderLength +
kMapBlobIndexLengthsSize;
@@ -788,12 +786,12 @@ Result<BlobFileBatchReader::MapBlobPayload>
BlobFileBatchReader::ReadMapBlobPayl
std::array<uint8_t, kMapBlobHeaderLength> header;
PAIMON_RETURN_NOT_OK(ReadBlobContentAt(payload_offset, header.size(),
header.data()));
const auto magic_number = ReadLittleEndian<int32_t>(header.data());
- if (magic_number != kMapBlobMagicNumber) {
+ if (magic_number != BlobDefs::kMapBlobMagicNumber) {
return Status::Invalid(
fmt::format("invalid MAP<..., BLOB> payload magic number: {}",
magic_number));
}
const auto version = static_cast<int8_t>(header[4]);
- if (version != kMapBlobVersion) {
+ if (version != BlobDefs::kMapBlobVersion) {
return Status::NotImplemented(
fmt::format("unsupported MAP<..., BLOB> payload version: {}",
version));
}
diff --git a/src/paimon/format/blob/blob_format_writer.cpp
b/src/paimon/format/blob/blob_format_writer.cpp
index b7673f06..7905be9d 100644
--- a/src/paimon/format/blob/blob_format_writer.cpp
+++ b/src/paimon/format/blob/blob_format_writer.cpp
@@ -23,10 +23,13 @@
#include <map>
#include <optional>
#include <string>
+#include <string_view>
+#include <unordered_set>
#include <vector>
#include "arrow/api.h"
#include "arrow/c/bridge.h"
+#include "arrow/util/utf8.h"
#include "paimon/common/data/blob_defs.h"
#include "paimon/common/data/blob_utils.h"
#include "paimon/common/io/offset_input_stream.h"
@@ -35,12 +38,92 @@
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/delta_varint_compressor.h"
+#include "paimon/common/utils/math.h"
#include "paimon/data/blob_descriptor.h"
+#include "paimon/data/decimal.h"
#include "paimon/fs/file_system.h"
#include "paimon/io/byte_array_input_stream.h"
#include "paimon/logging.h"
namespace paimon::blob {
+namespace {
+
+template <typename T>
+void AppendLittleEndian(T value, std::string* out) {
+ const T little_endian = ToLittleEndian(value);
+ out->append(reinterpret_cast<const char*>(&little_endian), sizeof(T));
+}
+
+/// Serialize the key at `index` of `keys` and append its bytes to `out`, as
Java's
+/// MapBlobElementSerializer serializes a MAP<..., BLOB> key: fixed-width
little-endian integers,
+/// one byte 0 or 1 for a boolean, the unscaled long of a compact decimal or
the big-endian
+/// two's-complement unscaled bytes of any other decimal, and the raw bytes of
a string or binary.
+Status SerializeMapBlobKey(const arrow::Array& keys, int64_t index,
std::string* out) {
+ switch (keys.type_id()) {
+ case arrow::Type::BOOL:
+ out->push_back(checked_cast<const
arrow::BooleanArray&>(keys).Value(index) ? 1 : 0);
+ return Status::OK();
+ case arrow::Type::INT8:
+ out->push_back(
+ static_cast<char>(checked_cast<const
arrow::Int8Array&>(keys).Value(index)));
+ return Status::OK();
+ case arrow::Type::INT16:
+ AppendLittleEndian(checked_cast<const
arrow::Int16Array&>(keys).Value(index), out);
+ return Status::OK();
+ case arrow::Type::INT32:
+ AppendLittleEndian(checked_cast<const
arrow::Int32Array&>(keys).Value(index), out);
+ return Status::OK();
+ case arrow::Type::INT64:
+ AppendLittleEndian(checked_cast<const
arrow::Int64Array&>(keys).Value(index), out);
+ return Status::OK();
+ case arrow::Type::DATE32:
+ AppendLittleEndian(checked_cast<const
arrow::Date32Array&>(keys).Value(index), out);
+ return Status::OK();
+ case arrow::Type::DECIMAL128: {
+ const auto& decimal_type = checked_cast<const
arrow::Decimal128Type&>(*keys.type());
+ const arrow::Decimal128 value(
+ checked_cast<const
arrow::Decimal128Array&>(keys).GetValue(index));
+ // Keep the key within its declared precision so the C++ reader
can read it back.
+ if (!value.FitsInPrecision(decimal_type.precision())) {
+ return Status::Invalid("MAP<..., BLOB> decimal key exceeds
declared precision");
+ }
+ const Decimal decimal(
+ decimal_type.precision(), decimal_type.scale(),
+ static_cast<Decimal::int128_t>(
+
static_cast<Decimal::uint128_t>(static_cast<uint64_t>(value.high_bits()))
+ << 64 |
+ value.low_bits()));
+ if (decimal.IsCompact()) {
+ AppendLittleEndian(decimal.ToUnscaledLong(), out);
+ return Status::OK();
+ }
+ const std::vector<char> bytes = decimal.ToUnscaledBytes();
+ out->append(bytes.data(), bytes.size());
+ return Status::OK();
+ }
+ case arrow::Type::STRING: {
+ const std::string_view key =
+ checked_cast<const arrow::StringArray&>(keys).GetView(index);
+ // Write only valid UTF-8 so the C++ reader can read the string
key back.
+ if (!arrow::util::ValidateUTF8(key)) {
+ return Status::Invalid("invalid UTF-8 in MAP<STRING, BLOB>
key");
+ }
+ out->append(key.data(), key.size());
+ return Status::OK();
+ }
+ case arrow::Type::BINARY: {
+ const std::string_view key =
+ checked_cast<const arrow::BinaryArray&>(keys).GetView(index);
+ out->append(key.data(), key.size());
+ return Status::OK();
+ }
+ default:
+ return Status::Invalid(
+ fmt::format("unsupported MAP<..., BLOB> key type: {}",
keys.type()->ToString()));
+ }
+}
+
+} // namespace
BlobFormatWriter::BlobFormatWriter(const std::shared_ptr<OutputStream>& out,
const std::string& uri,
const std::shared_ptr<arrow::DataType>&
data_type,
@@ -55,13 +138,16 @@ BlobFormatWriter::BlobFormatWriter(const
std::shared_ptr<OutputStream>& out, con
pool_(pool),
write_null_on_missing_file_(write_null_on_missing_file),
write_null_on_fetch_failure_(write_null_on_fetch_failure) {
- // Create() has already checked that data_type has exactly one BLOB or
ARRAY<BLOB> field.
+ // Create() has already checked that data_type has exactly one BLOB,
ARRAY<BLOB> or
+ // MAP<..., BLOB> field.
blob_field_name_ = data_type_->field(0)->name();
+ is_map_blob_field_ = BlobUtils::IsMapBlobField(data_type_->field(0));
metrics_ = std::make_shared<MetricsImpl>();
tmp_buffer_ = Bytes::AllocateBytes(kTmpBufferSize, pool_.get());
magic_number_bytes_ =
IntegerToLittleEndian<int32_t>(BlobDefs::kMagicNumber, pool_);
array_magic_number_bytes_ =
IntegerToLittleEndian<int32_t>(BlobDefs::kArrayBlobMagicNumber, pool_);
+ map_magic_number_bytes_ =
IntegerToLittleEndian<int32_t>(BlobDefs::kMapBlobMagicNumber, pool_);
logger_ = Logger::GetLogger("BlobFormatWriter");
}
@@ -85,10 +171,20 @@ Result<std::unique_ptr<BlobFormatWriter>>
BlobFormatWriter::Create(
return Status::Invalid(
fmt::format("blob data type field number {} is not 1",
data_type->num_fields()));
}
- if (!BlobUtils::IsBlobField(data_type->field(0)) &&
- !BlobUtils::IsArrayBlobField(data_type->field(0))) {
+ const std::shared_ptr<arrow::Field>& field = data_type->field(0);
+ if (!BlobUtils::IsAnyBlobField(field)) {
return Status::Invalid(
- fmt::format("field {} is not BLOB or ARRAY<BLOB>",
data_type->field(0)->ToString()));
+ fmt::format("field {} is not BLOB, ARRAY<BLOB> or MAP<..., BLOB>",
field->ToString()));
+ }
+ if (BlobUtils::IsMapBlobField(field)) {
+ const auto& map_type = checked_cast<const
arrow::MapType&>(*field->type());
+ if (!BlobUtils::IsSupportedMapBlobKeyType(*map_type.key_type())) {
+ return Status::Invalid(fmt::format("unsupported MAP<..., BLOB> key
type: {}",
+
map_type.key_type()->ToString()));
+ }
+ if (map_type.key_type()->id() == arrow::Type::STRING) {
+ arrow::util::InitializeUTF8();
+ }
}
PAIMON_ASSIGN_OR_RAISE(std::string uri, out->GetUri());
return std::unique_ptr<BlobFormatWriter>(new BlobFormatWriter(
@@ -130,6 +226,16 @@ Status BlobFormatWriter::AddBatch(ArrowArray* batch) {
UpdateMetrics();
return Status::OK();
}
+ if (child_array->type_id() == arrow::Type::type::MAP) {
+ const auto& map_array = checked_cast<const
arrow::MapArray&>(*child_array);
+ if (BlobUtils::IsMapBlobPlaceholder(map_array, 0)) {
+ bin_lengths_.push_back(BlobDefs::kPlaceholderBinLength);
+ return Status::OK();
+ }
+ PAIMON_RETURN_NOT_OK(WriteMapBlob(map_array));
+ UpdateMetrics();
+ return Status::OK();
+ }
if (child_array->type_id() != arrow::Type::type::LARGE_BINARY) {
return Status::Invalid("BlobFormatWriter only support large binary
type.");
}
@@ -217,22 +323,8 @@ Status BlobFormatWriter::WriteArrayBlob(const
arrow::ListArray& list_array) {
IntegerToLittleEndian<int32_t>(element_count, pool_);
PAIMON_RETURN_NOT_OK(WriteWithCrc32(element_count_bytes->data(),
element_count_bytes->size()));
- // Each element is opened before any of its bytes are written, so an
element converted to NULL
- // leaves no partial data. As in Java, any other failure leaves a partial
entry and fails the
- // write.
- std::vector<int64_t> element_lengths(element_count,
BlobDefs::kNullBinLength);
- for (int32_t i = 0; i < element_count; ++i) {
- const int64_t value_index = element_offset + i;
- if (blob_values.IsNull(value_index)) {
- continue;
- }
- PAIMON_ASSIGN_OR_RAISE(BlobCopySource source,
-
OpenBlobInputStream(blob_values.GetView(value_index), i));
- if (source.stream == nullptr) {
- continue;
- }
- PAIMON_ASSIGN_OR_RAISE(element_lengths[i], WriteBlobData(source, i));
- }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> element_lengths,
+ WriteBlobElements(blob_values, element_offset,
element_count));
const std::vector<char> index_bytes =
DeltaVarintCompressor::Compress(element_lengths);
PAIMON_ASSIGN_OR_RAISE(int32_t index_length,
ToIndexLength(index_bytes.size()));
@@ -243,6 +335,88 @@ Status BlobFormatWriter::WriteArrayBlob(const
arrow::ListArray& list_array) {
return FinishEntry(entry_pos);
}
+Status BlobFormatWriter::WriteMapBlob(const arrow::MapArray& map_array) {
+ const std::shared_ptr<arrow::Array>& items = map_array.items();
+ if (items->type_id() != arrow::Type::type::LARGE_BINARY) {
+ return Status::Invalid("BlobFormatWriter only support large binary
MAP<..., BLOB> values.");
+ }
+ const auto& blob_values = checked_cast<const
arrow::LargeBinaryArray&>(*items);
+ const arrow::Array& keys = *map_array.keys();
+ const int64_t entry_offset = map_array.value_offset(0);
+ const int32_t entry_count = map_array.value_length(0);
+
+ // As in Java, the keys are serialized and validated before the entry is
begun, so an invalid
+ // key fails the write without leaving partial data.
+ std::string key_data;
+ std::vector<int64_t> key_lengths(entry_count);
+ std::unordered_set<std::string> serialized_keys;
+ serialized_keys.reserve(entry_count);
+ for (int32_t i = 0; i < entry_count; ++i) {
+ const int64_t key_index = entry_offset + i;
+ if (keys.IsNull(key_index)) {
+ return Status::Invalid(
+ fmt::format("invalid {}: MAP<..., BLOB> keys cannot be null",
DescribeMapKey(i)));
+ }
+ std::string key;
+ Status status = SerializeMapBlobKey(keys, key_index, &key);
+ if (!status.ok()) {
+ return status.WithMessage("invalid ", DescribeMapKey(i), ": ",
status.message());
+ }
+ key_data.append(key);
+ key_lengths[i] = static_cast<int64_t>(key.size());
+ if (!serialized_keys.insert(std::move(key)).second) {
+ return Status::Invalid(
+ fmt::format("invalid {}: MAP<..., BLOB> keys must be unique",
DescribeMapKey(i)));
+ }
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(int64_t entry_pos, BeginEntry());
+ PAIMON_RETURN_NOT_OK(
+ WriteWithCrc32(map_magic_number_bytes_->data(),
map_magic_number_bytes_->size()));
+ PAIMON_RETURN_NOT_OK(WriteWithCrc32(reinterpret_cast<const
char*>(&BlobDefs::kMapBlobVersion),
+ sizeof(BlobDefs::kMapBlobVersion)));
+ PAIMON_UNIQUE_PTR<Bytes> entry_count_bytes =
IntegerToLittleEndian<int32_t>(entry_count, pool_);
+ PAIMON_RETURN_NOT_OK(WriteWithCrc32(entry_count_bytes->data(),
entry_count_bytes->size()));
+ PAIMON_RETURN_NOT_OK(WriteWithCrc32(key_data.data(), key_data.size()));
+
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> value_lengths,
+ WriteBlobElements(blob_values, entry_offset,
entry_count));
+
+ const std::vector<char> key_index_bytes =
DeltaVarintCompressor::Compress(key_lengths);
+ const std::vector<char> value_index_bytes =
DeltaVarintCompressor::Compress(value_lengths);
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_index_length,
ToIndexLength(key_index_bytes.size()));
+ PAIMON_ASSIGN_OR_RAISE(int32_t value_index_length,
ToIndexLength(value_index_bytes.size()));
+ PAIMON_RETURN_NOT_OK(WriteWithCrc32(key_index_bytes.data(),
key_index_bytes.size()));
+ PAIMON_RETURN_NOT_OK(WriteWithCrc32(value_index_bytes.data(),
value_index_bytes.size()));
+ PAIMON_UNIQUE_PTR<Bytes> key_index_length_bytes =
+ IntegerToLittleEndian<int32_t>(key_index_length, pool_);
+ PAIMON_RETURN_NOT_OK(
+ WriteWithCrc32(key_index_length_bytes->data(),
key_index_length_bytes->size()));
+ PAIMON_UNIQUE_PTR<Bytes> value_index_length_bytes =
+ IntegerToLittleEndian<int32_t>(value_index_length, pool_);
+ PAIMON_RETURN_NOT_OK(
+ WriteWithCrc32(value_index_length_bytes->data(),
value_index_length_bytes->size()));
+ return FinishEntry(entry_pos);
+}
+
+Result<std::vector<int64_t>> BlobFormatWriter::WriteBlobElements(
+ const arrow::LargeBinaryArray& values, int64_t offset, int32_t count) {
+ std::vector<int64_t> lengths(count, BlobDefs::kNullBinLength);
+ for (int32_t i = 0; i < count; ++i) {
+ const int64_t value_index = offset + i;
+ if (values.IsNull(value_index)) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(BlobCopySource source,
+
OpenBlobInputStream(values.GetView(value_index), i));
+ if (source.stream == nullptr) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(lengths[i], WriteBlobData(source, i));
+ }
+ return lengths;
+}
+
Result<BlobFormatWriter::BlobCopySource> BlobFormatWriter::OpenBlobInputStream(
std::string_view blob_data, std::optional<int32_t> element_index) {
// Whether blob_data is a serialized BlobDescriptor is detected by its
magic header rather
@@ -474,10 +648,19 @@ std::string
BlobFormatWriter::DescribeValue(std::optional<int32_t> element_index
if (!element_index) {
return fmt::format("BLOB field {} in row {} of blob file {}",
blob_field_name_, row, uri_);
}
+ if (is_map_blob_field_) {
+ return fmt::format("value of entry {} of MAP<..., BLOB> field {} in
row {} of blob file {}",
+ *element_index, blob_field_name_, row, uri_);
+ }
return fmt::format("element {} of ARRAY<BLOB> field {} in row {} of blob
file {}",
*element_index, blob_field_name_, row, uri_);
}
+std::string BlobFormatWriter::DescribeMapKey(int32_t entry_index) const {
+ return fmt::format("key of entry {} of MAP<..., BLOB> field {} in row {}
of blob file {}",
+ entry_index, blob_field_name_, bin_lengths_.size(),
uri_);
+}
+
Result<int32_t> BlobFormatWriter::ToIndexLength(size_t index_size) const {
if (index_size > static_cast<size_t>(std::numeric_limits<int32_t>::max()))
{
return Status::Invalid(
diff --git a/src/paimon/format/blob/blob_format_writer.h
b/src/paimon/format/blob/blob_format_writer.h
index 1e26c7cd..7293bc1f 100644
--- a/src/paimon/format/blob/blob_format_writer.h
+++ b/src/paimon/format/blob/blob_format_writer.h
@@ -37,7 +37,9 @@
namespace arrow {
class DataType;
+class LargeBinaryArray;
class ListArray;
+class MapArray;
} // namespace arrow
struct ArrowArray;
@@ -53,12 +55,12 @@ namespace paimon::blob {
class BlobMetrics {
public:
- /// Number of BLOB values (rows, or ARRAY<BLOB> elements) written as NULL
because their
- /// referenced file did not exist.
+ /// Number of BLOB values (rows, ARRAY<BLOB> elements or MAP<..., BLOB>
values) written as NULL
+ /// because their referenced file did not exist.
static inline const char WRITE_NULL_ON_MISSING_FILE_COUNT[] =
"blob.write.null-on-missing-file.count";
- /// Number of BLOB values (rows, or ARRAY<BLOB> elements) written as NULL
because their
- /// referenced data could not be reached.
+ /// Number of BLOB values (rows, ARRAY<BLOB> elements or MAP<..., BLOB>
values) written as NULL
+ /// because their referenced data could not be reached.
static inline const char WRITE_NULL_ON_FETCH_FAILURE_COUNT[] =
"blob.write.null-on-fetch-failure.count";
};
@@ -66,8 +68,9 @@ class BlobMetrics {
// Blob format:
//
https://cwiki.apache.org/confluence/display/PAIMON/PIP-35%3A+Introduce+Blob+to+store+multimodal+data
//
-// The single field of a blob file is BLOB or ARRAY<BLOB>. An ARRAY<BLOB>
entry stores the nested
-// payload of the Paimon BLOB file spec, where a null element has length -1.
+// The single field of a blob file is BLOB, ARRAY<BLOB> or MAP<..., BLOB>. An
ARRAY<BLOB> or
+// MAP<..., BLOB> entry stores the nested payload of the Paimon BLOB file
spec, where a null element
+// or value has length -1.
class BlobFormatWriter : public FormatWriter {
public:
/// `write_null_on_missing_file` converts a descriptor whose referenced
file does not exist
@@ -78,10 +81,17 @@ class BlobFormatWriter : public FormatWriter {
/// `write_null_on_fetch_failure` like any other failed open.
/// See Options::BLOB_WRITE_NULL_ON_MISSING_FILE /
BLOB_WRITE_NULL_ON_FETCH_FAILURE.
///
- /// A value exactly equal to BlobDefs::kPlaceholderSentinel, or an
ARRAY<BLOB> whose only
- /// element is that sentinel, is persisted as a placeholder entry
(bin_length -2, no data bytes)
- /// in every write, as Java persists its placeholder objects; any other
value is written as
- /// usual.
+ /// A value exactly equal to BlobDefs::kPlaceholderSentinel, an
ARRAY<BLOB> whose only element
+ /// is that sentinel, or a MAP<..., BLOB> of exactly two entries with
equal keys and null values
+ /// (see BlobUtils::IsMapBlobPlaceholder), is persisted as a placeholder
entry (bin_length -2,
+ /// no data bytes) in every write, as Java persists its placeholder
objects; any other value is
+ /// written as usual.
+ ///
+ /// Any other MAP<..., BLOB> row fails the write before any of its bytes
are written when its
+ /// keys are not unique by their serialized bytes, as in Java's
MapBlobElementSerializer, when
+ /// it has a null key, which Java allows once but Arrow map keys cannot
hold, or when it has a
+ /// key the C++ reader rejects: a decimal key exceeding its precision or a
string key that is
+ /// not valid UTF-8. The key type must satisfy
BlobUtils::IsSupportedMapBlobKeyType.
static Result<std::unique_ptr<BlobFormatWriter>> Create(
const std::shared_ptr<OutputStream>& out, const
std::shared_ptr<arrow::DataType>& data_type,
bool write_null_on_missing_file, bool write_null_on_fetch_failure,
@@ -114,6 +124,15 @@ class BlobFormatWriter : public FormatWriter {
Status WriteArrayBlob(const arrow::ListArray& list_array);
+ Status WriteMapBlob(const arrow::MapArray& map_array);
+
+ /// Copy `count` blob values of `values` from `offset` into the current
entry, returning the
+ /// length of each one, or -1 for a null value. Each value is opened
before any of its bytes are
+ /// written, so a value converted to NULL leaves no partial data. As in
Java, any other failure
+ /// leaves a partial entry and fails the write.
+ Result<std::vector<int64_t>> WriteBlobElements(const
arrow::LargeBinaryArray& values,
+ int64_t offset, int32_t
count);
+
/// The input stream of a blob value, as Java's BlobCopySource. A `reused`
stream is a view on
/// the kept source stream and must not be closed; any other stream is the
value's own.
struct BlobCopySource {
@@ -124,8 +143,8 @@ class BlobFormatWriter : public FormatWriter {
/// Open an input stream on a blob value, which is either a serialized
BlobDescriptor or the
/// raw blob bytes. Returns a null stream when a failure to reach the
referenced data is
/// converted to a NULL value by `write_null_on_missing_file_` or
- /// `write_null_on_fetch_failure_`. `element_index` identifies an
ARRAY<BLOB> element in logs
- /// and errors.
+ /// `write_null_on_fetch_failure_`. `element_index` identifies an
ARRAY<BLOB> element or a
+ /// MAP<..., BLOB> value in logs and errors.
Result<BlobCopySource> OpenBlobInputStream(std::string_view blob_data,
std::optional<int32_t>
element_index);
@@ -188,10 +207,14 @@ class BlobFormatWriter : public FormatWriter {
Status AddFailureContext(const Status& status, const std::string& action,
std::optional<int32_t> element_index) const;
- /// Describe the value being written for logs and errors: the BLOB field
or one element of the
- /// ARRAY<BLOB> field, with its row in the blob file.
+ /// Describe the value being written for logs and errors: the BLOB field,
one element of the
+ /// ARRAY<BLOB> field or the value of one entry of the MAP<..., BLOB>
field, with its row in the
+ /// blob file.
std::string DescribeValue(std::optional<int32_t> element_index) const;
+ /// Describe the key of one entry of the MAP<..., BLOB> field being
written for errors.
+ std::string DescribeMapKey(int32_t entry_index) const;
+
/// Convert the size of a compressed index to the signed 32-bit index
length stored after it,
/// failing when the index is too large for that field.
Result<int32_t> ToIndexLength(size_t index_size) const;
@@ -214,8 +237,10 @@ class BlobFormatWriter : public FormatWriter {
PAIMON_UNIQUE_PTR<Bytes> tmp_buffer_;
PAIMON_UNIQUE_PTR<Bytes> magic_number_bytes_;
PAIMON_UNIQUE_PTR<Bytes> array_magic_number_bytes_;
+ PAIMON_UNIQUE_PTR<Bytes> map_magic_number_bytes_;
std::shared_ptr<arrow::DataType> data_type_;
std::string blob_field_name_;
+ bool is_map_blob_field_ = false;
std::shared_ptr<FileSystem> fs_;
std::shared_ptr<MemoryPool> pool_;
std::shared_ptr<Metrics> metrics_;
diff --git a/src/paimon/format/blob/blob_format_writer_test.cpp
b/src/paimon/format/blob/blob_format_writer_test.cpp
index 10de8a60..49fed766 100644
--- a/src/paimon/format/blob/blob_format_writer_test.cpp
+++ b/src/paimon/format/blob/blob_format_writer_test.cpp
@@ -27,6 +27,7 @@
#include <vector>
#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
#include "fmt/format.h"
#include "gtest/gtest.h"
#include "paimon/common/data/blob_defs.h"
@@ -319,6 +320,13 @@ class BlobFormatWriterTestBase : public ::testing::Test {
return path;
}
+ Result<std::string> DescriptorOf(const std::string& path, int64_t offset =
0,
+ int64_t length = -1) const {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
Blob::FromPath(path, offset, length));
+ PAIMON_UNIQUE_PTR<Bytes> descriptor = blob->ToDescriptor(pool_);
+ return std::string(descriptor->data(), descriptor->size());
+ }
+
protected:
std::shared_ptr<MemoryPool> pool_;
std::unique_ptr<paimon::test::UniqueTestDirectory> dir_;
@@ -465,7 +473,7 @@ TEST_P(BlobFormatWriterTest,
TestCreateWithInvalidParameters) {
BlobFormatWriter::Create(output_stream_, non_blob_type,
/*write_null_on_missing_file=*/false,
/*write_null_on_fetch_failure=*/false,
file_system_, pool_),
- "field regular_col: binary is not BLOB or ARRAY<BLOB>");
+ "field regular_col: binary is not BLOB, ARRAY<BLOB> or MAP<...,
BLOB>");
auto plain_binary_array_type =
arrow::struct_({arrow::field("array_col",
arrow::list(arrow::large_binary()))});
@@ -473,14 +481,32 @@ TEST_P(BlobFormatWriterTest,
TestCreateWithInvalidParameters) {
BlobFormatWriter::Create(output_stream_, plain_binary_array_type,
/*write_null_on_missing_file=*/false,
/*write_null_on_fetch_failure=*/false,
file_system_, pool_),
- "is not BLOB or ARRAY<BLOB>");
+ "is not BLOB, ARRAY<BLOB> or MAP<..., BLOB>");
auto nested_array_type = arrow::struct_({arrow::field(
"nested_col", arrow::list(arrow::list(BlobUtils::ToArrowField("item",
true))))});
ASSERT_NOK_WITH_MSG(
BlobFormatWriter::Create(output_stream_, nested_array_type,
/*write_null_on_missing_file=*/false,
/*write_null_on_fetch_failure=*/false,
file_system_, pool_),
- "is not BLOB or ARRAY<BLOB>");
+ "is not BLOB, ARRAY<BLOB> or MAP<..., BLOB>");
+ auto plain_binary_map_type =
+ arrow::struct_({arrow::field("map_col", arrow::map(arrow::utf8(),
arrow::binary()))});
+ ASSERT_NOK_WITH_MSG(
+ BlobFormatWriter::Create(output_stream_, plain_binary_map_type,
+ /*write_null_on_missing_file=*/false,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_),
+ "is not BLOB, ARRAY<BLOB> or MAP<..., BLOB>");
+
+ for (const auto& key_type : {arrow::float32(),
arrow::time32(arrow::TimeUnit::MILLI),
+ arrow::timestamp(arrow::TimeUnit::MILLI)}) {
+ auto map_blob_type = arrow::struct_({arrow::field(
+ "map_col", arrow::map(key_type, BlobUtils::ToArrowField("value",
true)))});
+ ASSERT_NOK_WITH_MSG(
+ BlobFormatWriter::Create(output_stream_, map_blob_type,
+ /*write_null_on_missing_file=*/false,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_),
+ "unsupported MAP<..., BLOB> key type: " + key_type->ToString());
+ }
}
TEST_P(BlobFormatWriterTest, TestInvalidCase) {
@@ -1363,13 +1389,6 @@ class BlobFormatWriterArrayBlobTest : public
BlobFormatWriterTestBase {
return AddBatchOnce(writer, array);
}
- Result<std::string> DescriptorOf(const std::string& path, int64_t offset =
0,
- int64_t length = -1) const {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
Blob::FromPath(path, offset, length));
- PAIMON_UNIQUE_PTR<Bytes> descriptor = blob->ToDescriptor(pool_);
- return std::string(descriptor->data(), descriptor->size());
- }
-
Result<std::shared_ptr<arrow::ListArray>> ReadBackArrays(
bool blob_as_descriptor, bool emit_placeholder_sentinel,
const std::string& file_name = "file.blob") const {
@@ -1764,4 +1783,445 @@ TEST_F(BlobFormatWriterArrayBlobTest,
TestArrayBlobOutputFailureFailsWrite) {
}
}
+class BlobFormatWriterMapBlobTest : public BlobFormatWriterTestBase {
+ public:
+ using MapEntries = std::vector<std::pair<std::string,
std::optional<std::string>>>;
+
+ void SetUp() override {
+ BlobFormatWriterTestBase::SetUp();
+ struct_type_ = MapBlobStructType(arrow::utf8());
+ }
+
+ static std::shared_ptr<arrow::DataType> MapBlobStructType(
+ const std::shared_ptr<arrow::DataType>& key_type) {
+ return arrow::struct_({arrow::field(
+ "map_blob_col", arrow::map(key_type,
BlobUtils::ToArrowField("value", true)), true)});
+ }
+
+ static Result<std::shared_ptr<arrow::Array>> MakeMapBlobRow(
+ const std::shared_ptr<arrow::DataType>& struct_type,
+ const std::shared_ptr<arrow::Array>& keys,
+ const std::vector<std::optional<std::string>>& values) {
+ arrow::LargeBinaryBuilder value_builder;
+ for (const std::optional<std::string>& value : values) {
+ if (value) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder.Append(*value));
+ } else {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder.AppendNull());
+ }
+ }
+ std::shared_ptr<arrow::Array> items;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder.Finish(&items));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> offsets,
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(),
+ fmt::format("[0, {}]",
keys->length())));
+ const std::shared_ptr<arrow::Field>& map_field = struct_type->field(0);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> map_array,
+ arrow::MapArray::FromArrays(map_field->type(), offsets, keys,
items));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
struct_array,
+
arrow::StructArray::Make({map_array}, {map_field}));
+ return struct_array;
+ }
+
+ static Result<std::vector<std::shared_ptr<arrow::Array>>> RowsFromJson(
+ const std::shared_ptr<arrow::DataType>& struct_type, const
std::string& json) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> array,
+ arrow::ipc::internal::json::ArrayFromJSON(struct_type, json));
+ std::vector<std::shared_ptr<arrow::Array>> rows;
+ for (int64_t i = 0; i < array->length(); ++i) {
+ rows.push_back(array->Slice(i, 1));
+ }
+ return rows;
+ }
+
+ Status AddMapBlobRow(const std::shared_ptr<BlobFormatWriter>& writer,
+ const std::optional<MapEntries>& entries) const {
+ if (!entries) {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<arrow::Array>>
rows,
+ RowsFromJson(struct_type_, "[[null]]"));
+ return AddBatchOnce(writer, rows[0]);
+ }
+ arrow::StringBuilder key_builder;
+ std::vector<std::optional<std::string>> values;
+ for (const auto& [key, value] : *entries) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder.Append(key));
+ values.push_back(value);
+ }
+ std::shared_ptr<arrow::Array> keys;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder.Finish(&keys));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> row,
+ MakeMapBlobRow(struct_type_, keys, values));
+ return AddBatchOnce(writer, row);
+ }
+
+ Result<std::shared_ptr<arrow::MapArray>> ReadBackMapArray(
+ bool blob_as_descriptor, bool emit_placeholder_sentinel,
+ const std::string& file_name = "file.blob") const {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::StructArray>
struct_array,
+ ReadBack(blob_as_descriptor,
emit_placeholder_sentinel, file_name));
+ return checked_pointer_cast<arrow::MapArray>(struct_array->field(0));
+ }
+
+ Result<std::vector<std::optional<MapEntries>>> ReadBackMaps(
+ bool blob_as_descriptor, const std::string& file_name = "file.blob")
const {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::MapArray> map_array,
+ ReadBackMapArray(blob_as_descriptor,
+
/*emit_placeholder_sentinel=*/false, file_name));
+ const auto& keys = checked_cast<const
arrow::StringArray&>(*map_array->keys());
+ const auto& values = checked_cast<const
arrow::LargeBinaryArray&>(*map_array->items());
+ std::vector<std::optional<MapEntries>> rows;
+ for (int64_t row = 0; row < map_array->length(); ++row) {
+ if (map_array->IsNull(row)) {
+ rows.emplace_back(std::nullopt);
+ continue;
+ }
+ MapEntries entries;
+ for (int64_t i = map_array->value_offset(row); i <
map_array->value_offset(row + 1);
+ ++i) {
+ if (values.IsNull(i)) {
+ entries.emplace_back(keys.GetString(i), std::nullopt);
+ continue;
+ }
+ std::string_view stored = values.GetView(i);
+ if (!blob_as_descriptor) {
+ entries.emplace_back(keys.GetString(i),
std::string(stored));
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
+ Blob::FromDescriptor(stored.data(),
stored.size()));
+ PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> data,
+ blob->ToData(file_system_, pool_));
+ entries.emplace_back(keys.GetString(i),
std::string(data->data(), data->size()));
+ }
+ rows.emplace_back(std::move(entries));
+ }
+ return rows;
+ }
+
+ static MapEntries PlaceholderEntries() {
+ return {{"key", std::nullopt}, {"key", std::nullopt}};
+ }
+};
+
+TEST_F(BlobFormatWriterMapBlobTest, TestMapBlobGoldenBytes) {
+ ASSERT_OK_AND_ASSIGN(std::string descriptor,
+ DescriptorOf(WriteSourceFile("source.bin",
"descriptor")));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
CreateDefaultWriter());
+
+ ASSERT_OK(AddMapBlobRow(writer, MapEntries{}));
+ ASSERT_OK(AddMapBlobRow(
+ writer,
+ MapEntries{
+ {"", ""}, {"inline", "data"}, {"descriptor", descriptor}, {"null",
std::nullopt}}));
+ ASSERT_OK(AddMapBlobRow(writer, std::nullopt));
+ ASSERT_OK(AddMapBlobRow(writer, PlaceholderEntries()));
+ ASSERT_OK(writer->Finish());
+
+ std::vector<uint8_t> expected = {
+ 0xcf, 0x11, 0x4e, 0x58, 0x42, 0x43, 0x42, 0x4d, 0x01, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x21, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x83,
+ 0x60, 0x59, 0x1e, 0xcf, 0x11, 0x4e, 0x58, 0x42, 0x43, 0x42, 0x4d,
0x01, 0x04, 0x00, 0x00,
+ 0x00, 0x69, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x64, 0x65, 0x73, 0x63,
0x72, 0x69, 0x70, 0x74,
+ 0x6f, 0x72, 0x6e, 0x75, 0x6c, 0x6c, 0x64, 0x61, 0x74, 0x61, 0x64,
0x65, 0x73, 0x63, 0x72,
+ 0x69, 0x70, 0x74, 0x6f, 0x72, 0x00, 0x0c, 0x08, 0x0b, 0x00, 0x08,
0x0c, 0x15, 0x04, 0x00,
+ 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x4b, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x3a,
+ 0xbb, 0x6e, 0xee, 0x42, 0x54, 0x97, 0x01, 0x01, 0x05, 0x00, 0x00,
0x00, 0x01};
+ std::string content;
+ ASSERT_OK(file_system_->ReadFile(dir_->Str() + "/file.blob", &content));
+ ASSERT_EQ(std::vector<uint8_t>(content.begin(), content.end()), expected);
+}
+
+TEST_F(BlobFormatWriterMapBlobTest, TestMatchesJavaWrittenFiles) {
+ const std::string table_dir =
+ paimon::test::GetDataDir() +
"/parquet/map_blob_java.db/map_blob_java/bucket-0/";
+ const std::string java_file_prefix =
"data-1be78d56-71bd-453f-ab1f-4312f3bf480e-";
+ const std::string null_rows = R"(, [null], [null], [null]])";
+ struct Case {
+ std::string java_file;
+ std::shared_ptr<arrow::DataType> key_type;
+ std::string json;
+ };
+ const std::vector<Case> cases = {
+ {java_file_prefix + "1.blob", arrow::utf8(),
+ R"([[[["", "string-empty"], ["alpha", "string-alpha"]]], [[]], [null],
+ [[["omega", "string-omega"]]]])"},
+ {java_file_prefix + "2.blob", arrow::boolean(),
+ R"([[[[false, "bool-false"], [true, "bool-true"]]])" + null_rows},
+ {java_file_prefix + "3.blob", arrow::int8(),
+ R"([[[[-128, "tiny-min"], [-1, "tiny-negative"], [127,
"tiny-max"]]])" + null_rows},
+ {java_file_prefix + "4.blob", arrow::int16(),
+ R"([[[[-32768, "small-min"], [-1, "small-negative"], [32767,
"small-max"]]])" + null_rows},
+ {java_file_prefix + "5.blob", arrow::int32(),
+ R"([[[[-2147483648, "int-min"], [-1, "int-negative"], [2147483647,
"int-max"]]])" +
+ null_rows},
+ {java_file_prefix + "6.blob", arrow::int64(),
+ R"([[[[-9223372036854775808, "big-min"], [-1, "big-negative"],
+ [9223372036854775807, "big-max"]]])" +
+ null_rows},
+ {java_file_prefix + "7.blob", arrow::date32(),
+ R"([[[[-1, "date-negative"], [0, "date-epoch"]]])" + null_rows},
+ {java_file_prefix + "8.blob", arrow::binary(), ""},
+ {java_file_prefix + "9.blob", arrow::decimal128(10, 2),
+ R"([[[["-99999999.99", "compact-negative"], ["99999999.99",
"compact-positive"]]])" +
+ null_rows},
+ {java_file_prefix + "10.blob", arrow::decimal128(20, 2),
+ R"([[[["-999999999999999999.99", "large-negative"],
+ ["999999999999999999.99", "large-positive"]]])" +
+ null_rows},
+ {"data-d5380318-987b-4af0-9073-9c8e8162b568-0.blob", arrow::utf8(),
+ R"([[[["key", null], ["key", null]]], [[["key", null], ["key",
null]]],
+ [[["key", null], ["key", null]]], [[["key", null], ["key",
null]]]])"}};
+ for (size_t i = 0; i < cases.size(); ++i) {
+ const Case& c = cases[i];
+ SCOPED_TRACE(fmt::format("case {}: {}", i, c.java_file));
+ const auto struct_type = MapBlobStructType(c.key_type);
+ std::vector<std::shared_ptr<arrow::Array>> rows;
+ if (c.key_type->id() == arrow::Type::BINARY) {
+ arrow::BinaryBuilder key_builder;
+ ASSERT_TRUE(key_builder.Append("").ok());
+ ASSERT_TRUE(key_builder.Append(std::string("\0\xff\1\2", 4)).ok());
+ std::shared_ptr<arrow::Array> keys;
+ ASSERT_TRUE(key_builder.Finish(&keys).ok());
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<arrow::Array> first_row,
+ MakeMapBlobRow(struct_type, keys, {"binary-empty",
"binary-bytes"}));
+ ASSERT_OK_AND_ASSIGN(rows, RowsFromJson(struct_type, "[[null],
[null], [null]]"));
+ rows.insert(rows.begin(), first_row);
+ } else {
+ ASSERT_OK_AND_ASSIGN(rows, RowsFromJson(struct_type, c.json));
+ }
+
+ const std::string file_path = dir_->Str() +
fmt::format("/java_{}.blob", i);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
+ file_system_->Create(file_path,
/*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<BlobFormatWriter> writer,
+ BlobFormatWriter::Create(out, struct_type,
/*write_null_on_missing_file=*/false,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_));
+ for (const std::shared_ptr<arrow::Array>& row : rows) {
+ ASSERT_OK(AddBatchOnce(writer, row));
+ }
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK(out->Close());
+
+ std::string actual;
+ ASSERT_OK(file_system_->ReadFile(file_path, &actual));
+ std::string expected;
+ ASSERT_OK(file_system_->ReadFile(table_dir + c.java_file, &expected));
+ ASSERT_EQ(std::vector<uint8_t>(actual.begin(), actual.end()),
+ std::vector<uint8_t>(expected.begin(), expected.end()));
+ }
+}
+
+TEST_F(BlobFormatWriterMapBlobTest, TestMapBlobRoundTrip) {
+ std::string xxhash_file = paimon::test::GetDataDir() + "/xxhash.data";
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Blob> blob_slice,
+ Blob::FromPath(xxhash_file, /*offset=*/92,
/*length=*/85));
+ PAIMON_UNIQUE_PTR<Bytes> slice_descriptor =
blob_slice->ToDescriptor(pool_);
+ ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> slice_data,
+ blob_slice->ToData(file_system_, pool_));
+ const std::string slice_bytes(slice_data->data(), slice_data->size());
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
CreateDefaultWriter());
+ const std::vector<std::optional<MapEntries>> rows = {
+ MapEntries{{"first", "1"}, {"empty", ""}, {"", "empty key"}, {"null",
std::nullopt}},
+ std::nullopt,
+ MapEntries{},
+ MapEntries{{"slice", std::string(slice_descriptor->data(),
slice_descriptor->size())},
+ {"tail", "tail"}},
+ MapEntries{{"large", std::string((1 << 20) + 7, 'x')}, {"small",
"small"}},
+ };
+ for (const std::optional<MapEntries>& row : rows) {
+ ASSERT_OK(AddMapBlobRow(writer, row));
+ }
+ ASSERT_OK(writer->Finish());
+
+ std::vector<std::optional<MapEntries>> expected = rows;
+ expected[3] = MapEntries{{"slice", slice_bytes}, {"tail", "tail"}};
+ for (bool blob_as_descriptor : {false, true}) {
+ ASSERT_OK_AND_ASSIGN(std::vector<std::optional<MapEntries>> actual,
+ ReadBackMaps(blob_as_descriptor));
+ ASSERT_EQ(actual, expected) << "blob_as_descriptor: " <<
blob_as_descriptor;
+ }
+}
+
+TEST_F(BlobFormatWriterMapBlobTest, TestMapBlobPlaceholder) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
CreateDefaultWriter());
+ ASSERT_OK(AddMapBlobRow(writer, MapEntries{{"key", "value"}}));
+ ASSERT_OK(AddMapBlobRow(writer, PlaceholderEntries()));
+ ASSERT_OK(AddMapBlobRow(writer, MapEntries{{"key", std::nullopt}}));
+ ASSERT_OK(AddMapBlobRow(writer, MapEntries{{"left", std::nullopt},
{"right", std::nullopt}}));
+ ASSERT_OK(writer->Finish());
+
+ ASSERT_NOK_WITH_MSG(ReadBackMapArray(/*blob_as_descriptor=*/false,
+ /*emit_placeholder_sentinel=*/false),
+ "placeholder");
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::MapArray> map_array,
+ ReadBackMapArray(/*blob_as_descriptor=*/false,
+ /*emit_placeholder_sentinel=*/true));
+ ASSERT_EQ(map_array->length(), 4);
+ for (int64_t row = 0; row < map_array->length(); ++row) {
+ ASSERT_EQ(BlobUtils::IsMapBlobPlaceholder(*map_array, row), row == 1)
<< "row " << row;
+ }
+ const auto& keys = checked_cast<const
arrow::StringArray&>(*map_array->keys());
+ const auto& values = checked_cast<const
arrow::LargeBinaryArray&>(*map_array->items());
+ ASSERT_EQ(map_array->value_length(0), 1);
+ ASSERT_EQ(keys.GetString(map_array->value_offset(0)), "key");
+ ASSERT_EQ(values.GetString(map_array->value_offset(0)), "value");
+ ASSERT_EQ(map_array->value_length(2), 1);
+ ASSERT_TRUE(values.IsNull(map_array->value_offset(2)));
+ ASSERT_EQ(map_array->value_length(3), 2);
+}
+
+TEST_F(BlobFormatWriterMapBlobTest,
TestInvalidKeysFailWriteWithoutPartialData) {
+ const std::string key_context =
+ "invalid key of entry 1 of MAP<..., BLOB> field map_blob_col in row 0
of blob file";
+ std::shared_ptr<arrow::Array> null_keys =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["key",
null])").ValueOrDie();
+ std::shared_ptr<arrow::ArrayData> null_key_data =
null_keys->data()->Copy();
+ null_key_data->SetNullCount(0);
+ std::shared_ptr<arrow::Array> keys = arrow::MakeArray(null_key_data);
+ ASSERT_EQ(keys->null_count(), 0);
+ ASSERT_TRUE(keys->IsNull(1));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> null_key_row,
+ MakeMapBlobRow(struct_type_, keys, {"a", "b"}));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> writer,
CreateDefaultWriter());
+ const std::vector<std::pair<Status, std::string>> failures = {
+ {AddMapBlobRow(writer, MapEntries{{"key", "first"}, {"key",
"second"}}),
+ "MAP<..., BLOB> keys must be unique"},
+ {AddMapBlobRow(writer,
+ MapEntries{{"key", std::nullopt}, {"key",
std::nullopt}, {"x", "x"}}),
+ "MAP<..., BLOB> keys must be unique"},
+ {AddMapBlobRow(writer, MapEntries{{"key", std::nullopt}, {"key",
"value"}}),
+ "MAP<..., BLOB> keys must be unique"},
+ {AddMapBlobRow(writer, MapEntries{{"key", "first"}, {"\xff",
"second"}}),
+ "invalid UTF-8 in MAP<STRING, BLOB> key"},
+ {AddBatchOnce(writer, null_key_row), "MAP<..., BLOB> keys cannot be
null"}};
+ for (size_t i = 0; i < failures.size(); ++i) {
+ SCOPED_TRACE(fmt::format("case {}", i));
+ ASSERT_NOK_WITH_MSG(failures[i].first, key_context);
+ ASSERT_NOK_WITH_MSG(failures[i].first, failures[i].second);
+ }
+ ASSERT_OK(AddMapBlobRow(writer, MapEntries{{"key", "value"}}));
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK_AND_ASSIGN(std::vector<std::optional<MapEntries>> actual,
+ ReadBackMaps(/*blob_as_descriptor=*/false));
+ const std::vector<std::optional<MapEntries>> expected =
{MapEntries{{"key", "value"}}};
+ ASSERT_EQ(actual, expected);
+
+ const auto decimal_struct_type = MapBlobStructType(arrow::decimal128(10,
2));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> decimal_out,
+ file_system_->Create(dir_->Str() + "/decimal.blob",
/*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<BlobFormatWriter> decimal_writer,
+ BlobFormatWriter::Create(decimal_out, decimal_struct_type,
+ /*write_null_on_missing_file=*/false,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_));
+ arrow::Decimal128Builder key_builder(arrow::decimal128(10, 2));
+ ASSERT_TRUE(key_builder.Append(arrow::Decimal128(100)).ok());
+ ASSERT_TRUE(key_builder.Append(arrow::Decimal128(12345678901LL)).ok());
+ std::shared_ptr<arrow::Array> decimal_keys;
+ ASSERT_TRUE(key_builder.Finish(&decimal_keys).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> decimal_row,
+ MakeMapBlobRow(decimal_struct_type, decimal_keys,
{"a", "b"}));
+ Status status = AddBatchOnce(decimal_writer, decimal_row);
+ ASSERT_NOK_WITH_MSG(status, key_context);
+ ASSERT_NOK_WITH_MSG(status, "MAP<..., BLOB> decimal key exceeds declared
precision");
+ ASSERT_OK_AND_ASSIGN(int64_t decimal_pos, decimal_out->GetPos());
+ ASSERT_EQ(decimal_pos, 0);
+ ASSERT_OK(decimal_out->Close());
+}
+
+TEST_F(BlobFormatWriterMapBlobTest, TestWriteNullOnUnreachableValues) {
+ ASSERT_OK_AND_ASSIGN(const std::string missing_descriptor,
+ DescriptorOf(dir_->Str() + "/not_exist_file"));
+ const MapEntries entries = {{"first", "1"}, {"missing",
missing_descriptor}, {"last", "2"}};
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlobFormatWriter> failing_writer,
CreateDefaultWriter());
+ Status status = AddMapBlobRow(failing_writer, entries);
+ ASSERT_NOK_WITH_MSG(
+ status, "value of entry 1 of MAP<..., BLOB> field map_blob_col in row
0 of blob file");
+ ASSERT_NOK_WITH_MSG(status, "not exists");
+
+ const std::string file_name = "write_null.blob";
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
+ file_system_->Create(dir_->Str() + "/" + file_name,
/*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<BlobFormatWriter> writer,
+ BlobFormatWriter::Create(out, struct_type_,
/*write_null_on_missing_file=*/true,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_));
+ ASSERT_OK(AddMapBlobRow(writer, entries));
+ ASSERT_OK_AND_ASSIGN(
+ uint64_t missing_nulls,
+
writer->GetWriterMetrics()->GetCounter(BlobMetrics::WRITE_NULL_ON_MISSING_FILE_COUNT));
+ ASSERT_EQ(missing_nulls, 1);
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK(out->Close());
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::optional<MapEntries>> actual,
+ ReadBackMaps(/*blob_as_descriptor=*/false,
file_name));
+ const std::vector<std::optional<MapEntries>> expected = {
+ MapEntries{{"first", "1"}, {"missing", std::nullopt}, {"last", "2"}}};
+ ASSERT_EQ(actual, expected);
+}
+
+TEST_F(BlobFormatWriterMapBlobTest, TestMapBlobOutputFailureFailsWrite) {
+ struct Case {
+ int64_t write_limit;
+ bool short_write;
+ bool fail_flush;
+ bool fails_on_finish;
+ std::string expected_error;
+ };
+ const int64_t key_data_begin = 13;
+ const int64_t value_data_begin = 16;
+ const int64_t key_index_begin = 21;
+ const int64_t value_index_begin = 22;
+ const int64_t key_index_length_begin = 23;
+ const int64_t value_index_length_begin = 27;
+ const int64_t entry_end = 43;
+ const std::string write_error = "failed to write blob file mock.blob: mock
write error";
+ const std::vector<Case> cases = {
+ {key_data_begin, false, false, false, write_error},
+ {key_data_begin + 1, true, false, false,
+ "failed to write blob file mock.blob: unexpected actual length 1 not
match with expect 3"},
+ {value_data_begin, false, false, false,
+ "failed to copy value of entry 0 of MAP<..., BLOB> field map_blob_col
in row 0 of blob "
+ "file mock.blob: " +
+ write_error},
+ {key_index_begin, false, false, false, write_error},
+ {value_index_begin, false, false, false, write_error},
+ {key_index_length_begin, false, false, false, write_error},
+ {value_index_length_begin, false, false, false, write_error},
+ {entry_end, false, false, true, write_error},
+ {std::numeric_limits<int64_t>::max(), false, true, true,
+ "failed to flush blob file mock.blob: mock flush error"}};
+ for (size_t i = 0; i < cases.size(); ++i) {
+ const Case& c = cases[i];
+ SCOPED_TRACE(fmt::format("case {}: write_limit={}, short_write={},
fail_flush={}", i,
+ c.write_limit, c.short_write, c.fail_flush));
+ auto out =
+ std::make_shared<FailingOutputStream>(c.write_limit,
c.short_write, c.fail_flush);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<BlobFormatWriter> writer,
+ BlobFormatWriter::Create(out, struct_type_,
/*write_null_on_missing_file=*/false,
+ /*write_null_on_fetch_failure=*/false,
file_system_, pool_));
+ Status status = AddMapBlobRow(writer, MapEntries{{"key", "value"}});
+ if (c.fails_on_finish) {
+ ASSERT_OK(status);
+ status = writer->Finish();
+ }
+ ASSERT_NOK_WITH_MSG(status, c.expected_error);
+ if (!c.fail_flush) {
+ ASSERT_OK_AND_ASSIGN(int64_t pos, out->GetPos());
+ ASSERT_EQ(pos, c.write_limit);
+ }
+ }
+}
+
} // namespace paimon::blob::test
diff --git a/test/inte/blob_table_inte_test.cpp
b/test/inte/blob_table_inte_test.cpp
index b4b05901..0d566595 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -611,6 +611,35 @@ class BlobTableInteTest : public testing::Test, public
::testing::WithParamInter
return array;
}
+ using MapBlobEntries = std::vector<std::pair<std::string,
std::optional<std::string>>>;
+
+ static Result<std::shared_ptr<arrow::Array>> MakeMapBlobArray(
+ const std::shared_ptr<arrow::DataType>& map_type,
+ const std::vector<std::optional<MapBlobEntries>>& rows) {
+ auto key_builder = std::make_shared<arrow::StringBuilder>();
+ auto item_builder = std::make_shared<arrow::LargeBinaryBuilder>();
+ arrow::MapBuilder map_builder(arrow::default_memory_pool(),
key_builder, item_builder,
+ map_type);
+ for (const std::optional<MapBlobEntries>& row : rows) {
+ if (!row) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.AppendNull());
+ continue;
+ }
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Append());
+ for (const auto& [key, value] : *row) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Append(key));
+ if (value) {
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(item_builder->Append(*value));
+ } else {
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(item_builder->AppendNull());
+ }
+ }
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Finish(&array));
+ return array;
+ }
+
void CheckArrayBlobColumn(const std::shared_ptr<arrow::StructArray>& rows,
const std::string& field_name, const
std::string& expected_json,
bool blob_as_descriptor) const {
@@ -1706,6 +1735,104 @@ TEST_P(BlobTableInteTest,
TestArrayBlobWithBlobFieldAcrossMultipleBlobFiles) {
ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
}
+TEST_P(BlobTableInteTest, TestMapBlobWriteAndPartialUpdate) {
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::int32()),
+ arrow::field(
+ "m0", arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value",
/*nullable=*/true)))};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, GetParam()},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::ROW_TRACKING_ENABLED, "true"},
+ {Options::DATA_EVOLUTION_ENABLED, "true"},
+ {Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE, "true"}};
+ CreateTable(fields, /*partition_keys=*/{}, options);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields);
+
+ auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [1, [["a", "va"], ["", "empty key"], ["null", null]]],
+ [2, null],
+ [3, []],
+ [4, [["b", "vb"]]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
+ WriteArray(table_path, {}, schema->field_names(),
{src_array}));
+ ASSERT_OK(Commit(table_path, commit_msgs0));
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), src_array));
+
+ auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[1]}),
R"([
+ [[["k", null], ["k", null]]],
+ [[["updated", "vu"], ["null", null]]],
+ [null],
+ [[["k", null], ["k", null]]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, {"m0"},
{update_array}));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
+ ASSERT_OK(Commit(table_path, commit_msgs1));
+
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [1, [["a", "va"], ["", "empty key"], ["null", null]]],
+ [2, [["updated", "vu"], ["null", null]]],
+ [3, null],
+ [4, [["b", "vb"]]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ const std::string alpha_path = blob_dir_->Str() + "/alpha.bin";
+ ASSERT_OK(std::make_shared<LocalFileSystem>()->WriteFile(alpha_path,
"alpha",
+
/*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> alpha_blob,
Blob::FromPath(alpha_path));
+ PAIMON_UNIQUE_PTR<Bytes> alpha_descriptor =
alpha_blob->ToDescriptor(pool_);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> missing_blob,
+ Blob::FromPath(blob_dir_->Str() + "/missing.bin"));
+ PAIMON_UNIQUE_PTR<Bytes> missing_descriptor =
missing_blob->ToDescriptor(pool_);
+ const MapBlobEntries placeholder = {{"k", std::nullopt}, {"k",
std::nullopt}};
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<arrow::Array> update_map,
+ MakeMapBlobArray(
+ fields[1]->type(),
+ {MapBlobEntries{
+ {"missing", std::string(missing_descriptor->data(),
missing_descriptor->size())},
+ {"desc", std::string(alpha_descriptor->data(),
alpha_descriptor->size())}},
+ placeholder, placeholder, placeholder}));
+ auto second_update_array = arrow::StructArray::Make({update_map},
{fields[1]}).ValueOrDie();
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
+ WriteArray(table_path, {}, {"m0"},
{second_update_array}));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs2);
+ ASSERT_OK(Commit(table_path, commit_msgs2));
+ blob_dir_.reset();
+
+ expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ [1, [["missing", null], ["desc", "alpha"]]],
+ [2, [["updated", "vu"], ["null", null]]],
+ [3, null],
+ [4, [["b", "vb"]]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
+ std::map<std::string, std::string> read_options =
{{Options::BLOB_AS_DESCRIPTOR, "true"}};
+ ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path,
schema->field_names(), plan,
+ /*predicate=*/nullptr,
read_options));
+ ASSERT_TRUE(result);
+ auto rows = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::Concatenate(result->chunks()).ValueOrDie());
+ ASSERT_TRUE(rows);
+ CheckMapBlobColumn(rows, "m0",
+ R"([[["missing", null], ["desc", "alpha"]],
+ [["updated", "vu"], ["null", null]], null, [["b",
"vb"]]])",
+ /*blob_as_descriptor=*/true);
+}
+
TEST_P(BlobTableInteTest, TestDataEvolutionBlobPartialUpdateMultipleLayers) {
arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::utf8()),
BlobUtils::ToArrowField("b0")};
