Add Iceberg metadata management properties

oneonestar · ebyhr · commit 3413c225da13 · 2024-12-03T15:58:26.000+09:00
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
@@ -271,6 +271,8 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.EXTRA_PROPERTIES_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FILE_FORMAT_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FORMAT_VERSION_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.OBJECT_STORE_LAYOUT_ENABLED_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.PARTITIONING_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.SORTED_BY_PROPERTY;
@@ -359,6 +361,8 @@
 import static org.apache.iceberg.TableProperties.DELETE_ISOLATION_LEVEL;
 import static org.apache.iceberg.TableProperties.DELETE_ISOLATION_LEVEL_DEFAULT;
 import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED;
+import static org.apache.iceberg.TableProperties.METADATA_PREVIOUS_VERSIONS_MAX;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_ENABLED;
 import static org.apache.iceberg.TableProperties.WRITE_DATA_LOCATION;
 import static org.apache.iceberg.TableProperties.WRITE_LOCATION_PROVIDER_IMPL;
@@ -383,6 +387,8 @@ public class IcebergMetadata
             .add(OBJECT_STORE_LAYOUT_ENABLED_PROPERTY)
             .add(DATA_LOCATION_PROPERTY)
             .add(PARTITIONING_PROPERTY)
+            .add(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)
+            .add(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)
             .add(SORTED_BY_PROPERTY)
             .build();
 
@@ -2177,6 +2183,17 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta
             if (!objectStoreEnabled) {
                 throw new TrinoException(INVALID_TABLE_PROPERTY, "Data location can only be set when object store layout is enabled");
             }
+            if (properties.containsKey(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)) {
+                boolean commitEnabled = (boolean) properties.get(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)
+                        .orElseThrow(() -> new IllegalArgumentException("The metadata_delete_after_commit_enabled property cannot be empty"));
+                updateProperties.set(METADATA_DELETE_AFTER_COMMIT_ENABLED, String.valueOf(commitEnabled));
+            }
+
+            if (properties.containsKey(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)) {
+                int metadataPerviousVersionMax = (int) properties.get(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)
+                        .orElseThrow(() -> new IllegalArgumentException("The metadata_previous_versions_max property cannot be empty"));
+                updateProperties.set(METADATA_PREVIOUS_VERSIONS_MAX, Integer.toString(metadataPerviousVersionMax));
+            }
             updateProperties.set(WRITE_DATA_LOCATION, dataLocation);
         }
 
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableProperties.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableProperties.java
@@ -61,6 +61,8 @@ public class IcebergTableProperties
     public static final String PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY = "parquet_bloom_filter_columns";
     public static final String OBJECT_STORE_LAYOUT_ENABLED_PROPERTY = "object_store_layout_enabled";
     public static final String DATA_LOCATION_PROPERTY = "data_location";
+    public static final String METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY = "metadata_delete_after_commit_enabled";
+    public static final String METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY = "metadata_previous_versions_max";
     public static final String EXTRA_PROPERTIES_PROPERTY = "extra_properties";
 
     public static final Set<String> SUPPORTED_PROPERTIES = ImmutableSet.<String>builder()
@@ -73,6 +75,8 @@ public class IcebergTableProperties
             .add(ORC_BLOOM_FILTER_FPP_PROPERTY)
             .add(OBJECT_STORE_LAYOUT_ENABLED_PROPERTY)
             .add(DATA_LOCATION_PROPERTY)
+            .add(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)
+            .add(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)
             .add(EXTRA_PROPERTIES_PROPERTY)
             .add(PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY)
             .build();
@@ -190,6 +194,15 @@ public IcebergTableProperties(
                         "File system location URI for the table's data files",
                         null,
                         false))
