From cfa74c207b7d4c9d556cf2a4224551473c268caf Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Thu, 30 Jul 2026 19:03:35 +0800 Subject: [PATCH] [core][spark] Support switching MAP storage layouts --- .../shredding/MapSharedShreddingReadPlan.java | 118 +++++++-- .../MapSharedShreddingReadPlanFactory.java | 5 +- .../shredding/MapSharedShreddingUtils.java | 21 +- .../MapSharedShreddingReadPlanTest.java | 56 +++- .../MapSharedShreddingUtilsTest.java | 105 ++++++++ .../apache/paimon/schema/SchemaManager.java | 46 ---- .../paimon/schema/SchemaManagerTest.java | 67 +++-- .../table/MapSharedShreddingTableTest.java | 241 +++++++++++++++--- ...lectedKeysSharedShreddingE2ETestBase.scala | 93 +++++++ 9 files changed, 621 insertions(+), 131 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java index f09c58d685d3..6cc8dae4e3b3 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java @@ -45,14 +45,17 @@ import static org.apache.paimon.utils.Preconditions.checkArgument; -/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */ +/** + * Read plan that rebuilds logical MAP values from shared-shredding physical ROW values and projects + * selected keys from legacy default-layout MAP values. + */ public class MapSharedShreddingReadPlan implements ShreddingReadPlan { private static final int FIELD_MAPPING_POSITION = 0; private final RowType logicalType; private final RowType physicalType; - private final Map contextByFieldIndex; + private final Map contextByFieldIndex; public MapSharedShreddingReadPlan( RowType logicalType, Map fieldMetas) { @@ -81,15 +84,21 @@ public ShreddingBatchAssembler batchAssembler() { return new MapSharedShreddingBatchAssembler(); } - private static Map createContexts( + private static Map createContexts( RowType logicalType, RowType physicalType, Map fieldMetas) { - Map contexts = new LinkedHashMap<>(); + Map contexts = new LinkedHashMap<>(); for (int i = 0; i < logicalType.getFieldCount(); i++) { DataField field = logicalType.getFields().get(i); MapSharedShreddingFieldMeta fieldMeta = fieldMetas.get(field.name()); if (fieldMeta == null) { + if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) { + contexts.put( + i, + new NormalMapSelectedKeysContext( + (RowType) field.type(), field.description())); + } continue; } @@ -117,14 +126,13 @@ private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssemble public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) { ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()]; for (int i = 0; i < logicalVectors.length; i++) { - SharedShreddingContext context = contextByFieldIndex.get(i); + MapReadContext context = contextByFieldIndex.get(i); if (context == null) { logicalVectors[i] = physicalBatch.columns[i]; } else { logicalVectors[i] = context.materialize( - (RowColumnVector) physicalBatch.columns[i], - physicalBatch.getNumRows()); + physicalBatch.columns[i], physicalBatch.getNumRows()); } } return physicalBatch.copy(logicalVectors); @@ -208,6 +216,37 @@ private static ColumnVector materializeSelectedKeysRowVector( return ColumnVectorUtils.createReadableColumnVector(selectedKeysType, rowVector); } + private static ColumnVector materializeSelectedKeysRowVector( + MapColumnVector physicalVector, NormalMapSelectedKeysContext context, int rowCount) { + RowType selectedKeysType = context.selectedKeysType; + WritableColumnVector[] selectedKeyVectors = + new WritableColumnVector[selectedKeysType.getFieldCount()]; + for (int i = 0; i < selectedKeyVectors.length; i++) { + selectedKeyVectors[i] = + ColumnVectorUtils.createWritableColumnVector( + rowCount, selectedKeysType.getTypeAt(i)); + } + HeapRowVector rowVector = new HeapRowVector(rowCount, selectedKeyVectors); + + for (int row = 0; row < rowCount; row++) { + if (physicalVector.isNullAt(row)) { + rowVector.appendNull(); + continue; + } + + InternalMap map = physicalVector.getMap(row); + InternalArray keys = map.keyArray(); + InternalArray values = map.valueArray(); + for (int ordinal = 0; ordinal < selectedKeyVectors.length; ordinal++) { + appendSelectedKeyValue( + keys, values, map.size(), context, ordinal, selectedKeyVectors[ordinal]); + } + rowVector.appendRow(); + } + + return ColumnVectorUtils.createReadableColumnVector(selectedKeysType, rowVector); + } + private static ColumnVector[] physicalChildren(RowColumnVector physicalVector) { ColumnVector[] physicalChildren = physicalVector.getChildren(); if (physicalChildren != null) { @@ -329,6 +368,23 @@ private static void appendSelectedKeyValue( valueVector.appendNull(); } + private static void appendSelectedKeyValue( + InternalArray keys, + InternalArray values, + int size, + NormalMapSelectedKeysContext context, + int ordinal, + WritableColumnVector valueVector) { + BinaryString selectedKey = context.selectedKeys[ordinal]; + for (int i = 0; i < size; i++) { + if (!keys.isNullAt(i) && selectedKey.equals(keys.getString(i))) { + appendValue(context.selectedValueConverters[ordinal], values, i, valueVector); + return; + } + } + valueVector.appendNull(); + } + private static void appendValue( RowToColumnConverter.ElementConverter converter, ColumnVector source, @@ -401,9 +457,9 @@ private static InternalMap overflowMap( return overflowVector.isNullAt(row) ? null : ((MapColumnVector) overflowVector).getMap(row); } - private interface SharedShreddingContext { + private interface MapReadContext { - ColumnVector materialize(RowColumnVector physicalVector, int rowCount); + ColumnVector materialize(ColumnVector physicalVector, int rowCount); } private static class SharedPhysicalContext { @@ -475,7 +531,7 @@ private static int physicalColumnIndex(String fieldName) { } } - private static class FullMapContext implements SharedShreddingContext { + private static class FullMapContext implements MapReadContext { private final SharedPhysicalContext physical; private final MapType mapType; @@ -491,12 +547,12 @@ private FullMapContext(SharedPhysicalContext physical, MapType mapType) { } @Override - public ColumnVector materialize(RowColumnVector physicalVector, int rowCount) { - return materializeLogicalMapVector(physicalVector, this, rowCount); + public ColumnVector materialize(ColumnVector physicalVector, int rowCount) { + return materializeLogicalMapVector((RowColumnVector) physicalVector, this, rowCount); } } - private static class SelectedKeysContext implements SharedShreddingContext { + private static class SelectedKeysContext implements MapReadContext { private final SharedPhysicalContext physical; private final RowType selectedKeysType; @@ -537,8 +593,9 @@ private SelectedKeysContext( } @Override - public ColumnVector materialize(RowColumnVector physicalVector, int rowCount) { - return materializeSelectedKeysRowVector(physicalVector, this, rowCount); + public ColumnVector materialize(ColumnVector physicalVector, int rowCount) { + return materializeSelectedKeysRowVector( + (RowColumnVector) physicalVector, this, rowCount); } private static int[] candidateColumns(MapSharedShreddingFieldMeta fieldMeta, int fieldId) { @@ -553,4 +610,35 @@ private static int[] candidateColumns(MapSharedShreddingFieldMeta fieldMeta, int return result; } } + + private static class NormalMapSelectedKeysContext implements MapReadContext { + + private final RowType selectedKeysType; + private final BinaryString[] selectedKeys; + private final RowToColumnConverter.ElementConverter[] selectedValueConverters; + + private NormalMapSelectedKeysContext( + RowType selectedKeysType, String selectedKeysMetadata) { + this.selectedKeysType = selectedKeysType; + List keys = MapSelectedKeysMetadataUtils.selectedKeys(selectedKeysMetadata); + checkArgument( + keys.size() == selectedKeysType.getFieldCount(), + "Selected-key metadata size %s does not match selected ROW field count %s.", + keys.size(), + selectedKeysType.getFieldCount()); + this.selectedKeys = new BinaryString[keys.size()]; + this.selectedValueConverters = new RowToColumnConverter.ElementConverter[keys.size()]; + for (int i = 0; i < keys.size(); i++) { + this.selectedKeys[i] = BinaryString.fromString(keys.get(i)); + this.selectedValueConverters[i] = + RowToColumnConverter.createElementConverter(selectedKeysType.getTypeAt(i)); + } + } + + @Override + public ColumnVector materialize(ColumnVector physicalVector, int rowCount) { + return materializeSelectedKeysRowVector( + (MapColumnVector) physicalVector, this, rowCount); + } + } } diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java index 36ee35b8f725..aeedf4057ead 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java @@ -27,7 +27,7 @@ import java.util.LinkedHashMap; import java.util.Map; -/** Creates per-file shared-shredding MAP read plans. */ +/** Creates per-file MAP read plans for shared-shredding and selected-key fallback reads. */ public class MapSharedShreddingReadPlanFactory implements ShreddingReadPlanFactory { private final RowType logicalRowType; @@ -46,7 +46,8 @@ public boolean shouldCreateReadPlan( Map> fieldMetadata, @Nullable Object fileSchema) { for (DataField field : logicalRowType.getFields()) { Map metadata = fieldMetadata.get(field.name()); - if (MapSharedShreddingUtils.hasShreddingMetadata(metadata)) { + if (MapSharedShreddingUtils.hasShreddingMetadata(metadata) + || MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) { return true; } } diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java index 6fc2d13b5a72..caf4bbc88732 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java @@ -28,6 +28,7 @@ import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.IntType; import org.apache.paimon.types.MapType; import org.apache.paimon.types.RowType; @@ -108,24 +109,30 @@ public static RowType logicalToPhysicalSchema( public static RowType buildPhysicalReadType( RowType logicalReadType, Map sharedShreddingFieldMetas) { - if (sharedShreddingFieldMetas.isEmpty()) { - return logicalReadType; - } - List physicalReadFields = new ArrayList<>(); boolean converted = false; for (DataField logicalReadField : logicalReadType.getFields()) { MapSharedShreddingFieldMeta fieldMeta = sharedShreddingFieldMetas.get(logicalReadField.name()); if (fieldMeta == null) { - physicalReadFields.add(logicalReadField); + if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(logicalReadField)) { + // Read a legacy/default-layout file while selected-key pushdown is active. + DataType valueType = selectedKeysValueType((RowType) logicalReadField.type()); + DataType physicalType = + DataTypes.MAP(DataTypes.STRING().notNull(), valueType) + .copy(logicalReadField.type().isNullable()); + physicalReadFields.add(logicalReadField.newType(physicalType)); + converted = true; + } else { + physicalReadFields.add(logicalReadField); + } continue; } DataType valueType; DataType physicalType; if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(logicalReadField)) { - // recall partial key with shared shredding pushdown + // Read only selected keys from a shared-shredding ROW. valueType = selectedKeysValueType((RowType) logicalReadField.type()); physicalType = buildSpecificPhysicalStructType( @@ -134,7 +141,7 @@ public static RowType buildPhysicalReadType( selectedKeysIncludeOverflow(logicalReadField, fieldMeta)) .copy(logicalReadField.type().isNullable()); } else { - // recall whole field without shared shredding pushdown + // Rebuild the whole MAP from a shared-shredding ROW. valueType = ((MapType) logicalReadField.type()).getValueType(); physicalType = buildPhysicalStructType(valueType, fieldMeta.numColumns()) diff --git a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java index 6b781be4177e..b376a11d9581 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java @@ -28,11 +28,13 @@ import org.apache.paimon.data.columnar.RowColumnVector; import org.apache.paimon.data.columnar.VectorizedColumnBatch; import org.apache.paimon.data.columnar.heap.HeapArrayVector; +import org.apache.paimon.data.columnar.heap.HeapBytesVector; import org.apache.paimon.data.columnar.heap.HeapIntVector; import org.apache.paimon.data.columnar.heap.HeapLongVector; import org.apache.paimon.data.columnar.heap.HeapMapVector; import org.apache.paimon.data.columnar.heap.HeapRowVector; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.MapType; import org.apache.paimon.types.RowType; import org.junit.jupiter.api.Test; @@ -185,6 +187,48 @@ void testReadSelectedKeysFromPrunedPhysicalColumns() { assertThat(selectedKeys.isNullAt(2)).isTrue(); } + @Test + void testReadSelectedKeysFromNormalMap() { + MapSharedShreddingReadPlan readPlan = + new MapSharedShreddingReadPlan(selectedKeysLogicalType(), Collections.emptyMap()); + assertThat(readPlan.physicalRowType().getTypeAt(0)).isInstanceOf(MapType.class); + + HeapBytesVector keys = new HeapBytesVector(3); + appendString(keys, "key2"); + appendString(keys, "key1"); + appendString(keys, "key1"); + HeapLongVector values = new HeapLongVector(3); + values.appendLong(20L); + values.appendLong(10L); + values.appendNull(); + HeapMapVector normalMaps = new HeapMapVector(3, keys, values); + normalMaps.putOffsetLength(0, 0, 2); + normalMaps.putOffsetLength(1, 2, 1); + normalMaps.setNullAt(2); + normalMaps.putOffsetLength(2, 3, 0); + + RowColumnVector selectedKeysVector = assembleSelectedKeysVector(readPlan, normalMaps, 3); + InternalRow first = selectedKeysVector.getRow(0); + assertThat(first.getLong(0)).isEqualTo(10L); + assertThat(first.getLong(1)).isEqualTo(20L); + assertThat(first.isNullAt(2)).isTrue(); + + InternalRow second = selectedKeysVector.getRow(1); + assertThat(second.isNullAt(0)).isTrue(); + assertThat(second.isNullAt(1)).isTrue(); + assertThat(second.isNullAt(2)).isTrue(); + assertThat(selectedKeysVector.isNullAt(2)).isTrue(); + } + + @Test + void testFactoryCreatesSelectedKeysPlanForNormalMap() { + MapSharedShreddingReadPlanFactory factory = + new MapSharedShreddingReadPlanFactory(selectedKeysLogicalType()); + + assertThat(factory.shouldCreateReadPlan(Collections.emptyMap(), null)).isTrue(); + assertThat(factory.createReadPlan(Collections.emptyMap(), null).isIdentity()).isFalse(); + } + private static InternalMap readMap( MapSharedShreddingFieldMeta fieldMeta, HeapRowVector physicalMap) { return assembleMapVector(fieldMeta, physicalMap).getMap(0); @@ -205,11 +249,14 @@ private static MapColumnVector assembleMapVector( private static RowColumnVector assembleSelectedKeysVector( MapSharedShreddingReadPlan readPlan, HeapRowVector physicalMap) { - assertThat(readPlan.physicalRowType().getTypeAt(0)).isInstanceOf(RowType.class); + return assembleSelectedKeysVector(readPlan, physicalMap, 1); + } + private static RowColumnVector assembleSelectedKeysVector( + MapSharedShreddingReadPlan readPlan, ColumnVector physicalMap, int rowCount) { VectorizedColumnBatch physicalBatch = new VectorizedColumnBatch(new ColumnVector[] {physicalMap}); - physicalBatch.setNumRows(1); + physicalBatch.setNumRows(rowCount); VectorizedColumnBatch logicalBatch = readPlan.batchAssembler().assemble(physicalBatch); return (RowColumnVector) logicalBatch.columns[0]; } @@ -282,6 +329,11 @@ private static HeapLongVector longVector(Long value) { return vector; } + private static void appendString(HeapBytesVector vector, String value) { + byte[] bytes = BinaryString.fromString(value).toBytes(); + vector.appendByteArray(bytes, 0, bytes.length); + } + private static HeapMapVector overflowMap(int key, Long value) { HeapIntVector keys = new HeapIntVector(1); keys.appendInt(key); diff --git a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java index dd628bc5a37f..12acb8b0996b 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -187,6 +188,110 @@ void testLogicalToPhysicalSchemaNoShreddingColumns() { .isEqualTo(logical); } + @Test + void testBuildPhysicalReadTypeWithoutSharedShreddingMetadata() { + RowType fullMapReadType = + DataTypes.ROW( + DataTypes.FIELD(0, "id", DataTypes.INT()), + DataTypes.FIELD( + 1, + "metrics", + DataTypes.MAP( + DataTypes.STRING().notNull(), + DataTypes.BIGINT().notNull()))); + assertThat( + MapSharedShreddingUtils.buildPhysicalReadType( + fullMapReadType, Collections.emptyMap())) + .isSameAs(fullMapReadType); + + DataField selectedKeysField = + MapSelectedKeysMetadataUtils.withSelectedKeys( + fullMapReadType.getField("metrics"), + DataTypes.ROW( + DataTypes.FIELD(0, "0", DataTypes.BIGINT().notNull()), + DataTypes.FIELD(1, "1", DataTypes.BIGINT().notNull()), + DataTypes.FIELD(2, "2", DataTypes.BIGINT().notNull())), + Arrays.asList("key1", "key2", "missing")); + RowType selectedKeysReadType = + new RowType( + false, Arrays.asList(fullMapReadType.getField("id"), selectedKeysField)); + + RowType physicalReadType = + MapSharedShreddingUtils.buildPhysicalReadType( + selectedKeysReadType, Collections.emptyMap()); + assertThat(physicalReadType.isNullable()).isFalse(); + assertThat(physicalReadType.getField("id")).isSameAs(fullMapReadType.getField("id")); + DataField physicalMapField = physicalReadType.getField("metrics"); + assertThat(physicalMapField.id()).isEqualTo(selectedKeysField.id()); + assertThat(physicalMapField.description()).isEqualTo(selectedKeysField.description()); + assertThat(physicalMapField.type()) + .isEqualTo( + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT().notNull())); + } + + @Test + void testBuildPhysicalReadTypeWithSharedShreddingMetadata() { + Map nameToId = new TreeMap<>(); + nameToId.put("key1", 0); + nameToId.put("key2", 1); + Map> fieldToColumns = new TreeMap<>(); + fieldToColumns.put(0, Collections.singletonList(2)); + MapSharedShreddingFieldMeta fieldMeta = + new MapSharedShreddingFieldMeta( + nameToId, + fieldToColumns, + new TreeSet(Collections.singletonList(1)), + 4, + 2); + Map fieldMetas = new HashMap<>(); + fieldMetas.put("metrics", fieldMeta); + + DataField fullMapField = + DataTypes.FIELD( + 1, + "metrics", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT()).notNull()); + RowType fullMapReadType = + new RowType( + false, + Arrays.asList(DataTypes.FIELD(0, "id", DataTypes.INT()), fullMapField)); + RowType fullPhysicalReadType = + MapSharedShreddingUtils.buildPhysicalReadType(fullMapReadType, fieldMetas); + DataField fullPhysicalMapField = fullPhysicalReadType.getField("metrics"); + assertThat(fullPhysicalMapField.id()).isEqualTo(fullMapField.id()); + assertThat(fullPhysicalMapField.type().isNullable()).isFalse(); + assertThat(((RowType) fullPhysicalMapField.type()).getFieldNames()) + .containsExactly( + "__field_mapping", + "__col_0", + "__col_1", + "__col_2", + "__col_3", + "__overflow"); + + DataField selectedKeysField = + MapSelectedKeysMetadataUtils.withSelectedKeys( + fullMapField, + DataTypes.ROW( + DataTypes.FIELD(0, "0", DataTypes.BIGINT()), + DataTypes.FIELD(1, "1", DataTypes.BIGINT()), + DataTypes.FIELD(2, "2", DataTypes.BIGINT())) + .notNull(), + Arrays.asList("key1", "key2", "missing")); + RowType selectedKeysReadType = + new RowType( + false, Arrays.asList(fullMapReadType.getField("id"), selectedKeysField)); + RowType selectedPhysicalReadType = + MapSharedShreddingUtils.buildPhysicalReadType(selectedKeysReadType, fieldMetas); + DataField selectedPhysicalMapField = selectedPhysicalReadType.getField("metrics"); + assertThat(selectedPhysicalMapField.id()).isEqualTo(selectedKeysField.id()); + assertThat(selectedPhysicalMapField.description()) + .isEqualTo(selectedKeysField.description()); + assertThat(selectedPhysicalMapField.type().isNullable()).isFalse(); + assertThat(((RowType) selectedPhysicalMapField.type()).getFieldNames()) + .containsExactly("__field_mapping", "__col_2", "__overflow"); + } + @Test void testBuildSpecificPhysicalStructType() { RowType physicalType = diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 5eadd040c7d7..4c238781b9b1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -637,56 +637,10 @@ protected void updateLastColumn( newSchema.primaryKeys(), newSchema.options(), newSchema.comment()); - checkMapStorageLayoutUnchanged(oldTableSchema, newTableSchema); SchemaValidation.validateTableSchema(newTableSchema); return newTableSchema; } - private static void checkMapStorageLayoutUnchanged( - TableSchema oldTableSchema, TableSchema newTableSchema) { - Map oldLayouts = - mapStorageLayoutByFieldId(oldTableSchema); - Map newLayouts = - mapStorageLayoutByFieldId(newTableSchema); - Map oldNames = fieldNameById(oldTableSchema); - Map newNames = fieldNameById(newTableSchema); - - for (Map.Entry oldLayout : oldLayouts.entrySet()) { - Integer fieldId = oldLayout.getKey(); - CoreOptions.MapStorageLayout newLayout = newLayouts.get(fieldId); - if (newLayout == null || oldLayout.getValue() == newLayout) { - continue; - } - - throw new UnsupportedOperationException( - String.format( - "Cannot change map storage layout for field id %s ('%s' -> '%s') from '%s' to '%s'.", - fieldId, - oldNames.get(fieldId), - newNames.get(fieldId), - oldLayout.getValue(), - newLayout)); - } - } - - private static Map mapStorageLayoutByFieldId( - TableSchema schema) { - CoreOptions options = new CoreOptions(schema.options()); - Map layouts = new HashMap<>(); - for (DataField field : schema.fields()) { - layouts.put(field.id(), options.mapStorageLayout(field.name())); - } - return layouts; - } - - private static Map fieldNameById(TableSchema schema) { - Map names = new HashMap<>(); - for (DataField field : schema.fields()) { - names.put(field.id(), field.name()); - } - return names; - } - // gets the rootType at the defined depth // ex: ARRAY>> // if we want to update ARRAY -> ARRAY diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java index 2b0f4f18bf66..aed3a9a3099f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java @@ -172,41 +172,52 @@ public void testUpdateOptions() throws Exception { } @Test - public void testCannotChangeMapStorageLayoutForExistingField() throws Exception { + public void testChangeMapStorageLayoutForExistingField() throws Exception { retryArtificialException(() -> manager.createTable(mapStorageLayoutSchema("default"))); - assertThatThrownBy( - () -> - retryArtificialException( - () -> - manager.commitChanges( - SchemaChange.setOption( - "fields.metrics.map.storage-layout", - "shared-shredding")))) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining( - "Cannot change map storage layout for field id 1 ('metrics' -> 'metrics') from 'default' to 'shared-shredding'."); + retryArtificialException( + () -> + manager.commitChanges( + SchemaChange.setOption( + "fields.metrics.map.storage-layout", "shared-shredding"))); + Optional sharedShredding = retryArtificialException(() -> manager.latest()); + assertThat(sharedShredding).isPresent(); + assertThat(sharedShredding.get().options()) + .containsEntry("fields.metrics.map.storage-layout", "shared-shredding") + .containsEntry("fields.metrics.map.shared-shredding.max-columns", "2"); + + retryArtificialException( + () -> + manager.commitChanges( + SchemaChange.setOption( + "fields.metrics.map.storage-layout", "default"))); + Optional defaultLayout = retryArtificialException(() -> manager.latest()); + assertThat(defaultLayout).isPresent(); + assertThat(defaultLayout.get().options()) + .containsEntry("fields.metrics.map.storage-layout", "default") + .containsEntry("fields.metrics.map.shared-shredding.max-columns", "2"); } @Test - public void testCannotChangeMapStorageLayoutByRenameColumn() throws Exception { + public void testChangeMapStorageLayoutByRenameColumn() throws Exception { retryArtificialException(() -> manager.createTable(mapStorageLayoutSchema(null))); - assertThatThrownBy( - () -> - retryArtificialException( - () -> - manager.commitChanges( - Arrays.asList( - SchemaChange.renameColumn( - "metrics", - "renamed_metrics"), - SchemaChange.setOption( - "fields.renamed_metrics.map.storage-layout", - "shared-shredding"))))) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining( - "Cannot change map storage layout for field id 1 ('metrics' -> 'renamed_metrics') from 'default' to 'shared-shredding'."); + retryArtificialException( + () -> + manager.commitChanges( + Arrays.asList( + SchemaChange.renameColumn("metrics", "renamed_metrics"), + SchemaChange.setOption( + "fields.renamed_metrics.map.storage-layout", + "shared-shredding")))); + + Optional latest = retryArtificialException(() -> manager.latest()); + assertThat(latest).isPresent(); + assertThat(latest.get().fields().get(1).id()).isEqualTo(1); + assertThat(latest.get().fields().get(1).name()).isEqualTo("renamed_metrics"); + assertThat(latest.get().options()) + .doesNotContainKey("fields.metrics.map.storage-layout") + .containsEntry("fields.renamed_metrics.map.storage-layout", "shared-shredding"); } @ParameterizedTest diff --git a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java index c300bb4a70d8..2a7adce1c4d7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java @@ -76,7 +76,6 @@ import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Table-level tests for MAP shared-shredding. */ public class MapSharedShreddingTableTest extends TableTestBase { @@ -248,6 +247,45 @@ public void testReadSelectedKeysAfterMapValueTypeEvolution(String format) throws assertThat(actual).containsExactly(Arrays.asList(10L, null, null)); } + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testReadSelectedKeysAfterDefaultToSharedAndValueTypeEvolution(String format) + throws Exception { + catalog.createTable( + identifier(format), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column( + "metrics", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.INT())) + .option("bucket", "-1") + .option("file.format", format) + .option(CoreOptions.WRITE_ONLY.key(), "true") + .build(), + true); + Table table = catalog.getTable(identifier(format)); + Map oldValues = new LinkedHashMap<>(); + oldValues.put(BinaryString.fromString("key1"), 10); + write(table, GenericRow.of(1, new GenericMap(oldValues))); + + catalog.alterTable( + identifier(format), + Arrays.asList( + SchemaChange.updateColumnType( + new String[] {"metrics", "value"}, DataTypes.BIGINT(), false), + SchemaChange.setOption( + "fields.metrics.map.storage-layout", "shared-shredding"), + SchemaChange.setOption( + "fields.metrics.map.shared-shredding.max-columns", "1")), + false); + table = catalog.getTable(identifier(format)); + write(table, GenericRow.of(2, mapOf("key2", 20L))); + + assertThat(readSelectedKeysById(table)) + .containsEntry(1, Arrays.asList(10L, null, null)) + .containsEntry(2, Arrays.asList(null, 20L, null)); + } + @ParameterizedTest @ValueSource(strings = {"orc", "parquet"}) public void testReadSelectedKeysAfterRenameColumn(String format) throws Exception { @@ -285,6 +323,36 @@ public void testReadSelectedKeysAfterRenameColumn(String format) throws Exceptio .containsEntry(2, Arrays.asList(null, 30L, null)); } + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testReadSelectedKeysAfterDefaultToSharedAndRenameColumn(String format) + throws Exception { + Table table = + createTableWithBucket( + format, + 1, + "1", + Collections.singletonList("metrics"), + Collections.emptyList()); + write(table, GenericRow.of(1, mapOf("key1", 10L, "key2", 20L))); + + catalog.alterTable( + identifier(format), + Arrays.asList( + SchemaChange.renameColumn("metrics", "renamed_metrics"), + SchemaChange.setOption( + "fields.renamed_metrics.map.storage-layout", "shared-shredding"), + SchemaChange.setOption( + "fields.renamed_metrics.map.shared-shredding.max-columns", "1")), + false); + table = catalog.getTable(identifier(format)); + write(table, GenericRow.of(2, mapOf("key2", 30L))); + + assertThat(readSelectedKeysById(table, "renamed_metrics")) + .containsEntry(1, Arrays.asList(10L, 20L, null)) + .containsEntry(2, Arrays.asList(null, 30L, null)); + } + @ParameterizedTest @ValueSource(strings = {"orc", "parquet"}) public void testAppendOnlyTableReadWriteWithTwoMapFields(String format) throws Exception { @@ -857,29 +925,89 @@ public void testPrimaryKeyStartsNewRollingWriterWithMaxColumnCount( @ParameterizedTest @ValueSource(strings = {"orc", "parquet"}) - public void testCannotSwitchMapLayoutAndUseMaxColumnsWithoutMetadata(String format) - throws Exception { - createTableWithBucket( - format, 4, "1", Arrays.asList("metrics", "labels"), Arrays.asList("labels")); + public void testSwitchMapLayoutAndInferColumns(String format) throws Exception { + Table table = + createTableWithBucket( + format, + 4, + "1", + Arrays.asList("metrics", "labels"), + Arrays.asList("labels")); - assertThatThrownBy( - () -> - catalog.alterTable( - identifier(format), - Arrays.asList( - SchemaChange.setOption( - "fields.metrics.map.storage-layout", - "shared-shredding"), - SchemaChange.setOption( - "fields.metrics.map.shared-shredding.max-columns", - "3"), - SchemaChange.setOption( - "fields.labels.map.storage-layout", - "default")), - false)) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining( - "Cannot change map storage layout for field id 1 ('metrics' -> 'metrics') from 'default' to 'shared-shredding'."); + write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L), mapOf("x", 21L))); + + catalog.alterTable( + identifier(format), + Arrays.asList( + SchemaChange.setOption( + "fields.metrics.map.storage-layout", "shared-shredding"), + SchemaChange.setOption( + "fields.metrics.map.shared-shredding.max-columns", "3"), + SchemaChange.setOption("fields.labels.map.storage-layout", "default")), + false); + table = catalog.getTable(identifier(format)); + + write(table, GenericRow.of(2, mapOf("c", 31L), mapOf("y", 41L, "z", 42L))); + + FileStoreTable fileStoreTable = (FileStoreTable) table; + List files = currentDataFiles(fileStoreTable); + files.sort(Comparator.comparingLong(file -> file.dataFile.minSequenceNumber())); + assertThat(files).hasSize(2); + + MapSharedShreddingFieldMeta metricsMeta = + readSharedShreddingFieldMeta(fileStoreTable, files.get(1), "metrics"); + assertThat(metricsMeta.numColumns()).isEqualTo(3); + assertThat(metricsMeta.maxRowWidth()).isEqualTo(1); + + Map>> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put( + row.getInt(0), + Arrays.asList( + row.isNullAt(1) ? null : toJavaMap(row.getMap(1)), + row.isNullAt(2) ? null : toJavaMap(row.getMap(2)))); + } + + assertThat(actual) + .containsEntry(1, Arrays.asList(javaMapOf("a", 11L, "b", 12L), javaMapOf("x", 21L))) + .containsEntry( + 2, Arrays.asList(javaMapOf("c", 31L), javaMapOf("y", 41L, "z", 42L))); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testSwitchSharedShreddingToDefaultMap(String format) throws Exception { + Table table = + createTableWithBucket( + format, + 1, + "1", + Collections.singletonList("metrics"), + Collections.singletonList("metrics")); + write(table, GenericRow.of(1, mapOf("key1", 10L, "key2", 20L))); + + catalog.alterTable( + identifier(format), + Collections.singletonList( + SchemaChange.setOption("fields.metrics.map.storage-layout", "default")), + false); + table = catalog.getTable(identifier(format)); + write(table, GenericRow.of(2, mapOf("key2", 30L, "cold", 40L))); + + assertThat(readMapsById(table.newReadBuilder())) + .containsOnlyKeys(1, 2) + .containsEntry(1, javaMapOf("key1", 10L, "key2", 20L)) + .containsEntry(2, javaMapOf("key2", 30L, "cold", 40L)); + assertThat(countSharedShreddingFiles((FileStoreTable) table)).isEqualTo(1); + + table = enableForcedCompaction(table); + compact(table, BinaryRow.EMPTY_ROW, 0); + assertThat(countSharedShreddingFiles((FileStoreTable) table)).isZero(); + assertThat(currentDataFiles((FileStoreTable) table)).hasSize(1); + assertThat(readMapsById(table.newReadBuilder())) + .containsOnlyKeys(1, 2) + .containsEntry(1, javaMapOf("key1", 10L, "key2", 20L)) + .containsEntry(2, javaMapOf("key2", 30L, "cold", 40L)); } @ParameterizedTest @@ -1099,6 +1227,13 @@ private Table createTable(String format, String... sharedShreddingFields) throws return createTable(format, 2, sharedShreddingFields); } + private Table enableForcedCompaction(Table table) { + Map options = new LinkedHashMap<>(); + options.put(CoreOptions.WRITE_ONLY.key(), "false"); + options.put(CoreOptions.COMPACTION_FORCE_REWRITE_ALL_FILES.key(), "true"); + return table.copy(options); + } + private Table createTable(String format, int maxColumns, String... sharedShreddingFields) throws Exception { return createTableWithBucket(format, maxColumns, "-1", sharedShreddingFields); @@ -1537,21 +1672,65 @@ private List currentDataFiles(FileStoreTable table) throws Ex return files; } + private int countSharedShreddingFiles(FileStoreTable table) throws Exception { + int count = 0; + for (DataFileWithSplit file : currentDataFiles(table)) { + Map> fieldMetadata = readFieldMetadata(table, file); + if (MapSharedShreddingUtils.hasShreddingMetadata(fieldMetadata.get("metrics"))) { + count++; + } + } + return count; + } + private MapSharedShreddingFieldMeta readSharedShreddingFieldMeta( FileStoreTable table, DataFileWithSplit file, String fieldName) throws Exception { + return MapSharedShreddingUtils.deserializeMetadata( + readFieldMetadata(table, file).get(fieldName)); + } + + private Map> readFieldMetadata( + FileStoreTable table, DataFileWithSplit file) throws Exception { DataFilePathFactory pathFactory = table.store().pathFactory().createDataFilePathFactory(file.partition, file.bucket); FileFormat fileFormat = FileFormatDiscover.of(new CoreOptions(table.options())) .discover(file.dataFile.fileFormat()); - Map> fieldMetadata = - ((SupportsFieldMetadata) fileFormat) - .readFieldMetadata( - new FormatReaderContext( - table.fileIO(), - pathFactory.toPath(file.dataFile), - file.dataFile.fileSize())); - return MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get(fieldName)); + return ((SupportsFieldMetadata) fileFormat) + .readFieldMetadata( + new FormatReaderContext( + table.fileIO(), + pathFactory.toPath(file.dataFile), + file.dataFile.fileSize())); + } + + private Map> readSelectedKeysById(Table table) throws Exception { + return readSelectedKeysById(table, "metrics"); + } + + private Map> readSelectedKeysById(Table table, String fieldName) + throws Exception { + ReadBuilder readBuilder = + table.newReadBuilder().withReadType(selectedKeysReadType(fieldName)); + Map> actual = new LinkedHashMap<>(); + try (RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan())) { + reader.forEachRemaining( + row -> { + if (row.isNullAt(1)) { + actual.put(row.getInt(0), null); + return; + } + InternalRow selectedKeys = row.getRow(1, 3); + actual.put( + row.getInt(0), + Arrays.asList( + selectedKeys.isNullAt(0) ? null : selectedKeys.getLong(0), + selectedKeys.isNullAt(1) ? null : selectedKeys.getLong(1), + selectedKeys.isNullAt(2) ? null : selectedKeys.getLong(2))); + }); + } + return actual; } private GenericMap mapOf(Object... entries) { diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala index f2351ac965b5..fa3f49b77c0b 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala @@ -64,6 +64,88 @@ abstract class MapSelectedKeysSharedShreddingE2ETestBase extends PaimonSparkTest } } + test(s"switch map storage layout in both directions for $format") { + withTable("T") { + sql(s""" + |CREATE TABLE T (id INT, attrs MAP) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'file.format' = '$format' + |) + |""".stripMargin) + sql(""" + |INSERT INTO T VALUES + | (1, map('key1', CAST(10 AS BIGINT), 'key2', CAST(20 AS BIGINT))) + |""".stripMargin) + + sql(""" + |ALTER TABLE T SET TBLPROPERTIES ( + | 'fields.attrs.map.storage-layout' = 'shared-shredding', + | 'fields.attrs.map.shared-shredding.max-columns' = '1' + |) + |""".stripMargin) + sql(""" + |INSERT INTO T VALUES + | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS BIGINT))) + |""".stripMargin) + + val firstSharedQuery = + sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id") + assert( + pushedMapSelectedKeys(firstSharedQuery).contains("attrs"), + s"Expected selected-key pushdown after switching $format to shared-shredding") + checkAnswer(firstSharedQuery, Row(1, 10L, 20L) :: Row(2, null, 30L) :: Nil) + + sql(""" + |ALTER TABLE T SET TBLPROPERTIES ( + | 'fields.attrs.map.storage-layout' = 'default' + |) + |""".stripMargin) + sql(""" + |INSERT INTO T VALUES + | (3, map('key1', CAST(50 AS BIGINT))) + |""".stripMargin) + + val defaultQuery = + sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id") + assert( + pushedMapSelectedKeys(defaultQuery).isEmpty, + s"Expected selected-key pushdown to be disabled after switching $format to default") + checkAnswer( + defaultQuery, + Row(1, 10L, 20L) :: Row(2, null, 30L) :: Row(3, 50L, null) :: Nil) + + sql(""" + |ALTER TABLE T SET TBLPROPERTIES ( + | 'fields.attrs.map.storage-layout' = 'shared-shredding' + |) + |""".stripMargin) + sql(""" + |INSERT INTO T VALUES + | (4, map('key2', CAST(60 AS BIGINT))) + |""".stripMargin) + + val secondSharedQuery = + sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id") + assert( + pushedMapSelectedKeys(secondSharedQuery).contains("attrs"), + s"Expected selected-key pushdown after switching $format back to shared-shredding") + checkAnswer( + secondSharedQuery, + Row(1, 10L, 20L) :: + Row(2, null, 30L) :: + Row(3, 50L, null) :: + Row(4, null, 60L) :: Nil) + checkAnswer( + sql("SELECT id, attrs FROM T ORDER BY id"), + Row(1, Map("key1" -> 10L, "key2" -> 20L)) :: + Row(2, Map("key2" -> 30L, "cold" -> 40L)) :: + Row(3, Map("key1" -> 50L)) :: + Row(4, Map("key2" -> 60L)) :: Nil + ) + } + } + Seq(false, true).foreach { thinMode => test(s"skip selected-key pushdown for merge_map from $format with thin mode $thinMode") { @@ -212,4 +294,15 @@ abstract class MapSelectedKeysSharedShreddingE2ETestBase extends PaimonSparkTest checkAnswer(query, Row(1, 10L, 30L) :: Row(2, 50L, null) :: Nil) } } + + private def pushedMapSelectedKeys( + query: org.apache.spark.sql.DataFrame): Map[String, Seq[String]] = { + val sparkPlan = query.queryExecution.sparkPlan + sparkPlan + .collectFirst { + case scan: BatchScanExec if scan.scan.isInstanceOf[PaimonScan] => + scan.scan.asInstanceOf[PaimonScan].pushedMapSelectedKeys + } + .getOrElse(fail(s"Expected a Paimon scan in physical plan:\n$sparkPlan")) + } }