feat(shredding): support selected-key pushdown by MapSharedShreddingAccessBuilder

include/paimon/data/shredding/map_shared_shredding_schema_utils.h

@@ -56,6 +56,48 @@ struct PAIMON_EXPORT MapSharedShreddingFieldMeta {
     }
 };
 
+/// Builds a selected-key projection field for a top-level shared-shredding MAP column.
+///
+/// The built field is a STRUCT which replaces the MAP field in the read schema. Each child
+/// corresponds to one selected key and contains that key's MAP value, or NULL when the key is
+/// absent. Children use the selected keys as their names and preserve insertion order.
+///
+/// Example: read keys "age" and "score" from MAP column `attributes`:
+///
+///     auto builder = MapSharedShreddingAccessBuilder::Create(attributes_field);
+///     builder->AddKey("age");
+///     builder->AddKey("score");
+///     auto field = builder->Build();
+///
+/// Use the returned field in `ReadContextBuilder::SetReadSchema`.
+class PAIMON_EXPORT MapSharedShreddingAccessBuilder {
+ public:
+    /// Creates a builder bound to the original MAP field.
+    ///
+    /// The field must be a MAP with STRING keys. Its name, nullability, and value type are
+    /// retained for the selected-key projection. Ownership of the Arrow C schema resources is
+    /// transferred to this method.
+    static Result<std::unique_ptr<MapSharedShreddingAccessBuilder>> Create(
+        struct ArrowSchema* map_field);
+
+    ~MapSharedShreddingAccessBuilder();
+
+    /// Adds a selected MAP key.
+    ///
+    /// @param key The string MAP key. Keys are returned in insertion order.
+    Status AddKey(const std::string& key);
+
+    /// Builds a STRUCT projection field which retains the original MAP field's name and
+    /// nullability. Every selected-key child uses the complete MAP value type and is nullable.
+    Result<std::unique_ptr<struct ArrowSchema>> Build() const;
+
+ private:
+    class Impl;
+    explicit MapSharedShreddingAccessBuilder(std::unique_ptr<Impl>&& impl);
+
+    std::unique_ptr<Impl> impl_;
+};
+
 class PAIMON_EXPORT MapSharedShreddingSchemaUtils {
  public:
     MapSharedShreddingSchemaUtils() = delete;

include/paimon/read_context.h

@@ -226,6 +226,12 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// key list, for example: "k1,k2". Only map fields with string key type
     /// (Arrow utf8) are supported.
     ///
+    /// Attaching this metadata to a MAP field only filters the returned MAP after
+    /// reading. To push down selected keys from a shared-shredding MAP and return
+    /// them as STRUCT children, build the field with
+    /// `MapSharedShreddingAccessBuilder`. To read selected paths from a VARIANT
+    /// field, build the field with `VariantAccessBuilder`.
+    ///
     /// Example:
     /// @code{.cpp}
     /// auto map_field = arrow::field("m", arrow::map(arrow::utf8(), arrow::int32()));

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

@@ -33,21 +33,51 @@
 
 namespace paimon {
 
-class MapSharedShreddingFileReader : public FileBatchReader {
+class MapFieldReadPlan {
+ public:
+    virtual ~MapFieldReadPlan() = default;
+
+    MapFieldReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
+                     const std::shared_ptr<arrow::Field>& physical_read_field)
+        : logical_field_(logical_field), physical_read_field_(physical_read_field) {}
+
+    const std::shared_ptr<arrow::Field>& LogicalField() const {
+        return logical_field_;
+    }
+
+    const std::shared_ptr<arrow::Field>& PhysicalReadField() const {
+        return physical_read_field_;
+    }
+
+    virtual Result<std::shared_ptr<arrow::Array>> Materialize(
+        const std::shared_ptr<arrow::Array>& physical_array,
+        arrow::MemoryPool* arrow_pool) const = 0;
+
+ private:
+    std::shared_ptr<arrow::Field> logical_field_;
+    std::shared_ptr<arrow::Field> physical_read_field_;
+};
+
+class MapFieldReadPlanFactory {
  public:
-    struct SharedShreddingContext {
-        SharedShreddingContext(const MapSharedShreddingFieldMeta& _meta,
-                               const std::vector<std::string>& _selected_keys,
-                               const std::shared_ptr<arrow::MapType>& _map_type)
-            : meta(_meta), selected_keys(_selected_keys), map_type(_map_type) {}
-        MapSharedShreddingFieldMeta meta;
-        std::vector<std::string> selected_keys;
-        std::shared_ptr<arrow::MapType> map_type;
-    };
+    static Result<std::unique_ptr<MapFieldReadPlan>> CreateMapReadPlan(
+        const std::shared_ptr<arrow::Field>& logical_map_field,
+        const MapSharedShreddingFieldMeta& meta);
+
+    static Result<std::unique_ptr<MapFieldReadPlan>> CreateSharedSelectedKeysReadPlan(
+        const std::shared_ptr<arrow::Field>& selected_keys_field,
+        const MapSharedShreddingFieldMeta& meta);
 
+    static Result<std::unique_ptr<MapFieldReadPlan>> CreateDefaultSelectedKeysReadPlan(
+        const std::shared_ptr<arrow::Field>& file_map_field,
+        const std::shared_ptr<arrow::Field>& selected_keys_field);
+};
+
+class MapSharedShreddingFileReader : public FileBatchReader {
+ public:
     MapSharedShreddingFileReader(
         std::unique_ptr<FileBatchReader>&& reader,
-        std::map<std::string, SharedShreddingContext>&& shared_shredding_name_to_context,
+        std::map<std::string, std::unique_ptr<MapFieldReadPlan>>&& field_read_plans,
         const std::shared_ptr<MemoryPool>& pool);
 
     Result<std::unique_ptr<::ArrowSchema>> GetFileSchema() const override;
@@ -70,25 +100,13 @@ class MapSharedShreddingFileReader : public FileBatchReader {
     bool SupportPreciseBitmapSelection() const override;
 
  private:
-    Result<std::shared_ptr<arrow::Array>> RebuildLogicalMapArray(
-        const std::shared_ptr<arrow::Field>& physical_field,
-        const std::shared_ptr<arrow::StructArray>& physical_struct_array) const;
-
-    static std::vector<std::pair<std::string, int32_t>> ResolveSelectedKeyIds(
-        const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>& selected_keys);
-
-    static void CollectPhysicalColumns(
-        const std::shared_ptr<arrow::StructArray>& physical_struct_array,
-        std::map<std::string, std::shared_ptr<arrow::Array>>* physical_column_name_to_array,
-        std::shared_ptr<arrow::MapArray>* overflow_array);
-
     static Result<std::shared_ptr<arrow::Field>> ToLogicalMapField(
         const std::shared_ptr<arrow::Field>& physical_field);
 
  private:
     std::shared_ptr<arrow::MemoryPool> arrow_pool_;
     std::unique_ptr<FileBatchReader> reader_;
-    std::map<std::string, SharedShreddingContext> shared_shredding_name_to_context_;
+    std::map<std::string, std::unique_ptr<MapFieldReadPlan>> field_read_plans_;
 };
 
 }  // namespace paimon

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

@@ -104,8 +104,7 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
         const std::optional<std::string>& selected_keys_str = std::nullopt) const {
         EXPECT_OK_AND_ASSIGN(auto c_file_schema, reader->GetFileSchema());
         auto file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOrDie();
-        std::map<std::string, MapSharedShreddingFileReader::SharedShreddingContext>
-            shared_shredding_name_to_context;
+        std::map<std::string, std::unique_ptr<MapFieldReadPlan>> field_read_plans;
         for (const auto& field : file_schema->fields()) {
             auto metadata = std::const_pointer_cast<arrow::KeyValueMetadata>(field->metadata());
             if (!MapSharedShreddingUtils::HasShreddingMetadata(metadata)) {
@@ -125,22 +124,17 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
             EXPECT_TRUE(item_field);
             auto map_type = arrow::internal::checked_pointer_cast<arrow::MapType>(arrow::map(
                 arrow::utf8(), arrow::field("value", item_field->type(), item_field->nullable())));
-            std::vector<std::string> selected_keys;
+            std::shared_ptr<arrow::Field> logical_map_field = field->WithType(map_type);
             if (selected_keys_str.has_value()) {
-                selected_keys = StringUtils::Split(selected_keys_str.value(), ",",
-                                                   /*ignore_empty=*/false);
-            } else {
-                selected_keys.reserve(meta.name_to_id.size());
-                for (const auto& [key_name, _] : meta.name_to_id) {
-                    selected_keys.push_back(key_name);
-                }
+                logical_map_field = logical_map_field->WithMetadata(arrow::KeyValueMetadata::Make(
+                    {DataField::MAP_SELECTED_KEYS}, {selected_keys_str.value()}));
             }
-            shared_shredding_name_to_context.emplace(
-                field->name(), MapSharedShreddingFileReader::SharedShreddingContext(
-                                   meta, selected_keys, map_type));
+            EXPECT_OK_AND_ASSIGN(auto field_read_plan, MapFieldReadPlanFactory::CreateMapReadPlan(
+                                                           logical_map_field, meta));
+            field_read_plans.emplace(field->name(), std::move(field_read_plan));
         }
-        return std::make_unique<MapSharedShreddingFileReader>(
-            std::move(reader), std::move(shared_shredding_name_to_context), pool_);
+        return std::make_unique<MapSharedShreddingFileReader>(std::move(reader),
+                                                              std::move(field_read_plans), pool_);
     }
 
     Result<std::unique_ptr<MapSharedShreddingFileReader>> CreateReader(
@@ -299,6 +293,118 @@ TEST_F(MapSharedShreddingFileReaderTest, TestAllExistSelectedKeysWithOverflow) {
     AssertChunkedArrayEquals(expected, actual);
 }
 
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjection) {
+    ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata());
+    ASSERT_OK_AND_ASSIGN(auto physical_array, PhysicalArray());
+    auto mock_reader = std::make_unique<MockFileBatchReader>(
+        physical_array, arrow::struct_(physical_schema->fields()), /*read_batch_size=*/10);
+    mock_reader->EnableRandomizeBatchSize(false);
+
+    auto selected_type =
+        arrow::struct_({arrow::field("a", arrow::int64()), arrow::field("c", arrow::int64()),
+                        arrow::field("missing", arrow::int64())});
+    auto selected_field = arrow::field(
+        "tags", selected_type, /*nullable=*/true,
+        arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,c,missing"}));
+    ASSERT_OK_AND_ASSIGN(
+        auto field_read_plan,
+        MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(selected_field, TagsMeta()));
+    std::map<std::string, std::unique_ptr<MapFieldReadPlan>> contexts;
+    contexts.emplace("tags", std::move(field_read_plan));
+    auto reader = std::make_unique<MapSharedShreddingFileReader>(std::move(mock_reader),
+                                                                 std::move(contexts), pool_);
+
+    auto read_schema =
+        ExportSchema(arrow::schema({arrow::field("id", arrow::int32()), selected_field}));
+    ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr,
+                                    /*selection_bitmap=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(reader.get()));
+
+    auto expected_type = arrow::struct_({arrow::field("id", arrow::int32()), selected_field});
+    std::shared_ptr<arrow::ChunkedArray> expected;
+    ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(expected_type, {R"([
+                        [1, [10, null, null]],
+                        [2, [40, 30, null]],
+                        [3, null],
+                        [4, [80, null, null]]
+                    ])"},
+                                                                 &expected)
+                    .ok());
+    AssertChunkedArrayEquals(expected, actual);
+}
+
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjectionFromDefaultMap) {
+    auto map_type = arrow::internal::checked_pointer_cast<arrow::MapType>(
+        arrow::map(arrow::utf8(), arrow::field("value", arrow::int64())));
+    auto file_schema =
+        arrow::schema({arrow::field("id", arrow::int32()), arrow::field("tags", map_type)});
+    auto file_array =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(file_schema->fields()),
+                                                  R"([
+                              [1, [["a", 10], ["c", null]]],
+                              [2, [["b", 20]]],
+                              [3, null]
+                          ])")
+            .ValueOrDie();
+    auto mock_reader = std::make_unique<MockFileBatchReader>(
+        file_array, arrow::struct_(file_schema->fields()), /*read_batch_size=*/10);
+    mock_reader->EnableRandomizeBatchSize(false);
+
+    auto selected_type = arrow::struct_(
+        {arrow::field("a", arrow::int64()), arrow::field("missing", arrow::int64())});
+    auto selected_field =
+        arrow::field("tags", selected_type, /*nullable=*/true,
+                     arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,missing"}));
+    ASSERT_OK_AND_ASSIGN(auto field_read_plan,
+                         MapFieldReadPlanFactory::CreateDefaultSelectedKeysReadPlan(
+                             file_schema->field(1), selected_field));
+    std::map<std::string, std::unique_ptr<MapFieldReadPlan>> contexts;
+    contexts.emplace("tags", std::move(field_read_plan));
+    auto reader = std::make_unique<MapSharedShreddingFileReader>(std::move(mock_reader),
+                                                                 std::move(contexts), pool_);
+
+    auto read_schema =
+        ExportSchema(arrow::schema({arrow::field("id", arrow::int32()), selected_field}));
+    ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr,
+                                    /*selection_bitmap=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(reader.get()));
+
+    auto expected_type = arrow::struct_({arrow::field("id", arrow::int32()), selected_field});
+    std::shared_ptr<arrow::ChunkedArray> expected;
+    ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(expected_type, {R"([
+                        [1, [10, null]],
+                        [2, [null, null]],
+                        [3, null]
+                    ])"},
+                                                                 &expected)
+                    .ok());
+    AssertChunkedArrayEquals(expected, actual);
+}
+
+TEST_F(MapSharedShreddingFileReaderTest, TestInvalidSelectedKeysStructProjection) {
+    auto file_map_field = arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64()));
+    auto mismatched_count_field =
+        arrow::field("tags", arrow::struct_({arrow::field("a", arrow::int64())}), /*nullable=*/true,
+                     arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,b"}));
+    ASSERT_NOK_WITH_MSG(MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(
+                            mismatched_count_field, TagsMeta()),
+                        "metadata size 2 does not match STRUCT field count 1");
+    ASSERT_NOK_WITH_MSG(MapFieldReadPlanFactory::CreateDefaultSelectedKeysReadPlan(
+                            file_map_field, mismatched_count_field),
+                        "metadata size 2 does not match STRUCT field count 1");
+
+    auto mismatched_type_field = arrow::field(
+        "tags",
+        arrow::struct_({arrow::field("a", arrow::int64()), arrow::field("b", arrow::utf8())}),
+        /*nullable=*/true, arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,b"}));
+    ASSERT_NOK_WITH_MSG(MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(
+                            mismatched_type_field, TagsMeta()),
+                        "must have the same value type");
+    ASSERT_NOK_WITH_MSG(MapFieldReadPlanFactory::CreateDefaultSelectedKeysReadPlan(
+                            file_map_field, mismatched_type_field),
+                        "must have the same value type");
+}
+
 TEST_F(MapSharedShreddingFileReaderTest, TestPartialExistSelectedKeys) {
     ASSERT_OK_AND_ASSIGN(auto reader,
                          CreateReader(/*physical_array=*/nullptr, /*physical_schema=*/nullptr,