+                .add(booleanProperty(
+                        METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY,
+                        "Delete old tracked metadata files after each table commit",
+                        null,
+                        false))
+                .add(integerProperty(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY,
+                        "The number of old metadata files to keep",
+                        null,
+                        false))
                 .build();
 
         checkState(SUPPORTED_PROPERTIES.containsAll(tableProperties.stream()
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergUtil.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergUtil.java
@@ -123,6 +123,8 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.FILE_FORMAT_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FORMAT_VERSION_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.LOCATION_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.OBJECT_STORE_LAYOUT_ENABLED_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_COLUMNS_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_FPP_PROPERTY;
@@ -174,6 +176,8 @@
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED;
+import static org.apache.iceberg.TableProperties.METADATA_PREVIOUS_VERSIONS_MAX;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_ENABLED;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_ENABLED_DEFAULT;
 import static org.apache.iceberg.TableProperties.ORC_BLOOM_FILTER_COLUMNS;
@@ -343,6 +347,15 @@ public static Map<String, Object> getIcebergTableProperties(Table icebergTable)
         Optional<String> dataLocation = Optional.ofNullable(icebergTable.properties().get(WRITE_DATA_LOCATION));
         dataLocation.ifPresent(location -> properties.put(DATA_LOCATION_PROPERTY, location));
 
+        String metadataDeleteAfterCommitEnabled = icebergTable.properties().get(METADATA_DELETE_AFTER_COMMIT_ENABLED);
+        if (metadataDeleteAfterCommitEnabled != null) {
+            properties.put(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY, Boolean.parseBoolean(metadataDeleteAfterCommitEnabled));
+        }
+        String metadataPreviousVersionsMax = icebergTable.properties().get(METADATA_PREVIOUS_VERSIONS_MAX);
+        if (metadataPreviousVersionsMax != null) {
+            properties.put(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY, Integer.parseInt(metadataPreviousVersionsMax));
+        }
+
         return properties.buildOrThrow();
     }
 
@@ -882,6 +895,13 @@ public static Map<String, String> createTableProperties(ConnectorTableMetadata t
             }
         }
 
+        if (tableMetadata.getProperties().containsKey(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)) {
+            propertiesBuilder.put(METADATA_DELETE_AFTER_COMMIT_ENABLED, ((Boolean) tableMetadata.getProperties().get(METADATA_DELETE_AFTER_COMMIT_ENABLED_PROPERTY)).toString());
+        }
+        if (tableMetadata.getProperties().containsKey(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)) {
+            propertiesBuilder.put(METADATA_PREVIOUS_VERSIONS_MAX, ((Integer) tableMetadata.getProperties().get(METADATA_PREVIOUS_VERSIONS_MAX_PROPERTY)).toString());
+        }
+
         if (tableMetadata.getComment().isPresent()) {
             propertiesBuilder.put(TABLE_COMMENT, tableMetadata.getComment().get());
         }
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/file/FileMetastoreTableOperations.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/file/FileMetastoreTableOperations.java
@@ -35,6 +35,7 @@
 import static io.trino.plugin.iceberg.IcebergTableName.tableNameFrom;
 import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;
 import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;
+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;
 
 @NotThreadSafe
 public class FileMetastoreTableOperations
@@ -56,7 +57,7 @@ public FileMetastoreTableOperations(
     protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
     {
         Table currentTable = getTable();
-        commitTableUpdate(currentTable, metadata, (table, newMetadataLocation) -> Table.builder(table)
+        commitTableUpdate(currentTable, base, metadata, (table, newMetadataLocation) -> Table.builder(table)
                 .apply(builder -> updateMetastoreTable(builder, metadata, newMetadataLocation, Optional.of(currentMetadataLocation)))
                 .build());
     }
@@ -65,12 +66,12 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
     protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata metadata)
     {
         Table materializedView = getTable(database, tableNameFrom(tableName));
-        commitTableUpdate(materializedView, metadata, (table, newMetadataLocation) -> Table.builder(table)
+        commitTableUpdate(materializedView, base, metadata, (table, newMetadataLocation) -> Table.builder(table)
                 .apply(builder -> builder.setParameter(METADATA_LOCATION_PROP, newMetadataLocation).setParameter(PREVIOUS_METADATA_LOCATION_PROP, currentMetadataLocation))
                 .build());
     }
 
-    private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)
+    private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)
     {
         checkState(currentMetadataLocation != null, "No current metadata location for existing table");
         String metadataLocation = table.getParameters().get(METADATA_LOCATION_PROP);
@@ -97,5 +98,6 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T
             }
             throw new CommitStateUnknownException(e);
         }
+        deleteRemovedMetadataFiles(io(), base, metadata);
     }
 }
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperations.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperations.java
@@ -62,6 +62,7 @@
 import static java.util.Objects.requireNonNull;
 import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;
 import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;
+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;
 
 public class GlueIcebergTableOperations
         extends AbstractIcebergTableOperations
