Allow skipping of data deletion in expire_snapshots

Author: tbaeg

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",
+                                "Expire snapshots without deleting the underlying data",
+                                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,44 @@ 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);
+        assertThat(query("SELECT sum(value), listagg(key, ' ') WITHIN GROUP (ORDER BY key) FROM " + tableName))
+                .matches("VALUES (BIGINT '3', VARCHAR 'one two')");
+        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).hasSize(initialFiles.size());
+    }
+
+    @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);
+        assertQuery("SELECT sum(col2) FROM " + tableName, "SELECT 1101");
+        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).hasSize(initialFiles.size());
+    }
+
     @Test
     public void testExpireSnapshotsOnSnapshot()
     {