[core] Support adaptive MAP and Variant shredding across rolling files

docs/generated/core_configuration.html

@@ -1711,6 +1711,24 @@
             <td>Boolean</td>
             <td>Whether to automatically infer the shredding schema when writing Variant columns.</td>
         </tr>
+        <tr>
+            <td><h5>variant.shredding.adaptive.maxInferBufferRow</h5></td>
+            <td style="word-wrap: break-word;">256</td>
+            <td>Integer</td>
+            <td>Maximum number of prefix rows sampled after the first file in an adaptive Variant shredding inference session.</td>
+        </tr>
+        <tr>
+            <td><h5>variant.shredding.adaptive.retentionRatio</h5></td>
+            <td style="word-wrap: break-word;">0.05</td>
+            <td>Double</td>
+            <td>Minimum combined presence ratio for retaining a Variant path selected in the previous file. This must not exceed 'variant.shredding.minFieldCardinalityRatio'.</td>
+        </tr>
+        <tr>
+            <td><h5>variant.shredding.inferenceMode</h5></td>
+            <td style="word-wrap: break-word;">per-file</td>
+            <td><p>Enum</p></td>
+            <td>The Variant shredding inference mode. PER_FILE infers each file independently. ADAPTIVE reuses bounded evidence within one rolling writer and samples a smaller prefix after the first file.<br /><br />Possible values:<ul><li>"per-file": Infer every file independently from its own prefix rows.</li><li>"adaptive": Reuse bounded inference evidence within one rolling writer and correct it with a smaller prefix from each subsequent file.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>variant.shredding.maxInferBufferRow</h5></td>
             <td style="word-wrap: break-word;">4096</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -441,6 +441,17 @@ public InlineElement getDescription() {
                     .withDescription(
                             "Whether to automatically infer the shredding schema when writing Variant columns.");
 
+    public static final ConfigOption<VariantShreddingInferenceMode>
+            VARIANT_SHREDDING_INFERENCE_MODE =
+                    key("variant.shredding.inferenceMode")
+                            .enumType(VariantShreddingInferenceMode.class)
+                            .defaultValue(VariantShreddingInferenceMode.PER_FILE)
+                            .withDescription(
+                                    "The Variant shredding inference mode. PER_FILE infers each "
+                                            + "file independently. ADAPTIVE reuses bounded evidence "
+                                            + "within one rolling writer and samples a smaller "
+                                            + "prefix after the first file.");
+
     public static final ConfigOption<Integer> VARIANT_SHREDDING_MAX_SCHEMA_WIDTH =
             key("variant.shredding.maxSchemaWidth")
                     .intType()
@@ -469,6 +480,23 @@ public InlineElement getDescription() {
                     .defaultValue(4096)
                     .withDescription("Maximum number of rows to buffer for schema inference.");
 
+    public static final ConfigOption<Integer> VARIANT_SHREDDING_ADAPTIVE_MAX_INFER_BUFFER_ROW =
+            key("variant.shredding.adaptive.maxInferBufferRow")
+                    .intType()
+                    .defaultValue(256)
+                    .withDescription(
+                            "Maximum number of prefix rows sampled after the first file in "
+                                    + "an adaptive Variant shredding inference session.");
+
+    public static final ConfigOption<Double> VARIANT_SHREDDING_ADAPTIVE_RETENTION_RATIO =
+            key("variant.shredding.adaptive.retentionRatio")
+                    .doubleType()
+                    .defaultValue(0.05)
+                    .withDescription(
+                            "Minimum combined presence ratio for retaining a Variant path selected "
+                                    + "in the previous file. This must not exceed "
+                                    + "'variant.shredding.minFieldCardinalityRatio'.");
+
     public static final ConfigOption<String> MANIFEST_FORMAT =
             key("manifest.format")
                     .stringType()
@@ -5417,6 +5445,33 @@ public InlineElement getDescription() {
         }
     }
 
+    /** Inference mode for Variant shredding schemas. */
+    public enum VariantShreddingInferenceMode implements DescribedEnum {
+        PER_FILE("per-file", "Infer every file independently from its own prefix rows."),
+        ADAPTIVE(
+                "adaptive",
+                "Reuse bounded inference evidence within one rolling writer and correct it with "
+                        + "a smaller prefix from each subsequent file.");
+
+        private final String value;
+        private final String description;
+
+        VariantShreddingInferenceMode(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
+
     /**
      * Action to take when an UPDATE (e.g. via MERGE INTO) modifies columns that are covered by a
      * global index.

paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java

@@ -78,4 +78,12 @@ private Map<String, Map<String, String>> buildFieldMetadata(String compression)
         }
         return Collections.unmodifiableMap(metadata);
     }
+
+    Map<String, Integer> fileMaxRowWidths() {
+        Map<String, Integer> result = new LinkedHashMap<>();
+        for (String fieldName : converter.shreddingFieldNames()) {
+            result.put(fieldName, converter.buildFieldMeta(fieldName).maxRowWidth());
+        }
+        return result;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java

@@ -20,7 +20,6 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.CoreOptions.MapSharedShreddingColumnPlacementPolicy;
-import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.shredding.ShreddingWritePlanFactory;
 import org.apache.paimon.options.Options;
@@ -35,12 +34,10 @@
 /** Creates per-file shared-shredding MAP write plans. */
 public class MapSharedShreddingWritePlanFactory implements ShreddingWritePlanFactory {
 
-    private static final int INFER_BUFFER_ROW_COUNT = 1;
-
     private final RowType logicalRowType;
     private final Map<String, Integer> fieldToMaxColumns;
     private final Map<String, MapSharedShreddingColumnPlacementPolicy> fieldToColumnPlacementPolicy;
-    private final Map<String, Integer> fieldToPosition;
+    private final MapSharedShreddingContext context;
 
     public MapSharedShreddingWritePlanFactory(RowType logicalRowType, Options options) {
         this.logicalRowType = logicalRowType;
@@ -50,12 +47,11 @@ public MapSharedShreddingWritePlanFactory(RowType logicalRowType, Options option
         this.fieldToMaxColumns =
                 MapSharedShreddingUtils.buildColumnToNumColumns(shreddingFields, coreOptions);
         this.fieldToColumnPlacementPolicy = new LinkedHashMap<>();
-        this.fieldToPosition = new LinkedHashMap<>();
         for (String field : shreddingFields) {
             fieldToColumnPlacementPolicy.put(
                     field, coreOptions.mapSharedShreddingColumnPlacementPolicy(field));
-            fieldToPosition.put(field, logicalRowType.getFieldIndex(field));
         }
+        this.context = new MapSharedShreddingContext(fieldToMaxColumns);
     }
 
     @Override
@@ -70,39 +66,29 @@ public boolean shouldCreateWritePlan() {
 
     @Override
     public boolean shouldInferWritePlan() {
-        return shouldCreateWritePlan();
+        return false;
     }
 
     @Override
     public int inferBufferRowCount() {
-        return INFER_BUFFER_ROW_COUNT;
+        return 0;
     }
 
     @Override
     public ShreddingWritePlan createWritePlan(List<InternalRow> sampleRows) {
         checkArgument(shouldCreateWritePlan(), "MAP shared-shredding write plan is not active.");
-
-        Map<String, Integer> fieldToNumColumns = new LinkedHashMap<>();
-        for (Map.Entry<String, Integer> entry : fieldToMaxColumns.entrySet()) {
-            int maxColumns = entry.getValue();
-            int numColumns = maxColumns;
-            if (!sampleRows.isEmpty()) {
-                int fieldPosition = fieldToPosition.get(entry.getKey());
-                int maxRowWidth = 0;
-                int sampleCount = Math.min(sampleRows.size(), INFER_BUFFER_ROW_COUNT);
-                for (int i = 0; i < sampleCount; i++) {
-                    InternalRow row = sampleRows.get(i);
-                    InternalMap map =
-                            row.isNullAt(fieldPosition) ? null : row.getMap(fieldPosition);
-                    maxRowWidth = Math.max(maxRowWidth, map == null ? 0 : map.size());
-                }
-                numColumns = Math.max(1, Math.min(maxRowWidth, maxColumns));
-            }
-            fieldToNumColumns.put(entry.getKey(), numColumns);
-        }
-
-        // TODO: Infer the column count from recent file metadata instead of current-file samples.
         return new MapSharedShreddingWritePlan(
-                logicalRowType, fieldToNumColumns, fieldToColumnPlacementPolicy);
+                logicalRowType, context.computeNextK(), fieldToColumnPlacementPolicy);
+    }
+
+    @Override
+    public void onFileCompleted(ShreddingWritePlan writePlan) {
+        checkArgument(
+                writePlan instanceof MapSharedShreddingWritePlan,
+                "Unexpected MAP shared-shredding write plan: %s",
+                writePlan.getClass().getName());
+        ((MapSharedShreddingWritePlan) writePlan)
+                .fileMaxRowWidths()
+                .forEach(context::reportFileStats);
     }
 }