Expire Snapshot and Remove Orphan files for Iceberg together with coordinator-only execute

lib/trino-plugin-toolkit/pom.xml

@@ -89,6 +89,11 @@
             <optional>true</optional>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.errorprone</groupId>
+            <artifactId>error_prone_annotations</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/Procedures.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.base.util;
+
+import com.google.errorprone.annotations.FormatMethod;
+import io.trino.spi.TrinoException;
+
+import static io.trino.spi.StandardErrorCode.INVALID_PROCEDURE_ARGUMENT;
+import static java.lang.String.format;
+
+public final class Procedures
+{
+    private Procedures() {}
+
+    @FormatMethod
+    public static void checkProcedureArgument(boolean condition, String message, Object... args)
+    {
+        if (!condition) {
+            throw new TrinoException(INVALID_PROCEDURE_ARGUMENT, format(message, args));
+        }
+    }
+}

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

@@ -27,12 +27,15 @@
 import static io.trino.plugin.hive.HiveCompressionCodec.ZSTD;
 import static io.trino.plugin.iceberg.CatalogType.HIVE_METASTORE;
 import static io.trino.plugin.iceberg.IcebergFileFormat.ORC;
+import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 public class IcebergConfig
 {
     public static final int FORMAT_VERSION_SUPPORT_MIN = 1;
     public static final int FORMAT_VERSION_SUPPORT_MAX = 2;
+    public static final String EXPIRE_SNAPSHOTS_MIN_RETENTION = "iceberg.expire_snapshots.min-retention";
+    public static final String DELETE_ORPHAN_FILES_MIN_RETENTION = "iceberg.delete_orphan_files.min-retention";
 
     private IcebergFileFormat fileFormat = ORC;
     private HiveCompressionCodec compressionCodec = ZSTD;
@@ -45,6 +48,8 @@ public class IcebergConfig
     private boolean projectionPushdownEnabled = true;
     private Optional<String> hiveCatalogName = Optional.empty();
     private int formatVersion = FORMAT_VERSION_SUPPORT_MIN;
+    private Duration expireSnapshotsMinRetention = new Duration(7, DAYS);
+    private Duration deleteOrphanFilesMinRetention = new Duration(7, DAYS);
 
     public CatalogType getCatalogType()
     {
@@ -202,4 +207,32 @@ public IcebergConfig setFormatVersion(int formatVersion)
         this.formatVersion = formatVersion;
         return this;
     }
+
+    @NotNull
+    public Duration getExpireSnapshotsMinRetention()
+    {
+        return expireSnapshotsMinRetention;
+    }
+
+    @Config(EXPIRE_SNAPSHOTS_MIN_RETENTION)
+    @ConfigDescription("Minimal retention period for expire_snapshot procedure")
+    public IcebergConfig setExpireSnapshotsMinRetention(Duration expireSnapshotsMinRetention)
+    {
+        this.expireSnapshotsMinRetention = expireSnapshotsMinRetention;
+        return this;
+    }
+
+    @NotNull
+    public Duration getDeleteOrphanFilesMinRetention()
+    {
+        return deleteOrphanFilesMinRetention;
+    }
+
+    @Config(DELETE_ORPHAN_FILES_MIN_RETENTION)
+    @ConfigDescription("Minimal retention period for delete_orphan_files procedure")
+    public IcebergConfig setDeleteOrphanFilesMinRetention(Duration deleteOrphanFilesMinRetention)
+    {
+        this.deleteOrphanFilesMinRetention = deleteOrphanFilesMinRetention;
+        return this;
+    }
 }