Flink: Add Flink SQL procedures

api/src/main/java/org/apache/iceberg/catalog/Catalog.java

@@ -32,6 +32,8 @@
 /** A Catalog API for table create, drop, and load operations. */
 public interface Catalog {
 
+  String SYSTEM_DATABASE_NAME = "sys";
+
   /**
    * Return the name for this catalog.
    *

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

@@ -45,13 +45,15 @@
 import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
 import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
 import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
 import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
 import org.apache.flink.table.catalog.exceptions.TableNotExistException;
 import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
 import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
 import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
 import org.apache.flink.table.expressions.Expression;
 import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.procedures.Procedure;
 import org.apache.flink.util.StringUtils;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.DataFile;
@@ -69,6 +71,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.flink.procedure.ProcedureUtil;
 import org.apache.iceberg.flink.util.FlinkAlterTableUtil;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
 import org.apache.iceberg.io.CloseableIterable;
@@ -869,4 +872,14 @@ public CatalogColumnStatistics getPartitionColumnStatistics(
       ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
     return CatalogColumnStatistics.UNKNOWN;
   }
+
+  @Override
+  public Procedure getProcedure(ObjectPath procedurePath)
+      throws ProcedureNotExistException, CatalogException {
+    if (!Catalog.SYSTEM_DATABASE_NAME.equals(procedurePath.getDatabaseName())) {
+      throw new ProcedureNotExistException(icebergCatalog.name(), procedurePath);
+    }
+
+    return ProcedureUtil.getProcedure(procedurePath.getObjectName(), icebergCatalog);
+  }
 }

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/procedure/CreateBranchProcedure.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.flink.procedure;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+
+/**
+ * Create branch procedure for given tag. Usage:
+ *
+ * <p>Create branch from the current snapshot:
+ *
+ * <pre><code>
+ *  CALL sys.create_branch('tableId', 'branchName')
+ * </code></pre>
+ *
+ * Create branch from the tagged snapshot:
+ *
+ * <pre><code>
+ *  CALL sys.create_branch('tableId', 'branchName', 'tagName')
+ * </code></pre>
+ */
+public class CreateBranchProcedure extends ProcedureBase {
+  public static final String PROCEDURE_NAME = "create_branch";
+
+  @ProcedureHint(
+      arguments = {
+        @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "branch", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "tag", type = @DataTypeHint("STRING"), isOptional = true)
+      })
+  public String[] call(
+      ProcedureContext procedureContext, String tableId, String branchName, String tagName)
+      throws NoSuchTableException {
+    Table table = table(tableId);
+    table.refresh();
+
+    createBranch(table, branchName, tagName);
+
+    return new String[] {"Success"};
+  }
+
+  private void createBranch(Table table, String branchName, String tagName)
+      throws NoSuchTableException {
+
+    if (tagName == null || tagName.isEmpty()) {
+      table.manageSnapshots().createBranch(branchName).commit();
+    } else {
+      Snapshot snapshot = table.snapshot(tagName);
+      table.manageSnapshots().createBranch(branchName, snapshot.snapshotId()).commit();
+    }
+  }
+
+  @Override
+  public String procedureName() {
+    return PROCEDURE_NAME;
+  }
+}

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/procedure/CreateTagFromTimestampProcedure.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.flink.procedure;
+
+import java.time.Duration;
+import javax.annotation.Nullable;
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+
+/**
+ * Create a tag for the first snapshot whose commit-time greater than the specified timestamp.
+ * Usage:
+ *
+ * <pre><code>
+ *  CALL sys.create_tag_from_timestamp('tableId', 'tagName', timestamp, 'timeRetained')
+ * </code></pre>
+ */
+public class CreateTagFromTimestampProcedure extends ProcedureBase {
+  public static final String PROCEDURE_NAME = "create_tag_from_timestamp";
+
+  @ProcedureHint(
+      arguments = {
+        @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "tag", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "timestamp", type = @DataTypeHint("BIGINT")),
+        @ArgumentHint(name = "time_retained", type = @DataTypeHint("STRING"), isOptional = true)
+      })
+  @DataTypeHint("ROW< tagName STRING, snapshot BIGINT, `commit_time` BIGINT>")
+  public Row[] call(
+      ProcedureContext procedureContext,
+      String tableId,
+      String tagName,
+      Long timestamp,
+      @Nullable String timeRetained)
+      throws NoSuchTableException {
+    Table table = table(tableId);
+    table.refresh();
+    Snapshot snapshot = createTagFromTimestamp(table, tagName, timestamp, toDuration(timeRetained));
+    return new Row[] {Row.of(tagName, snapshot.snapshotId(), snapshot.timestampMillis())};
+  }
+
+  private Snapshot createTagFromTimestamp(
+      Table table, String tagName, Long timestamp, Duration timeRetained) {
+    Snapshot snapshot = getFirstSnapshotGreaterThanTimestamp(table, timestamp);
+
+    if (snapshot == null) {
+      throw new IllegalArgumentException(
+          "Could not find any snapshot whose commit-time later than " + timestamp);
+    }
+
+    if (timeRetained == null) {
+      table.manageSnapshots().createTag(tagName, snapshot.snapshotId()).commit();
+    } else {
+      table
+          .manageSnapshots()
+          .createTag(tagName, snapshot.snapshotId())
+          .setMaxRefAgeMs(tagName, timeRetained.toMillis())
+          .commit();
+    }
+
+    return snapshot;
+  }
+
+  private Snapshot getFirstSnapshotGreaterThanTimestamp(Table table, Long timestamp) {
+    Iterable<Snapshot> snapshots = table.snapshots();
+    Snapshot target = null;
+    for (Snapshot snapshot : snapshots) {
+      if (snapshot.timestampMillis() > timestamp) {
+        if (target == null || target.timestampMillis() >= snapshot.timestampMillis()) {
+          target = snapshot;
+        }
+      }
+    }
+
+    return target;
+  }
+
+  @Override
+  public String procedureName() {
+    return PROCEDURE_NAME;
+  }
+}

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/procedure/CreateTagProcedure.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.flink.procedure;
+
+import java.time.Duration;
+import javax.annotation.Nullable;
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+
+/**
+ * Create tag procedure. Usage:
+ *
+ * <pre><code>
+ *  CALL sys.create_tag('tableId', 'tagName', snapshotId, 'timeRetained')
+ * </code></pre>
+ */
+public class CreateTagProcedure extends ProcedureBase {
+  public static final String PROCEDURE_NAME = "create_tag";
+
+  @ProcedureHint(
+      arguments = {
+        @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "tag", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "snapshot_id", type = @DataTypeHint("BIGINT"), isOptional = true),
+        @ArgumentHint(name = "time_retained", type = @DataTypeHint("STRING"), isOptional = true)
+      })
+  public String[] call(
+      ProcedureContext procedureContext,
+      String tableId,
+      String tagName,
+      @Nullable Long snapshotId,
+      @Nullable String timeRetained)
+      throws NoSuchTableException {
+    Table table = table(tableId);
+    table.refresh();
+    createTag(table, tagName, snapshotId, toDuration(timeRetained));
+    return new String[] {"Success"};
+  }
+
+  void createTag(Table table, String tagName, Long snapshotId, Duration timeRetained) {
+    Long requiredSnapshotId = snapshotId;
+    if (requiredSnapshotId == null) {
+      requiredSnapshotId = table.currentSnapshot().snapshotId();
+    }
+
+    if (timeRetained == null) {
+      table.manageSnapshots().createTag(tagName, requiredSnapshotId).commit();
+    } else {
+      table
+          .manageSnapshots()
+          .createTag(tagName, requiredSnapshotId)
+          .setMaxRefAgeMs(tagName, timeRetained.toMillis())
+          .commit();
+    }
+  }
+
+  @Override
+  public String procedureName() {
+    return PROCEDURE_NAME;
+  }
+}

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/procedure/DeleteBranchProcedure.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.flink.procedure;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+
+/**
+ * Delete branch procedure. Usage:
+ *
+ * <pre><code>
+ *  CALL sys.delete_branch('tableId', 'branchName')
+ * </code></pre>
+ */
+public class DeleteBranchProcedure extends ProcedureBase {
+  public static final String PROCEDURE_NAME = "delete_branch";
+
+  @ProcedureHint(
+      arguments = {
+        @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+        @ArgumentHint(name = "branch", type = @DataTypeHint("STRING"))
+      })
+  public String[] call(ProcedureContext procedureContext, String tableId, String branchName)
+      throws NoSuchTableException {
+    Table table = table(tableId);
+    table.refresh();
+
+    deleteBranch(table, branchName);
+
+    return new String[] {"Success"};
+  }
+
+  private void deleteBranch(Table table, String branchName) {
+    table.manageSnapshots().removeBranch(branchName).commit();
+  }
+
+  @Override
+  public String procedureName() {
+    return PROCEDURE_NAME;
+  }
+}