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

@@ -71,7 +71,8 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
         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);
+            MapSharedShreddingUtils::LogicalToPhysicalSchema(logical_schema_, field_to_num_columns)
+                .value();
         auto metadata = std::make_shared<arrow::KeyValueMetadata>();
         EXPECT_OK(MapSharedShreddingUtils::SerializeMetadata(
             meta, MapSharedShreddingDefine::kDefaultDictCompression, metadata.get()));
@@ -458,8 +459,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()));

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<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema(
+    std::unique_ptr<::ArrowSchema> logical_schema,
+    const std::map<std::string, int32_t>& field_to_num_columns) {
+    if (!logical_schema) {
+        return Status::Invalid("logical schema is null");
+    }
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_logical_schema,
+                                      arrow::ImportSchema(logical_schema.get()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> 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<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
+    std::unique_ptr<::ArrowSchema> physical_schema,
+    const std::map<std::string, MapSharedShreddingFieldMeta>& 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::Schema> 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<arrow::KeyValueMetadata>();
+        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<MapSharedShreddingFieldMeta> 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::Schema> 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<arrow::KeyValueMetadata>();
+    return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression);
+}
+
+}  // namespace paimon