feat(shredding): support adaptive schemas across rolling files

include/paimon/data/shredding/map_shared_shredding_schema_utils.h

@@ -85,11 +85,9 @@ class PAIMON_EXPORT MapSharedShreddingSchemaUtils {
     /// @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);
+        std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name);
 };
 
 }  // namespace paimon

include/paimon/defs.h

@@ -117,6 +117,12 @@ struct PAIMON_EXPORT Options {
     /// append table: the default value is 256 MB.
     static const char TARGET_FILE_SIZE[];
 
+    /// "target-file-row-num" - Target number of rows per newly written data file. Disabled by
+    /// default. A file rolls when this or target-file-size is reached, whichever comes first.
+    /// This limit is enforced at write-batch granularity, so a file may exceed the target by up
+    /// to one batch.
+    static const char TARGET_FILE_ROW_NUM[];
+
     /// "blob.target-file-size" - Target size of a blob file. Default is TARGET_FILE_SIZE.
     static const char BLOB_TARGET_FILE_SIZE[];
 
@@ -433,6 +439,8 @@ struct PAIMON_EXPORT Options {
     /// "variant.inferShreddingSchema" - Whether to automatically infer the shredding schema when
     /// writing Variant columns. Default value is "false".
     static const char VARIANT_INFER_SHREDDING_SCHEMA[];
+    /// "variant.shredding.inferenceMode" - "per-file" or "adaptive". Default is "per-file".
+    static const char VARIANT_SHREDDING_INFERENCE_MODE[];
     /// "variant.shredding.maxSchemaWidth" - Maximum number of shredded fields allowed in an
     /// inferred schema. Default value is 300.
     static const char VARIANT_SHREDDING_MAX_SCHEMA_WIDTH[];
@@ -446,6 +454,12 @@ struct PAIMON_EXPORT Options {
     /// "variant.shredding.maxInferBufferRow" - Maximum number of rows to buffer for schema
     /// inference. Default value is 4096.
     static const char VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW[];
+    /// "variant.shredding.adaptive.maxInferBufferRow" - Maximum prefix rows sampled after the
+    /// first file in an adaptive session. Default value is 256.
+    static const char VARIANT_SHREDDING_ADAPTIVE_MAX_INFER_BUFFER_ROW[];
+    /// "variant.shredding.adaptive.retentionRatio" - Minimum combined ratio for retaining a
+    /// previously selected path. Default value is 0.05.
+    static const char VARIANT_SHREDDING_ADAPTIVE_RETENTION_RATIO[];
 
     /// "blob-as-descriptor" - Read blob field using blob descriptor rather than blob
     /// bytes. Default value is "false".

src/paimon/CMakeLists.txt

@@ -40,6 +40,7 @@ set(PAIMON_COMMON_SRCS
     common/data/timestamp.cpp
     common/data/variant/generic_variant.cpp
     common/data/variant/infer_variant_shredding_schema.cpp
+    common/data/variant/variant_shredding_inference_session.cpp
     common/data/variant/variant.cpp
     common/data/variant/variant_binary_util.cpp
     common/data/variant/variant_builder.cpp
@@ -264,7 +265,6 @@ set(PAIMON_CORE_SRCS
     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/multiple_blob_file_writer.cpp

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

@@ -337,6 +337,17 @@ Result<MapSharedShreddingFieldMeta> MapSharedShreddingBatchConverter::BuildField
         "cannot find field_name '{}' in MapSharedShreddingBatchConverter contexts", field_name));
 }
 
+Result<int32_t> MapSharedShreddingBatchConverter::GetMaxRowWidth(
+    const std::string& field_name) const {
+    for (const auto& context : contexts_) {
+        if (context.field_name == field_name) {
+            return context.allocator->GetMaxRowWidth();
+        }
+    }
+    return Status::Invalid(fmt::format(
+        "cannot find field_name '{}' in MapSharedShreddingBatchConverter contexts", field_name));
+}
+
 const std::vector<std::string>& MapSharedShreddingBatchConverter::GetShreddingColumnNames() const {
     return shredding_field_names_;
 }

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

@@ -71,6 +71,9 @@ class MapSharedShreddingBatchConverter : public ShreddingBatchConverter {
     /// Called at file close to serialize metadata.
     Result<MapSharedShreddingFieldMeta> BuildFieldMeta(const std::string& field_name) const;
 
+    /// Returns the maximum number of entries observed in one row for a shredding column.
+    Result<int32_t> GetMaxRowWidth(const std::string& field_name) const;
+
     /// Returns all shredding column field names.
     const std::vector<std::string>& GetShreddingColumnNames() const;
 

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

@@ -28,6 +28,7 @@
 #include "gtest/gtest.h"
 #include "paimon/common/data/shredding/map_shared_shredding_context.h"
 #include "paimon/common/data/shredding/map_shared_shredding_utils.h"
+#include "paimon/common/data/shredding/map_shared_shredding_write_plan_factory.h"
 #include "paimon/common/data/shredding/map_shredding_defs.h"
 #include "paimon/core/core_options.h"
 #include "paimon/memory/memory_pool.h"
@@ -118,6 +119,77 @@ TEST_F(MapSharedShreddingBatchConverterTest, BasicConversion) {
     ASSERT_EQ(expected_meta, converter->BuildFieldMeta("tags").value());
 }
 
+TEST_F(MapSharedShreddingBatchConverterTest, MapWithNullValue) {
+    auto logical_schema =
+        arrow::schema({arrow::field("metrics", arrow::map(arrow::utf8(), arrow::int64()))});
+    auto context =
+        std::make_shared<MapSharedShreddingContext>(std::map<std::string, int32_t>{{"metrics", 2}});
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, MakeCoreOptions({{"metrics", "plain"}}));
+    ASSERT_OK_AND_ASSIGN(auto converter, MapSharedShreddingBatchConverter::Create(
+                                             logical_schema, context, options, pool_));
+    auto logical_type = arrow::struct_(logical_schema->fields());
+    auto physical_type = arrow::struct_(converter->GetPhysicalSchema()->fields());
+
+    auto actual =
+        RunConvert(logical_type, R"([[[["a", null], ["b", 20]]]])", physical_type, converter.get());
+    auto expected = ArrayFromJSON(physical_type, R"([[[[0, 1], null, 20, null]]])").ValueOrDie();
+    AssertArrayEquals(expected, actual);
+
+    MapSharedShreddingFieldMeta expected_meta;
+    expected_meta.name_to_id = {{"a", 0}, {"b", 1}};
+    expected_meta.field_to_columns = {{0, {0}}, {1, {1}}};
+    expected_meta.num_columns = 2;
+    expected_meta.max_row_width = 2;
+    ASSERT_EQ(expected_meta, converter->BuildFieldMeta("metrics").value());
+}
+
+TEST_F(MapSharedShreddingBatchConverterTest, OverflowWhenExceedK) {
+    auto logical_schema =
+        arrow::schema({arrow::field("metrics", arrow::map(arrow::utf8(), arrow::int64()))});
+    auto context =
+        std::make_shared<MapSharedShreddingContext>(std::map<std::string, int32_t>{{"metrics", 2}});
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, MakeCoreOptions({{"metrics", "plain"}}));
+    ASSERT_OK_AND_ASSIGN(auto converter, MapSharedShreddingBatchConverter::Create(
+                                             logical_schema, context, options, pool_));
+    auto logical_type = arrow::struct_(logical_schema->fields());
+    auto physical_type = arrow::struct_(converter->GetPhysicalSchema()->fields());
+
+    auto actual = RunConvert(logical_type, R"([[[["a", 10], ["b", 20], ["c", 30]]]])",
+                             physical_type, converter.get());
+    auto expected = ArrayFromJSON(physical_type, R"([[[[0, 1], 10, 20, [[2, 30]]]]])").ValueOrDie();
+    AssertArrayEquals(expected, actual);
+
+    MapSharedShreddingFieldMeta expected_meta;
+    expected_meta.name_to_id = {{"a", 0}, {"b", 1}, {"c", 2}};
+    expected_meta.field_to_columns = {{0, {0}}, {1, {1}}};
+    expected_meta.overflow_field_set = {2};
+    expected_meta.num_columns = 2;
+    expected_meta.max_row_width = 3;
+    ASSERT_EQ(expected_meta, converter->BuildFieldMeta("metrics").value());
+}
+
+TEST_F(MapSharedShreddingBatchConverterTest, EmptyAndNullMaps) {
+    auto logical_schema =
+        arrow::schema({arrow::field("metrics", arrow::map(arrow::utf8(), arrow::int64()))});
+    auto context =
+        std::make_shared<MapSharedShreddingContext>(std::map<std::string, int32_t>{{"metrics", 2}});
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, MakeCoreOptions({{"metrics", "plain"}}));
+    ASSERT_OK_AND_ASSIGN(auto converter, MapSharedShreddingBatchConverter::Create(
+                                             logical_schema, context, options, pool_));
+    auto logical_type = arrow::struct_(logical_schema->fields());
+    auto physical_type = arrow::struct_(converter->GetPhysicalSchema()->fields());
+
+    auto actual = RunConvert(logical_type, R"([[null], [[]]])", physical_type, converter.get());
+    auto expected =
+        ArrayFromJSON(physical_type, R"([[null], [[[-1, -1], null, null, null]]])").ValueOrDie();
+    AssertArrayEquals(expected, actual);
+
+    MapSharedShreddingFieldMeta expected_meta;
+    expected_meta.num_columns = 2;
+    expected_meta.max_row_width = 0;
+    ASSERT_EQ(expected_meta, converter->BuildFieldMeta("metrics").value());
+}
+
 TEST_F(MapSharedShreddingBatchConverterTest, NestedValueStruct) {
     // MAP<STRING, STRUCT<x:INT32, y:DOUBLE>>, K=2
     auto value_type = arrow::struct_({
@@ -408,13 +480,18 @@ TEST_F(MapSharedShreddingBatchConverterTest, BuildFieldMetaInvalidFieldName) {
 
     // Valid case: "tags" exists
     ASSERT_OK_AND_ASSIGN([[maybe_unused]] auto meta, converter->BuildFieldMeta("tags"));
+    ASSERT_OK_AND_ASSIGN(int32_t max_row_width, converter->GetMaxRowWidth("tags"));
+    ASSERT_EQ(0, max_row_width);
 
     // Invalid case: "id" is not a shredding field
     ASSERT_NOK_WITH_MSG(converter->BuildFieldMeta("id"), "cannot find field_name 'id'");
+    ASSERT_NOK_WITH_MSG(converter->GetMaxRowWidth("id"), "cannot find field_name 'id'");
 
     // Invalid case: nonexistent field name
     ASSERT_NOK_WITH_MSG(converter->BuildFieldMeta("nonexistent"),
                         "cannot find field_name 'nonexistent'");
+    ASSERT_NOK_WITH_MSG(converter->GetMaxRowWidth("nonexistent"),
+                        "cannot find field_name 'nonexistent'");
 }
 
 TEST_F(MapSharedShreddingBatchConverterTest, SequentialPlacementUsesSmallestColumn) {
@@ -496,4 +573,57 @@ TEST_F(MapSharedShreddingBatchConverterTest, LruPlacementPreservesResidentColumn
     ASSERT_EQ(expected_meta, converter->BuildFieldMeta("tags").value());
 }
 
+TEST_F(MapSharedShreddingBatchConverterTest, FactoryUsesMaxColumnCountForFirstFile) {
+    auto logical_schema =
+        arrow::schema({arrow::field("tags", arrow::map(arrow::utf8(), arrow::int32()))});
+    ASSERT_OK_AND_ASSIGN(
+        CoreOptions options,
+        CoreOptions::FromMap({{"fields.tags.map.storage-layout", "shared-shredding"},
+                              {"fields.tags.map.shared-shredding.max-columns", "4"}}));
+    ASSERT_OK_AND_ASSIGN(
+        auto factory, MapSharedShreddingWritePlanFactory::Create(options, logical_schema, pool_));
+
+    ASSERT_TRUE(factory->ShouldCreateWritePlan());
+    ASSERT_FALSE(factory->ShouldInferWritePlan());
+    ASSERT_EQ(0, factory->InferBufferRowCount());
+    ASSERT_OK_AND_ASSIGN(auto converter, factory->CreateConverter("parquet", {}));
+    ASSERT_OK_AND_ASSIGN(auto expected, MapSharedShreddingUtils::LogicalToPhysicalSchema(
+                                            logical_schema, {{"tags", 4}}));
+    ASSERT_TRUE(converter->GetPhysicalSchema()->Equals(*expected));
+}
+
+TEST_F(MapSharedShreddingBatchConverterTest, FactoryUsesCompletedFileStatsForNextFile) {
+    auto logical_schema =
+        arrow::schema({arrow::field("tags", arrow::map(arrow::utf8(), arrow::int32()))});
+    ASSERT_OK_AND_ASSIGN(
+        CoreOptions options,
+        CoreOptions::FromMap({{"fields.tags.map.storage-layout", "shared-shredding"},
+                              {"fields.tags.map.shared-shredding.max-columns", "8"}}));
+    ASSERT_OK_AND_ASSIGN(
+        auto factory, MapSharedShreddingWritePlanFactory::Create(options, logical_schema, pool_));
+    ASSERT_OK_AND_ASSIGN(auto first, factory->CreateConverter("parquet", {}));
+
+    auto logical_type = arrow::struct_(logical_schema->fields());
+    auto first_physical_type = arrow::struct_(first->GetPhysicalSchema()->fields());
+    auto input = ArrayFromJSON(logical_type, R"([
+        [[["a", 1], ["b", 2]]],
+        [[["c", 3], ["d", 4], ["e", 5]]]
+    ])")
+                     .ValueOrDie();
+    ArrowArray c_input;
+    ASSERT_TRUE(arrow::ExportArray(*input, &c_input).ok());
+    ASSERT_OK_AND_ASSIGN(auto c_output, first->Convert(&c_input));
+    auto first_output = arrow::ImportArray(c_output.get(), first_physical_type).ValueOrDie();
+    ASSERT_EQ(2, first_output->length());
+    ASSERT_OK(factory->OnFileCompleted(first));
+
+    ASSERT_OK_AND_ASSIGN(auto second, factory->CreateConverter("parquet", {}));
+    ASSERT_OK_AND_ASSIGN(auto expected_first, MapSharedShreddingUtils::LogicalToPhysicalSchema(
+                                                  logical_schema, {{"tags", 8}}));
+    ASSERT_OK_AND_ASSIGN(auto expected_second, MapSharedShreddingUtils::LogicalToPhysicalSchema(
+                                                   logical_schema, {{"tags", 3}}));
+    ASSERT_TRUE(first->GetPhysicalSchema()->Equals(*expected_first));
+    ASSERT_TRUE(second->GetPhysicalSchema()->Equals(*expected_second));
+}
+
 }  // namespace paimon

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

@@ -25,7 +25,7 @@ namespace paimon {
 
 /// Cross-file shared context for shared-shredding MAP columns.
 ///
-/// Lifetime: same as the owning writer (e.g. AppendOnlyWriter).
+/// Lifetime: same as one rolling writer / write-plan factory.
 /// Holds per-column K_max and a sliding window of recent max_row_width
 /// values to support adaptive K sizing across files.
 ///

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

@@ -108,9 +108,7 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
             if (!MapSharedShreddingUtils::HasShreddingMetadata(metadata)) {
                 continue;
             }
-            EXPECT_OK_AND_ASSIGN(auto meta,
-                                 MapSharedShreddingUtils::DeserializeMetadata(
-                                     metadata, MapSharedShreddingDefine::kDefaultDictCompression));
+            EXPECT_OK_AND_ASSIGN(auto meta, MapSharedShreddingUtils::DeserializeMetadata(metadata));
             auto physical_type =
                 arrow::internal::checked_pointer_cast<arrow::StructType>(field->type());
             std::shared_ptr<arrow::Field> item_field;
@@ -219,12 +217,9 @@ class MapSharedShreddingFileReaderTest : public ::testing::Test {
         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_);
+                                                  compact_manager, pool_);
     }
 
  private:

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

@@ -75,8 +75,7 @@ Result<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::AttachMeta
 }
 
 Result<MapSharedShreddingFieldMeta> MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
