[flink][spark] Support dry_run in drop_global_index procedure (#8309)

Author: XiaoHongbo-Hope

docs/docs/flink/procedures.md

@@ -1053,14 +1053,16 @@ All available procedures are listed below.
             `table` => 'table',<br/>
             `index_column` => 'columnName',<br/>
             `index_type` => 'indexType',<br/>
-            `partitions` => 'partitions')<br/>
+            `partitions` => 'partitions',<br/>
+            `dry_run` => dryRun)<br/>
       </td>
       <td>
          To drop global index files from a table. Arguments:
             <li>table(required): the target table identifier.</li>
             <li>index_column(required): the column name for which to drop the index.</li>
             <li>index_type(required): the type of global index to drop, e.g., 'btree'.</li>
             <li>partitions(optional): partition specification for selective index deletion.</li>
+            <li>dry_run(optional): when true, report how many index files would be dropped without committing any change. Default is false.</li>
       </td>
       <td>
          -- Drop all btree indexes for column 'name'<br/>
@@ -1073,7 +1075,13 @@ All available procedures are listed below.
             `table` => 'default.T',<br/>
             `index_column` => 'name',<br/>
             `index_type` => 'btree',<br/>
-            `partitions` => 'pt=p1;pt=p2')
+            `partitions` => 'pt=p1;pt=p2')<br/><br/>
+         -- Preview what would be dropped without deleting<br/>
+         CALL sys.drop_global_index(<br/>
+            `table` => 'default.T',<br/>
+            `index_column` => 'name',<br/>
+            `index_type` => 'btree',<br/>
+            `dry_run` => true)
       </td>
    </tr>
    <tr>

docs/docs/spark/procedures.md

@@ -538,9 +538,12 @@ This section introduce all available spark procedures about paimon.
             <li>index_column: the name of the indexed column. Cannot be empty.</li>
             <li>index_type: type of the index to drop, e.g. 'btree'. Cannot be empty.</li>
             <li>partitions: partition filter to limit the partitions from which to drop the index. The comma (",") represents "AND", the semicolon (";") represents "OR". Left empty for all partitions.</li>
+            <li>dry_run: when true, return the number of index files that would be dropped without committing any change. Default is false.</li>
       </td>
       <td>
-         CALL sys.drop_global_index(table => 'default.T', index_column => 'name', index_type => 'btree', partitions => 'pt=p1')
+         CALL sys.drop_global_index(table => 'default.T', index_column => 'name', index_type => 'btree', partitions => 'pt=p1')<br/><br/>
+         -- Preview what would be dropped without deleting<br/>
+         CALL sys.drop_global_index(table => 'default.T', index_column => 'name', index_type => 'btree', dry_run => true)
       </td>
    </tr>
    <tr>

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java

