[flink] Add paimon procedures in flink generic catalog (apache#2651)

lmagic233 · Jan 8, 2024 · 0aad580 · 0aad580
1 parent 58ee0fd
commit 0aad580
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 0 deletions.
diff --git a/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java b/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
@@ -18,6 +18,8 @@
 
 package org.apache.paimon.flink;
 
+import org.apache.paimon.flink.procedure.ProcedureUtil;
+
 import org.apache.flink.table.catalog.AbstractCatalog;
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CatalogBaseTable;
@@ -36,6 +38,7 @@
 import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
 import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
 import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+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;
@@ -46,6 +49,7 @@
 import org.apache.flink.table.factories.Factory;
 import org.apache.flink.table.factories.FunctionDefinitionFactory;
 import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.table.procedures.Procedure;
 
 import java.util.List;
 import java.util.Optional;
@@ -477,4 +481,24 @@ public List<CatalogColumnStatistics> bulkGetPartitionColumnStatistics(
             return flink.bulkGetPartitionColumnStatistics(tablePath, partitionSpecs);
         }
     }
+
+    /**
+     * Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.17-.
+     */
+    public List<String> listProcedures(String dbName)
+            throws DatabaseNotExistException, CatalogException {
+        if (paimon.databaseExists(dbName)) {
+            return ProcedureUtil.listProcedures();
+        }
+        return flink.listProcedures(dbName);
+    }
+
+    /**
+     * Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.17-.
+     */
+    public Procedure getProcedure(ObjectPath procedurePath)
+            throws ProcedureNotExistException, CatalogException {
+        return ProcedureUtil.getProcedure(paimon.catalog(), procedurePath)
+                .orElse(flink.getProcedure(procedurePath));
+    }
 }
diff --git a/...hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java b/...hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java
@@ -137,4 +137,12 @@ public void testReadPaimonSystemTable() {
 
         assertThat(result).containsExactly(Row.of(1L, 0L, "APPEND"), Row.of(2L, 0L, "APPEND"));
     }
+
+    @Test
+    public void testReadPaimonAllProcedures() {
+        List<Row> result = sql("SHOW PROCEDURES");
+
+        assertThat(result)
+                .contains(Row.of("compact"), Row.of("merge_into"), Row.of("migrate_table"));
+    }
 }