feat(shared-shredding): Expose shared-shredding utilities

Author: dalingmeng

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 <cstdint>
+#include <map>
+#include <memory>
+#include <set>
+#include <string>
+#include <vector>
+
+#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<std::string, int32_t> name_to_id;
+    /// field_id -> set of physical column indices S
+    std::map<int32_t, std::vector<int32_t>> field_to_columns;
+    /// Set of field_ids that ever spilled into __overflow
+    std::set<int32_t> 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<STRING, T> 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<std::unique_ptr<::ArrowSchema>> LogicalToPhysicalSchema(
+        std::unique_ptr<::ArrowSchema> logical_schema,
+        const std::map<std::string, int32_t>& 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<std::unique_ptr<::ArrowSchema>> AttachMetadataToSchema(
+        std::unique_ptr<::ArrowSchema> physical_schema,
+        const std::map<std::string, MapSharedShreddingFieldMeta>& 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<MapSharedShreddingFieldMeta> ExtractMetadataFromField(
+        std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name,
+        const std::string& compression);
+};
+
+}  // namespace paimon

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

src/paimon/common/data/shredding/map_shared_shredding_batch_converter.cpp

@@ -65,8 +65,9 @@ Result<std::shared_ptr<MapSharedShreddingBatchConverter>> MapSharedShreddingBatc
     const std::shared_ptr<MapSharedShreddingContext>& context, const CoreOptions& options,
     const std::shared_ptr<MemoryPool>& pool) {
     std::map<std::string, int32_t> field_to_num_columns = context->ComputeNextK();
-    std::shared_ptr<arrow::Schema> physical_schema =
-        MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, field_to_num_columns);
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<arrow::Schema> physical_schema,
+        MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema, field_to_num_columns));
     std::vector<ColumnContext> contexts;
     std::vector<std::string> shredding_field_names;
     contexts.reserve(field_to_num_columns.size());

src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp

@@ -63,26 +63,28 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
         return meta;
     }
 
-    std::shared_ptr<arrow::Schema> PhysicalSchemaWithMetadata() const {
+    Result<std::shared_ptr<arrow::Schema>> PhysicalSchemaWithMetadata() const {
         return PhysicalSchemaWithMetadata(TagsMeta());
     }
 
-    std::shared_ptr<arrow::Schema> PhysicalSchemaWithMetadata(
+    Result<std::shared_ptr<arrow::Schema>> PhysicalSchemaWithMetadata(
         const MapSharedShreddingFieldMeta& meta) const {
         std::map<std::string, int32_t> 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<arrow::KeyValueMetadata>();
-        EXPECT_OK(MapSharedShreddingUtils::SerializeMetadata(
+        PAIMON_RETURN_NOT_OK(MapSharedShreddingUtils::SerializeMetadata(
             meta, MapSharedShreddingDefine::kDefaultDictCompression, metadata.get()));
 
         arrow::FieldVector fields = physical_schema->fields();
         fields[1] = fields[1]->WithMetadata(metadata);
         return arrow::schema(std::move(fields));
     }
 
-    std::shared_ptr<arrow::Array> PhysicalArray() const {
-        std::shared_ptr<arrow::Schema> physical_schema = PhysicalSchemaWithMetadata();
+    Result<std::shared_ptr<arrow::Array>> PhysicalArray() const {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> 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<MapSharedShreddingFileReader> CreateReader(
+    Result<std::unique_ptr<MapSharedShreddingFileReader>> CreateReader(
         std::shared_ptr<arrow::Array> physical_array = nullptr,
         std::shared_ptr<arrow::Schema> physical_schema = nullptr,
         const std::optional<std::string>& 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<MockFileBatchReader>(
             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<std::string, int32_t> 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<arrow::KeyValueMetadata>();
     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<arrow::KeyValueMetadata>();
     read_metadata->Append("paimon.map.selected-keys", "a,c");