From fb2d83a77c79b4162360a54914e404c351225750 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Wed, 2 Sep 2026 15:49:37 +0530 Subject: [PATCH 1/2] update --- .../apache/paimon/schema/SchemaManager.java | 35 +++++++++--- .../paimon/schema/SchemaManagerTest.java | 56 +++++++++++++++++++ 2 files changed, 83 insertions(+), 8 deletions(-) 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 1a76668448b7..cd74cfeae8c1 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 @@ -44,6 +44,7 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeCasts; import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.ReassignFieldId; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.BranchManager; @@ -533,7 +534,7 @@ protected void updateLastColumn( "Column type %s[%s] cannot be converted to %s without losing information.", field.name(), sourceRootType, targetRootType)); DataType newFieldType = - getArrayMapTypeWithTargetTypeRoot( + getNestedTypeWithTargetTypeRoot( field.type(), targetRootType, depth, @@ -572,7 +573,7 @@ protected void updateLastColumn( return new DataField( field.id(), field.name(), - getArrayMapTypeWithTargetTypeRoot( + getNestedTypeWithTargetTypeRoot( field.type(), sourceRootType, depth, @@ -668,6 +669,8 @@ private static DataType getRootType(DataType type, int currDepth, int maxDepth) return getRootType(((ArrayType) type).getElementType(), currDepth + 1, maxDepth); case MAP: return getRootType(((MapType) type).getValueType(), currDepth + 1, maxDepth); + case MULTISET: + return getRootType(((MultisetType) type).getElementType(), currDepth + 1, maxDepth); default: return type; } @@ -677,7 +680,7 @@ private static DataType getRootType(DataType type, int currDepth, int maxDepth) // ex: ARRAY>> -> ARRAY>> // here we only need to update type of ARRAY to ARRAY and rest of the type // remains same. This function achieves this. - private static DataType getArrayMapTypeWithTargetTypeRoot( + private static DataType getNestedTypeWithTargetTypeRoot( DataType source, DataType target, int currDepth, int maxDepth) { if (currDepth == maxDepth - 1) { return target; @@ -686,7 +689,7 @@ private static DataType getArrayMapTypeWithTargetTypeRoot( case ARRAY: return new ArrayType( source.isNullable(), - getArrayMapTypeWithTargetTypeRoot( + getNestedTypeWithTargetTypeRoot( ((ArrayType) source).getElementType(), target, currDepth + 1, @@ -695,11 +698,19 @@ private static DataType getArrayMapTypeWithTargetTypeRoot( return new MapType( source.isNullable(), ((MapType) source).getKeyType(), - getArrayMapTypeWithTargetTypeRoot( + getNestedTypeWithTargetTypeRoot( ((MapType) source).getValueType(), target, currDepth + 1, maxDepth)); + case MULTISET: + return new MultisetType( + source.isNullable(), + getNestedTypeWithTargetTypeRoot( + ((MultisetType) source).getElementType(), + target, + currDepth + 1, + maxDepth)); default: return target; } @@ -1074,14 +1085,14 @@ private void updateIntermediateColumn( updateLastColumn(depth, newFields, updateFieldNames[depth]); return; } else if (depth >= updateFieldNames.length) { - // to handle the case of ARRAY or MAP type evolution + // to handle the case of ARRAY, MAP or MULTISET type evolution // for instance : ARRAY -> ARRAY // the updateFieldNames in this case is [v, element] where v is array field name // the depth returned by extractRowDataFields is 2 which will overflow. // So the logic is to go to previous depth and update the column using previous // fields which will have DataFields from prevDepth - // The reason for this handling is the addition of element and value for array - // and map type in FlinkCatalog as dummy column name + // The reason for this handling is the addition of element and value for array, + // map and multiset types in FlinkCatalog as dummy column names updateLastColumn(prevDepth, previousFields, updateFieldNames[prevDepth]); return; } @@ -1126,6 +1137,10 @@ private int extractRowDataFields(DataType type, List nestedFields) { + 1; case MAP: return extractRowDataFields(((MapType) type).getValueType(), nestedFields) + 1; + case MULTISET: + return extractRowDataFields( + ((MultisetType) type).getElementType(), nestedFields) + + 1; default: return 1; } @@ -1145,6 +1160,10 @@ private DataType wrapNewRowType(DataType type, List nestedFields) { type.isNullable(), mapType.getKeyType(), wrapNewRowType(mapType.getValueType(), nestedFields)); + case MULTISET: + return new MultisetType( + type.isNullable(), + wrapNewRowType(((MultisetType) type).getElementType(), nestedFields)); default: return type; } 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 78396006a7fe..bec3f6e09ab3 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 @@ -40,6 +40,7 @@ import org.apache.paimon.types.DoubleType; import org.apache.paimon.types.IntType; import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; import org.apache.paimon.types.VarCharType; import org.apache.paimon.types.VariantType; @@ -1569,6 +1570,61 @@ public void testUpdateRowTypeInArrayAndMap() throws Exception { assertThat(manager.latest().get().logicalRowType()).isEqualTo(outerType); } + @Test + public void testUpdateMultisetElementType() throws Exception { + MultisetType oldType = new MultisetType(DataTypes.INT()); + RowType rowType = + RowType.of( + new DataField(0, "k", DataTypes.INT()), new DataField(1, "items", oldType)); + Schema schema = + new Schema( + rowType.getFields(), + Collections.singletonList("k"), + Collections.emptyList(), + new HashMap<>(), + ""); + SchemaManager manager = new SchemaManager(LocalFileIO.create(), path); + manager.createTable(schema); + + MultisetType newType = new MultisetType(DataTypes.BIGINT()); + List changes = new ArrayList<>(); + NestedSchemaUtils.generateNestedColumnUpdates( + Collections.singletonList("items"), oldType, newType, changes); + manager.commitChanges(changes); + + assertThat(manager.latest().get().fields().get(1).type()).isEqualTo(newType); + } + + @Test + public void testUpdateRowTypeInMultiset() throws Exception { + RowType oldElementType = RowType.of(new DataField(2, "id", DataTypes.INT())); + MultisetType oldType = new MultisetType(oldElementType); + RowType rowType = + RowType.of( + new DataField(0, "k", DataTypes.INT()), new DataField(1, "items", oldType)); + Schema schema = + new Schema( + rowType.getFields(), + Collections.singletonList("k"), + Collections.emptyList(), + new HashMap<>(), + ""); + SchemaManager manager = new SchemaManager(LocalFileIO.create(), path); + manager.createTable(schema); + + RowType newElementType = + RowType.of( + new DataField(2, "id", DataTypes.BIGINT()), + new DataField(3, "name", DataTypes.STRING())); + MultisetType newType = new MultisetType(newElementType); + List changes = new ArrayList<>(); + NestedSchemaUtils.generateNestedColumnUpdates( + Collections.singletonList("items"), oldType, newType, changes); + manager.commitChanges(changes); + + assertThat(manager.latest().get().fields().get(1).type()).isEqualTo(newType); + } + @Test public void testAlterDeletionVectorsMode() throws Exception { // create table From 6b3c6906116a76b5447497e65814018d3841c293 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Tue, 8 Sep 2026 16:43:01 +0530 Subject: [PATCH 2/2] update --- .../org/apache/paimon/casting/CastedMap.java | 20 ++++-- .../paimon/schema/SchemaEvolutionUtil.java | 15 ++++ .../paimon/table/SchemaEvolutionTest.java | 68 +++++++++++++++++++ 3 files changed, 98 insertions(+), 5 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/casting/CastedMap.java b/paimon-common/src/main/java/org/apache/paimon/casting/CastedMap.java index 4068407ca71c..e6e866931964 100644 --- a/paimon-common/src/main/java/org/apache/paimon/casting/CastedMap.java +++ b/paimon-common/src/main/java/org/apache/paimon/casting/CastedMap.java @@ -30,11 +30,17 @@ */ public class CastedMap implements InternalMap { - private final CastedArray castedValueArray; + private final CastedArray castedArray; + private final boolean castKey; private InternalMap map; protected CastedMap(CastElementGetter castValueGetter) { - this.castedValueArray = CastedArray.from(castValueGetter); + this(castValueGetter, false); + } + + private CastedMap(CastElementGetter castElementGetter, boolean castKey) { + this.castedArray = CastedArray.from(castElementGetter); + this.castKey = castKey; } /** @@ -47,8 +53,12 @@ public static CastedMap from(CastElementGetter castValueGetter) { return new CastedMap(castValueGetter); } + public static CastedMap fromKey(CastElementGetter castKeyGetter) { + return new CastedMap(castKeyGetter, true); + } + public CastedMap replaceMap(InternalMap map) { - this.castedValueArray.replaceArray(map.valueArray()); + this.castedArray.replaceArray(castKey ? map.keyArray() : map.valueArray()); this.map = map; return this; } @@ -60,11 +70,11 @@ public int size() { @Override public InternalArray keyArray() { - return map.keyArray(); + return castKey ? castedArray : map.keyArray(); } @Override public InternalArray valueArray() { - return castedValueArray; + return castKey ? map.valueArray() : castedArray; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java index 05fb9be9f87c..d0d41e664989 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java @@ -36,6 +36,7 @@ import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.InternalRowUtils; import org.apache.paimon.utils.ProjectedRow; @@ -258,6 +259,8 @@ private static CastFieldGetter[] createCastFieldGetterMapping( return createArrayCastExecutor((ArrayType) inputType, (ArrayType) targetType); } else if (inputType instanceof MapType && targetType instanceof MapType) { return createMapCastExecutor((MapType) inputType, (MapType) targetType); + } else if (inputType instanceof MultisetType && targetType instanceof MultisetType) { + return createMultisetCastExecutor((MultisetType) inputType, (MultisetType) targetType); } else { return checkNotNull( CastExecutors.resolve(inputType, targetType), @@ -314,4 +317,16 @@ private static CastExecutor createMapCastExecutor( CastedMap castedMap = CastedMap.from(castElementGetter); return castedMap::replaceMap; } + + private static CastExecutor createMultisetCastExecutor( + MultisetType inputType, MultisetType targetType) { + CastElementGetter castElementGetter = + new CastElementGetter( + InternalArray.createElementGetter(inputType.getElementType()), + createCastExecutor( + inputType.getElementType(), targetType.getElementType())); + + CastedMap castedMap = CastedMap.fromKey(castElementGetter); + return castedMap::replaceMap; + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java index 59acdec83ee3..040175febce6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java @@ -23,13 +23,16 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.DataFormatTestUtil; +import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalMap; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.variant.GenericVariant; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.schema.NestedSchemaUtils; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; @@ -44,6 +47,7 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.LazyField; @@ -61,6 +65,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; @@ -656,6 +661,69 @@ public void testUpdateFieldType() throws Exception { assertThat(tableSchema.fields().get(0).type()).isEqualTo(DataTypes.STRING()); } + @Test + public void testUpdateMultisetElementType() throws Exception { + MultisetType oldIntMultiset = new MultisetType(DataTypes.INT()); + RowType oldElementType = RowType.of(new DataField(3, "id", DataTypes.INT())); + MultisetType oldRowMultiset = new MultisetType(oldElementType); + schemaManager.createTable( + new Schema( + RowType.of( + new DataField(0, "k", DataTypes.INT()), + new DataField(1, "int_items", oldIntMultiset), + new DataField(2, "row_items", oldRowMultiset)) + .getFields(), + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyMap(), + "")); + + Map intItems = new LinkedHashMap<>(); + intItems.put(1, 2); + intItems.put(2, 1); + Map rowItems = new LinkedHashMap<>(); + rowItems.put(GenericRow.of(3), 4); + + FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), tablePath); + try (StreamTableWrite write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, new GenericMap(intItems), new GenericMap(rowItems))); + commit.commit(0, write.prepareCommit(true, 0)); + } + + MultisetType newIntMultiset = new MultisetType(DataTypes.BIGINT()); + RowType newElementType = + RowType.of( + new DataField(3, "id", DataTypes.BIGINT()), + new DataField(4, "name", DataTypes.STRING())); + MultisetType newRowMultiset = new MultisetType(newElementType); + List changes = new ArrayList<>(); + NestedSchemaUtils.generateNestedColumnUpdates( + Collections.singletonList("int_items"), oldIntMultiset, newIntMultiset, changes); + NestedSchemaUtils.generateNestedColumnUpdates( + Collections.singletonList("row_items"), oldRowMultiset, newRowMultiset, changes); + schemaManager.commitChanges(changes); + + table = FileStoreTableFactory.create(LocalFileIO.create(), tablePath); + int[] rowCount = {0}; + forEachRemaining( + table, + null, + row -> { + rowCount[0]++; + InternalMap intMultiset = row.getMap(1); + assertThat(intMultiset.keyArray().toLongArray()).containsExactly(1L, 2L); + assertThat(intMultiset.valueArray().toIntArray()).containsExactly(2, 1); + + InternalMap rowMultiset = row.getMap(2); + InternalRow element = rowMultiset.keyArray().getRow(0, 2); + assertThat(element.getLong(0)).isEqualTo(3L); + assertThat(element.isNullAt(1)).isTrue(); + assertThat(rowMultiset.valueArray().toIntArray()).containsExactly(4); + }); + assertThat(rowCount[0]).isEqualTo(1); + } + @Test public void testUpdatePrimaryKeyType() throws Exception { Schema schema =