From fb2d83a77c79b4162360a54914e404c351225750 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Wed, 2 Sep 2026 15:49:37 +0530 Subject: [PATCH] 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