feat(index): support dry_run in drop_global_index

Author: JunRuiLee

crates/integrations/datafusion/src/procedures.rs

@@ -26,7 +26,7 @@
 //! - `CALL sys.create_global_index(table => '...', index_column => '...', index_type => 'btree')`
 //! - `CALL sys.create_global_index(table => '...', index_column => '...', index_type => 'bitmap')`
 //! - `CALL sys.create_global_index(table => '...', index_column => '...', index_type => 'ivf-pq')`
-//! - `CALL sys.drop_global_index(table => '...', index_column => '...', index_type => 'btree')` (also 'bitmap', 'lumina', or a vindex type such as 'ivf-pq')
+//! - `CALL sys.drop_global_index(table => '...', index_column => '...', index_type => 'btree')` (also 'bitmap', 'lumina', or a vindex type such as 'ivf-pq'; optional `dry_run => true` previews the matched count without committing)
 //! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
 
 use std::collections::HashMap;
@@ -599,17 +599,27 @@ async fn proc_drop_global_index(
             "drop_global_index partitions are not supported yet".to_string(),
         ));
     }
-    if args.contains_key("dry_run") {
-        return Err(DataFusionError::NotImplemented(
-            "drop_global_index dry_run is not supported yet".to_string(),
-        ));
-    }
+    let dry_run = match args.get("dry_run") {
+        None => false,
+        Some(v) if v.eq_ignore_ascii_case("true") => true,
+        Some(v) if v.eq_ignore_ascii_case("false") => false,
+        Some(v) => {
+            return Err(DataFusionError::Plan(format!(
+                "drop_global_index dry_run must be 'true' or 'false', got '{v}'"
+            )));
+        }
+    };
 
     let mut builder = table.new_global_index_drop_builder();
     builder.with_index_column(index_column);
     builder.with_index_type(index_type);
