From 73d6c02e4b39d462dbf9d3625eaf8c0c614d3c1a Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Wed, 8 Jul 2026 21:43:38 +0800 Subject: [PATCH 1/2] feat(index): support dry_run in drop_global_index --- .../integrations/datafusion/src/procedures.rs | 35 +++++--- .../datafusion/tests/procedures.rs | 54 +++++++++++ .../src/table/global_index_drop_builder.rs | 89 +++++++++++++++++++ 3 files changed, 166 insertions(+), 12 deletions(-) diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 32b7989a..d4de0c98 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/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> { } fn ok_result(ctx: &SessionContext) -> DFResult { + message_result(ctx, "OK".to_string()) +} + +fn message_result(ctx: &SessionContext, message: String) -> DFResult { 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) } diff --git a/crates/integrations/datafusion/tests/procedures.rs b/crates/integrations/datafusion/tests/procedures.rs index 90d4d44e..7f3c9f29 100644 --- a/crates/integrations/datafusion/tests/procedures.rs +++ b/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; diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs index 9d6f557b..338b94d1 100644 --- a/crates/paimon/src/table/global_index_drop_builder.rs +++ b/crates/paimon/src/table/global_index_drop_builder.rs @@ -28,6 +28,7 @@ pub struct GlobalIndexDropBuilder<'a> { table: &'a Table, index_column: Option, 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 { 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::>(); + 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); + } } From 89f545373fdbb1f8fd8d225f7a434efcdca8fc97 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Fri, 10 Jul 2026 19:34:22 +0800 Subject: [PATCH 2/2] fix(index): drop_global_index matches full indexed field list Match global index files by the ordered [index_field_id] + extra_field_ids list instead of the primary field only, consistent with Java DropGlobalIndexProcedure. Fixes over-delete of composite indexes (drop('id') no longer removes an [id, name] index) and lets comma-separated index_column target composite indexes. --- .../src/table/global_index_drop_builder.rs | 292 +++++++++++++++++- 1 file changed, 289 insertions(+), 3 deletions(-) diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs index 338b94d1..62a9f1b0 100644 --- a/crates/paimon/src/table/global_index_drop_builder.rs +++ b/crates/paimon/src/table/global_index_drop_builder.rs @@ -19,7 +19,7 @@ use super::global_index_types::{ normalize_global_index_type_for_drop, BTREE_GLOBAL_INDEX_TYPE, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP, }; -use crate::spec::{DataField, FileKind, IndexFileMeta, IndexManifest}; +use crate::spec::{DataField, FileKind, GlobalIndexMeta, IndexFileMeta, IndexManifest}; use crate::table::{CommitMessage, SnapshotManager, Table, TableCommit}; use crate::{Error, Result}; use std::collections::HashMap; @@ -73,7 +73,10 @@ impl<'a> GlobalIndexDropBuilder<'a> { message: "Global index column is required".to_string(), source: None, })?; - let index_field = find_index_field(self.table, index_column)?; + let requested_field_ids = parse_index_field_ids(self.table, index_column)?; + if requested_field_ids.is_empty() { + return Ok(0); + } let snapshot_manager = SnapshotManager::new( self.table.file_io().clone(), @@ -106,7 +109,7 @@ impl<'a> GlobalIndexDropBuilder<'a> { let Some(global_meta) = entry.index_file.global_index_meta.as_ref() else { continue; }; - if global_meta.index_field_id != index_field.id() { + if global_meta_indexed_field_ids(global_meta) != requested_field_ids { continue; } dropped += 1; @@ -164,6 +167,28 @@ fn find_index_field<'a>(table: &'a Table, column: &str) -> Result<&'a DataField> }) } +/// Ordered field-id list requested for a drop, parsed from a comma-separated +/// `index_column`. Mirrors Java `DropGlobalIndexProcedure`: split on `,`, trim, +/// drop empty fragments, resolve each name to its field id (unknown -> error). +fn parse_index_field_ids(table: &Table, index_column: &str) -> Result> { + index_column + .split(',') + .map(str::trim) + .filter(|column| !column.is_empty()) + .map(|column| find_index_field(table, column).map(DataField::id)) + .collect() +} + +/// Full indexed-field-id list of a global index file: `[index_field_id]` followed +/// by `extra_field_ids`. Drop-side mirror of the identity built in `TableCommit`. +fn global_meta_indexed_field_ids(global_meta: &GlobalIndexMeta) -> Vec { + let mut ids = vec![global_meta.index_field_id]; + if let Some(extra_field_ids) = global_meta.extra_field_ids.as_ref() { + ids.extend(extra_field_ids.iter().copied()); + } + ids +} + #[cfg(test)] mod tests { use super::*; @@ -229,6 +254,27 @@ mod tests { } } + fn composite_global_index_file( + index_type: &str, + name: &str, + index_field_id: i32, + extra_field_ids: Vec, + row_range_start: i64, + row_range_end: i64, + ) -> IndexFileMeta { + let mut file = global_index_file( + index_type, + name, + index_field_id, + row_range_start, + row_range_end, + ); + if let Some(meta) = file.global_index_meta.as_mut() { + meta.extra_field_ids = Some(extra_field_ids); + } + file + } + fn hash_index_file(name: &str) -> IndexFileMeta { IndexFileMeta { index_type: "HASH".to_string(), @@ -566,4 +612,244 @@ mod tests { .unwrap(); assert_eq!(would_drop, 0); } + + #[tokio::test] + async fn test_drop_single_column_does_not_delete_composite() { + let table = test_table("memory:/test_drop_single_vs_composite"); + 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), + composite_global_index_file( + BTREE_GLOBAL_INDEX_TYPE, + "btree-id-name.index", + 0, + vec![1], + 0, + 9, + ), + ]; + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(vec![message]) + .await + .unwrap(); + + let dropped = table + .new_global_index_drop_builder() + .with_index_column("id") + .execute() + .await + .unwrap(); + assert_eq!(dropped, 1); + + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|e| e.index_file.file_name) + .collect::>(); + assert_eq!(remaining, vec!["btree-id-name.index".to_string()]); + } + + #[tokio::test] + async fn test_drop_composite_column_matches_only_composite() { + let table = test_table("memory:/test_drop_composite_hit"); + 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), + composite_global_index_file( + BTREE_GLOBAL_INDEX_TYPE, + "btree-id-name.index", + 0, + vec![1], + 0, + 9, + ), + ]; + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(vec![message]) + .await + .unwrap(); + + let dropped = table + .new_global_index_drop_builder() + .with_index_column("id,name") + .execute() + .await + .unwrap(); + assert_eq!(dropped, 1); + + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|e| e.index_file.file_name) + .collect::>(); + assert_eq!(remaining, vec!["btree-id.index".to_string()]); + } + + #[tokio::test] + async fn test_drop_composite_field_order_is_significant() { + let table = test_table("memory:/test_drop_composite_order"); + 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![composite_global_index_file( + BTREE_GLOBAL_INDEX_TYPE, + "btree-id-name.index", + 0, + vec![1], + 0, + 9, + )]; + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(vec![message]) + .await + .unwrap(); + + // Requested [name, id] = [1, 0] must NOT match stored [id, name] = [0, 1]. + let dropped = table + .new_global_index_drop_builder() + .with_index_column("name,id") + .execute() + .await + .unwrap(); + assert_eq!(dropped, 0); + + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|e| e.index_file.file_name) + .collect::>(); + assert_eq!(remaining, vec!["btree-id-name.index".to_string()]); + } + + #[tokio::test] + async fn test_drop_unknown_column_in_list_errors_on_trimmed_name() { + let table = test_table("memory:/test_drop_unknown_column"); + 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(); + + let err = table + .new_global_index_drop_builder() + .with_index_column("id, bogus") + .execute() + .await + .expect_err("unknown column must error"); + assert!(matches!(err, Error::ColumnNotExist { column, .. } if column == "bogus")); + } + + #[tokio::test] + async fn test_drop_trims_and_filters_empty_fragments() { + let table = test_table("memory:/test_drop_trim_filter"); + 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), + composite_global_index_file( + BTREE_GLOBAL_INDEX_TYPE, + "btree-id-name.index", + 0, + vec![1], + 0, + 9, + ), + ]; + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(vec![message]) + .await + .unwrap(); + + // "id, name," -> trim + drop empty trailing fragment -> [id, name] = [0, 1]. + let dropped = table + .new_global_index_drop_builder() + .with_index_column("id, name,") + .execute() + .await + .unwrap(); + assert_eq!(dropped, 1); + + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|e| e.index_file.file_name) + .collect::>(); + assert_eq!(remaining, vec!["btree-id.index".to_string()]); + } + + #[tokio::test] + async fn test_drop_all_empty_column_returns_zero_without_touching_manifest() { + let table = test_table("memory:/test_drop_all_empty_column"); + 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(); + + // An all-empty index_column parses to an empty field list -> Ok(0), + // returned before the manifest is read, so nothing is removed. + for empty in [",", ", ,", " "] { + let dropped = table + .new_global_index_drop_builder() + .with_index_column(empty) + .execute() + .await + .unwrap(); + assert_eq!(dropped, 0, "index_column {empty:?} should match nothing"); + + let remaining = latest_index_entries(&table) + .await + .into_iter() + .map(|e| e.index_file.file_name) + .collect::>(); + assert_eq!(remaining, vec!["btree-id.index".to_string()]); + } + } }