-    std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name,
-    const std::string& compression) {
+    std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name) {
     if (!physical_schema) {
         return Status::Invalid("physical schema is null");
     }
@@ -90,7 +89,7 @@ Result<MapSharedShreddingFieldMeta> MapSharedShreddingSchemaUtils::ExtractMetada
 
     auto metadata =
         field->metadata() ? field->metadata()->Copy() : std::shared_ptr<arrow::KeyValueMetadata>();
-    return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression);
+    return MapSharedShreddingUtils::DeserializeMetadata(metadata);
 }
 
 }  // namespace paimon

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

@@ -62,7 +62,7 @@ TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaBasic) {
     auto tags_metadata = updated_schema->GetFieldByName("tags")->metadata()->Copy();
     ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(tags_metadata));
     ASSERT_OK_AND_ASSIGN(auto deserialized,
-                         MapSharedShreddingUtils::DeserializeMetadata(tags_metadata, "none"));
+                         MapSharedShreddingUtils::DeserializeMetadata(tags_metadata));
     ASSERT_EQ(deserialized, tags_meta);
 }
 
@@ -159,7 +159,7 @@ TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaOverwritesExisting
     }
     ASSERT_EQ(storage_layout_key_count, 1);
     ASSERT_OK_AND_ASSIGN(auto deserialized,
-                         MapSharedShreddingUtils::DeserializeMetadata(updated_metadata, "none"));
+                         MapSharedShreddingUtils::DeserializeMetadata(updated_metadata));
     ASSERT_EQ(deserialized, tags_meta);
 }
 
