This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1ca84c0d feat(index): support dry run when dropping global indexes
(#1078)
1ca84c0d is described below
commit 1ca84c0d069f80c2f6715722993d61c29d359c92
Author: QuakeWang <[email protected]>
AuthorDate: Thu Oct 8 20:40:32 2026 +0800
feat(index): support dry run when dropping global indexes (#1078)
---
crates/integrations/datafusion/src/procedures.rs | 22 ++-
crates/integrations/datafusion/tests/procedures.rs | 141 ++++++++++++++-
.../paimon/src/table/global_index_drop_builder.rs | 189 +++++++++++++++++++--
docs/src/sql.md | 19 +++
4 files changed, 351 insertions(+), 20 deletions(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 84f014c7..18416eec 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -176,8 +176,8 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
"rename_branch" => &["table", "from_branch", "to_branch"],
"create_global_index" => &["table", "index_column", "index_type",
"options"],
- // `partitions`/`dry_run` are declared but not yet implemented; they
still reach
- // their own "not supported yet" error rather than being reported as
unknown.
+ // `partitions` is declared but not yet implemented; it still reaches
its
+ // own "not supported yet" error rather than being reported as unknown.
"drop_global_index" => &[
"table",
"index_column",
@@ -810,16 +810,22 @@ 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 = bool_arg(args, "dry_run")?;
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)?;
+ builder.with_dry_run(dry_run);
+ let dropped = builder.execute().await.map_err(to_datafusion_error)?;
+ if dry_run {
+ return utf8_result(
+ ctx,
+ &[("result", false)],
+ vec![vec![Some(format!(
+ "Would drop {dropped} global index file(s)"
+ ))]],
+ );
+ }
ok_result(ctx)
}
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index f2b21646..d89c4547 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -19,12 +19,14 @@ mod common;
use common::{
assert_sql_error, collect_id_name, create_sql_context, create_test_env,
exec, row_count,
- setup_sql_context,
+ setup_sql_context, string_value,
};
use datafusion::arrow::array::{Array, Int64Array, StringArray};
use paimon::catalog::Identifier;
use paimon::table::BranchManager;
use paimon::Catalog;
+use std::collections::BTreeMap;
+use std::path::{Path, PathBuf};
async fn setup_table_with_snapshots() -> (tempfile::TempDir,
paimon_datafusion::SQLContext) {
let (tmp, sql_context) = setup_sql_context().await;
@@ -1009,6 +1011,143 @@ async fn
test_create_full_text_global_index_rejects_non_string_column() {
.await;
}
+fn table_files(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
+ let mut files = BTreeMap::new();
+ for entry in std::fs::read_dir(root).unwrap() {
+ let path = entry.unwrap().path();
+ if path.is_dir() {
+ files.extend(table_files(&path));
+ } else {
+ let content = std::fs::read(&path).unwrap();
+ files.insert(path, content);
+ }
+ }
+ files
+}
+
+#[tokio::test]
+async fn test_drop_global_index_dry_run_then_drop() {
+ for (table_name, dry_run, drop_arg) in [
+ ("dry_run_false", "true", ", dry_run => false"),
+ ("dry_run_default", "' TrUe '", ""),
+ ] {
+ let (tmp, sql_context) =
setup_btree_global_index_table(table_name).await;
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.{table_name} (id, name) VALUES (1,
'alice'), (2, 'bob')"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
+ &format!(
+ "CALL sys.create_global_index(table => 'test_db.{table_name}',
index_column => 'id')"
+ ),
+ )
+ .await;
+ let index_sql = format!("SELECT * FROM
paimon.test_db.`{table_name}$table_indexes`");
+ let matched = row_count(&sql_context, &index_sql).await;
+ assert!(matched > 0);
+ let before = table_files(tmp.path());
+
+ let result = sql_context
+ .sql(&format!(
+ "CALL sys.drop_global_index(table => 'test_db.{table_name}',
index_column => 'id', dry_run => {dry_run})"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column_by_name("result").unwrap().as_ref(),
0),
+ format!("Would drop {matched} global index file(s)")
+ );
+ assert_eq!(row_count(&sql_context, &index_sql).await, matched);
+ assert_eq!(table_files(tmp.path()), before);
+
+ let result = sql_context
+ .sql(&format!(
+ "CALL sys.drop_global_index(table => 'test_db.{table_name}',
index_column => 'id'{drop_arg})"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column_by_name("result").unwrap().as_ref(),
0),
+ "OK"
+ );
+ assert_eq!(row_count(&sql_context, &index_sql).await, 0);
+ assert_ne!(table_files(tmp.path()), before);
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_dry_run_without_match() {
+ let (tmp, sql_context) =
setup_btree_global_index_table("dry_run_empty").await;
+ for insert_data in [false, true] {
+ if insert_data {
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.dry_run_empty (id, name) VALUES
(1, 'alice')",
+ )
+ .await;
+ }
+ let before = table_files(tmp.path());
+ let result = sql_context
+ .sql("CALL sys.drop_global_index(table => 'test_db.dry_run_empty',
index_column => 'id', dry_run => true)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ string_value(result[0].column(0).as_ref(), 0),
+ "Would drop 0 global index file(s)"
+ );
+ assert_eq!(table_files(tmp.path()), before);
+ }
+}
+
+#[tokio::test]
+async fn test_drop_global_index_dry_run_rejects_invalid_args() {
+ let (tmp, sql_context) =
setup_btree_global_index_table("dry_run_invalid").await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.dry_run_invalid (id, name) VALUES (1,
'alice')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table => 'test_db.dry_run_invalid',
index_column => 'id')",
+ )
+ .await;
+ let before = table_files(tmp.path());
+ for (args, expected_error) in [
+ ("dry_run => 'maybe'", "Invalid dry_run"),
+ ("dry_run => 1", "Invalid dry_run"),
+ ("dry_run => ''", "Invalid dry_run"),
+ ("dry_run => NULL", "Unsupported argument value"),
+ (
+ "dry_run => true, partitions => 'pt=p1'",
+ "drop_global_index partitions are not supported yet",
+ ),
+ ] {
+ assert_sql_error(
+ &sql_context,
+ &format!(
+ "CALL sys.drop_global_index(table =>
'test_db.dry_run_invalid', index_column => 'id', {args})"
+ ),
+ expected_error,
+ )
+ .await;
+ assert_eq!(table_files(tmp.path()), before);
+ }
+}
+
#[tokio::test]
async fn test_drop_global_index_removes_btree_and_reads_fallback() {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop").await;
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index 3a21f2f2..155a708e 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -29,6 +29,7 @@ pub struct GlobalIndexDropBuilder<'a> {
table: &'a Table,
index_column: Option<IndexColumns>,
index_type: String,
+ dry_run: bool,
}
impl<'a> GlobalIndexDropBuilder<'a> {
@@ -37,6 +38,7 @@ impl<'a> GlobalIndexDropBuilder<'a> {
table,
index_column: None,
index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
+ dry_run: false,
}
}
@@ -58,6 +60,12 @@ impl<'a> GlobalIndexDropBuilder<'a> {
self
}
+ /// Return the matched index file count without committing when enabled.
+ 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> {
// Dropping an index reads the index manifest.
crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
@@ -135,8 +143,8 @@ impl<'a> GlobalIndexDropBuilder<'a> {
.or_default()
.push(entry.index_file);
}
- if dropped == 0 {
- return Ok(0);
+ if self.dry_run || dropped == 0 {
+ return Ok(dropped);
}
let mut groups = deletions_by_partition_bucket
@@ -182,6 +190,7 @@ mod tests {
use crate::table::TableCommit;
use chrono::{DateTime, Utc};
use indexmap::IndexMap;
+ use std::collections::BTreeMap;
#[tokio::test]
async fn test_query_auth_table_refuses_index_builds_and_drops() {
@@ -195,15 +204,18 @@ mod tests {
);
};
- refused(
- table
- .new_global_index_drop_builder()
- .with_index_type("btree")
- .execute()
- .await
- .unwrap_err(),
- "dropping a global index",
- );
+ for dry_run in [false, true] {
+ refused(
+ table
+ .new_global_index_drop_builder()
+ .with_index_type("btree")
+ .with_dry_run(dry_run)
+ .execute()
+ .await
+ .unwrap_err(),
+ "dropping a global index",
+ );
+ }
refused(
table
.new_btree_global_index_build_builder()
@@ -365,6 +377,161 @@ mod tests {
.unwrap()
}
+ async fn table_files(table: &Table) -> BTreeMap<String, Vec<u8>> {
+ let mut files = BTreeMap::new();
+ for status in table
+ .file_io()
+ .list_status_recursive(&format!("{}/", table.location()))
+ .await
+ .unwrap()
+ {
+ let content = table
+ .file_io()
+ .new_input(&status.path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ files.insert(status.path, content.to_vec());
+ }
+ files
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_dry_run_preserves_files() {
+ for index_type in [
+ "btree",
+ "bitmap",
+ "multivalue",
+ "fm",
+ "full-text",
+ "lumina",
+ "ivf-flat",
+ "ivf-pq",
+ "ivf-sq",
+ "ivf-rq",
+ "diskann",
+ ] {
+ let table = test_table(&format!("memory:/dry_run_{index_type}"));
+ setup_dirs(&table).await;
+ let mut message = CommitMessage::new(vec![], 0,
vec![data_file("data.parquet")]);
+ message.new_index_files = vec![
+ global_index_file(index_type, "id-0.index", 0, 0, 9),
+ global_index_file(index_type, "id-1.index", 0, 10, 19),
+ global_index_file(index_type, "name.index", 1, 0, 19),
+ hash_index_file("hash.index"),
+ ];
+ TableCommit::new(table.clone(), "test-user".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ let before = table_files(&table).await;
+
+ let mut builder = table.new_global_index_drop_builder();
+ builder
+ .with_index_column("id")
+ .with_index_type(index_type)
+ .with_dry_run(true);
+ assert_eq!(builder.execute().await.unwrap(), 2, "{index_type}");
+ assert_eq!(table_files(&table).await, before, "{index_type}");
+ assert_eq!(latest_index_entries(&table).await.len(), 4);
+
+ builder.with_dry_run(false);
+ assert_eq!(builder.execute().await.unwrap(), 2, "{index_type}");
+ assert_eq!(latest_index_entries(&table).await.len(), 2);
+ }
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_dry_run_matches_full_column_list() {
+ let table = test_table("memory:/dry_run_composite");
+ setup_dirs(&table).await;
+ let mut composite = global_index_file("btree", "id-name.index", 0, 0,
9);
+ composite
+ .global_index_meta
+ .as_mut()
+ .unwrap()
+ .extra_field_ids = Some(vec![1]);
+ let mut reversed = global_index_file("btree", "name-id.index", 1, 0,
9);
+ reversed.global_index_meta.as_mut().unwrap().extra_field_ids =
Some(vec![0]);
+ let mut message = CommitMessage::new(vec![], 0,
vec![data_file("data.parquet")]);
+ message.new_index_files = vec![
+ global_index_file("btree", "id.index", 0, 0, 9),
+ composite,
+ reversed,
+ global_index_file("bitmap", "bitmap-id.index", 0, 0, 9),
+ ];
+ TableCommit::new(table.clone(), "test-user".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ let before = table_files(&table).await;
+
+ for columns in ["id", "id,name", "name,id"] {
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_column(columns)
+ .with_dry_run(true)
+ .execute()
+ .await
+ .unwrap(),
+ 1,
+ "{columns}"
+ );
+ }
+ assert_eq!(table_files(&table).await, before);
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_dry_run_without_match() {
+ for (case, index_files) in [
+ ("no_snapshot", None),
+ ("no_index_manifest", Some(vec![])),
+ (
+ "no_match",
+ Some(vec![global_index_file("btree", "name.index", 1, 0, 9)]),
+ ),
+ ] {
+ let table = test_table(&format!("memory:/dry_run_{case}"));
+ setup_dirs(&table).await;
+ if let Some(index_files) = index_files {
+ let mut message = CommitMessage::new(vec![], 0,
vec![data_file("data.parquet")]);
+ message.new_index_files = index_files;
+ TableCommit::new(table.clone(), "test-user".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ }
+ let before = table_files(&table).await;
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_column("id")
+ .with_dry_run(true)
+ .execute()
+ .await
+ .unwrap(),
+ 0,
+ "{case}"
+ );
+ assert_eq!(table_files(&table).await, before, "{case}");
+ }
+ }
+
+ #[tokio::test]
+ async fn test_drop_global_index_dry_run_validates_columns() {
+ let table = test_table("memory:/dry_run_invalid_columns");
+ for column in [None, Some(""), Some(" "), Some(", ,"),
Some("missing")] {
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_dry_run(true);
+ if let Some(column) = column {
+ builder.with_index_column(column);
+ }
+ assert!(builder.execute().await.is_err(), "{column:?}");
+ }
+ }
+
#[tokio::test]
async fn test_drop_btree_global_index_only_removes_target_entries() {
let table = test_table("memory:/test_drop_btree_global_index");
diff --git a/docs/src/sql.md b/docs/src/sql.md
index a37f1774..5bcd622f 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1412,6 +1412,25 @@ To drop a composite BTree definition, pass the same
ordered `index_column`
list used to build it, such as `'category,item_number'`. Scalar indexes and
definitions with a different column order remain available.
+Set `dry_run => true` to return the number of matching index files without
+committing changes to the table:
+
+```sql
+CALL sys.drop_global_index(
+ table => 'paimon.my_db.my_table',
+ index_column => 'id',
+ index_type => 'btree',
+ dry_run => true
+);
+```
+
+The `result` column contains `Would drop N global index file(s)`, including
+when `N` is zero. `dry_run` defaults to `false`; an explicit `false` or an
+omitted argument performs the drop and returns `OK`. Other values are rejected.
+The preview uses the latest snapshot at the time of the call; concurrent
changes
+can affect the number of files matched by a later drop. Reads retain the usual
+local-cache behavior. Partition filtering is not supported by this procedure.
+
### create_lumina_index
Build and commit a Lumina global vector index for a table column:
