Skip to content

[core] Fix schema evolution for multiset elements #9545

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ArnavBalyan wants to merge 1 commit into
base: master
Choose a base branch
from
+83 −8
Open

[core] Fix schema evolution for multiset elements #9545

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 27 additions & 8 deletions paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -572,7 +573,7 @@ protected void updateLastColumn(
return new DataField(
field.id(),
field.name(),
getArrayMapTypeWithTargetTypeRoot(
getNestedTypeWithTargetTypeRoot(
field.type(),
sourceRootType,
depth,
Expand Down Expand Up @@ -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;
}
Expand All @@ -677,7 +680,7 @@ private static DataType getRootType(DataType type, int currDepth, int maxDepth)
// ex: ARRAY<MAP<STRING, ARRAY<INT>>> -> ARRAY<MAP<STRING, ARRAY<BIGINT>>>
// here we only need to update type of ARRAY<INT> to ARRAY<BIGINT> 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;
Expand All @@ -686,7 +689,7 @@ private static DataType getArrayMapTypeWithTargetTypeRoot(
case ARRAY:
return new ArrayType(
source.isNullable(),
getArrayMapTypeWithTargetTypeRoot(
getNestedTypeWithTargetTypeRoot(
((ArrayType) source).getElementType(),
target,
currDepth + 1,
Expand All @@ -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;
}
Expand Down Expand Up @@ -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<INT> -> ARRAY<BIGINT>
// 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;
}
Expand Down Expand Up @@ -1126,6 +1137,10 @@ private int extractRowDataFields(DataType type, List<DataField> nestedFields) {
+ 1;
case MAP:
return extractRowDataFields(((MapType) type).getValueType(), nestedFields) + 1;
case MULTISET:
return extractRowDataFields(
((MultisetType) type).getElementType(), nestedFields)
+ 1;
default:
return 1;
}
Expand All @@ -1145,6 +1160,10 @@ private DataType wrapNewRowType(DataType type, List<DataField> nestedFields) {
type.isNullable(),
mapType.getKeyType(),
wrapNewRowType(mapType.getValueType(), nestedFields));
case MULTISET:
return new MultisetType(
type.isNullable(),
wrapNewRowType(((MultisetType) type).getElementType(), nestedFields));
default:
return type;
}
Expand Down
56 changes: 56 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<SchemaChange> 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<SchemaChange> 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
Expand Down
Loading