[flink] Procedure sorting compact a table without partitions filter. (#…

…4405)
apache · Oct 30, 2024 · 1488e1c · 1488e1c
1 parent 74634b9
commit 1488e1c
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 0 deletions.
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
@@ -71,6 +71,7 @@ All available procedures are listed below.
          -- Use indexed argument<br/>
          CALL [catalog.]sys.compact('table') <br/><br/>
          CALL [catalog.]sys.compact('table', 'partitions') <br/><br/> 
+         CALL [catalog.]sys.compact('table', 'order_strategy', 'order_by') <br/><br/>
          CALL [catalog.]sys.compact('table', 'partitions', 'order_strategy', 'order_by') <br/><br/>
          CALL [catalog.]sys.compact('table', 'partitions', 'order_strategy', 'order_by', 'options') <br/><br/>
          CALL [catalog.]sys.compact('table', 'partitions', 'order_strategy', 'order_by', 'options', 'where') <br/><br/>

diff --git a/...k/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java b/...k/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java
@@ -44,6 +44,9 @@
  *  CALL sys.compact('tableId', 'pt1=A,pt2=a;pt1=B,pt2=b')
  *
  *  -- compact a table with sorting
+ *  CALL sys.compact('tableId', 'ORDER/ZORDER', 'col1,col2')
+ *
+ *  -- compact specific partitions with sorting
  *  CALL sys.compact('tableId', 'partitions', 'ORDER/ZORDER', 'col1,col2', 'sink.parallelism=6')
  *
  * </code></pre>
@@ -61,6 +64,15 @@ public String[] call(ProcedureContext procedureContext, String tableId, String p
         return call(procedureContext, tableId, partitions, "", "", "", "");
     }
 
+    public String[] call(
+            ProcedureContext procedureContext,
+            String tableId,
+            String orderStrategy,
+            String orderByColumns)
+            throws Exception {
+        return call(procedureContext, tableId, "", orderStrategy, orderByColumns, "", "");
+    }
+
     public String[] call(
             ProcedureContext procedureContext,
             String tableId,

diff --git a/...8/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java b/...8/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
@@ -56,6 +56,8 @@ public void testCompactDatabaseAndTable() {
         assertThatCode(() -> sql("CALL sys.compact('default.T')")).doesNotThrowAnyException();
         assertThatCode(() -> sql("CALL sys.compact('default.T', 'pt=1')"))
                 .doesNotThrowAnyException();
+        assertThatCode(() -> sql("CALL sys.compact('default.T', '', '')"))
+                .doesNotThrowAnyException();
         assertThatCode(() -> sql("CALL sys.compact('default.T', 'pt=1', '', '')"))
                 .doesNotThrowAnyException();
         assertThatCode(() -> sql("CALL sys.compact('default.T', '', '', '', 'sink.parallelism=1')"))