[iceberg] support altering table properties

Author: nastra

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java

@@ -32,6 +32,8 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -119,8 +121,27 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co
     @Override
     public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Context context)
             throws TableNotExistException {
-        throw new UnsupportedOperationException(
-                "Alter table is not supported for Iceberg at the moment");
+        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;
+                    updateProperties.set(
+                            convertFlussPropertyKeyToIceberg(option.getKey()), option.getValue());
+                } else if (tableChange instanceof TableChange.ResetOption) {
+                    TableChange.ResetOption option = (TableChange.ResetOption) tableChange;
+                    updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+                } else {
+                    throw new UnsupportedOperationException(
+                            "Unsupported table change: " + tableChange.getClass());
+                }
+            }
+
+            updateProperties.commit();
+        } catch (TableNotExistException e) {
+            throw new TableNotExistException("Table " + tablePath + " not exists.");
+        }
     }
 
     private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) {
@@ -249,6 +270,14 @@ private void setFlussPropertyToIceberg(
         }
     }
 
+    private static String convertFlussPropertyKeyToIceberg(String key) {
+        if (key.startsWith(ICEBERG_CONF_PREFIX)) {
+            return key.substring(ICEBERG_CONF_PREFIX.length());
+        } else {
+            return FLUSS_CONF_PREFIX + key;
+        }
+    }
+
     private void createDatabase(String databaseName) {
         Namespace namespace = Namespace.of(databaseName);
         if (icebergCatalog instanceof SupportsNamespaces) {

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java

@@ -22,6 +22,7 @@
 import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
 import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableChange;
 import org.apache.fluss.metadata.TableDescriptor;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.types.DataTypes;
@@ -30,6 +31,7 @@
 import org.apache.iceberg.SortDirection;
 import org.apache.iceberg.SortField;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.types.Types;
 import org.assertj.core.api.Assertions;
@@ -455,4 +457,88 @@ void testIllegalPartitionKeyType(boolean isPrimaryKeyTable) throws Exception {
                 .hasMessage(
                         "Partition key only support string type for iceberg currently. Column `c1` is not string type.");
     }
+
+    @Test
+    void alterTableProperties() {
+        String database = "test_alter_table_db";
+        String tableName = "test_alter_table";
+
+        Schema flussSchema = Schema.newBuilder().column("id", DataTypes.BIGINT()).build();
+
+        TableDescriptor tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(flussSchema)
+                        .distributedBy(3)
+                        .property("iceberg.commit.retry.num-retries", "5")
+                        .property("table.datalake.freshness", "30s")
+                        .build();
+
+        TablePath tablePath = TablePath.of(database, tableName);
+        TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+        flussIcebergCatalog.createTable(tablePath, tableDescriptor, context);
+
+        Catalog catalog = flussIcebergCatalog.getIcebergCatalog();
+        assertThat(catalog.loadTable(TableIdentifier.of(database, tableName)).properties())
+                .containsEntry("commit.retry.num-retries", "5")
+                .containsEntry("fluss.table.datalake.freshness", "30s")
+                .doesNotContainKeys("iceberg.commit.retry.num-retries", "table.datalake.freshness");
+
+        // set new iceberg property
+        flussIcebergCatalog.alterTable(
+                tablePath,
+                List.of(TableChange.set("iceberg.commit.retry.min-wait-ms", "1000")),
+                context);
+        assertThat(catalog.loadTable(TableIdentifier.of(database, tableName)).properties())
+                .containsEntry("commit.retry.min-wait-ms", "1000")
+                .containsEntry("commit.retry.num-retries", "5")
+                .containsEntry("fluss.table.datalake.freshness", "30s")
+                .doesNotContainKeys(
+                        "iceberg.commit.retry.min-wait-ms",
+                        "iceberg.commit.retry.num-retries",
+                        "table.datalake.freshness");
+
+        // update existing properties
+        flussIcebergCatalog.alterTable(
+                tablePath,
+                List.of(
+                        TableChange.set("iceberg.commit.retry.num-retries", "10"),
+                        TableChange.set("table.datalake.freshness", "23s")),
+                context);
+        assertThat(catalog.loadTable(TableIdentifier.of(database, tableName)).properties())
+                .containsEntry("commit.retry.min-wait-ms", "1000")
+                .containsEntry("commit.retry.num-retries", "10")
+                .containsEntry("fluss.table.datalake.freshness", "23s")
+                .doesNotContainKeys(
+                        "iceberg.commit.retry.min-wait-ms",
+                        "iceberg.commit.retry.num-retries",
+                        "table.datalake.freshness");
+
+        // remove existing properties
+        flussIcebergCatalog.alterTable(
+                tablePath,
+                List.of(
+                        TableChange.reset("iceberg.commit.retry.min-wait-ms"),
+                        TableChange.reset("table.datalake.freshness")),
+                context);
+        assertThat(catalog.loadTable(TableIdentifier.of(database, tableName)).properties())
+                .containsEntry("commit.retry.num-retries", "10")
+                .doesNotContainKeys(
+                        "commit.retry.min-wait-ms",
+                        "iceberg.commit.retry.min-wait-ms",
+                        "table.datalake.freshness",
+                        "fluss.table.datalake.freshness");
+
+        // remove non-existing property
+        flussIcebergCatalog.alterTable(
+                tablePath, List.of(TableChange.reset("iceberg.non-existing.property")), context);
+        assertThat(catalog.loadTable(TableIdentifier.of(database, tableName)).properties())
+                .containsEntry("commit.retry.num-retries", "10")
+                .doesNotContainKeys(
+                        "non-existing.property",
+                        "iceberg.non-existing.property",
+                        "commit.retry.min-wait-ms",
+                        "iceberg.commit.retry.min-wait-ms",
+                        "table.datalake.freshness",
+                        "fluss.table.datalake.freshness");
+    }
 }