Allow skipping of data deletion in expire_snapshots

Author: tbaeg

docs/src/main/sphinx/connector/iceberg.md

@@ -881,7 +881,7 @@ with the `retention_threshold` parameter.
 `expire_snapshots` can be run as follows:
 
 ```sql
-ALTER TABLE test_table EXECUTE expire_snapshots(retention_threshold => '7d');
+ALTER TABLE test_table EXECUTE expire_snapshots(retention_threshold => '7d', delete_files => true);
 ```
 
 The value for `retention_threshold` must be higher than or equal to
@@ -890,6 +890,10 @@ procedure fails with a similar message: `Retention specified (1.00d) is shorter
 than the minimum retention configured in the system (7.00d)`. The default value
 for this property is `7d`.
 
+The value for `delete_files` can be `true` or `false`. When set to `false` 
+files associated with the expired snapshots will NOT be deleted. The default 
+value for this property is `true`.
+
 (iceberg-remove-orphan-files)=
 ##### remove_orphan_files
 

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -416,6 +416,7 @@ public class IcebergMetadata
     private static final int OPTIMIZE_MAX_SUPPORTED_TABLE_VERSION = 2;
     private static final int CLEANING_UP_PROCEDURES_MAX_SUPPORTED_TABLE_VERSION = 2;
     private static final String RETENTION_THRESHOLD = "retention_threshold";
+    private static final String DELETE_FILES = "delete_files";
     private static final String UNKNOWN_SNAPSHOT_TOKEN = "UNKNOWN";
     public static final Set<String> UPDATABLE_TABLE_PROPERTIES = ImmutableSet.<String>builder()
             .add(EXTRA_PROPERTIES_PROPERTY)
@@ -1721,12 +1722,13 @@ private Optional<ConnectorTableExecuteHandle> getTableHandleForDropExtendedStats
     private Optional<ConnectorTableExecuteHandle> getTableHandleForExpireSnapshots(ConnectorSession session, IcebergTableHandle tableHandle, Map<String, Object> executeProperties)
     {
         Duration retentionThreshold = (Duration) executeProperties.get(RETENTION_THRESHOLD);
+        boolean deleteFiles = (boolean) executeProperties.get(DELETE_FILES);
         Table icebergTable = catalog.loadTable(session, tableHandle.getSchemaTableName());
 
         return Optional.of(new IcebergTableExecuteHandle(
                 tableHandle.getSchemaTableName(),
                 EXPIRE_SNAPSHOTS,
-                new IcebergExpireSnapshotsHandle(retentionThreshold),
+                new IcebergExpireSnapshotsHandle(retentionThreshold, deleteFiles),
                 icebergTable.location(),
                 icebergTable.io().properties()));
     }
@@ -2198,6 +2200,7 @@ private void executeExpireSnapshots(ConnectorSession session, IcebergTableExecut
             table.expireSnapshots()
                     .expireOlderThan(expireTimestampMillis)
                     .deleteWith(deleteFunction)
+                    .cleanExpiredFiles(expireSnapshotsHandle.deleteFiles())
                     .commit();
 
             fileSystem.deleteFiles(pathsToDelete);

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/procedure/ExpireSnapshotsTableProcedure.java

@@ -21,6 +21,7 @@
 import static io.trino.plugin.base.session.PropertyMetadataUtil.durationProperty;
 import static io.trino.plugin.iceberg.procedure.IcebergTableProcedureId.EXPIRE_SNAPSHOTS;
 import static io.trino.spi.connector.TableProcedureExecutionMode.coordinatorOnly;
+import static io.trino.spi.session.PropertyMetadata.booleanProperty;
 
 public class ExpireSnapshotsTableProcedure
         implements Provider<TableProcedureMetadata>
@@ -36,6 +37,11 @@ public TableProcedureMetadata get()
                                 "retention_threshold",
                                 "Only snapshots older than threshold should be removed",
                                 Duration.valueOf("7d"),
+                                false),
+                        booleanProperty(
+                                "delete_files",
+                                "Delete underlying files associated to the expired snapshot(s)",
+                                true,
                                 false)));
     }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/procedure/IcebergExpireSnapshotsHandle.java

