From 0f4cd637adabd2b19b6411913ffac33d97773c33 Mon Sep 17 00:00:00 2001 From: "menglingda.mld" Date: Thu, 9 Jul 2026 15:42:54 +0800 Subject: [PATCH 1/2] feat(shared-shredding): Expose shared-shredding utilities --- .../map_shared_shredding_schema_utils.h | 94 +++++++++ src/paimon/CMakeLists.txt | 2 + .../map_shared_shredding_batch_converter.cpp | 5 +- .../map_shared_shredding_file_reader_test.cpp | 70 ++++--- .../map_shared_shredding_schema_utils.cpp | 96 +++++++++ ...map_shared_shredding_schema_utils_test.cpp | 197 ++++++++++++++++++ .../shredding/map_shared_shredding_utils.cpp | 7 +- .../shredding/map_shared_shredding_utils.h | 19 +- .../map_shared_shredding_utils_test.cpp | 20 +- .../data/shredding/map_shredding_defs.h | 28 +-- .../core/append/append_only_writer_test.cpp | 63 +++--- .../core/mergetree/merge_tree_writer_test.cpp | 16 +- .../append_only_file_store_write_test.cpp | 17 +- .../key_value_file_store_write_test.cpp | 5 +- .../postpone/postpone_bucket_writer_test.cpp | 4 +- 15 files changed, 518 insertions(+), 125 deletions(-) create mode 100644 include/paimon/data/shredding/map_shared_shredding_schema_utils.h create mode 100644 src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp create mode 100644 src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp diff --git a/include/paimon/data/shredding/map_shared_shredding_schema_utils.h b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h new file mode 100644 index 000000000..a9cfa4518 --- /dev/null +++ b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h @@ -0,0 +1,94 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "paimon/arrow/abi.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// Parsed file-level meta for one shared-shredding MAP column. +struct PAIMON_EXPORT MapSharedShreddingFieldMeta { + /// field_name -> field_id + std::map name_to_id; + /// field_id -> set of physical column indices S + std::map> field_to_columns; + /// Set of field_ids that ever spilled into __overflow + std::set overflow_field_set; + /// Number of physical columns K in this file + int32_t num_columns = 0; + /// Maximum row width observed in this file + int32_t max_row_width = 0; + + bool operator==(const MapSharedShreddingFieldMeta& other) const { + if (this == &other) { + return true; + } + return name_to_id == other.name_to_id && field_to_columns == other.field_to_columns && + overflow_field_set == other.overflow_field_set && num_columns == other.num_columns && + max_row_width == other.max_row_width; + } +}; + +class PAIMON_EXPORT MapSharedShreddingSchemaUtils { + public: + MapSharedShreddingSchemaUtils() = delete; + ~MapSharedShreddingSchemaUtils() = delete; + + /// Converts a logical schema to a physical schema by replacing shredding MAP columns + /// with their physical Struct representation. + /// @param logical_schema The original Arrow C schema with MAP columns. + /// Ownership of schema resources is transferred to this method. + /// @param field_to_num_columns Map from field name to its physical column count K. + /// Each shredding column can have its own width. + /// @return The exported Arrow C schema for file writing. + static Result> LogicalToPhysicalSchema( + std::unique_ptr<::ArrowSchema> logical_schema, + const std::map& field_to_num_columns); + + /// Attaches shared-shredding metadata to fields in a physical schema. + /// @param physical_schema The Arrow C physical schema whose fields should receive metadata. + /// Ownership of schema resources is transferred to this method. + /// @param field_name_to_meta Map from physical field name to its shared-shredding metadata. + /// @param compression Compression codec name for field_dict serialization. + /// @return A new Arrow C schema with shared-shredding metadata attached to matching fields. + static Result> AttachMetadataToSchema( + std::unique_ptr<::ArrowSchema> physical_schema, + const std::map& field_name_to_meta, + const std::string& compression); + + /// Extracts shared-shredding metadata from a named field in a physical schema. + /// @param physical_schema The Arrow C physical schema that contains the target field. + /// Ownership of schema resources is transferred to this method. + /// @param field_name The physical field name whose metadata should be extracted. + /// @param compression Compression codec name for field_dict deserialization. + /// @return Parsed shared-shredding metadata for the field. + static Result ExtractMetadataFromField( + std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name, + const std::string& compression); +}; + +} // namespace paimon diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index b044badea..fb332ef75 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -139,6 +139,7 @@ set(PAIMON_COMMON_SRCS common/utils/crc32c.cpp common/utils/decimal_utils.cpp common/data/shredding/map_shared_shredding_utils.cpp + common/data/shredding/map_shared_shredding_schema_utils.cpp common/data/shredding/map_shared_shredding_context.cpp common/data/shredding/map_shared_shredding_batch_converter.cpp common/data/shredding/map_shared_shredding_column_allocator.cpp @@ -547,6 +548,7 @@ if(PAIMON_BUILD_TESTS) common/utils/threadsafe_queue_test.cpp common/utils/generic_lru_cache_test.cpp common/data/shredding/map_shared_shredding_utils_test.cpp + common/data/shredding/map_shared_shredding_schema_utils_test.cpp common/data/shredding/map_shared_shredding_batch_converter_test.cpp common/data/shredding/lru_map_shared_shredding_column_allocator_test.cpp common/data/shredding/plain_map_shared_shredding_column_allocator_test.cpp diff --git a/src/paimon/common/data/shredding/map_shared_shredding_batch_converter.cpp b/src/paimon/common/data/shredding/map_shared_shredding_batch_converter.cpp index c29145f69..9072dae5d 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_batch_converter.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_batch_converter.cpp @@ -65,8 +65,9 @@ Result> MapSharedShreddingBatc const std::shared_ptr& context, const CoreOptions& options, const std::shared_ptr& pool) { std::map field_to_num_columns = context->ComputeNextK(); - std::shared_ptr physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, field_to_num_columns); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, field_to_num_columns)); std::vector contexts; std::vector shredding_field_names; contexts.reserve(field_to_num_columns.size()); diff --git a/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp b/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp index b3e23afa6..ab386088d 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp @@ -63,17 +63,18 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test { return meta; } - std::shared_ptr PhysicalSchemaWithMetadata() const { + Result> PhysicalSchemaWithMetadata() const { return PhysicalSchemaWithMetadata(TagsMeta()); } - std::shared_ptr PhysicalSchemaWithMetadata( + Result> PhysicalSchemaWithMetadata( const MapSharedShreddingFieldMeta& meta) const { std::map field_to_num_columns = {{"tags", 2}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema_, field_to_num_columns); + PAIMON_ASSIGN_OR_RAISE(auto physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema_, field_to_num_columns)); auto metadata = std::make_shared(); - EXPECT_OK(MapSharedShreddingUtils::SerializeMetadata( + PAIMON_RETURN_NOT_OK(MapSharedShreddingUtils::SerializeMetadata( meta, MapSharedShreddingDefine::kDefaultDictCompression, metadata.get())); arrow::FieldVector fields = physical_schema->fields(); @@ -81,8 +82,9 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test { return arrow::schema(std::move(fields)); } - std::shared_ptr PhysicalArray() const { - std::shared_ptr physical_schema = PhysicalSchemaWithMetadata(); + Result> PhysicalArray() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr physical_schema, + PhysicalSchemaWithMetadata()); std::string json = R"([ [1, [[0, 1], 10, 20, null]], [2, [[2, 0], 30, 40, null]], @@ -140,15 +142,15 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test { std::move(reader), std::move(shared_shredding_name_to_context), pool_); } - std::unique_ptr CreateReader( + Result> CreateReader( std::shared_ptr physical_array = nullptr, std::shared_ptr physical_schema = nullptr, const std::optional& selected_keys = std::nullopt) const { if (!physical_schema) { - physical_schema = PhysicalSchemaWithMetadata(); + PAIMON_ASSIGN_OR_RAISE(physical_schema, PhysicalSchemaWithMetadata()); } if (!physical_array) { - physical_array = PhysicalArray(); + PAIMON_ASSIGN_OR_RAISE(physical_array, PhysicalArray()); } auto mock_reader = std::make_unique( physical_array, arrow::struct_(physical_schema->fields()), /*read_batch_size=*/10); @@ -243,7 +245,7 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test { }; TEST_F(MapSharedShreddingFileReaderTest, TestGetFileSchemaReturnsLogicalMapSchema) { - auto reader = CreateReader(); + ASSERT_OK_AND_ASSIGN(auto reader, CreateReader()); ASSERT_OK_AND_ASSIGN(auto c_schema, reader->GetFileSchema()); auto schema = arrow::ImportSchema(c_schema.get()).ValueOrDie(); @@ -256,8 +258,9 @@ TEST_F(MapSharedShreddingFileReaderTest, TestGetFileSchemaReturnsLogicalMapSchem } TEST_F(MapSharedShreddingFileReaderTest, TestAllExistSelectedKeysWithoutOverflow) { - auto reader = CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, - /*selected_keys=*/"b"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, + /*selected_keys=*/"b")); auto read_schema = ExportSchema(ReadSchema("b")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -277,8 +280,9 @@ TEST_F(MapSharedShreddingFileReaderTest, TestAllExistSelectedKeysWithoutOverflow } TEST_F(MapSharedShreddingFileReaderTest, TestAllExistSelectedKeysWithOverflow) { - auto reader = CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, - /*selected_keys=*/"a,c"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, + /*selected_keys=*/"a,c")); auto read_schema = ExportSchema(ReadSchema("a,c")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -298,8 +302,9 @@ TEST_F(MapSharedShreddingFileReaderTest, TestAllExistSelectedKeysWithOverflow) { } TEST_F(MapSharedShreddingFileReaderTest, TestPartialExistSelectedKeys) { - auto reader = CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, - /*selected_keys=*/"a,c,missing"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, + /*selected_keys=*/"a,c,missing")); auto read_schema = ExportSchema(ReadSchema("a,c,missing")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -320,7 +325,7 @@ TEST_F(MapSharedShreddingFileReaderTest, TestPartialExistSelectedKeys) { } TEST_F(MapSharedShreddingFileReaderTest, TestMissingSelectedKeysReadsWholeMap) { - auto reader = CreateReader(); + ASSERT_OK_AND_ASSIGN(auto reader, CreateReader()); auto read_schema = ExportSchema(ReadSchema(std::nullopt)); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -345,7 +350,7 @@ TEST_F(MapSharedShreddingFileReaderTest, TestSpecialSelectedKeys) { meta.field_to_columns = {{0, {0}}, {1, {1}}, {2, {0}}, {3, {1}}}; meta.num_columns = 2; meta.max_row_width = 2; - auto physical_schema = PhysicalSchemaWithMetadata(meta); + ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata(meta)); std::string json = R"([ [1, [[0, 1], 10, 20, null]], [2, [[2, 3], 30, 40, null]], @@ -356,7 +361,8 @@ TEST_F(MapSharedShreddingFileReaderTest, TestSpecialSelectedKeys) { .ValueOrDie(); auto assert_read = [&](const std::string& selected_keys, const std::string& expected_json) { - auto reader = CreateReader(physical_array, physical_schema, selected_keys); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(physical_array, physical_schema, selected_keys)); auto read_schema = ExportSchema(ReadSchema(selected_keys)); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -392,8 +398,9 @@ TEST_F(MapSharedShreddingFileReaderTest, TestSpecialSelectedKeys) { } TEST_F(MapSharedShreddingFileReaderTest, TestUnknownSelectedKeyReturnsEmptyMap) { - auto reader = CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, - /*selected_keys=*/"missing"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr, + /*selected_keys=*/"missing")); auto read_schema = ExportSchema(ReadSchema("missing")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -414,14 +421,15 @@ TEST_F(MapSharedShreddingFileReaderTest, TestUnknownSelectedKeyReturnsEmptyMap) } TEST_F(MapSharedShreddingFileReaderTest, TestInvalidNullFieldMappingField) { - auto physical_schema = PhysicalSchemaWithMetadata(); + ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata()); std::string json = R"([ [1, [null, 10, null, null]] ])"; auto physical_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(physical_schema->fields()), json) .ValueOrDie(); - auto reader = CreateReader(physical_array, physical_schema, /*selected_keys=*/"a"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(physical_array, physical_schema, /*selected_keys=*/"a")); auto read_schema = ExportSchema(ReadSchema("a")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -430,14 +438,15 @@ TEST_F(MapSharedShreddingFileReaderTest, TestInvalidNullFieldMappingField) { } TEST_F(MapSharedShreddingFileReaderTest, TestInvalidNullFieldMappingFieldElement) { - auto physical_schema = PhysicalSchemaWithMetadata(); + ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata()); std::string json = R"([ [1, [[0, null], 10, null, null]] ])"; auto physical_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(physical_schema->fields()), json) .ValueOrDie(); - auto reader = CreateReader(physical_array, physical_schema, /*selected_keys=*/"b"); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(physical_array, physical_schema, /*selected_keys=*/"b")); auto read_schema = ExportSchema(ReadSchema("b")); ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt)); @@ -458,8 +467,8 @@ TEST_F(MapSharedShreddingFileReaderTest, TestListValue) { meta.max_row_width = 3; std::map field_to_num_columns = {{"tags", 2}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, field_to_num_columns); + ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, field_to_num_columns)); auto metadata = std::make_shared(); ASSERT_OK(MapSharedShreddingUtils::SerializeMetadata( meta, MapSharedShreddingDefine::kDefaultDictCompression, metadata.get())); @@ -475,8 +484,9 @@ TEST_F(MapSharedShreddingFileReaderTest, TestListValue) { [4, [[1, 0], [8], [9, 10], [[2, [null]]]]] ])") .ValueOrDie(); - auto reader = CreateReader(physical_array, physical_schema, - /*selected_keys=*/"a,c"); // NOLINT(whitespace/comma) + ASSERT_OK_AND_ASSIGN(auto reader, + CreateReader(physical_array, physical_schema, + /*selected_keys=*/"a,c")); // NOLINT(whitespace/comma) auto read_metadata = std::make_shared(); read_metadata->Append("paimon.map.selected-keys", "a,c"); diff --git a/src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp new file mode 100644 index 000000000..f713053f7 --- /dev/null +++ b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp @@ -0,0 +1,96 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/data/shredding/map_shared_shredding_schema_utils.h" + +#include "arrow/c/bridge.h" +#include "arrow/type.h" +#include "arrow/util/key_value_metadata.h" +#include "fmt/format.h" +#include "paimon/common/data/shredding/map_shared_shredding_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" + +namespace paimon { + +Result> MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema( + std::unique_ptr<::ArrowSchema> logical_schema, + const std::map& field_to_num_columns) { + if (!logical_schema) { + return Status::Invalid("logical schema is null"); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_logical_schema, + arrow::ImportSchema(logical_schema.get())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(arrow_logical_schema, + field_to_num_columns)); + auto c_schema = std::make_unique<::ArrowSchema>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*physical_schema, c_schema.get())); + return c_schema; +} + +Result> MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::unique_ptr<::ArrowSchema> physical_schema, + const std::map& field_name_to_meta, + const std::string& compression) { + if (!physical_schema) { + return Status::Invalid("physical schema is null"); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_physical_schema, + arrow::ImportSchema(physical_schema.get())); + + arrow::FieldVector updated_fields = arrow_physical_schema->fields(); + for (const auto& [field_name, field_meta] : field_name_to_meta) { + int32_t field_index = arrow_physical_schema->GetFieldIndex(field_name); + if (field_index < 0) { + return Status::Invalid(fmt::format( + "Shared-shredding field '{}' not found in physical schema.", field_name)); + } + + const auto& field = arrow_physical_schema->field(field_index); + auto metadata = field->metadata() ? field->metadata()->Copy() + : std::make_shared(); + PAIMON_RETURN_NOT_OK( + MapSharedShreddingUtils::SerializeMetadata(field_meta, compression, metadata.get())); + updated_fields[field_index] = field->WithMetadata(metadata); + } + + auto updated_schema = + arrow::schema(std::move(updated_fields), arrow_physical_schema->metadata()); + auto c_schema = std::make_unique<::ArrowSchema>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*updated_schema, c_schema.get())); + return c_schema; +} + +Result MapSharedShreddingSchemaUtils::ExtractMetadataFromField( + std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name, + const std::string& compression) { + if (!physical_schema) { + return Status::Invalid("physical schema is null"); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_physical_schema, + arrow::ImportSchema(physical_schema.get())); + const auto& field = arrow_physical_schema->GetFieldByName(field_name); + if (!field) { + return Status::Invalid( + fmt::format("Shared-shredding field '{}' not found in physical schema.", field_name)); + } + + auto metadata = + field->metadata() ? field->metadata()->Copy() : std::shared_ptr(); + return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression); +} + +} // namespace paimon diff --git a/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp new file mode 100644 index 000000000..cb1182596 --- /dev/null +++ b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp @@ -0,0 +1,197 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/data/shredding/map_shared_shredding_schema_utils.h" + +#include +#include + +#include "arrow/c/bridge.h" +#include "arrow/type.h" +#include "arrow/util/key_value_metadata.h" +#include "gtest/gtest.h" +#include "paimon/common/data/shredding/map_shared_shredding_utils.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaBasic) { + MapSharedShreddingFieldMeta tags_meta; + tags_meta.name_to_id = {{"host", 0}, {"region", 1}}; + tags_meta.field_to_columns = {{0, {0}}, {1, {1}}}; + tags_meta.num_columns = 2; + tags_meta.max_row_width = 2; + + auto id_metadata = std::make_shared(); + id_metadata->Append("paimon.field.id", "1"); + auto schema_metadata = std::make_shared(); + schema_metadata->Append("schema.key", "schema.value"); + auto schema = arrow::schema( + {arrow::field("id", arrow::int32(), true, id_metadata), + arrow::field("tags", arrow::struct_({arrow::field("__field_mapping", + arrow::list(arrow::int32()), true), + arrow::field("__col_0", arrow::utf8(), true), + arrow::field("__col_1", arrow::utf8(), true)}))}, + schema_metadata); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_OK_AND_ASSIGN(auto c_updated_schema, + MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::move(c_schema), {{"tags", tags_meta}}, "none")); + ASSERT_TRUE(c_updated_schema); + ASSERT_TRUE(c_updated_schema->release); + auto updated_schema = arrow::ImportSchema(c_updated_schema.get()).ValueOrDie(); + + ASSERT_TRUE(updated_schema->metadata()->Equals(*schema_metadata)); + ASSERT_TRUE(updated_schema->field(0)->metadata()->Equals(*id_metadata)); + + auto tags_metadata = updated_schema->GetFieldByName("tags")->metadata()->Copy(); + ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(tags_metadata)); + ASSERT_OK_AND_ASSIGN(auto deserialized, + MapSharedShreddingUtils::DeserializeMetadata(tags_metadata, "none")); + ASSERT_EQ(deserialized, tags_meta); +} + +TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaMissingField) { + MapSharedShreddingFieldMeta tags_meta; + tags_meta.name_to_id = {{"host", 0}}; + tags_meta.field_to_columns = {{0, {0}}}; + tags_meta.num_columns = 1; + tags_meta.max_row_width = 1; + + auto schema = arrow::schema({arrow::field("id", arrow::int32())}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::move(c_schema), {{"tags", tags_meta}}, "none"), + "Shared-shredding field 'tags' not found in physical schema."); +} + +TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaPreservesExistingFieldMetadata) { + MapSharedShreddingFieldMeta tags_meta; + tags_meta.name_to_id = {{"host", 0}}; + tags_meta.field_to_columns = {{0, {0}}}; + tags_meta.num_columns = 1; + tags_meta.max_row_width = 1; + + auto tags_metadata = std::make_shared(); + tags_metadata->Append("paimon.field.id", "7"); + tags_metadata->Append("description", "original tags field"); + auto schema = arrow::schema({arrow::field( + "tags", + arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true), + arrow::field("__col_0", arrow::utf8(), true)}), + true, tags_metadata)}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_OK_AND_ASSIGN(auto c_updated_schema, + MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::move(c_schema), {{"tags", tags_meta}}, "none")); + ASSERT_TRUE(c_updated_schema); + ASSERT_TRUE(c_updated_schema->release); + auto updated_schema = arrow::ImportSchema(c_updated_schema.get()).ValueOrDie(); + + auto updated_metadata = updated_schema->field(0)->metadata()->Copy(); + ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("paimon.field.id")), "7"); + ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("description")), + "original tags field"); + ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(updated_metadata)); +} + +TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromField) { + MapSharedShreddingFieldMeta tags_meta; + tags_meta.name_to_id = {{"host", 0}, {"region", 1}}; + tags_meta.field_to_columns = {{0, {0}}, {1, {1}}}; + tags_meta.num_columns = 2; + tags_meta.max_row_width = 2; + + auto metadata = std::make_shared(); + ASSERT_OK(MapSharedShreddingUtils::SerializeMetadata(tags_meta, "none", metadata.get())); + auto field = arrow::field( + "tags", + arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true), + arrow::field("__col_0", arrow::utf8(), true), + arrow::field("__col_1", arrow::utf8(), true)}), + true, metadata); + auto schema = arrow::schema({arrow::field("id", arrow::int32()), field}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_OK_AND_ASSIGN(auto parsed_meta, MapSharedShreddingSchemaUtils::ExtractMetadataFromField( + std::move(c_schema), "tags", "none")); + ASSERT_EQ(parsed_meta, tags_meta); +} + +TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldNoShreddingMetadata) { + auto schema = + arrow::schema({arrow::field("id", arrow::int32()), arrow::field("tags", arrow::utf8())}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema), + "tags", "none"), + "metadata is null or storage layout is not shared-shredding"); +} + +TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaNonMapField) { + auto schema = + arrow::schema({arrow::field("ts", arrow::int64()), arrow::field("tags", arrow::int32())}); + std::map field_to_num_columns = {{"tags", 2}}; + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema( + std::move(c_schema), field_to_num_columns), + "Field 'tags' is expected to be MAP type"); +} + +TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaNestedListValue) { + // MAP, b: array, c: array>> + auto nested_value = arrow::struct_({arrow::field("a", arrow::list(arrow::int32())), + arrow::field("b", arrow::list(arrow::int32())), + arrow::field("c", arrow::list(arrow::int32()))}); + auto map_type = arrow::map(arrow::utf8(), nested_value); + auto schema = + arrow::schema({arrow::field("ts", arrow::int64()), arrow::field("data", map_type)}); + + std::map field_to_num_columns = {{"data", 3}}; + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_OK_AND_ASSIGN(auto c_physical_schema, + MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema( + std::move(c_schema), field_to_num_columns)); + ASSERT_TRUE(c_physical_schema); + ASSERT_TRUE(c_physical_schema->release); + auto physical_schema = arrow::ImportSchema(c_physical_schema.get()).ValueOrDie(); + + auto expected_struct = arrow::struct_({ + arrow::field("__field_mapping", arrow::list(arrow::int32()), true), + arrow::field("__col_0", nested_value, true), + arrow::field("__col_1", nested_value, true), + arrow::field("__col_2", nested_value, true), + arrow::field("__overflow", arrow::map(arrow::int32(), nested_value), true), + }); + auto expected_schema = arrow::schema( + {arrow::field("ts", arrow::int64()), arrow::field("data", expected_struct, true)}); + ASSERT_TRUE(physical_schema->Equals(expected_schema)); +} + +} // namespace paimon::test diff --git a/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp b/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp index ffda4c596..391378bde 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp @@ -125,7 +125,7 @@ std::shared_ptr MapSharedShreddingUtils::InnerBuildSpecificPhys return arrow::struct_(std::move(struct_fields)); } -std::shared_ptr MapSharedShreddingUtils::LogicalToPhysicalSchema( +Result> MapSharedShreddingUtils::LogicalToPhysicalSchema( const std::shared_ptr& logical_schema, const std::map& field_to_num_columns) { arrow::FieldVector physical_fields; @@ -135,6 +135,11 @@ std::shared_ptr MapSharedShreddingUtils::LogicalToPhysicalSchema( const auto& field = logical_schema->field(i); auto it = field_to_num_columns.find(field->name()); if (it != field_to_num_columns.end()) { + if (field->type()->id() != arrow::Type::MAP) { + return Status::Invalid( + fmt::format("Field '{}' is expected to be MAP type, but got '{}'.", + field->name(), field->type()->name())); + } auto map_type = std::static_pointer_cast(field->type()); auto value_type = map_type->item_type(); bool value_nullable = map_type->item_field()->nullable(); diff --git a/src/paimon/common/data/shredding/map_shared_shredding_utils.h b/src/paimon/common/data/shredding/map_shared_shredding_utils.h index a7fa59cb9..6f3a8d06e 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_utils.h +++ b/src/paimon/common/data/shredding/map_shared_shredding_utils.h @@ -26,6 +26,7 @@ #include "arrow/type.h" #include "paimon/common/data/shredding/map_shredding_defs.h" +#include "paimon/data/shredding/map_shared_shredding_schema_utils.h" #include "paimon/result.h" #include "paimon/status.h" @@ -68,7 +69,7 @@ class MapSharedShreddingUtils { /// @param field_to_num_columns Map from field name to its physical column count K. /// Each shredding column can have its own width. /// @return The physical schema for file writing. - static std::shared_ptr LogicalToPhysicalSchema( + static Result> LogicalToPhysicalSchema( const std::shared_ptr& logical_schema, const std::map& field_to_num_columns); @@ -83,6 +84,14 @@ class MapSharedShreddingUtils { // ---- Metadata serialization ---- + /// Serializes shredding metadata and appends entries to an existing KeyValueMetadata. + /// @param field_meta The field-level shredding metadata to serialize. + /// @param compression Compression codec name for field_dict compression. + /// @param[out] metadata The KeyValueMetadata to append entries to. + static Status SerializeMetadata(const MapSharedShreddingFieldMeta& field_meta, + const std::string& compression, + arrow::KeyValueMetadata* metadata); + /// Deserializes shredding metadata from file footer KeyValueMetadata (per field). /// @param metadata The KeyValueMetadata from file footer. /// @param compression Compression codec name. @@ -139,14 +148,6 @@ class MapSharedShreddingUtils { static Result> BuildColumnToNumColumns( const std::vector& shredding_field_names, const CoreOptions& options); - /// Serializes shredding metadata and appends entries to an existing KeyValueMetadata. - /// @param field_meta The field-level shredding metadata to serialize. - /// @param compression Compression codec name for field_dict compression. - /// @param[out] metadata The KeyValueMetadata to append entries to. - static Status SerializeMetadata(const MapSharedShreddingFieldMeta& field_meta, - const std::string& compression, - arrow::KeyValueMetadata* metadata); - /// Builds the physical Arrow type for one shredding MAP column. /// @param value_type The value type of the original MAP. /// @param num_columns Number of physical columns K. diff --git a/src/paimon/common/data/shredding/map_shared_shredding_utils_test.cpp b/src/paimon/common/data/shredding/map_shared_shredding_utils_test.cpp index 7a1e87f2b..533186dfb 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_utils_test.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_utils_test.cpp @@ -88,8 +88,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaBasic) { }); std::map field_to_num_columns = {{"tags", 4}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, field_to_num_columns); + ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + schema, field_to_num_columns)); // Build expected schema for comparison auto expected_struct = arrow::struct_({ @@ -116,8 +116,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNestedValue) { auto schema = arrow::schema({arrow::field("data", map_type)}); std::map field_to_num_columns = {{"data", 2}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, field_to_num_columns); + ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + schema, field_to_num_columns)); auto expected_struct = arrow::struct_({ arrow::field("__field_mapping", arrow::list(arrow::int32()), true), @@ -135,7 +135,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNullable) { auto schema_nullable = arrow::schema({arrow::field("m", nullable_map)}); std::map col_map = {{"m", 2}}; - auto physical = MapSharedShreddingUtils::LogicalToPhysicalSchema(schema_nullable, col_map); + ASSERT_OK_AND_ASSIGN( + auto physical, MapSharedShreddingUtils::LogicalToPhysicalSchema(schema_nullable, col_map)); auto struct_type = physical->field(0)->type(); ASSERT_TRUE(struct_type->field(0)->nullable()); ASSERT_TRUE(struct_type->field(1)->nullable()); @@ -145,7 +146,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNullable) { auto non_nullable_map = arrow::map(arrow::utf8(), arrow::field("item", arrow::int64(), false)); auto schema_non_nullable = arrow::schema({arrow::field("m", non_nullable_map)}); - auto physical2 = MapSharedShreddingUtils::LogicalToPhysicalSchema(schema_non_nullable, col_map); + ASSERT_OK_AND_ASSIGN(auto physical2, MapSharedShreddingUtils::LogicalToPhysicalSchema( + schema_non_nullable, col_map)); auto struct_type2 = physical2->field(0)->type(); ASSERT_FALSE(struct_type2->field(1)->nullable()); ASSERT_FALSE(struct_type2->field(2)->nullable()); @@ -159,7 +161,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaPreservesFieldMetadata) auto schema = arrow::schema({arrow::field("m", map_type, false, metadata)}); std::map col_map = {{"m", 2}}; - auto physical_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, col_map); + ASSERT_OK_AND_ASSIGN(auto physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, col_map)); ASSERT_FALSE(physical_schema->field(0)->nullable()); ASSERT_TRUE(physical_schema->field(0)->metadata()->Equals(*metadata)); @@ -172,7 +175,8 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNoShreddingColumns) { }); std::map empty_map; - auto physical_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, empty_map); + ASSERT_OK_AND_ASSIGN(auto physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, empty_map)); ASSERT_TRUE(physical_schema->Equals(schema)); } diff --git a/src/paimon/common/data/shredding/map_shredding_defs.h b/src/paimon/common/data/shredding/map_shredding_defs.h index 82ac15a14..580d98ad4 100644 --- a/src/paimon/common/data/shredding/map_shredding_defs.h +++ b/src/paimon/common/data/shredding/map_shredding_defs.h @@ -17,10 +17,9 @@ #pragma once #include -#include -#include #include -#include + +#include "paimon/data/shredding/map_shared_shredding_schema_utils.h" namespace paimon { @@ -71,27 +70,4 @@ struct MapSharedShreddingDefine { } }; -/// Parsed file-level meta for one shared-shredding MAP column. -struct MapSharedShreddingFieldMeta { - /// field_name -> field_id - std::map name_to_id; - /// field_id -> set of physical column indices S - std::map> field_to_columns; - /// Set of field_ids that ever spilled into __overflow - std::set overflow_field_set; - /// Number of physical columns K in this file - int32_t num_columns = 0; - /// Maximum row width observed in this file - int32_t max_row_width = 0; - - bool operator==(const MapSharedShreddingFieldMeta& other) const { - if (this == &other) { - return true; - } - return name_to_id == other.name_to_id && field_to_columns == other.field_to_columns && - overflow_field_set == other.overflow_field_set && num_columns == other.num_columns && - max_row_width == other.max_row_width; - } -}; - } // namespace paimon diff --git a/src/paimon/core/append/append_only_writer_test.cpp b/src/paimon/core/append/append_only_writer_test.cpp index a73e735dd..2d3d05427 100644 --- a/src/paimon/core/append/append_only_writer_test.cpp +++ b/src/paimon/core/append/append_only_writer_test.cpp @@ -971,8 +971,9 @@ TEST_P(AppendOnlyWriterShreddingTest, TestWriteSharedShreddingMapFieldContent) { // Check shared-shredding map metadata: a=0, b=1, c=2; K=3, max_row_width=3, no overflow. std::map column_to_k = {{"tags", 3}}; - auto expected_physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k); + ASSERT_OK_AND_ASSIGN( + auto expected_physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k)); MapSharedShreddingFieldMeta expected_meta; expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}}; @@ -1034,8 +1035,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapAllEmptyFirstFile) { path_factory->ToPath(inc.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map first_file_k = {{"tags", 3}}; - auto first_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, first_file_k); + ASSERT_OK_AND_ASSIGN(auto first_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, first_file_k)); MapSharedShreddingFieldMeta empty_meta; empty_meta.num_columns = 3; empty_meta.max_row_width = 0; @@ -1090,8 +1091,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapAllNullThenAllEmptyF path_factory->ToPath(null_inc.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map first_file_k = {{"tags", 3}}; - auto first_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, first_file_k); + ASSERT_OK_AND_ASSIGN(auto first_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, first_file_k)); MapSharedShreddingFieldMeta empty_meta; empty_meta.num_columns = 3; empty_meta.max_row_width = 0; @@ -1122,8 +1123,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapAllNullThenAllEmptyF // Previous file observed max_row_width=0, but the next file must still keep at least one // physical value column so shared-shredding never produces a K=0 schema. std::map second_file_k = {{"tags", 1}}; - auto second_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, second_file_k); + ASSERT_OK_AND_ASSIGN(auto second_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, second_file_k)); empty_meta.num_columns = 1; CheckShreddingFileSchema(empty_file_path, format, second_schema, /*field_index=*/1, empty_meta, options.GetFileCompression()); @@ -1216,8 +1217,9 @@ TEST_P(AppendOnlyWriterShreddingTest, TestWriteSharedShreddingMapWithOverflow) { path_factory->ToPath(inc.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k = {{"tags", 2}}; - auto expected_physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k); + ASSERT_OK_AND_ASSIGN( + auto expected_physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k)); std::string compression = options.GetFileCompression(); // Verify metadata: a=0,b=1,c=2,d=3,e=4,f=5; K=2, max_row_width=4 @@ -1285,8 +1287,9 @@ TEST_P(AppendOnlyWriterShreddingTest, TestWriteSharedShreddingMapWithLruPlacemen path_factory->ToPath(inc.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k = {{"tags", 3}}; - auto expected_physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k); + ASSERT_OK_AND_ASSIGN( + auto expected_physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k)); MapSharedShreddingFieldMeta expected_meta; expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}, {"d", 3}}; @@ -1350,8 +1353,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapKAdaptationAcrossFil // File 1 should have K=10 (first file uses K_max). std::map column_to_k_file1 = {{"tags", 10}}; - auto phys_schema1 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k_file1); + ASSERT_OK_AND_ASSIGN(auto phys_schema1, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, column_to_k_file1)); // Verify file1 physical schema has 10 columns. auto struct_type1 = std::static_pointer_cast(phys_schema1->field(1)->type()); ASSERT_EQ(12, struct_type1->num_fields()); // mapping + 10 cols + overflow @@ -1379,8 +1382,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapKAdaptationAcrossFil // File 2 should have K=3 (adapted from file1's max_row_width=3). std::map column_to_k_file2 = {{"tags", 3}}; - auto phys_schema2 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k_file2); + ASSERT_OK_AND_ASSIGN(auto phys_schema2, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, column_to_k_file2)); auto struct_type2 = std::static_pointer_cast(phys_schema2->field(1)->type()); ASSERT_EQ(5, struct_type2->num_fields()); // mapping + 3 cols + overflow @@ -1418,8 +1421,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapKAdaptationAcrossFil // File 3 should have K=5 (window max grew from file2's max_row_width=5). std::map column_to_k_file3 = {{"tags", 5}}; - auto phys_schema3 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k_file3); + ASSERT_OK_AND_ASSIGN(auto phys_schema3, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, column_to_k_file3)); auto struct_type3 = std::static_pointer_cast(phys_schema3->field(1)->type()); ASSERT_EQ(7, struct_type3->num_fields()); // mapping + 5 cols + overflow @@ -1482,8 +1485,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapUsesInitialContextFo path_factory->ToPath(inc.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k = {{"tags", 2}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, column_to_k); + ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, column_to_k)); MapSharedShreddingFieldMeta expected_meta; expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}}; expected_meta.field_to_columns = {{0, {0}}, {1, {1}}}; @@ -1552,8 +1555,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestMultipleSharedShreddingMapFieldsWithKA // Verify file1: tags K=8, attrs K=4 (first file uses K_max). std::map col_to_k_file1 = {{"tags", 8}, {"attrs", 4}}; - auto phys_schema1 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, col_to_k_file1); + ASSERT_OK_AND_ASSIGN(auto phys_schema1, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, col_to_k_file1)); MapSharedShreddingFieldMeta meta1_tags; meta1_tags.name_to_id = {{"a", 0}, {"b", 1}}; @@ -1584,8 +1587,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestMultipleSharedShreddingMapFieldsWithKA path_factory->ToPath(inc2.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map col_to_k_file2 = {{"tags", 2}, {"attrs", 1}}; - auto phys_schema2 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, col_to_k_file2); + ASSERT_OK_AND_ASSIGN(auto phys_schema2, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, col_to_k_file2)); MapSharedShreddingFieldMeta meta2_tags; meta2_tags.name_to_id = {{"c", 0}, {"d", 1}, {"e", 2}}; @@ -1618,8 +1621,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestMultipleSharedShreddingMapFieldsWithKA path_factory->ToPath(inc3.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map col_to_k_file3 = {{"tags", 3}, {"attrs", 3}}; - auto phys_schema3 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, col_to_k_file3); + ASSERT_OK_AND_ASSIGN(auto phys_schema3, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, col_to_k_file3)); MapSharedShreddingFieldMeta meta3_tags; meta3_tags.name_to_id = {{"f", 0}, {"g", 1}}; @@ -1701,7 +1704,8 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapDataFileMetaInfo) { // Verify the written file has correct shared-shredding map content. std::string file_path = path_factory->ToPath(actual_meta->file_name); std::map col_to_k = {{"tags", 3}}; - auto phys_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, col_to_k); + ASSERT_OK_AND_ASSIGN(auto phys_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, col_to_k)); auto physical_type = arrow::struct_(phys_schema->fields()); std::shared_ptr expected_array; @@ -1788,8 +1792,9 @@ TEST_P(AppendOnlyWriterShreddingTest, TestSharedShreddingMapWithBlobSeparation) arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())), }); std::map col_to_k = {{"tags", 3}}; - auto expected_physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(main_logical_schema, col_to_k); + ASSERT_OK_AND_ASSIGN( + auto expected_physical_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(main_logical_schema, col_to_k)); MapSharedShreddingFieldMeta expected_meta; expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}}; diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp b/src/paimon/core/mergetree/merge_tree_writer_test.cpp index ee46215e8..558b7047c 100644 --- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp @@ -448,8 +448,8 @@ TEST_P(MergeTreeWriterTest, TestSharedShreddingMapDataFileMetaInfo) { options.GetFileSystem()->GetFileStatus(expected_data_file_path)); std::map column_to_k = {{"tags", 3}}; - auto physical_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, column_to_k); + ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + write_schema, column_to_k)); auto physical_type = arrow::struct_(physical_schema->fields()); std::shared_ptr expected_array; @@ -541,8 +541,8 @@ TEST_P(MergeTreeWriterTest, TestSharedShreddingMultipleMapFieldsWithKAdaptation) path_factory->ToPath(commit_increment1.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k_file1 = {{"tags", 8}, {"attrs", 4}}; - auto physical_schema1 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, column_to_k_file1); + ASSERT_OK_AND_ASSIGN(auto physical_schema1, MapSharedShreddingUtils::LogicalToPhysicalSchema( + write_schema, column_to_k_file1)); MapSharedShreddingFieldMeta tags_meta1; tags_meta1.name_to_id = {{"a", 0}, {"b", 1}}; tags_meta1.field_to_columns = {{0, {0}}, {1, {1}}}; @@ -569,8 +569,8 @@ TEST_P(MergeTreeWriterTest, TestSharedShreddingMultipleMapFieldsWithKAdaptation) path_factory->ToPath(commit_increment2.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k_file2 = {{"tags", 2}, {"attrs", 1}}; - auto physical_schema2 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, column_to_k_file2); + ASSERT_OK_AND_ASSIGN(auto physical_schema2, MapSharedShreddingUtils::LogicalToPhysicalSchema( + write_schema, column_to_k_file2)); MapSharedShreddingFieldMeta tags_meta2; tags_meta2.name_to_id = {{"c", 0}, {"d", 1}, {"e", 2}}; tags_meta2.field_to_columns = {{0, {0}}, {1, {1}}}; @@ -599,8 +599,8 @@ TEST_P(MergeTreeWriterTest, TestSharedShreddingMultipleMapFieldsWithKAdaptation) path_factory->ToPath(commit_increment3.GetNewFilesIncrement().NewFiles()[0]->file_name); std::map column_to_k_file3 = {{"tags", 3}, {"attrs", 3}}; - auto physical_schema3 = - MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, column_to_k_file3); + ASSERT_OK_AND_ASSIGN(auto physical_schema3, MapSharedShreddingUtils::LogicalToPhysicalSchema( + write_schema, column_to_k_file3)); MapSharedShreddingFieldMeta tags_meta3; tags_meta3.name_to_id = {{"f", 0}, {"g", 1}}; tags_meta3.field_to_columns = {{0, {0}}, {1, {1}}}; diff --git a/src/paimon/core/operation/append_only_file_store_write_test.cpp b/src/paimon/core/operation/append_only_file_store_write_test.cpp index 80eae3c43..270545358 100644 --- a/src/paimon/core/operation/append_only_file_store_write_test.cpp +++ b/src/paimon/core/operation/append_only_file_store_write_test.cpp @@ -322,8 +322,9 @@ TEST_F(AppendOnlyFileStoreWriteTest, TestSharedShreddingMapRestoreInitializesNex ReadDataFileSchema(table_path, OnlyNewFile(second_commit_msgs), options); auto second_meta = ShreddingMeta(second_file_schema, /*field_index=*/1); - auto expected_second_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, {{"tags", 2}}); + ASSERT_OK_AND_ASSIGN( + auto expected_second_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, {{"tags", 2}})); ASSERT_TRUE(second_file_schema->Equals(*expected_second_schema, /*check_metadata=*/false)); ASSERT_EQ(2, second_meta.num_columns); ASSERT_EQ(3, second_meta.max_row_width); @@ -366,8 +367,8 @@ TEST_F(AppendOnlyFileStoreWriteTest, TestSharedShreddingRestoreIgnoresAvroFileWi ReadDataFileSchema(table_path, OnlyNewFile(second_commit_msgs), shredding_options); auto second_meta = ShreddingMeta(second_file_schema, /*field_index=*/1); - auto expected_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, {{"tags", 10}}); + ASSERT_OK_AND_ASSIGN(auto expected_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, {{"tags", 10}})); ASSERT_TRUE(second_file_schema->Equals(*expected_schema, /*check_metadata=*/false)); ASSERT_EQ(10, second_meta.num_columns); ASSERT_EQ(3, second_meta.max_row_width); @@ -427,8 +428,8 @@ TEST_F(AppendOnlyFileStoreWriteTest, TestSharedShreddingRestoreMultipleMapColumn auto tags_meta = ShreddingMeta(full_file_schema, /*field_index=*/1); auto attrs_meta = ShreddingMeta(full_file_schema, /*field_index=*/2); - auto expected_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema( - logical_schema, {{"tags", 2}, {"attrs", 4}}); + ASSERT_OK_AND_ASSIGN(auto expected_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, {{"tags", 2}, {"attrs", 4}})); ASSERT_TRUE(full_file_schema->Equals(*expected_schema, /*check_metadata=*/false)); ASSERT_EQ(2, tags_meta.num_columns); ASSERT_EQ(3, tags_meta.max_row_width); @@ -479,8 +480,8 @@ TEST_F(AppendOnlyFileStoreWriteTest, TestSharedShreddingRestoreUsesDefaultForMis auto tags_meta = ShreddingMeta(full_file_schema, /*field_index=*/1); auto attrs_meta = ShreddingMeta(full_file_schema, /*field_index=*/2); - auto expected_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema( - logical_schema, {{"tags", 2}, {"attrs", 10}}); + ASSERT_OK_AND_ASSIGN(auto expected_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + logical_schema, {{"tags", 2}, {"attrs", 10}})); ASSERT_TRUE(full_file_schema->Equals(*expected_schema, /*check_metadata=*/false)); ASSERT_EQ(2, tags_meta.num_columns); ASSERT_EQ(3, tags_meta.max_row_width); diff --git a/src/paimon/core/operation/key_value_file_store_write_test.cpp b/src/paimon/core/operation/key_value_file_store_write_test.cpp index e79a59635..2fa199af1 100644 --- a/src/paimon/core/operation/key_value_file_store_write_test.cpp +++ b/src/paimon/core/operation/key_value_file_store_write_test.cpp @@ -352,8 +352,9 @@ TEST_F(KeyValueFileStoreWriteTest, TestSharedShreddingMapRestoreInitializesNextW ReadDataFileSchema(table_path, OnlyNewFile(second_commit_msgs), options); auto second_meta = ShreddingMeta(second_file_schema, /*field_index=*/3); - auto expected_second_schema = - MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, {{"tags", 2}}); + ASSERT_OK_AND_ASSIGN( + auto expected_second_schema, + MapSharedShreddingUtils::LogicalToPhysicalSchema(write_schema, {{"tags", 2}})); ASSERT_TRUE(second_file_schema->Equals(*expected_second_schema, /*check_metadata=*/false)); ASSERT_EQ(2, second_meta.num_columns); ASSERT_EQ(3, second_meta.max_row_width); diff --git a/src/paimon/core/postpone/postpone_bucket_writer_test.cpp b/src/paimon/core/postpone/postpone_bucket_writer_test.cpp index 362303a77..0e4e7f287 100644 --- a/src/paimon/core/postpone/postpone_bucket_writer_test.cpp +++ b/src/paimon/core/postpone/postpone_bucket_writer_test.cpp @@ -366,8 +366,8 @@ TEST_F(PostponeBucketWriterTest, TestSharedShreddingMap) { arrow::FieldVector write_fields = {arrow::field("_SEQUENCE_NUMBER", arrow::int64()), arrow::field("_VALUE_KIND", arrow::int8())}; write_fields.insert(write_fields.end(), fields.begin(), fields.end()); - auto expected_schema = MapSharedShreddingUtils::LogicalToPhysicalSchema( - arrow::schema(write_fields), {{"tags", 3}}); + ASSERT_OK_AND_ASSIGN(auto expected_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema( + arrow::schema(write_fields), {{"tags", 3}})); MapSharedShreddingFieldMeta expected_meta; expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}}; From 0e16cf396d0af58d0c3a80b6e576a1f70cc49a0b Mon Sep 17 00:00:00 2001 From: "menglingda.mld" Date: Fri, 10 Jul 2026 15:53:32 +0800 Subject: [PATCH 2/2] Address shared-shredding schema utils review comments --- .../map_shared_shredding_schema_utils.h | 3 +- ...map_shared_shredding_schema_utils_test.cpp | 75 ++++++++++++++++++- .../shredding/map_shared_shredding_utils.cpp | 32 ++++---- 3 files changed, 93 insertions(+), 17 deletions(-) diff --git a/include/paimon/data/shredding/map_shared_shredding_schema_utils.h b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h index a9cfa4518..db47be311 100644 --- a/include/paimon/data/shredding/map_shared_shredding_schema_utils.h +++ b/include/paimon/data/shredding/map_shared_shredding_schema_utils.h @@ -34,7 +34,7 @@ namespace paimon { struct PAIMON_EXPORT MapSharedShreddingFieldMeta { /// field_name -> field_id std::map name_to_id; - /// field_id -> set of physical column indices S + /// field_id -> ordered physical column indices std::map> field_to_columns; /// Set of field_ids that ever spilled into __overflow std::set overflow_field_set; @@ -73,6 +73,7 @@ class PAIMON_EXPORT MapSharedShreddingSchemaUtils { /// @param physical_schema The Arrow C physical schema whose fields should receive metadata. /// Ownership of schema resources is transferred to this method. /// @param field_name_to_meta Map from physical field name to its shared-shredding metadata. + /// Existing shared-shredding metadata keys on matching fields are overwritten. /// @param compression Compression codec name for field_dict serialization. /// @return A new Arrow C schema with shared-shredding metadata attached to matching fields. static Result> AttachMetadataToSchema( diff --git a/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp index cb1182596..f534cdbd3 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp @@ -66,7 +66,11 @@ TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaBasic) { ASSERT_EQ(deserialized, tags_meta); } -TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaMissingField) { +TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaInvalidInput) { + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::unique_ptr<::ArrowSchema>(), {}, "none"), + "physical schema is null"); + MapSharedShreddingFieldMeta tags_meta; tags_meta.name_to_id = {{"host", 0}}; tags_meta.field_to_columns = {{0, {0}}}; @@ -114,6 +118,51 @@ TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaPreservesExistingF ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(updated_metadata)); } +TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaOverwritesExistingShreddingMetadata) { + MapSharedShreddingFieldMeta old_meta; + old_meta.name_to_id = {{"old", 0}}; + old_meta.field_to_columns = {{0, {0}}}; + old_meta.num_columns = 1; + old_meta.max_row_width = 1; + + MapSharedShreddingFieldMeta tags_meta; + tags_meta.name_to_id = {{"host", 0}, {"region", 1}}; + tags_meta.field_to_columns = {{0, {0}}, {1, {1}}}; + tags_meta.overflow_field_set = {1}; + tags_meta.num_columns = 2; + tags_meta.max_row_width = 2; + + auto tags_metadata = std::make_shared(); + tags_metadata->Append("paimon.field.id", "7"); + ASSERT_OK(MapSharedShreddingUtils::SerializeMetadata(old_meta, "none", tags_metadata.get())); + auto schema = arrow::schema({arrow::field( + "tags", + arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true), + arrow::field("__col_0", arrow::utf8(), true), + arrow::field("__col_1", arrow::utf8(), true)}), + true, tags_metadata)}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + ASSERT_OK_AND_ASSIGN(auto c_updated_schema, + MapSharedShreddingSchemaUtils::AttachMetadataToSchema( + std::move(c_schema), {{"tags", tags_meta}}, "none")); + auto updated_schema = arrow::ImportSchema(c_updated_schema.get()).ValueOrDie(); + auto updated_metadata = updated_schema->field(0)->metadata()->Copy(); + + ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("paimon.field.id")), "7"); + int32_t storage_layout_key_count = 0; + for (const auto& key : updated_metadata->keys()) { + if (key == MapShreddingDefine::kStorageLayout) { + ++storage_layout_key_count; + } + } + ASSERT_EQ(storage_layout_key_count, 1); + ASSERT_OK_AND_ASSIGN(auto deserialized, + MapSharedShreddingUtils::DeserializeMetadata(updated_metadata, "none")); + ASSERT_EQ(deserialized, tags_meta); +} + TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromField) { MapSharedShreddingFieldMeta tags_meta; tags_meta.name_to_id = {{"host", 0}, {"region", 1}}; @@ -150,10 +199,30 @@ TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldNoShreddingMetad "metadata is null or storage layout is not shared-shredding"); } -TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaNonMapField) { +TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldInvalidInput) { + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField( + std::unique_ptr<::ArrowSchema>(), "tags", "none"), + "physical schema is null"); + + auto schema = arrow::schema({arrow::field("id", arrow::int32())}); + + auto c_schema = std::make_unique<::ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); + + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema), + "tags", "none"), + "Shared-shredding field 'tags' not found in physical schema."); +} + +TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaInvalidInput) { + std::map field_to_num_columns = {{"tags", 2}}; + + ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema( + std::unique_ptr<::ArrowSchema>(), field_to_num_columns), + "logical schema is null"); + auto schema = arrow::schema({arrow::field("ts", arrow::int64()), arrow::field("tags", arrow::int32())}); - std::map field_to_num_columns = {{"tags", 2}}; auto c_schema = std::make_unique<::ArrowSchema>(); ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok()); diff --git a/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp b/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp index 391378bde..d9cd0290c 100644 --- a/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp +++ b/src/paimon/common/data/shredding/map_shared_shredding_utils.cpp @@ -27,6 +27,7 @@ #include "paimon/common/compression/block_decompressor.h" #include "paimon/common/data/shredding/map_shared_shredding_batch_converter.h" #include "paimon/common/data/shredding/map_shared_shredding_context.h" +#include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/string_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/options/map_storage_layout.h" @@ -357,23 +358,28 @@ Result> DeserializeOverflowSet(const std::string& json_str) { Status MapSharedShreddingUtils::SerializeMetadata(const MapSharedShreddingFieldMeta& field_meta, const std::string& compression, arrow::KeyValueMetadata* metadata) { - metadata->Append(MapShreddingDefine::kStorageLayout, - MapShreddingDefine::kStorageLayoutSharedShredding); - metadata->Append(MapSharedShreddingDefine::kVersion, - std::to_string(MapSharedShreddingDefine::kCurrentVersion)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(metadata->Set( + MapShreddingDefine::kStorageLayout, MapShreddingDefine::kStorageLayoutSharedShredding)); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + metadata->Set(MapSharedShreddingDefine::kVersion, + std::to_string(MapSharedShreddingDefine::kCurrentVersion))); std::string field_dict_json = SerializeFieldDict(field_meta); - metadata->Append(MapSharedShreddingDefine::kFieldDictOriginalSize, - std::to_string(field_dict_json.size())); + PAIMON_RETURN_NOT_OK_FROM_ARROW(metadata->Set(MapSharedShreddingDefine::kFieldDictOriginalSize, + std::to_string(field_dict_json.size()))); PAIMON_ASSIGN_OR_RAISE(std::string compressed_dict, CompressString(field_dict_json, compression)); - metadata->Append(MapSharedShreddingDefine::kFieldDict, std::move(compressed_dict)); - - metadata->Append(MapSharedShreddingDefine::kFieldColumns, SerializeFieldColumns(field_meta)); - metadata->Append(MapSharedShreddingDefine::kOverflowSet, SerializeOverflowSet(field_meta)); - metadata->Append(MapSharedShreddingDefine::kNumColumns, std::to_string(field_meta.num_columns)); - metadata->Append(MapSharedShreddingDefine::kMaxRowWidth, - std::to_string(field_meta.max_row_width)); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + metadata->Set(MapSharedShreddingDefine::kFieldDict, std::move(compressed_dict))); + + PAIMON_RETURN_NOT_OK_FROM_ARROW( + metadata->Set(MapSharedShreddingDefine::kFieldColumns, SerializeFieldColumns(field_meta))); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + metadata->Set(MapSharedShreddingDefine::kOverflowSet, SerializeOverflowSet(field_meta))); + PAIMON_RETURN_NOT_OK_FROM_ARROW(metadata->Set(MapSharedShreddingDefine::kNumColumns, + std::to_string(field_meta.num_columns))); + PAIMON_RETURN_NOT_OK_FROM_ARROW(metadata->Set(MapSharedShreddingDefine::kMaxRowWidth, + std::to_string(field_meta.max_row_width))); return Status::OK(); }