@@ -41,6 +41,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -71,14 +73,16 @@ public String identifier() {
                 @ArgumentHint(
                         name = "partitions",
                         type = @DataTypeHint("STRING"),
-                        isOptional = true)
+                        isOptional = true),
+                @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true)
             })
     public String[] call(
             ProcedureContext procedureContext,
             String tableId,
             String indexColumn,
             String indexType,
-            String partitions)
+            @Nullable String partitions,
+            @Nullable Boolean dryRun)
             throws Exception {
 
         FileStoreTable table = (FileStoreTable) table(tableId);
@@ -142,6 +146,21 @@ public String[] call(
                 columnsDesc,
                 table.name());
 
+        // Dry run: report what would be dropped without committing any change.
+        if (dryRun != null && dryRun) {
+            return new String[] {
+                "Dry run: "
+                        + waitToDelete.size()
+                        + " "
+                        + indexTypeLower
+                        + " global index files would be dropped for columns '"
+                        + columnsDesc
+                        + "' on table '"
+                        + table.name()
+                        + "'"
+            };
+        }
+
         if (waitToDelete.isEmpty()) {
             return new String[] {
                 "No " + indexTypeLower + " global index found for columns '" + columnsDesc + "'"

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java

@@ -115,6 +115,153 @@ public void testDropBtreeGlobalIndex() throws Exception {
         assertThat(btreeEntries).isEmpty();
     }
 
+    @Test
+    public void testDropGlobalIndexDryRun() throws Exception {
+        sql(
+                "CREATE TABLE T ("
+                        + " id INT,"
+                        + " name STRING"
+                        + ") WITH ("
+                        + " 'bucket' = '-1',"
+                        + " 'global-index.row-count-per-shard' = '10000',"
+                        + " 'row-tracking.enabled' = 'true',"
+                        + " 'data-evolution.enabled' = 'true'"
+                        + ")");
+
+        FileStoreTable table = paimonTable("T");
+        BatchWriteBuilder builder = table.newBatchWriteBuilder();
+        try (BatchTableWrite batchTableWrite = builder.newWrite()) {
+            for (int i = 0; i < 100000; i++) {
+                batchTableWrite.write(GenericRow.of(i, BinaryString.fromString("name_" + i)));
+            }
+            List<CommitMessage> commitMessages = batchTableWrite.prepareCommit();
+            BatchTableCommit commit = builder.newCommit();
+            commit.commit(commitMessages);
+            commit.close();
+        }
+
+        tEnv.getConfig()
+                .set(org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYNC, true);
+        sql(
+                "CALL sys.create_global_index(`table` => 'default.T', "
+                        + "`index_column` => 'name', "
+                        + "`index_type` => 'btree')");
+        table = paimonTable("T");
+        List<IndexManifestEntry> btreeEntries =
+                table.store().newIndexFileHandler().scanEntries().stream()
+                        .filter(entry -> entry.indexFile().indexType().equals("btree"))
+                        .collect(Collectors.toList());
+        assertThat(btreeEntries).isNotEmpty();
+
+        // Dry run: should report how many would be dropped, but keep the index intact.
+        List<Row> dryRunResult =
+                sql(
+                        "CALL sys.drop_global_index(`table` => 'default.T', "
+                                + "`index_column` => 'name', "
+                                + "`index_type` => 'btree', "
+                                + "`dry_run` => true)");
+        assertThat(dryRunResult).hasSize(1);
+        assertThat(dryRunResult.get(0).getField(0))
+                .isInstanceOf(String.class)
+                .asString()
+                .contains("Dry run")
+                .contains(String.valueOf(btreeEntries.size()))
+                .contains("btree")
+                .contains("name");
+
+        // Index files must still be present after a dry run.
+        table = paimonTable("T");
+        List<IndexManifestEntry> afterDryRun =
+                table.store().newIndexFileHandler().scanEntries().stream()
+                        .filter(entry -> entry.indexFile().indexType().equals("btree"))
+                        .collect(Collectors.toList());
+        assertThat(afterDryRun).hasSameSizeAs(btreeEntries);
+    }
+
+    @Test
+    public void testDropGlobalIndexDryRunWithPartition() throws Exception {
+        sql(
+                "CREATE TABLE T ("
+                        + " id INT,"
+                        + " name STRING,"
+                        + " pt STRING"
+                        + ") PARTITIONED BY (pt) WITH ("
+                        + " 'bucket' = '-1',"
+                        + " 'global-index.row-count-per-shard' = '10000',"
+                        + " 'row-tracking.enabled' = 'true',"
+                        + " 'data-evolution.enabled' = 'true'"
+                        + ")");
+
+        FileStoreTable table = paimonTable("T");
+        BatchWriteBuilder builder = table.newBatchWriteBuilder();
+        try (BatchTableWrite batchTableWrite = builder.newWrite()) {
+            for (int i = 0; i < 20000; i++) {
+                batchTableWrite.write(
+                        GenericRow.of(
+                                i,
+                                BinaryString.fromString("name_" + i),
+                                BinaryString.fromString("p0")));
+            }
+            for (int i = 0; i < 20000; i++) {
+                batchTableWrite.write(
+                        GenericRow.of(
+                                i,
+                                BinaryString.fromString("name_" + i),
+                                BinaryString.fromString("p1")));
+            }
+            List<CommitMessage> commitMessages = batchTableWrite.prepareCommit();
+            BatchTableCommit commit = builder.newCommit();
+            commit.commit(commitMessages);
+            commit.close();
+        }
+
+        tEnv.getConfig()
+                .set(org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYNC, true);
+        sql(
+                "CALL sys.create_global_index(`table` => 'default.T', "
+                        + "`index_column` => 'name', "
+                        + "`index_type` => 'btree')");
+
+        table = paimonTable("T");
+        List<IndexManifestEntry> before =
+                table.store().newIndexFileHandler().scanEntries().stream()
+                        .filter(entry -> entry.indexFile().indexType().equals("btree"))
+                        .collect(Collectors.toList());
+        assertThat(before).isNotEmpty();
+
+        // Dry run scoped to one partition.
+        String partitionMsg =
+                (String)
+                        sql("CALL sys.drop_global_index(`table` => 'default.T', "
+                                        + "`index_column` => 'name', "
+                                        + "`index_type` => 'btree', "
+                                        + "`partitions` => 'pt=p1', "
+                                        + "`dry_run` => true)")
+                                .get(0)
+                                .getField(0);
+        assertThat(partitionMsg).contains("Dry run").contains("btree");
+
+        // Dry run over all partitions reports a different (larger) count, proving the
+        // partition filter narrows the preview.
+        String allMsg =
+                (String)
+                        sql("CALL sys.drop_global_index(`table` => 'default.T', "
+                                        + "`index_column` => 'name', "
+                                        + "`index_type` => 'btree', "
+                                        + "`dry_run` => true)")
+                                .get(0)
+                                .getField(0);
+        assertThat(allMsg).contains("Dry run");
+        assertThat(partitionMsg).isNotEqualTo(allMsg);
+
+        // Neither dry run committed anything.
+        List<IndexManifestEntry> after =
+                table.store().newIndexFileHandler().scanEntries().stream()
+                        .filter(entry -> entry.indexFile().indexType().equals("btree"))
+                        .collect(Collectors.toList());
+        assertThat(after).hasSameSizeAs(before);
+    }
+
     @Test
     public void testDropBtreeGlobalIndexWithPartition() throws Exception {
         sql(

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java

@@ -67,12 +67,15 @@ public class DropGlobalIndexProcedure extends BaseProcedure {
                 ProcedureParameter.required("index_column", DataTypes.StringType),
                 ProcedureParameter.required("index_type", DataTypes.StringType),
                 ProcedureParameter.optional("partitions", StringType),
+                ProcedureParameter.optional("dry_run", DataTypes.BooleanType),
             };
 
     private static final StructType OUTPUT_TYPE =
             new StructType(
                     new StructField[] {
-                        new StructField("result", DataTypes.BooleanType, true, Metadata.empty())
+                        new StructField("result", DataTypes.BooleanType, true, Metadata.empty()),
+                        new StructField(
+                                "dropped_file_count", DataTypes.LongType, true, Metadata.empty())
                     });
 
     protected DropGlobalIndexProcedure(TableCatalog tableCatalog) {
@@ -103,6 +106,7 @@ public InternalRow[] call(InternalRow args) {
                 (args.isNullAt(3) || StringUtils.isNullOrWhitespaceOnly(args.getString(3)))
                         ? null
                         : args.getString(3);
+        boolean dryRun = !args.isNullAt(4) && args.getBoolean(4);
 
         String finalWhere = partitions != null ? SparkProcedureUtils.toWhere(partitions) : null;
 
@@ -172,6 +176,18 @@ public InternalRow[] call(InternalRow args) {
                                 "Waiting for global index to be deleted size: "
                                         + waitDelete.size());
 
+                        // Dry run: report how many would be dropped, commit nothing.
+                        if (dryRun) {
+                            return new InternalRow[] {
+                                newInternalRow(true, (long) waitDelete.size())
+                            };
+                        }
+
+                        // Nothing matched: avoid committing an empty change.
+                        if (waitDelete.isEmpty()) {
+                            return new InternalRow[] {newInternalRow(true, 0L)};
+                        }
+
                         Map<BinaryRow, List<IndexFileMeta>> deleteEntries =
                                 waitDelete.stream()
                                         .map(IndexManifestEntry::toDeleteEntry)
@@ -202,7 +218,7 @@ public InternalRow[] call(InternalRow args) {
                             commit.commit(commitMessages);
                         }
 
-                        return new InternalRow[] {newInternalRow(true)};
+                        return new InternalRow[] {newInternalRow(true, (long) waitDelete.size())};
                     } catch (Exception e) {
                         throw new RuntimeException(
                                 String.format(

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/DropGlobalIndexProcedureTest.scala

@@ -63,12 +63,14 @@ class DropGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest {
       val totalRowCount = btreeEntries.map(_.indexFile().rowCount()).sum
       assert(totalRowCount == 100000L)
 
+      val droppedCount = btreeEntries.size
       output = spark
         .sql("CALL sys.drop_global_index(table => 'test.T', index_column => 'name', index_type => 'btree')")
         .collect()
         .head
 
       assert(output.getBoolean(0))
+      assert(output.getLong(1) == droppedCount)
 
       table = loadTable("T")
       btreeEntries = table
@@ -81,6 +83,54 @@ class DropGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest {
     }
   }
 
+  test("drop btree global index dry run") {
+    withTable("T") {
+      spark.sql("""
+                  |CREATE TABLE T (id INT, name STRING)
+                  |TBLPROPERTIES (
+                  |  'bucket' = '-1',
+                  |  'global-index.row-count-per-shard' = '10000',
+                  |  'row-tracking.enabled' = 'true',
+                  |  'data-evolution.enabled' = 'true')
+                  |""".stripMargin)
+
+      val values =
+        (0 until 100000).map(i => s"($i, 'name_$i')").mkString(",")
+      spark.sql(s"INSERT INTO T VALUES $values")
+
+      spark
+        .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'name', index_type => 'btree')")
+        .collect()
+
+      var table = loadTable("T")
+      val before = table
+        .store()
+        .newIndexFileHandler()
+        .scanEntries()
+        .asScala
+        .filter(_.indexFile().indexType() == "btree")
+      assert(before.nonEmpty)
+
+      // Dry run: reports the would-drop count but commits nothing.
+      val output = spark
+        .sql("CALL sys.drop_global_index(table => 'test.T', index_column => 'name', index_type => 'btree', dry_run => true)")
+        .collect()
+        .head
+      assert(output.getBoolean(0))
+      assert(output.getLong(1) == before.size)
+
+      // Index files must still be present after a dry run.
+      table = loadTable("T")
+      val after = table
+        .store()
+        .newIndexFileHandler()
+        .scanEntries()
+        .asScala
+        .filter(_.indexFile().indexType() == "btree")
+      assert(after.size == before.size)
+    }
+  }
+
   test("create btree global index with partition") {
     withTable("T") {
       spark.sql("""