fresh-borzoni commented on code in PR #2593:
URL: https://github.com/apache/fluss/pull/2593#discussion_r3296975914
##########
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:
nah
--
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]
