Copilot commented on code in PR #2593:
URL: https://github.com/apache/fluss/pull/2593#discussion_r3279763684
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java:
##########
@@ -133,35 +138,157 @@ public void alterTable(TablePath tablePath,
List<TableChange> tableChanges, Cont
throws TableNotExistException {
try {
Table table =
icebergCatalog.loadTable(toIcebergTableIdentifier(tablePath));
- UpdateProperties updateProperties = table.updateProperties();
- for (TableChange tableChange : tableChanges) {
- if (tableChange instanceof TableChange.SetOption) {
- TableChange.SetOption option = (TableChange.SetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot set table property '%s'",
- option.getKey());
- updateProperties.set(
- convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
- } else if (tableChange instanceof TableChange.ResetOption) {
- TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot reset table property '%s'",
- option.getKey());
-
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+
+ List<TableChange> schemaChanges = new ArrayList<>();
+ List<TableChange> propertyChanges = new ArrayList<>();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SchemaChange) {
+ schemaChanges.add(change);
} else {
- throw new UnsupportedOperationException(
- "Unsupported table change: " +
tableChange.getClass());
+ propertyChanges.add(change);
}
}
- updateProperties.commit();
+ if (!schemaChanges.isEmpty()) {
+ applySchemaChanges(table, schemaChanges, context);
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ applyPropertyChanges(table, propertyChanges);
+ }
} catch (NoSuchTableException e) {
throw new TableNotExistException("Table " + tablePath + " does not
exist.");
}
}
+ private void applyPropertyChanges(Table table, List<TableChange>
propertyChanges) {
+ UpdateProperties updateProperties = table.updateProperties();
+ for (TableChange tableChange : propertyChanges) {
+ if (tableChange instanceof TableChange.SetOption) {
+ TableChange.SetOption option = (TableChange.SetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot set table property '%s'",
+ option.getKey());
+ updateProperties.set(
+ convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
+ } else if (tableChange instanceof TableChange.ResetOption) {
+ TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot reset table property '%s'",
+ option.getKey());
+
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported table change: " + tableChange.getClass());
+ }
+ }
+ updateProperties.commit();
+ }
+
+ private void applySchemaChanges(Table table, List<TableChange>
schemaChanges, Context context) {
+ Schema currentIcebergSchema = table.schema();
+
+ // Check schema compatibility to handle crash recovery idempotency.
+ boolean skipAddColumns;
+ if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getCurrentTable())) {
+ // Iceberg schema matches current Fluss schema, apply all changes.
+ skipAddColumns = false;
+ } else if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getExpectedTable())) {
+ // Iceberg schema already matches expected (post-alter) schema,
+ // skip AddColumn changes since they were already applied.
+ skipAddColumns = true;
+ } else {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Iceberg schema is not compatible with Fluss
schema: "
+ + "Iceberg schema: %s, Fluss schema: %s. "
+ + "therefore you need to add the diff
columns all at once, "
+ + "rather than applying other table
changes: %s.",
Review Comment:
The exception message has a couple of clarity/grammar issues (e.g., new
sentence starts with lowercase “therefore”, and “diff columns” is ambiguous).
Consider rephrasing/capitalizing to make the failure reason and required action
clearer for users.
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java:
##########
@@ -133,35 +138,157 @@ public void alterTable(TablePath tablePath,
List<TableChange> tableChanges, Cont
throws TableNotExistException {
try {
Table table =
icebergCatalog.loadTable(toIcebergTableIdentifier(tablePath));
- UpdateProperties updateProperties = table.updateProperties();
- for (TableChange tableChange : tableChanges) {
- if (tableChange instanceof TableChange.SetOption) {
- TableChange.SetOption option = (TableChange.SetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot set table property '%s'",
- option.getKey());
- updateProperties.set(
- convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
- } else if (tableChange instanceof TableChange.ResetOption) {
- TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot reset table property '%s'",
- option.getKey());
-
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+
+ List<TableChange> schemaChanges = new ArrayList<>();
+ List<TableChange> propertyChanges = new ArrayList<>();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SchemaChange) {
+ schemaChanges.add(change);
} else {
- throw new UnsupportedOperationException(
- "Unsupported table change: " +
tableChange.getClass());
+ propertyChanges.add(change);
}
}
- updateProperties.commit();
+ if (!schemaChanges.isEmpty()) {
+ applySchemaChanges(table, schemaChanges, context);
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ applyPropertyChanges(table, propertyChanges);
+ }
} catch (NoSuchTableException e) {
throw new TableNotExistException("Table " + tablePath + " does not
exist.");
}
}
+ private void applyPropertyChanges(Table table, List<TableChange>
propertyChanges) {
+ UpdateProperties updateProperties = table.updateProperties();
+ for (TableChange tableChange : propertyChanges) {
+ if (tableChange instanceof TableChange.SetOption) {
+ TableChange.SetOption option = (TableChange.SetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot set table property '%s'",
+ option.getKey());
+ updateProperties.set(
+ convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
+ } else if (tableChange instanceof TableChange.ResetOption) {
+ TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot reset table property '%s'",
+ option.getKey());
+
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported table change: " + tableChange.getClass());
+ }
+ }
+ updateProperties.commit();
+ }
+
+ private void applySchemaChanges(Table table, List<TableChange>
schemaChanges, Context context) {
+ Schema currentIcebergSchema = table.schema();
+
+ // Check schema compatibility to handle crash recovery idempotency.
+ boolean skipAddColumns;
+ if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getCurrentTable())) {
+ // Iceberg schema matches current Fluss schema, apply all changes.
+ skipAddColumns = false;
+ } else if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getExpectedTable())) {
+ // Iceberg schema already matches expected (post-alter) schema,
+ // skip AddColumn changes since they were already applied.
+ skipAddColumns = true;
+ } else {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Iceberg schema is not compatible with Fluss
schema: "
+ + "Iceberg schema: %s, Fluss schema: %s. "
+ + "therefore you need to add the diff
columns all at once, "
+ + "rather than applying other table
changes: %s.",
+ currentIcebergSchema,
+ context.getCurrentTable().getSchema(),
+ schemaChanges));
+ }
+
+ UpdateSchema updateSchema = table.updateSchema();
+ String firstSystemColumnName =
SYSTEM_COLUMNS.keySet().iterator().next();
+ boolean hasChanges = false;
+
+ for (TableChange tableChange : schemaChanges) {
+ if (tableChange instanceof TableChange.AddColumn) {
+ if (skipAddColumns) {
+ continue;
+ }
+ TableChange.AddColumn addColumn = (TableChange.AddColumn)
tableChange;
+
+ if (!(addColumn.getPosition() instanceof TableChange.Last)) {
+ throw new UnsupportedOperationException(
+ "Only support to add column at last for iceberg
table.");
+ }
+
+ org.apache.fluss.types.DataType flussDataType =
addColumn.getDataType();
+ if (!flussDataType.isNullable()) {
+ throw new UnsupportedOperationException(
+ "Only support to add nullable column for iceberg
table.");
+ }
+
+ Type icebergType = flussDataType.accept(new
FlussDataTypeToIcebergDataType());
+ updateSchema.addColumn(addColumn.getName(), icebergType,
addColumn.getComment());
+ updateSchema.moveBefore(addColumn.getName(),
firstSystemColumnName);
Review Comment:
`FlussDataTypeToIcebergDataType` assigns nested field IDs starting from
`nextId`. Here `new FlussDataTypeToIcebergDataType()` starts at 0, which can
collide with existing field IDs in the table schema when adding a complex-typed
column (MAP/ARRAY/ROW). This can cause Iceberg schema updates to fail or
produce invalid schemas. Consider assigning fresh IDs for the new column type
starting after the current schema’s maximum field ID (e.g., compute max ID from
`currentIcebergSchema` and initialize the converter with that start value, or
use an Iceberg helper that assigns fresh IDs for a `Type`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
