[core] Support column placement policies for MAP shared-shredding

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

@@ -100,6 +100,9 @@ public class CoreOptions implements Serializable {
     public static final String MAP_SHARED_SHREDDING_MAX_COLUMNS =
             "map.shared-shredding.max-columns";
 
+    public static final String MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY =
+            "map.shared-shredding.column-placement-policy";
+
     public static final String FILE_INDEX = "file-index";
 
     public static final String COLUMNS = "columns";
@@ -5256,6 +5259,18 @@ public int mapSharedShreddingMaxColumns(String fieldName) {
         return maxColumns;
     }
 
+    public MapSharedShreddingColumnPlacementPolicy mapSharedShreddingColumnPlacementPolicy(
+            String fieldName) {
+        return options.get(
+                key(FIELDS_PREFIX
+                                + "."
+                                + fieldName
+                                + "."
+                                + MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY)
+                        .enumType(MapSharedShreddingColumnPlacementPolicy.class)
+                        .defaultValue(MapSharedShreddingColumnPlacementPolicy.LRU));
+    }
+
     /** MAP storage layout. */
     public enum MapStorageLayout implements DescribedEnum {
         DEFAULT(
@@ -5288,6 +5303,40 @@ public InlineElement getDescription() {
         }
     }
 
+    /** Physical column placement policy for shared-shredding MAP fields. */
+    public enum MapSharedShreddingColumnPlacementPolicy implements DescribedEnum {
+        PLAIN(
+                "plain",
+                "Keep each MAP row's input key order and place the first K keys into physical "
+                        + "columns."),
+        SEQUENTIAL(
+                "sequential",
+                "Order keys by their field dictionary IDs and place the first K keys into physical "
+                        + "columns."),
+        LRU(
+                "lru",
+                "Reuse physical columns for recently seen keys and evict the least recently used "
+                        + "column when necessary.");
+
+        private final String value;
+        private final String description;
+
+        MapSharedShreddingColumnPlacementPolicy(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/LruMapSharedShreddingColumnAllocator.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Allocator that retains resident field-to-column assignments and evicts the least recently used
+ * column when necessary.
+ */
+public class LruMapSharedShreddingColumnAllocator extends MapSharedShreddingColumnAllocator {
+
+    private final int[] residentFieldByColumn;
+    private final long[] lastUsed;
+    private long lruClock;
+
+    public LruMapSharedShreddingColumnAllocator(int numColumns) {
+        super(numColumns);
+        this.residentFieldByColumn = emptyColumnMapping();
+        this.lastUsed = new long[numColumns];
+    }
+
+    @Override
+    public RowAllocation allocateRow(List<Integer> fieldIds) {
+        List<Integer> sortedFieldIds = new ArrayList<>(fieldIds);
+        Collections.sort(sortedFieldIds);
+
+        int[] colToField = emptyColumnMapping();
+        int[] nextResidentFieldByColumn = residentFieldByColumn.clone();
+        boolean[] usedColumns = new boolean[numColumns];
+        List<Integer> unassignedFields = new ArrayList<>();
+
+        for (Integer fieldId : sortedFieldIds) {
+            int column = findResidentColumn(fieldId);
+            if (column == -1) {
+                unassignedFields.add(fieldId);
+            } else {
+                usedColumns[column] = true;
+                colToField[column] = fieldId;
+            }
+        }
+
+        List<Integer> overflowFields = new ArrayList<>();
+        for (Integer fieldId : unassignedFields) {
+            int column = selectColumn(usedColumns, nextResidentFieldByColumn);
+            if (column == -1) {
+                overflowFields.add(fieldId);
+                continue;
+            }
+
+            usedColumns[column] = true;
+            colToField[column] = fieldId;
+            nextResidentFieldByColumn[column] = fieldId;
+        }
+
+        RowAllocation allocation = new RowAllocation(colToField, overflowFields);
+        updateLastUsed(colToField);
+        System.arraycopy(nextResidentFieldByColumn, 0, residentFieldByColumn, 0, numColumns);
+        commitRow(allocation, sortedFieldIds);
+        return allocation;
+    }
+
+    private int findResidentColumn(int fieldId) {
+        for (int column = 0; column < numColumns; column++) {
+            if (residentFieldByColumn[column] == fieldId) {
+                return column;
+            }
+        }
+        return -1;
+    }
+
+    private int selectColumn(boolean[] usedColumns, int[] plannedResidentFieldByColumn) {
+        int selectedColumn = -1;
+        long selectedLastUsed = Long.MAX_VALUE;
+        for (int column = 0; column < numColumns; column++) {
+            if (usedColumns[column]) {
+                continue;
+            }
+            if (plannedResidentFieldByColumn[column] == -1) {
+                return column;
+            }
+            if (lastUsed[column] < selectedLastUsed) {
+                selectedColumn = column;
+                selectedLastUsed = lastUsed[column];
+            }
+        }
+        return selectedColumn;
+    }
+
+    private void updateLastUsed(int[] colToField) {
+        boolean touched = false;
+        for (int column = 0; column < numColumns; column++) {
+            if (colToField[column] != -1) {
+                lastUsed[column] = lruClock;
+                touched = true;
+            }
+        }
+        if (touched) {
+            lruClock++;
+        }
+    }
+}

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

@@ -29,44 +29,43 @@
 /**
  * Per-row physical column allocator for one shared-shredding MAP column.
  *
- * <p>This is a simple temporary implementation which assigns fields to physical columns by row
- * order. A later version will use a more sophisticated LRU-style allocator to improve column reuse
- * across rows.
+ * <p>Implementations decide the physical column placement for each row. This base class accumulates
+ * the file-level metadata shared by all placement policies.
  */
-public class MapSharedShreddingColumnAllocator {
+public abstract class MapSharedShreddingColumnAllocator {
 
-    private final int numColumns;
+    protected final int numColumns;
     private final Map<Integer, Set<Integer>> fieldToColumns = new TreeMap<>();
     private final Set<Integer> overflowFieldSet = new TreeSet<>();
     private int maxRowWidth = 0;
 
-    public MapSharedShreddingColumnAllocator(int numColumns) {
+    protected MapSharedShreddingColumnAllocator(int numColumns) {
         this.numColumns = numColumns;
     }
 
-    public RowAllocation allocateRow(List<Integer> fieldIds) {
+    /** Allocates physical columns for one row's field IDs. */
+    public abstract RowAllocation allocateRow(List<Integer> fieldIds);
+
+    /** Commits one row allocation and updates accumulated file-level metadata. */
+    protected void commitRow(RowAllocation allocation, List<Integer> fieldIds) {
         maxRowWidth = Math.max(maxRowWidth, fieldIds.size());
 
-        int[] colToField = new int[numColumns];
-        for (int i = 0; i < numColumns; i++) {
-            colToField[i] = -1;
+        for (int column = 0; column < numColumns; column++) {
+            int fieldId = allocation.colToField[column];
+            if (fieldId != -1) {
+                fieldToColumns.computeIfAbsent(fieldId, ignored -> new TreeSet<>()).add(column);
+            }
         }
 
-        int assignLimit = Math.min(fieldIds.size(), numColumns);
-        for (int i = 0; i < assignLimit; i++) {
-            int fieldId = fieldIds.get(i);
-            colToField[i] = fieldId;
-            fieldToColumns.computeIfAbsent(fieldId, ignored -> new TreeSet<>()).add(i);
-        }
+        overflowFieldSet.addAll(allocation.overflowFields);
+    }
 
-        List<Integer> overflowFields = new ArrayList<>();
-        for (int i = assignLimit; i < fieldIds.size(); i++) {
-            int fieldId = fieldIds.get(i);
-            overflowFields.add(fieldId);
-            overflowFieldSet.add(fieldId);
+    protected int[] emptyColumnMapping() {
+        int[] colToField = new int[numColumns];
+        for (int i = 0; i < numColumns; i++) {
+            colToField[i] = -1;
         }
-
-        return new RowAllocation(colToField, overflowFields);
+        return colToField;
     }
 
     public Map<Integer, List<Integer>> fieldToColumns() {
@@ -97,9 +96,9 @@ public static class RowAllocation {
         private final int[] colToField;
         private final List<Integer> overflowFields;
 
-        private RowAllocation(int[] colToField, List<Integer> overflowFields) {
-            this.colToField = colToField;
-            this.overflowFields = Collections.unmodifiableList(overflowFields);
+        RowAllocation(int[] colToField, List<Integer> overflowFields) {
+            this.colToField = colToField.clone();
+            this.overflowFields = Collections.unmodifiableList(new ArrayList<>(overflowFields));
         }
 
         public int[] colToField() {

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

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.data.shredding;
 
+import org.apache.paimon.CoreOptions.MapSharedShreddingColumnPlacementPolicy;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Blob;
 import org.apache.paimon.data.Decimal;
@@ -42,6 +43,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Converts logical rows containing shared-shredding MAP fields into physical rows. */
 public class MapSharedShreddingRowConverter {
 
@@ -52,7 +55,9 @@ public class MapSharedShreddingRowConverter {
     private final List<String> shreddingFieldNames;
 
     public MapSharedShreddingRowConverter(
-            RowType logicalType, Map<String, Integer> fieldToNumColumns) {
+            RowType logicalType,
+            Map<String, Integer> fieldToNumColumns,
+            Map<String, MapSharedShreddingColumnPlacementPolicy> fieldToColumnPlacementPolicy) {
         this.logicalType = logicalType;
         this.physicalType =
                 MapSharedShreddingUtils.logicalToPhysicalSchema(logicalType, fieldToNumColumns);
@@ -68,7 +73,14 @@ public MapSharedShreddingRowConverter(
             }
 
             MapType mapType = (MapType) field.type();
-            ColumnContext context = new ColumnContext(field.name(), numColumns, mapType);
+            MapSharedShreddingColumnPlacementPolicy placementPolicy =
+                    fieldToColumnPlacementPolicy.get(field.name());
+            checkArgument(
+                    placementPolicy != null,
+                    "Missing column placement policy for shared-shredding field '%s'.",
+                    field.name());
+            ColumnContext context =
+                    new ColumnContext(field.name(), numColumns, mapType, placementPolicy);
             contextByFieldName.put(field.name(), context);
             contextByFieldPos[i] = context;
             shreddingFieldNames.add(field.name());
@@ -289,14 +301,33 @@ private static class ColumnContext {
         private final MapSharedShreddingFieldDict dict;
         private final MapSharedShreddingColumnAllocator allocator;
 
-        private ColumnContext(String fieldName, int numColumns, MapType mapType) {
+        private ColumnContext(
+                String fieldName,
+                int numColumns,
+                MapType mapType,
+                MapSharedShreddingColumnPlacementPolicy placementPolicy) {
             this.fieldName = fieldName;
             this.numColumns = numColumns;
             this.keyGetter = InternalArray.createElementGetter(mapType.getKeyType());
             DataType valueType = mapType.getValueType();
             this.valueGetter = InternalArray.createElementGetter(valueType);
             this.dict = new MapSharedShreddingFieldDict();
-            this.allocator = new MapSharedShreddingColumnAllocator(numColumns);
+            this.allocator = createAllocator(numColumns, placementPolicy);
+        }
+
+        private static MapSharedShreddingColumnAllocator createAllocator(
+                int numColumns, MapSharedShreddingColumnPlacementPolicy placementPolicy) {
+            switch (placementPolicy) {
+                case PLAIN:
+                    return new PlainMapSharedShreddingColumnAllocator(numColumns);
+                case SEQUENTIAL:
+                    return new SequentialMapSharedShreddingColumnAllocator(numColumns);
+                case LRU:
+                    return new LruMapSharedShreddingColumnAllocator(numColumns);
+                default:
+                    throw new IllegalArgumentException(
+                            "Unknown shared-shredding column placement policy: " + placementPolicy);
+            }
         }
     }
 }

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

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.data.shredding;
 
+import org.apache.paimon.CoreOptions.MapSharedShreddingColumnPlacementPolicy;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.types.RowType;
 
@@ -35,9 +36,13 @@ public class MapSharedShreddingWritePlan implements ShreddingWritePlan {
     @Nullable private Map<String, Map<String, String>> fieldMetadata;
 
     public MapSharedShreddingWritePlan(
-            RowType logicalRowType, Map<String, Integer> fieldToNumColumns) {
+            RowType logicalRowType,
+            Map<String, Integer> fieldToNumColumns,
+            Map<String, MapSharedShreddingColumnPlacementPolicy> fieldToColumnPlacementPolicy) {
         this.logicalRowType = logicalRowType;
-        this.converter = new MapSharedShreddingRowConverter(logicalRowType, fieldToNumColumns);
+        this.converter =
+                new MapSharedShreddingRowConverter(
+                        logicalRowType, fieldToNumColumns, fieldToColumnPlacementPolicy);
     }
 
     @Override