@@ -161,6 +162,7 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
     {
         commitTableUpdate(
                 getTable(database, tableName, false),
+                base,
                 metadata,
                 (table, newMetadataLocation) ->
                         getTableInput(
@@ -179,6 +181,7 @@ protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata m
     {
         commitTableUpdate(
                 getTable(database, tableNameFrom(tableName), false),
+                base,
                 metadata,
                 (table, newMetadataLocation) -> {
                     Map<String, String> parameters = new HashMap<>(getTableParameters(table));
@@ -193,7 +196,7 @@ protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata m
                 });
     }
 
-    private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, TableInput> tableUpdateFunction)
+    private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, TableInput> tableUpdateFunction)
     {
         String newMetadataLocation = writeNewMetadata(metadata, version.orElseThrow() + 1);
         TableInput tableInput = tableUpdateFunction.apply(table, newMetadataLocation);
@@ -218,6 +221,7 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T
             // regardless of the exception thrown (e.g. : timeout exception) or it actually failed
             throw new CommitStateUnknownException(e);
         }
+        deleteRemovedMetadataFiles(io(), base, metadata);
         shouldRefresh = true;
     }
 
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/hms/HiveMetastoreTableOperations.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/hms/HiveMetastoreTableOperations.java
@@ -39,6 +39,7 @@
 import static java.util.Objects.requireNonNull;
 import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;
 import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;
+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;
 
 @NotThreadSafe
 public class HiveMetastoreTableOperations
@@ -65,7 +66,7 @@ public HiveMetastoreTableOperations(
     protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
     {
         Table currentTable = getTable();
-        commitTableUpdate(currentTable, metadata, (table, newMetadataLocation) -> Table.builder(table)
+        commitTableUpdate(currentTable, base, metadata, (table, newMetadataLocation) -> Table.builder(table)
                 .apply(builder -> updateMetastoreTable(builder, metadata, newMetadataLocation, Optional.of(currentMetadataLocation)))
                 .build());
     }
@@ -74,14 +75,14 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
     protected final void commitMaterializedViewRefresh(TableMetadata base, TableMetadata metadata)
     {
         Table materializedView = getTable(database, tableNameFrom(tableName));
-        commitTableUpdate(materializedView, metadata, (table, newMetadataLocation) -> Table.builder(table)
+        commitTableUpdate(materializedView, base, metadata, (table, newMetadataLocation) -> Table.builder(table)
                 .apply(builder -> builder
                         .setParameter(METADATA_LOCATION_PROP, newMetadataLocation)
                         .setParameter(PREVIOUS_METADATA_LOCATION_PROP, currentMetadataLocation))
                 .build());
     }
 
-    private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)
+    private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)
     {
         String newMetadataLocation = writeNewMetadata(metadata, version.orElseThrow() + 1);
         long lockId = thriftMetastore.acquireTableExclusiveLock(
@@ -126,6 +127,7 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T
             }
         }
 
+        deleteRemovedMetadataFiles(io(), base, metadata);
         shouldRefresh = true;
     }
 }
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/IcebergJdbcTableOperations.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/IcebergJdbcTableOperations.java
@@ -24,6 +24,7 @@
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static java.util.Objects.requireNonNull;
+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;
 
 public class IcebergJdbcTableOperations
         extends AbstractIcebergTableOperations
@@ -65,6 +66,7 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
         checkState(currentMetadataLocation != null, "No current metadata location for existing table");
         String newMetadataLocation = writeNewMetadata(metadata, version.orElseThrow() + 1);
         jdbcClient.alterTable(database, tableName, newMetadataLocation, currentMetadataLocation);
+        deleteRemovedMetadataFiles(io(), base, metadata);
         shouldRefresh = true;
     }
 
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/IcebergNessieTableOperations.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/IcebergNessieTableOperations.java
@@ -38,6 +38,7 @@
 import static io.trino.plugin.iceberg.catalog.nessie.IcebergNessieUtil.toIdentifier;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;
 
 public class IcebergNessieTableOperations
         extends AbstractIcebergTableOperations
@@ -149,6 +150,7 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)
             // CommitFailedException is handled as a special case in the Iceberg library. This commit will automatically retry
             throw new CommitFailedException(e, "Cannot commit: ref hash is out of date. Update the ref '%s' and try again", nessieClient.refName());
         }
+        deleteRemovedMetadataFiles(io(), base, metadata);
         shouldRefresh = true;
     }
 
diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@`
`35`	`35`	`import static io.trino.plugin.iceberg.IcebergTableName.tableNameFrom;`
`36`	`36`	`import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;`
`37`	`37`	`import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;`
	`38`	`+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;`
`38`	`39`
`39`	`40`	`@NotThreadSafe`
`40`	`41`	`public class FileMetastoreTableOperations`
`@@ -56,7 +57,7 @@ public FileMetastoreTableOperations(`
`56`	`57`	`protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`57`	`58`	`{`
`58`	`59`	`Table currentTable = getTable();`
`59`		`- commitTableUpdate(currentTable, metadata, (table, newMetadataLocation) -> Table.builder(table)`
	`60`	`+ commitTableUpdate(currentTable, base, metadata, (table, newMetadataLocation) -> Table.builder(table)`
`60`	`61`	`.apply(builder -> updateMetastoreTable(builder, metadata, newMetadataLocation, Optional.of(currentMetadataLocation)))`
`61`	`62`	`.build());`
`62`	`63`	`}`
`@@ -65,12 +66,12 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`65`	`66`	`protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata metadata)`
`66`	`67`	`{`
`67`	`68`	`Table materializedView = getTable(database, tableNameFrom(tableName));`
`68`		`- commitTableUpdate(materializedView, metadata, (table, newMetadataLocation) -> Table.builder(table)`
	`69`	`+ commitTableUpdate(materializedView, base, metadata, (table, newMetadataLocation) -> Table.builder(table)`
`69`	`70`	`.apply(builder -> builder.setParameter(METADATA_LOCATION_PROP, newMetadataLocation).setParameter(PREVIOUS_METADATA_LOCATION_PROP, currentMetadataLocation))`
`70`	`71`	`.build());`
`71`	`72`	`}`
`72`	`73`
`73`		`- private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)`
	`74`	`+ private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)`
`74`	`75`	`{`
`75`	`76`	`checkState(currentMetadataLocation != null, "No current metadata location for existing table");`
`76`	`77`	`String metadataLocation = table.getParameters().get(METADATA_LOCATION_PROP);`
`@@ -97,5 +98,6 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T`
`97`	`98`	`}`
`98`	`99`	`throw new CommitStateUnknownException(e);`
`99`	`100`	`}`
	`101`	`+ deleteRemovedMetadataFiles(io(), base, metadata);`
`100`	`102`	`}`
`101`	`103`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,6 +62,7 @@`
`62`	`62`	`import static java.util.Objects.requireNonNull;`
`63`	`63`	`import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;`
`64`	`64`	`import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;`
	`65`	`+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;`