@@ -17,7 +17,7 @@
 
 import static java.util.Objects.requireNonNull;
 
-public record IcebergExpireSnapshotsHandle(Duration retentionThreshold)
+public record IcebergExpireSnapshotsHandle(Duration retentionThreshold, boolean deleteFiles)
         implements IcebergProcedureHandle
 {
     public IcebergExpireSnapshotsHandle

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -6550,6 +6550,41 @@ public void testExpireSnapshotsPartitionedTable()
         assertThat(updatedSnapshots.size()).isLessThan(initialSnapshots.size());
     }
 
+    @Test
+    public void testExpireSnapshotsKeepFiles()
+            throws Exception
+    {
+        String tableName = "test_expiring_snapshots_" + randomNameSuffix();
+        Session sessionWithShortRetentionUnlocked = prepareCleanUpSession();
+        assertUpdate("CREATE TABLE " + tableName + " (key varchar, value integer)");
+        assertUpdate("INSERT INTO " + tableName + " VALUES ('one', 1)", 1);
+        assertUpdate("INSERT INTO " + tableName + " VALUES ('two', 2)", 1);
+        List<String> initialFiles = getAllDataFilesFromTableDirectory(tableName);
+
+        assertQuerySucceeds(sessionWithShortRetentionUnlocked, "ALTER TABLE " + tableName + " EXECUTE EXPIRE_SNAPSHOTS (retention_threshold => '0s', delete_files => false)");
+
+        List<String> updatedDataFiles = getAllDataFilesFromTableDirectory(tableName);
+        assertThat(updatedDataFiles).containsExactlyInAnyOrderElementsOf(initialFiles);
+    }
+
+    @Test
+    public void testExpireSnapshotsPartitionedTableKeepFiles()
+            throws Exception
+    {
+        String tableName = "test_expiring_snapshots_partitioned_table" + randomNameSuffix();
+        Session sessionWithShortRetentionUnlocked = prepareCleanUpSession();
+        assertUpdate("CREATE TABLE " + tableName + " (col1 BIGINT, col2 BIGINT) WITH (partitioning = ARRAY['col1'])");
+        assertUpdate("INSERT INTO " + tableName + " VALUES(1, 100), (1, 101), (1, 102), (2, 200), (2, 201), (3, 300)", 6);
+        assertUpdate("DELETE FROM " + tableName + " WHERE col1 = 1", 3);
+        assertUpdate("INSERT INTO " + tableName + " VALUES(4, 400)", 1);
+        List<String> initialFiles = getAllDataFilesFromTableDirectory(tableName);
+
+        assertQuerySucceeds(sessionWithShortRetentionUnlocked, "ALTER TABLE " + tableName + " EXECUTE EXPIRE_SNAPSHOTS (retention_threshold => '0s', delete_files => false)");
+
+        List<String> updatedDataFiles = getAllDataFilesFromTableDirectory(tableName);
+        assertThat(updatedDataFiles).containsExactlyInAnyOrderElementsOf(initialFiles);
+    }
+
     @Test
     public void testExpireSnapshotsOnSnapshot()
     {
@@ -6584,7 +6619,7 @@ public void testExplainExpireSnapshotOutput()
         assertUpdate("INSERT INTO " + tableName + " VALUES ('two', 2)", 1);
 
         assertExplain("EXPLAIN ALTER TABLE " + tableName + " EXECUTE EXPIRE_SNAPSHOTS (retention_threshold => '0s')",
-                "SimpleTableExecute\\[table = iceberg:schemaTableName:tpch.test_expiring_snapshots.*\\[retentionThreshold=0\\.00s].*");
+                "SimpleTableExecute\\[table = iceberg:schemaTableName:tpch.test_expiring_snapshots.*\\[retentionThreshold=0\\.00s, deleteFiles=true].*");
     }
 
     @Test