feat(shredding): support shared shredding file reader (alibaba#376)

Author: lszskye

src/paimon/CMakeLists.txt

@@ -142,6 +142,7 @@ set(PAIMON_COMMON_SRCS
     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
+    common/data/shredding/shared_shredding_file_reader.cpp
     common/utils/delta_varint_compressor.cpp
     common/utils/fields_comparator.cpp
     common/utils/path_util.cpp
@@ -541,6 +542,7 @@ if(PAIMON_BUILD_TESTS)
                     common/data/shredding/map_shared_shredding_column_allocator_test.cpp
                     common/data/shredding/map_shared_shredding_field_dict_test.cpp
                     common/data/shredding/map_shared_shredding_context_test.cpp
+                    common/data/shredding/shared_shredding_file_reader_test.cpp
                     STATIC_LINK_LIBS
                     paimon_shared
                     test_utils_static

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

@@ -35,8 +35,23 @@
 #include "rapidjson/writer.h"
 
 namespace paimon {
-// ---- Column detection ----
+Result<std::vector<int32_t>> MapSharedShreddingUtils::GetPhysicalColumnIndices(
+    const MapSharedShreddingFieldMeta& meta, const std::string& name) {
+    auto name_iter = meta.name_to_id.find(name);
+    if (name_iter == meta.name_to_id.end()) {
+        return Status::Invalid(
+            fmt::format("cannot find field {} in map shared shredding meta", name));
+    }
+    auto id_iter = meta.field_to_columns.find(name_iter->second);
+    if (id_iter == meta.field_to_columns.end()) {
+        return Status::Invalid(
+            fmt::format("cannot find field id {} in field_to_columns in map shared shredding meta",
+                        name_iter->second));
+    }
+    return id_iter->second;
+}
 
+// ---- Column detection ----
 bool MapSharedShreddingUtils::IsShreddingKeyMap(
     const std::shared_ptr<arrow::DataType>& arrow_type) {
     if (arrow_type->id() != arrow::Type::MAP) {
@@ -75,24 +90,38 @@ Result<std::shared_ptr<MapSharedShreddingContext>> MapSharedShreddingUtils::Crea
 }
 
 // ---- Schema conversion ----
+std::shared_ptr<arrow::DataType> MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+    const std::shared_ptr<arrow::DataType>& value_type, const std::set<int32_t>& physical_col_ids,
+    bool value_nullable, bool include_overflow) {
+    std::vector<int32_t> sorted_cols(physical_col_ids.begin(), physical_col_ids.end());
+    return InnerBuildSpecificPhysicalStructType(value_type, sorted_cols, value_nullable,
+                                                include_overflow);
+}
 
 std::shared_ptr<arrow::DataType> MapSharedShreddingUtils::BuildPhysicalStructType(
     const std::shared_ptr<arrow::DataType>& value_type, int32_t num_columns, bool value_nullable) {
-    arrow::FieldVector struct_fields;
-    struct_fields.reserve(num_columns + 2);
+    std::vector<int32_t> sorted_cols(num_columns);
+    std::iota(sorted_cols.begin(), sorted_cols.end(), 0);
+    return InnerBuildSpecificPhysicalStructType(value_type, sorted_cols, value_nullable,
+                                                /*include_overflow=*/true);
+}
 
+std::shared_ptr<arrow::DataType> MapSharedShreddingUtils::InnerBuildSpecificPhysicalStructType(
+    const std::shared_ptr<arrow::DataType>& value_type, const std::vector<int32_t>& sorted_cols,
+    bool value_nullable, bool include_overflow) {
+    arrow::FieldVector struct_fields;
+    struct_fields.reserve(sorted_cols.size() + 2);
     struct_fields.push_back(
         arrow::field(MapSharedShreddingDefine::kFieldMapping, arrow::list(arrow::int32()), true));
-
-    for (int32_t i = 0; i < num_columns; ++i) {
-        struct_fields.push_back(arrow::field(MapSharedShreddingDefine::PhysicalColumnName(i),
+    for (const auto& col : sorted_cols) {
+        struct_fields.push_back(arrow::field(MapSharedShreddingDefine::PhysicalColumnName(col),
                                              value_type, value_nullable));
     }
-
-    struct_fields.push_back(arrow::field(
-        MapSharedShreddingDefine::kOverflow,
-        arrow::map(arrow::int32(), arrow::field("value", value_type, value_nullable)), true));
-
+    if (include_overflow) {
+        struct_fields.push_back(arrow::field(
+            MapSharedShreddingDefine::kOverflow,
+            arrow::map(arrow::int32(), arrow::field("value", value_type, value_nullable)), true));
+    }
     return arrow::struct_(std::move(struct_fields));
 }
 
@@ -400,6 +429,16 @@ bool MapSharedShreddingUtils::HasShreddingMetadata(
     return metadata->value(index) == MapShreddingDefine::kStorageLayoutSharedShredding;
 }
 
+Result<bool> MapSharedShreddingUtils::IsOverflowField(const MapSharedShreddingFieldMeta& meta,
+                                                      const std::string& name) {
+    auto name_iter = meta.name_to_id.find(name);
+    if (name_iter == meta.name_to_id.end()) {
+        return Status::Invalid(
+            fmt::format("cannot find field {} in map shared shredding meta", name));
+    }
+    return meta.overflow_field_set.count(name_iter->second) > 0;
+}
+
 std::function<Result<std::shared_ptr<arrow::Schema>>()>
 MapSharedShreddingUtils::BuildMetadataFinalizer(
     const std::shared_ptr<MapSharedShreddingBatchConverter>& converter,

src/paimon/common/data/shredding/map_shared_shredding_utils.h

@@ -20,6 +20,7 @@
 #include <functional>
 #include <map>
 #include <memory>
+#include <set>
 #include <string>
 #include <vector>
 
@@ -45,6 +46,13 @@ class MapSharedShreddingUtils {
     MapSharedShreddingUtils() = delete;
     ~MapSharedShreddingUtils() = delete;
 
+    /// Returns the physical column indices for the given field name from the shredding meta.
+    /// @param meta The shredding field meta parsed from file footer.
+    /// @param name The field name to look up.
+    /// @return Vector of physical column indices assigned to this field,
+    ///         or Status::Invalid if the field name or field id is not found.
+    static Result<std::vector<int32_t>> GetPhysicalColumnIndices(
+        const MapSharedShreddingFieldMeta& meta, const std::string& name);
     // ---- Column detection ----
 
     /// Checks whether a given arrow field is MAP<STRING, T> (the type prerequisite for shredding).
@@ -80,6 +88,15 @@ class MapSharedShreddingUtils {
         const std::shared_ptr<arrow::Schema>& logical_schema,
         const std::map<std::string, int32_t>& field_to_num_columns);
 
+    /// Builds the physical Arrow type for one shredding MAP column with physical_col_ids.
+    /// @param value_type The value type of the original MAP.
+    /// @param physical_col_ids The set of physical column ids to include.
+    /// @param value_nullable Whether the MAP's value field is nullable.
+    /// @param include_overflow Whether to include __overflow column.
+    static std::shared_ptr<arrow::DataType> BuildSpecificPhysicalStructType(
+        const std::shared_ptr<arrow::DataType>& value_type,
+        const std::set<int32_t>& physical_col_ids, bool value_nullable, bool include_overflow);
+
     /// Builds field_to_num_columns map from DetectShreddingColumns result and CoreOptions.
     /// @param shredding_field_names Field names returned by DetectShreddingColumns.
     /// @param options CoreOptions containing per-column max-columns config.
@@ -107,6 +124,10 @@ class MapSharedShreddingUtils {
     /// Checks whether a KeyValueMetadata contains shredding MAP metadata.
     static bool HasShreddingMetadata(const std::shared_ptr<arrow::KeyValueMetadata>& metadata);
 
+    /// Checks whether a field in MapSharedShreddingFieldMeta is a overflow field.
+    static Result<bool> IsOverflowField(const MapSharedShreddingFieldMeta& meta,
+                                        const std::string& name);
+
     // ---- Writer helpers ----
 
     /// Builds a MetadataFinalizer that serializes shredding metadata into per-field
@@ -131,6 +152,15 @@ class MapSharedShreddingUtils {
     static std::shared_ptr<arrow::DataType> BuildPhysicalStructType(
         const std::shared_ptr<arrow::DataType>& value_type, int32_t num_columns,
         bool value_nullable);
+
+    /// Builds the physical Arrow type for one shredding MAP column with sorted_cols.
+    /// @param value_type The value type of the original MAP.
+    /// @param sorted_cols The vector of physical column ids to include.
+    /// @param value_nullable Whether the MAP's value field is nullable.
+    /// @param include_overflow Whether to include __overflow column.
+    static std::shared_ptr<arrow::DataType> InnerBuildSpecificPhysicalStructType(
+        const std::shared_ptr<arrow::DataType>& value_type, const std::vector<int32_t>& sorted_cols,
+        bool value_nullable, bool include_overflow);
 };
 
 }  // namespace paimon

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

@@ -16,6 +16,8 @@
 
 #include "paimon/common/data/shredding/map_shared_shredding_utils.h"
 
+#include <set>
+
 #include "arrow/type.h"
 #include "arrow/util/key_value_metadata.h"
 #include "gtest/gtest.h"
@@ -162,6 +164,38 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNoShreddingColumns) {
     ASSERT_TRUE(physical_schema->Equals(schema));
 }
 
+TEST(MapSharedShreddingUtilsTest, BuildSpecificPhysicalStructTypeWithOverflow) {
+    auto actual = MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+        arrow::int64(), /*physical_col_ids=*/{3, 1}, /*value_nullable=*/false,
+        /*include_overflow=*/true);
+
+    auto expected = arrow::struct_({
+        arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
+        arrow::field("__col_1", arrow::int64(), false),
+        arrow::field("__col_3", arrow::int64(), false),
+        arrow::field("__overflow",
+                     arrow::map(arrow::int32(), arrow::field("value", arrow::int64(), false)),
+                     true),
+    });
+    ASSERT_TRUE(actual->Equals(*expected)) << "Expected:\n"
+                                           << expected->ToString() << "\nActual:\n"
+                                           << actual->ToString();
+}
+
+TEST(MapSharedShreddingUtilsTest, BuildSpecificPhysicalStructTypeWithoutOverflow) {
+    auto actual = MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+        arrow::utf8(), /*physical_col_ids=*/{3}, /*value_nullable=*/true,
+        /*include_overflow=*/false);
+
+    auto expected = arrow::struct_({
+        arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
+        arrow::field("__col_3", arrow::utf8(), true),
+    });
+    ASSERT_TRUE(actual->Equals(*expected)) << "Expected:\n"
+                                           << expected->ToString() << "\nActual:\n"
+                                           << actual->ToString();
+}
+
 // ---- BuildColumnToNumColumns ----
 
 TEST(MapSharedShreddingUtilsTest, BuildColumnToNumColumns) {
@@ -362,4 +396,94 @@ TEST(MapSharedShreddingUtilsTest, PhysicalColumnName) {
     ASSERT_EQ(MapSharedShreddingDefine::PhysicalColumnName(99), "__col_99");
 }
 
+// ---- GetPhysicalColumnIndices ----
+
+// Normal: single physical column per field
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesSingleColumn) {
+    MapSharedShreddingFieldMeta meta;
+    meta.name_to_id = {{"age", 0}, {"name", 1}};
+    meta.field_to_columns = {{0, {2}}, {1, {5}}};
+
+    ASSERT_OK_AND_ASSIGN(auto cols_age,
+                         MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "age"));
+    ASSERT_EQ(cols_age, (std::vector<int32_t>{2}));
+
+    ASSERT_OK_AND_ASSIGN(auto cols_name,
+                         MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "name"));
+    ASSERT_EQ(cols_name, (std::vector<int32_t>{5}));
+}
+
+// Normal: multiple physical columns for one field
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesMultipleColumns) {
+    MapSharedShreddingFieldMeta meta;
+    meta.name_to_id = {{"tags", 0}};
+    meta.field_to_columns = {{0, {0, 3, 7}}};
+
+    ASSERT_OK_AND_ASSIGN(auto cols,
+                         MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "tags"));
+    ASSERT_EQ(cols, (std::vector<int32_t>{0, 3, 7}));
+}
+
+// Normal: many fields each mapping to different physical columns
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesMultipleFields) {
+    MapSharedShreddingFieldMeta meta;
+    meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}};
+    meta.field_to_columns = {{0, {0, 1}}, {1, {2, 3, 4}}, {2, {5}}};
+
+    ASSERT_OK_AND_ASSIGN(auto cols_a, MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "a"));
+    ASSERT_EQ(cols_a, (std::vector<int32_t>{0, 1}));
+
+    ASSERT_OK_AND_ASSIGN(auto cols_b, MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "b"));
+    ASSERT_EQ(cols_b, (std::vector<int32_t>{2, 3, 4}));
+
+    ASSERT_OK_AND_ASSIGN(auto cols_c, MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "c"));
+    ASSERT_EQ(cols_c, (std::vector<int32_t>{5}));
+}
+
+// Error: field name not found in name_to_id
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesFieldNotFound) {
+    MapSharedShreddingFieldMeta meta;
+    meta.name_to_id = {{"age", 0}};
+    meta.field_to_columns = {{0, {1}}};
+
+    ASSERT_NOK_WITH_MSG(MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "nonexistent"),
+                        "cannot find field nonexistent in map shared shredding meta");
+}
+
+// Error: field name not found in empty meta
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesEmptyMeta) {
+    MapSharedShreddingFieldMeta meta;
+    ASSERT_NOK_WITH_MSG(MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "any"),
+                        "cannot find field any in map shared shredding meta");
+}
+
+// Error: field id exists in name_to_id but is missing from field_to_columns
+TEST(MapSharedShreddingUtilsTest, GetPhysicalColumnIndicesFieldIdMissingInFieldToColumns) {
+    MapSharedShreddingFieldMeta meta;
+    // "score" -> field_id 42, but field_to_columns has no entry for 42
+    meta.name_to_id = {{"score", 42}};
+    meta.field_to_columns = {};
+
+    ASSERT_NOK_WITH_MSG(MapSharedShreddingUtils::GetPhysicalColumnIndices(meta, "score"),
+                        "cannot find field id 42 in field_to_columns in map shared shredding meta");
+}
+
+TEST(MapSharedShreddingUtilsTest, IsOverflowField) {
+    MapSharedShreddingFieldMeta meta;
+    meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}};
+    meta.overflow_field_set = {0, 2};
+
+    ASSERT_OK_AND_ASSIGN(bool a_overflow, MapSharedShreddingUtils::IsOverflowField(meta, "a"));
+    ASSERT_TRUE(a_overflow);
+
+    ASSERT_OK_AND_ASSIGN(bool b_overflow, MapSharedShreddingUtils::IsOverflowField(meta, "b"));
+    ASSERT_FALSE(b_overflow);
+
+    ASSERT_OK_AND_ASSIGN(bool c_overflow, MapSharedShreddingUtils::IsOverflowField(meta, "c"));
+    ASSERT_TRUE(c_overflow);
+
+    ASSERT_NOK_WITH_MSG(MapSharedShreddingUtils::IsOverflowField(meta, "missing"),
+                        "cannot find field missing in map shared shredding meta");
+}
+
 }  // namespace paimon::test