This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 25b7b0d204 [flink][spark] Support dry_run in drop_global_index
procedure (#8309)
25b7b0d204 is described below
commit 25b7b0d20496c2b6e770d02c8fb1012c9d655acf
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jun 22 18:06:43 2026 +0800
[flink][spark] Support dry_run in drop_global_index procedure (#8309)
---
docs/docs/flink/procedures.md | 12 +-
docs/docs/spark/procedures.md | 5 +-
.../flink/procedure/DropGlobalIndexProcedure.java | 23 +++-
.../procedure/DropGlobalIndexProcedureITCase.java | 147 +++++++++++++++++++++
.../spark/procedure/DropGlobalIndexProcedure.java | 20 ++-
.../procedure/DropGlobalIndexProcedureTest.scala | 50 +++++++
6 files changed, 250 insertions(+), 7 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index a59306df18..9ae1b176b0 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -1053,7 +1053,8 @@ 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:
@@ -1061,6 +1062,7 @@ All available procedures are listed below.
<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>
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 0fd5f733b6..b264c49fc2 100644
--- a/docs/docs/spark/procedures.md
+++ b/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>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
index 92bde693ea..098c30cbe4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
@@ -41,6 +41,8 @@ import org.apache.flink.table.procedure.ProcedureContext;
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 class DropGlobalIndexProcedure extends ProcedureBase
{
@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 class DropGlobalIndexProcedure extends
ProcedureBase {
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 + "'"
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
index a348b5af7e..84de80acd6 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
@@ -115,6 +115,153 @@ public class DropGlobalIndexProcedureITCase extends
CatalogITCaseBase {
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(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
index bd218eb68d..ec8b1a2c85 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
+++
b/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 class DropGlobalIndexProcedure extends BaseProcedure
{
(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 class DropGlobalIndexProcedure extends
BaseProcedure {
"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 class DropGlobalIndexProcedure extends BaseProcedure
{
commit.commit(commitMessages);
}
- return new InternalRow[] {newInternalRow(true)};
+ return new InternalRow[] {newInternalRow(true, (long)
waitDelete.size())};
} catch (Exception e) {
throw new RuntimeException(
String.format(
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/DropGlobalIndexProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/DropGlobalIndexProcedureTest.scala
index fd76da2cb8..57ecb4645f 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/DropGlobalIndexProcedureTest.scala
+++
b/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("""