`65`	`66`
`66`	`67`	`public class GlueIcebergTableOperations`
`67`	`68`	`extends AbstractIcebergTableOperations`
`@@ -161,6 +162,7 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`161`	`162`	`{`
`162`	`163`	`commitTableUpdate(`
`163`	`164`	`getTable(database, tableName, false),`
	`165`	`+ base,`
`164`	`166`	`metadata,`
`165`	`167`	`(table, newMetadataLocation) ->`
`166`	`168`	`getTableInput(`
`@@ -179,6 +181,7 @@ protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata m`
`179`	`181`	`{`
`180`	`182`	`commitTableUpdate(`
`181`	`183`	`getTable(database, tableNameFrom(tableName), false),`
	`184`	`+ base,`
`182`	`185`	`metadata,`
`183`	`186`	`(table, newMetadataLocation) -> {`
`184`	`187`	`Map<String, String> parameters = new HashMap<>(getTableParameters(table));`
`@@ -193,7 +196,7 @@ protected void commitMaterializedViewRefresh(TableMetadata base, TableMetadata m`
`193`	`196`	`});`
`194`	`197`	`}`
`195`	`198`
`196`		`- private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, TableInput> tableUpdateFunction)`
	`199`	`+ private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, TableInput> tableUpdateFunction)`
`197`	`200`	`{`
`198`	`201`	`String newMetadataLocation = writeNewMetadata(metadata, version.orElseThrow() + 1);`
`199`	`202`	`TableInput tableInput = tableUpdateFunction.apply(table, newMetadataLocation);`
`@@ -218,6 +221,7 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T`
`218`	`221`	`// regardless of the exception thrown (e.g. : timeout exception) or it actually failed`
`219`	`222`	`throw new CommitStateUnknownException(e);`
`220`	`223`	`}`
	`224`	`+ deleteRemovedMetadataFiles(io(), base, metadata);`
`221`	`225`	`shouldRefresh = true;`
`222`	`226`	`}`
`223`	`227`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@`
`39`	`39`	`import static java.util.Objects.requireNonNull;`
`40`	`40`	`import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP;`
`41`	`41`	`import static org.apache.iceberg.BaseMetastoreTableOperations.PREVIOUS_METADATA_LOCATION_PROP;`
	`42`	`+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;`
`42`	`43`
`43`	`44`	`@NotThreadSafe`
`44`	`45`	`public class HiveMetastoreTableOperations`
`@@ -65,7 +66,7 @@ public HiveMetastoreTableOperations(`
`65`	`66`	`protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`66`	`67`	`{`
`67`	`68`	`Table currentTable = getTable();`
`68`		`- commitTableUpdate(currentTable, metadata, (table, newMetadataLocation) -> Table.builder(table)`
	`69`	`+ commitTableUpdate(currentTable, base, metadata, (table, newMetadataLocation) -> Table.builder(table)`
`69`	`70`	`.apply(builder -> updateMetastoreTable(builder, metadata, newMetadataLocation, Optional.of(currentMetadataLocation)))`
`70`	`71`	`.build());`
`71`	`72`	`}`
`@@ -74,14 +75,14 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`74`	`75`	`protected final void commitMaterializedViewRefresh(TableMetadata base, TableMetadata metadata)`
`75`	`76`	`{`
`76`	`77`	`Table materializedView = getTable(database, tableNameFrom(tableName));`
`77`		`- commitTableUpdate(materializedView, metadata, (table, newMetadataLocation) -> Table.builder(table)`
	`78`	`+ commitTableUpdate(materializedView, base, metadata, (table, newMetadataLocation) -> Table.builder(table)`
`78`	`79`	`.apply(builder -> builder`
`79`	`80`	`.setParameter(METADATA_LOCATION_PROP, newMetadataLocation)`
`80`	`81`	`.setParameter(PREVIOUS_METADATA_LOCATION_PROP, currentMetadataLocation))`
`81`	`82`	`.build());`
`82`	`83`	`}`
`83`	`84`
`84`		`- private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)`
	`85`	`+ private void commitTableUpdate(Table table, TableMetadata base, TableMetadata metadata, BiFunction<Table, String, Table> tableUpdateFunction)`
`85`	`86`	`{`
`86`	`87`	`String newMetadataLocation = writeNewMetadata(metadata, version.orElseThrow() + 1);`
`87`	`88`	`long lockId = thriftMetastore.acquireTableExclusiveLock(`
`@@ -126,6 +127,7 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T`
`126`	`127`	`}`
`127`	`128`	`}`
`128`	`129`
	`130`	`+ deleteRemovedMetadataFiles(io(), base, metadata);`
`129`	`131`	`shouldRefresh = true;`
`130`	`132`	`}`
`131`	`133`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@`
`38`	`38`	`import static io.trino.plugin.iceberg.catalog.nessie.IcebergNessieUtil.toIdentifier;`
`39`	`39`	`import static java.lang.String.format;`
`40`	`40`	`import static java.util.Objects.requireNonNull;`
	`41`	`+import static org.apache.iceberg.CatalogUtil.deleteRemovedMetadataFiles;`
`41`	`42`
`42`	`43`	`public class IcebergNessieTableOperations`
`43`	`44`	`extends AbstractIcebergTableOperations`
`@@ -149,6 +150,7 @@ protected void commitToExistingTable(TableMetadata base, TableMetadata metadata)`
`149`	`150`	`// CommitFailedException is handled as a special case in the Iceberg library. This commit will automatically retry`
`150`	`151`	`throw new CommitFailedException(e, "Cannot commit: ref hash is out of date. Update the ref '%s' and try again", nessieClient.refName());`
`151`	`152`	`}`
	`153`	`+ deleteRemovedMetadataFiles(io(), base, metadata);`
`152`	`154`	`shouldRefresh = true;`
`153`	`155`	`}`
`154`	`156`