feat(shredding): improve shared-shredding adaptive width restore & refactor write (alibaba#377)

Author: lxy-9602

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

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"));
 }

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

@@ -139,7 +139,7 @@ Result<std::shared_ptr<arrow::Schema>> MapSharedShreddingUtils::LogicalToPhysica
             auto value_type = map_type->item_type();
             bool value_nullable = map_type->item_field()->nullable();
             auto physical_type = BuildPhysicalStructType(value_type, it->second, value_nullable);
-            auto physical_field = arrow::field(field->name(), physical_type, field->nullable());
+            auto physical_field = field->WithType(physical_type);
             physical_fields.push_back(physical_field);
         } else {
             physical_fields.push_back(field);
@@ -451,6 +451,10 @@ MapSharedShreddingUtils::BuildMetadataFinalizer(
         arrow::FieldVector updated_fields = physical_schema->fields();
         for (const std::string& field_name : shredding_field_names) {
             int32_t col_index = physical_schema->GetFieldIndex(field_name);
+            if (col_index < 0) {
+                return Status::Invalid(fmt::format(
+                    "Shared-shredding field '{}' not found in physical schema.", field_name));
+            }
             const auto& field = physical_schema->field(col_index);
             auto metadata = field->metadata() ? field->metadata()->Copy()
                                               : std::make_shared<arrow::KeyValueMetadata>();

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

@@ -138,6 +138,7 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNullable) {
     ASSERT_OK_AND_ASSIGN(
         auto physical, MapSharedShreddingUtils::LogicalToPhysicalSchema(schema_nullable, col_map));
     auto struct_type = physical->field(0)->type();
+    ASSERT_TRUE(struct_type->field(0)->nullable());
     ASSERT_TRUE(struct_type->field(1)->nullable());
     ASSERT_TRUE(struct_type->field(2)->nullable());
 
@@ -152,6 +153,21 @@ TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNullable) {
     ASSERT_FALSE(struct_type2->field(2)->nullable());
 }
 
+TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaPreservesFieldMetadata) {
+    auto metadata = std::make_shared<arrow::KeyValueMetadata>();
+    metadata->Append("paimon.field.id", "7");
+    metadata->Append("description", "original map field");
+    auto map_type = arrow::map(arrow::utf8(), arrow::int64());
+    auto schema = arrow::schema({arrow::field("m", map_type, false, metadata)});
+    std::map<std::string, int32_t> col_map = {{"m", 2}};
+
+    ASSERT_OK_AND_ASSIGN(auto physical_schema,
+                         MapSharedShreddingUtils::LogicalToPhysicalSchema(schema, col_map));
+
+    ASSERT_FALSE(physical_schema->field(0)->nullable());
+    ASSERT_TRUE(physical_schema->field(0)->metadata()->Equals(*metadata));
+}
+
 TEST(MapSharedShreddingUtilsTest, LogicalToPhysicalSchemaNoShreddingColumns) {
     auto schema = arrow::schema({
         arrow::field("id", arrow::int32()),

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

@@ -166,6 +166,20 @@ class SharedShreddingFileReaderTest : public ::testing::Test {
         return reader_builder->Build(input_stream).value();
     }
 
+    Result<std::unique_ptr<AppendOnlyWriter>> CreateAppendOnlyWriter(
+        const CoreOptions& core_options, int64_t schema_id,
+        const std::shared_ptr<arrow::Schema>& logical_schema,
+        const std::optional<std::vector<std::string>>& write_cols, int64_t max_sequence_number,
+        const std::shared_ptr<DataFilePathFactory>& path_factory,
+        const std::shared_ptr<CompactManager>& compact_manager) const {
+        PAIMON_ASSIGN_OR_RAISE(
+            std::shared_ptr<MapSharedShreddingContext> shredding_context,
+            MapSharedShreddingUtils::CreateShreddingContext(logical_schema, core_options));
+        return std::make_unique<AppendOnlyWriter>(core_options, schema_id, logical_schema,
+                                                  write_cols, max_sequence_number, path_factory,
+                                                  compact_manager, shredding_context, pool_);
+    }
+
  private:
     std::shared_ptr<MemoryPool> pool_;
     std::shared_ptr<arrow::Schema> logical_schema_ = arrow::schema({
@@ -473,9 +487,9 @@ TEST_F(SharedShreddingFileReaderTest, TestOrcDictionaryEncodedStringValue) {
     auto compact_manager = std::make_shared<NoopCompactManager>();
     ASSERT_OK_AND_ASSIGN(
         auto writer,
-        AppendOnlyWriter::Create(core_options, /*schema_id=*/0, logical_schema,
-                                 /*write_cols=*/std::nullopt,
-                                 /*max_sequence_number=*/-1, path_factory, compact_manager, pool_));
+        CreateAppendOnlyWriter(core_options, /*schema_id=*/0, logical_schema,
+                               /*write_cols=*/std::nullopt,
+                               /*max_sequence_number=*/-1, path_factory, compact_manager));
     auto batch = CreateBatch(logical_schema, R"([
         [1, [["a", "red"], ["b", "blue"]]],
         [2, [["c", "green"], ["a", "red"], ["b", "blue"]]],
@@ -530,9 +544,9 @@ TEST_F(SharedShreddingFileReaderTest, TestReadsRealFormatFile) {
     auto compact_manager = std::make_shared<NoopCompactManager>();
     ASSERT_OK_AND_ASSIGN(
         auto writer,
-        AppendOnlyWriter::Create(core_options, /*schema_id=*/0, logical_schema_,
-                                 /*write_cols=*/std::nullopt,
-                                 /*max_sequence_number=*/-1, path_factory, compact_manager, pool_));
+        CreateAppendOnlyWriter(core_options, /*schema_id=*/0, logical_schema_,
+                               /*write_cols=*/std::nullopt,
+                               /*max_sequence_number=*/-1, path_factory, compact_manager));
     auto batch = CreateBatch(logical_schema_, R"([
         [1, [["a", 1], ["b", 2]]],
         [2, [["c", 3], ["a", 4], ["b", 5]]],