Merge branch 'main' into codex/fix-review-minor-issues

Author: lxy-9602

cmake_modules/CorrosionFetch.cmake

@@ -16,7 +16,7 @@
 # targets. Used to bring in third_party/tantivy_ffi for the tantivy-fulltext
 # global index (see docs/dev/tantivy_fts_migration_plan.md).
 #
-# Pinned to v0.5.0 (stable release). Requires CMake >= 3.22.
+# Pinned to v0.5.2 (stable release). Requires CMake >= 3.22.
 
 include(FetchContent)
 

include/paimon/read_context.h

@@ -23,6 +23,7 @@
 #include <string>
 #include <vector>
 
+#include "arrow/c/abi.h"
 #include "paimon/cache/cache.h"
 #include "paimon/predicate/predicate.h"
 #include "paimon/result.h"
@@ -44,7 +45,7 @@ class FileSystem;
 class PAIMON_EXPORT ReadContext {
  public:
     ReadContext(const std::string& path, const std::string& branch,
-                const std::vector<std::string>& read_schema,
+                const std::vector<std::string>& read_field_names,
                 const std::vector<int32_t>& read_field_ids,
                 const std::shared_ptr<Predicate>& predicate, bool enable_predicate_filter,
                 bool enable_prefetch, uint32_t prefetch_batch_count,
@@ -75,8 +76,8 @@ class PAIMON_EXPORT ReadContext {
         return options_;
     }
 
-    const std::vector<std::string>& GetReadSchema() const {
-        return read_schema_;
+    const std::vector<std::string>& GetReadFieldNames() const {
+        return read_field_names_;
     }
 
     const std::vector<int32_t>& GetReadFieldIds() const {
@@ -130,10 +131,26 @@ class PAIMON_EXPORT ReadContext {
         return cache_;
     }
 
+    /// Whether a read schema (C ArrowSchema) for nested column pruning was provided.
+    bool HasReadSchema() const {
+        return read_schema_ != nullptr && read_schema_->release != nullptr;
+    }
+
+    /// Get the read schema as a mutable C ArrowSchema pointer.
+    /// ImportSchema will consume (release) the schema content.
+    ArrowSchema* GetReadSchema() {
+        return read_schema_.get();
+    }
+
+    /// Set the read schema from a C ArrowSchema unique_ptr and take ownership of
+    /// schema resources (released via ArrowSchema::release in destructor).
+    /// Called internally by ReadContextBuilder.
+    void SetReadSchema(std::unique_ptr<ArrowSchema> schema);
+
  private:
     std::string path_;
     std::string branch_;
-    std::vector<std::string> read_schema_;
+    std::vector<std::string> read_field_names_;
     std::vector<int32_t> read_field_ids_;
     std::shared_ptr<Predicate> predicate_;
     bool enable_predicate_filter_;
@@ -151,6 +168,8 @@ class PAIMON_EXPORT ReadContext {
     PrefetchCacheMode prefetch_cache_mode_;
     CacheConfig cache_config_;
     std::shared_ptr<Cache> cache_;
+    // Owns schema resources and releases ArrowSchema::release in destructor.
+    std::unique_ptr<ArrowSchema> read_schema_;
 };
 
 /// `ReadContextBuilder` used to build a `ReadContext`, has input validation.
@@ -173,9 +192,9 @@ class PAIMON_EXPORT ReadContextBuilder {
     ///
     /// @param read_field_names Vector of field names to read from the table.
     /// @return Reference to this builder for method chaining.
-    /// @note Currently supports top-level field selection. Future versions may support
-    ///       nested field selection using ArrowSchema for more granular projection
-    ReadContextBuilder& SetReadSchema(const std::vector<std::string>& read_field_names);
+    /// @note Currently supports top-level field selection. For nested field selection
+    ///       use SetReadSchema(std::unique_ptr<ArrowSchema>) instead.
+    ReadContextBuilder& SetReadFieldNames(const std::vector<std::string>& read_field_names);
     /// Set the schema fields to read from the table.
     ///
     /// If not set, all fields from the table schema will be read. This is useful for
@@ -184,12 +203,51 @@ class PAIMON_EXPORT ReadContextBuilder {
     ///
     /// @param read_field_ids Vector of field ids to read from the table.
     /// @return Reference to this builder for method chaining.
-    /// @note Currently supports top-level field selection. Future versions may support
-    ///       nested field selection using ArrowSchema for more granular projection.
-    /// @note SetReadFieldIds() and SetReadSchema() are mutually exclusive.
-    ///       Calling both will ignore the read schema set by SetReadSchema().
+    /// @note Currently supports top-level field selection.
+    /// @note SetReadFieldIds() and SetReadFieldNames() are mutually exclusive.
+    ///       Calling both will ignore the read schema set by SetReadFieldNames().
     ReadContextBuilder& SetReadFieldIds(const std::vector<int32_t>& read_field_ids);
 
+    /// Set the read Arrow Schema for nested column pruning.
+    ///
+    /// The read schema is an Arrow C Data Interface schema where STRUCT types
+    /// may contain only a subset of the original sub-fields, enabling nested column
+    /// pruning to reduce I/O. Field matching is based on field name: the system
+    /// looks up each field by name in the table schema and rebuilds the aligned
+    /// schema using the table schema's type and metadata. Metadata propagation
+    /// from the user-provided schema is whitelist-based: currently only
+    /// "paimon.map.selected-keys" is preserved and merged into the final aligned
+    /// schema.
+    ///
+    /// To prune map entries by key, attach metadata "paimon.map.selected-keys"
+    /// to the target map field in read schema. The value is a comma-separated
+    /// key list, for example: "k1,k2". Only map fields with string key type
+    /// (Arrow utf8) are supported.
+    ///
+    /// Example:
+    /// @code{.cpp}
+    /// auto map_field = arrow::field("m", arrow::map(arrow::utf8(), arrow::int32()));
+    /// auto map_meta = arrow::KeyValueMetadata::Make(
+    ///     {"paimon.map.selected-keys"}, {"k1,k2"});
+    /// auto projected_schema = arrow::schema({
+    ///     arrow::field("id", arrow::int64()),
+    ///     map_field->WithMetadata(map_meta),
+    /// });
+    ///
+    /// auto c_schema = std::make_unique<ArrowSchema>();
+    /// arrow::ExportSchema(*projected_schema, c_schema.get());
+    ///
+    /// ReadContextBuilder builder("/path/to/table");
+    /// builder.SetReadSchema(std::move(c_schema));
+    /// @endcode
+    ///
+    /// @param read_schema Arrow C Schema. Ownership of schema resources is transferred
+    ///        to the built ReadContext.
+    /// @return Reference to this builder for method chaining.
+    /// @note Priority: read_schema > read_field_ids > read_field_names.
+    ///       When set, read_field_ids and read_field_names are ignored.
+    ReadContextBuilder& SetReadSchema(std::unique_ptr<ArrowSchema> read_schema);
+
     /// Set a configuration options map to set some option entries which are not defined in the
     /// table schema or whose values you want to overwrite.
     /// @note The options map will clear the options added by `AddOption()` before.

src/paimon/CMakeLists.txt

@@ -220,17 +220,24 @@ set(PAIMON_CORE_SRCS
     core/io/data_file_meta.cpp
     core/io/data_file_meta_serializer.cpp
     core/io/data_file_path_factory.cpp
+    core/io/append_data_file_writer_factory.cpp
+    core/io/blob_data_file_writer_factory.cpp
+    core/io/data_file_writer_factory.cpp
     core/io/data_file_writer.cpp
     core/io/field_mapping_reader.cpp
     core/io/complete_row_tracking_fields_reader.cpp
     core/io/file_index_evaluator.cpp
     core/io/key_value_data_file_record_reader.cpp
+    core/io/key_value_data_file_writer_factory.cpp
     core/io/key_value_data_file_writer.cpp
     core/io/key_value_in_memory_record_reader.cpp
     core/io/merged_key_value_record_reader.cpp
     core/io/key_value_meta_projection_consumer.cpp
     core/io/key_value_projection_consumer.cpp
     core/io/key_value_projection_reader.cpp
+    core/io/map_shared_shredding_core_utils.cpp
+    core/io/shredding_append_data_file_writer_factory.cpp
+    core/io/shredding_key_value_data_file_writer_factory.cpp
     core/io/external_storage_blob_writer.cpp
     core/io/multiple_blob_file_writer.cpp
     core/io/rolling_blob_file_writer.cpp
@@ -343,6 +350,7 @@ set(PAIMON_CORE_SRCS
     core/utils/blob_view_lookup.cpp
     core/utils/consumer_manager.cpp
     core/utils/field_mapping.cpp
+    core/utils/nested_projection_utils.cpp
     core/utils/file_store_path_factory.cpp
     core/utils/file_utils.cpp
     core/utils/manifest_meta_reader.cpp
@@ -737,6 +745,7 @@ if(PAIMON_BUILD_TESTS)
                     core/utils/consumer_manager_test.cpp
                     core/utils/file_store_path_factory_cache_test.cpp
                     core/utils/field_mapping_test.cpp
+                    core/utils/nested_projection_utils_test.cpp
                     core/utils/file_store_path_factory_test.cpp
                     core/utils/file_utils_test.cpp
                     core/utils/manifest_meta_reader_test.cpp

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

@@ -17,6 +17,7 @@
 #include "paimon/common/data/shredding/map_shared_shredding_context.h"
 
 #include <algorithm>
+#include <cmath>
 
 namespace paimon {
 
@@ -32,8 +33,8 @@ std::map<std::string, int32_t> MapSharedShreddingContext::ComputeNextK() const {
             // First file — no history, use K_max.
             result[field_name] = k_max;
         } else {
-            int32_t window_max = ComputeWindowMax(it->second);
-            result[field_name] = std::max(1, std::min(window_max, k_max));
+            int32_t adaptive_width = ComputeAdaptiveWidth(it->second);
+            result[field_name] = std::max(1, std::min(adaptive_width, k_max));
         }
     }
     return result;
@@ -57,12 +58,27 @@ std::vector<std::string> MapSharedShreddingContext::GetShreddingColumnNames() co
     return names;
 }
 
-int32_t MapSharedShreddingContext::ComputeWindowMax(const std::vector<int32_t>& values) {
+int32_t MapSharedShreddingContext::ComputeAdaptiveWidth(const std::vector<int32_t>& values) {
     if (values.empty()) {
         return 0;
     }
-    // TODO(xinyu.lxy): support P99
-    return *std::max_element(values.begin(), values.end());
+
+    std::vector<int32_t> sorted_values(values.begin(), values.end());
+    std::sort(sorted_values.begin(), sorted_values.end());
+
+    int32_t max_width = sorted_values.back();
+    auto percentile_rank = static_cast<int64_t>(std::ceil(kPercentileRatio * sorted_values.size()));
+    percentile_rank = std::clamp<int64_t>(percentile_rank, 1, sorted_values.size());
+    int32_t percentile_width = sorted_values[percentile_rank - 1];
+
+    // Use P90 to ignore far outliers, but keep max when it is close enough to normal rows.
+    auto relative_close_threshold = static_cast<int32_t>(
+        std::ceil(static_cast<double>(percentile_width) * kMaxCloseRelativeRatio));
+    if (max_width - percentile_width <= kMaxCloseAbsoluteSlack ||
+        max_width <= relative_close_threshold) {
+        return max_width;
+    }
+    return percentile_width;
 }
 
 }  // namespace paimon

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

@@ -30,7 +30,7 @@ namespace paimon {
 /// values to support adaptive K sizing across files.
 ///
 /// - First file: K = K_max (no history).
-/// - Subsequent files: K = min(max(recent_max_row_widths), K_max).
+/// - Subsequent files: K = min(adaptive_width(recent_max_row_widths), K_max).
 class MapSharedShreddingContext {
  public:
     /// @param column_to_k_max Map from field name to its K_max (from options).
@@ -50,9 +50,12 @@ class MapSharedShreddingContext {
     std::vector<std::string> GetShreddingColumnNames() const;
 
  private:
-    static constexpr int32_t kWindowSize = 100;
+    static constexpr int32_t kWindowSize = 20;
+    static constexpr double kPercentileRatio = 0.90;
+    static constexpr int32_t kMaxCloseAbsoluteSlack = 4;
+    static constexpr double kMaxCloseRelativeRatio = 1.25;
 
-    static int32_t ComputeWindowMax(const std::vector<int32_t>& values);
+    static int32_t ComputeAdaptiveWidth(const std::vector<int32_t>& values);
 
     /// K_max per shared-shredding field, from options.
     std::map<std::string, int32_t> column_to_k_max_;

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

@@ -37,7 +37,7 @@ TEST(MapSharedShreddingContextTest, FirstFileUsesKMax) {
 
 TEST(MapSharedShreddingContextTest, AdaptKAfterOneFile) {
     // After reporting stats from one file, K should adapt to
-    // min(max_row_width, K_max).
+    // min(adaptive width, K_max).
     std::map<std::string, int32_t> field_to_k_max = {{"m", 10}};
     MapSharedShreddingContext context(field_to_k_max);
 
@@ -64,20 +64,71 @@ TEST(MapSharedShreddingContextTest, AdaptKCappedByKMax) {
     ASSERT_EQ(5, next_k.at("m"));
 }
 
-TEST(MapSharedShreddingContextTest, WindowMaxTracksLargest) {
-    // K should use the max of all recent max_row_widths within the window.
+TEST(MapSharedShreddingContextTest, WindowP90UsesMaxWhenSamplesAreClose) {
+    // Small sample windows still use max because max and P90 are close.
     std::map<std::string, int32_t> field_to_k_max = {{"m", 20}};
     MapSharedShreddingContext context(field_to_k_max);
 
     context.ReportFileStats("m", 3);
     context.ReportFileStats("m", 7);
     context.ReportFileStats("m", 5);
 
-    // max of {3, 7, 5} = 7, capped by K_max=20 → K=7.
     auto next_k = context.ComputeNextK();
     ASSERT_EQ(7, next_k.at("m"));
 }
 
+TEST(MapSharedShreddingContextTest, WindowP90IgnoresSingleFarOutlier) {
+    std::map<std::string, int32_t> field_to_k_max = {{"m", 2000}};
+    MapSharedShreddingContext context(field_to_k_max);
+
+    for (int32_t i = 0; i < 19; ++i) {
+        context.ReportFileStats("m", 3);
+    }
+    context.ReportFileStats("m", 1000);
+
+    auto next_k = context.ComputeNextK();
+    ASSERT_EQ(3, next_k.at("m"));
+}
+
+TEST(MapSharedShreddingContextTest, WindowP90UsesMaxWithinAbsoluteSlack) {
+    std::map<std::string, int32_t> field_to_k_max = {{"m", 20}};
+    MapSharedShreddingContext context(field_to_k_max);
+
+    for (int32_t i = 0; i < 19; ++i) {
+        context.ReportFileStats("m", 3);
+    }
+    context.ReportFileStats("m", 7);
+
+    auto next_k = context.ComputeNextK();
+    ASSERT_EQ(7, next_k.at("m"));
+}
+
+TEST(MapSharedShreddingContextTest, WindowP90UsesMaxWithinRelativeSlack) {
+    std::map<std::string, int32_t> field_to_k_max = {{"m", 200}};
+    MapSharedShreddingContext context(field_to_k_max);
+
+    for (int32_t i = 0; i < 19; ++i) {
+        context.ReportFileStats("m", 100);
+    }
+    context.ReportFileStats("m", 125);
+
+    auto next_k = context.ComputeNextK();
+    ASSERT_EQ(125, next_k.at("m"));
+}
+
+TEST(MapSharedShreddingContextTest, WindowP90IgnoresMaxBeyondBothSlacks) {
+    std::map<std::string, int32_t> field_to_k_max = {{"m", 200}};
+    MapSharedShreddingContext context(field_to_k_max);
+
+    for (int32_t i = 0; i < 19; ++i) {
+        context.ReportFileStats("m", 100);
+    }
+    context.ReportFileStats("m", 130);
+
+    auto next_k = context.ComputeNextK();
+    ASSERT_EQ(100, next_k.at("m"));
+}
+
 TEST(MapSharedShreddingContextTest, MultipleColumnsIndependent) {
     // Each field adapts independently.
     std::map<std::string, int32_t> field_to_k_max = {{"tags", 10}, {"attrs", 6}};
@@ -101,8 +152,7 @@ TEST(MapSharedShreddingContextTest, MultipleColumnsIndependent) {
     context.ReportFileStats("attrs", 6);
 
     auto k3 = context.ComputeNextK();
-    // tags: max(4,8)=8, capped by 10 → 8
-    // attrs: max(2,6)=6, capped by 6 → 6
+    // tags and attrs are close sample windows, so adaptive width keeps max.
     ASSERT_EQ(8, k3.at("tags"));
     ASSERT_EQ(6, k3.at("attrs"));
 }
@@ -116,7 +166,7 @@ TEST(MapSharedShreddingContextTest, GetShreddingColumnNames) {
 }
 
 TEST(MapSharedShreddingContextTest, SlidingWindowEvictsOldEntries) {
-    // The window size is 100. After filling 100 entries, adding one more
+    // The window size is 20. After filling 20 entries, adding one more
     // should evict the oldest. Verify that the evicted value no longer
     // affects ComputeNextK.
     std::map<std::string, int32_t> field_to_k_max = {{"m", 256}};
@@ -125,19 +175,19 @@ TEST(MapSharedShreddingContextTest, SlidingWindowEvictsOldEntries) {
     // Insert a large value as the first entry.
     context.ReportFileStats("m", 200);
 
-    // Fill the remaining 99 slots with small values.
-    for (int i = 0; i < 99; ++i) {
+    // Fill the remaining 19 slots with small values.
+    for (int32_t i = 0; i < 19; ++i) {
         context.ReportFileStats("m", 3);
     }
 
-    // Window = [200, 3, 3, ..., 3] (100 entries). Max = 200.
+    // Window = [200, 3, 3, ..., 3] (20 entries). P90 = 3, max is a far outlier.
     auto k_before = context.ComputeNextK();
-    ASSERT_EQ(200, k_before.at("m"));
+    ASSERT_EQ(3, k_before.at("m"));
 
     // Push one more — evicts the 200.
     context.ReportFileStats("m", 5);
 
-    // Window = [3, 3, ..., 3, 5] (100 entries). Max = 5.
+    // Window = [3, 3, ..., 3, 5] (20 entries). Max is within the absolute slack.
     auto k_after = context.ComputeNextK();
     ASSERT_EQ(5, k_after.at("m"));
 }