-    builder.execute().await.map_err(to_datafusion_error)?;
-    ok_result(ctx)
+    builder.with_dry_run(dry_run);
+    let matched = builder.execute().await.map_err(to_datafusion_error)?;
+    if dry_run {
+        message_result(ctx, format!("Would drop {matched} global index file(s)"))
+    } else {
+        ok_result(ctx)
+    }
 }
 
 fn is_sorted_global_index_type(index_type: &str) -> bool {
@@ -636,15 +646,16 @@ fn parse_key_value_options(options: &str) -> DFResult<HashMap<String, String>> {
 }
 
 fn ok_result(ctx: &SessionContext) -> DFResult<DataFrame> {
+    message_result(ctx, "OK".to_string())
+}
+
+fn message_result(ctx: &SessionContext, message: String) -> DFResult<DataFrame> {
     let schema = Arc::new(Schema::new(vec![Field::new(
         "result",
         ArrowDataType::Utf8,
         false,
     )]));
-    let batch = RecordBatch::try_new(
-        schema.clone(),
-        vec![Arc::new(StringArray::from(vec!["OK"]))],
-    )?;
+    let batch = RecordBatch::try_new(schema, vec![Arc::new(StringArray::from(vec![message]))])?;
     ctx.read_batch(batch)
 }
 

crates/integrations/datafusion/tests/procedures.rs

@@ -325,6 +325,60 @@ async fn test_drop_global_index_removes_btree_and_reads_fallback() {
     assert_eq!(rows, vec![(2, "bob".to_string()), (3, "carol".to_string())]);
 }
 
+#[tokio::test]
+async fn test_drop_global_index_dry_run_previews_without_removing() {
+    let (_tmp, sql_context) = setup_btree_global_index_table("btree_dry_run").await;
+    exec(
+        &sql_context,
+        "INSERT INTO paimon.test_db.btree_dry_run (id, name) VALUES (1, 'alice'), (2, 'bob')",
+    )
+    .await;
+    exec(
+        &sql_context,
+        "CALL sys.create_global_index(table => 'test_db.btree_dry_run', index_column => 'id')",
+    )
+    .await;
+
+    // dry-run: must NOT remove the index.
+    exec(
+        &sql_context,
+        "CALL sys.drop_global_index(table => 'test_db.btree_dry_run', index_column => 'id', index_type => 'btree', dry_run => true)",
+    )
+    .await;
+    let after_dry = row_count(
+        &sql_context,
+        "SELECT * FROM paimon.test_db.`btree_dry_run$table_indexes` \
+         WHERE index_type = 'btree' AND index_field_name = 'id'",
+    )
+    .await;
+    assert_eq!(after_dry, 1);
+
+    // real drop removes it.
+    exec(
+        &sql_context,
+        "CALL sys.drop_global_index(table => 'test_db.btree_dry_run', index_column => 'id', index_type => 'btree')",
+    )
+    .await;
+    let after_real = row_count(
+        &sql_context,
+        "SELECT * FROM paimon.test_db.`btree_dry_run$table_indexes` \
+         WHERE index_type = 'btree' AND index_field_name = 'id'",
+    )
+    .await;
+    assert_eq!(after_real, 0);
+}
+
+#[tokio::test]
+async fn test_drop_global_index_rejects_invalid_dry_run() {
+    let (_tmp, sql_context) = setup_btree_global_index_table("btree_bad_dry_run").await;
+    assert_sql_error(
+        &sql_context,
+        "CALL sys.drop_global_index(table => 'test_db.btree_bad_dry_run', index_column => 'id', dry_run => 'maybe')",
+        "dry_run",
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn test_drop_global_index_removes_bitmap() {
     let (_tmp, sql_context) = setup_btree_global_index_table("bitmap_drop").await;

crates/paimon/src/table/global_index_drop_builder.rs

@@ -28,6 +28,7 @@ pub struct GlobalIndexDropBuilder<'a> {
     table: &'a Table,
     index_column: Option<String>,
     index_type: String,
+    dry_run: bool,
 }
 
 impl<'a> GlobalIndexDropBuilder<'a> {
@@ -36,6 +37,7 @@ impl<'a> GlobalIndexDropBuilder<'a> {
             table,
             index_column: None,
             index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
+            dry_run: false,
         }
     }
 
@@ -49,6 +51,11 @@ impl<'a> GlobalIndexDropBuilder<'a> {
         self
     }
 
+    pub fn with_dry_run(&mut self, dry_run: bool) -> &mut Self {
+        self.dry_run = dry_run;
+        self
+    }
+
     pub async fn execute(&self) -> Result<usize> {
         let index_type =
             normalize_global_index_type_for_drop(&self.index_type).ok_or_else(|| {
@@ -108,6 +115,9 @@ impl<'a> GlobalIndexDropBuilder<'a> {
                 .or_default()
                 .push(entry.index_file);
         }
+        if self.dry_run {
+            return Ok(dropped);
+        }
         if dropped == 0 {
             return Ok(0);
         }
@@ -477,4 +487,83 @@ mod tests {
                 if message.contains("unsupported global index type") && message.contains("ivf-pq")
         ));
     }
+
+    #[tokio::test]
+    async fn test_dry_run_previews_without_committing() {
+        let table = test_table("memory:/test_drop_dry_run_preview");
+        setup_dirs(&table).await;
+
+        let mut message = CommitMessage::new(
+            BinaryRow::new(0).to_serialized_bytes(),
+            0,
+            vec![data_file("data-0.parquet")],
+        );
+        message.new_index_files = vec![global_index_file(
+            BTREE_GLOBAL_INDEX_TYPE,
+            "btree-id.index",
+            0,
+            0,
+            9,
+        )];
+        TableCommit::new(table.clone(), "test-user".to_string())
+            .commit(vec![message])
+            .await
+            .unwrap();
+
+        // dry-run: returns the matched count, commits nothing.
+        let would_drop = table
+            .new_global_index_drop_builder()
+            .with_index_column("id")
+            .with_dry_run(true)
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(would_drop, 1);
+
+        // Entry is still present (no commit happened).
+        let after_dry = latest_index_entries(&table)
+            .await
+            .into_iter()
+            .map(|e| e.index_file.file_name)
+            .collect::<Vec<_>>();
+        assert_eq!(after_dry, vec!["btree-id.index".to_string()]);
+
+        // A real drop (dry_run defaults false) removes it.
+        let dropped = table
+            .new_global_index_drop_builder()
+            .with_index_column("id")
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(dropped, 1);
+        assert!(latest_index_entries(&table).await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_dry_run_without_match_returns_zero() {
+        let table = test_table("memory:/test_drop_dry_run_no_match");
+        setup_dirs(&table).await;
+
+        let mut message = CommitMessage::new(vec![], 0, vec![data_file("data-0.parquet")]);
+        message.new_index_files = vec![global_index_file(
+            BTREE_GLOBAL_INDEX_TYPE,
+            "btree-name.index",
+            1,
+            0,
+            9,
+        )];
+        TableCommit::new(table.clone(), "test-user".to_string())
+            .commit(vec![message])
+            .await
+            .unwrap();
+
+        let would_drop = table
+            .new_global_index_drop_builder()
+            .with_index_column("id")
+            .with_dry_run(true)
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(would_drop, 0);
+    }
 }