@@ -183,7 +183,7 @@ TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromField) {
     auto c_schema = std::make_unique<::ArrowSchema>();
     ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
     ASSERT_OK_AND_ASSIGN(auto parsed_meta, MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
-                                               std::move(c_schema), "tags", "none"));
+                                               std::move(c_schema), "tags"));
     ASSERT_EQ(parsed_meta, tags_meta);
 }
 
@@ -194,24 +194,24 @@ TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldNoShreddingMetad
     auto c_schema = std::make_unique<::ArrowSchema>();
     ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
 
-    ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema),
-                                                                                "tags", "none"),
-                        "metadata is null or storage layout is not shared-shredding");
+    ASSERT_NOK_WITH_MSG(
+        MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema), "tags"),
+        "metadata is null or storage layout is not shared-shredding");
 }
 
 TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldInvalidInput) {
     ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
-                            std::unique_ptr<::ArrowSchema>(), "tags", "none"),
+                            std::unique_ptr<::ArrowSchema>(), "tags"),
                         "physical schema is null");
 
     auto schema = arrow::schema({arrow::field("id", arrow::int32())});
 
     auto c_schema = std::make_unique<::ArrowSchema>();
     ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
 
-    ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema),
-                                                                                "tags", "none"),
-                        "Shared-shredding field 'tags' not found in physical schema.");
+    ASSERT_NOK_WITH_MSG(
+        MapSharedShreddingSchemaUtils::ExtractMetadataFromField(std::move(c_schema), "tags"),
+        "Shared-shredding field 'tags' not found in physical schema.");
 }
 
 TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaInvalidInput) {