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 e312ca6 feat(index): add btree global index drop (#472)
e312ca6 is described below
commit e312ca63165cddf00e72120205a6baca2150269f
Author: QuakeWang <[email protected]>
AuthorDate: Tue Jul 7 20:38:28 2026 +0800
feat(index): add btree global index drop (#472)
---
crates/integrations/datafusion/src/procedures.rs | 36 +++
crates/integrations/datafusion/tests/procedures.rs | 95 ++++++
.../src/table/btree_global_index_drop_builder.rs | 357 +++++++++++++++++++++
crates/paimon/src/table/mod.rs | 6 +
4 files changed, 494 insertions(+)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 89a3593..6587983 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -24,6 +24,7 @@
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'btree')`
+//! - `CALL sys.drop_global_index(table => '...', index_column => '...',
index_type => 'btree')`
//! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
use std::collections::HashMap;
@@ -150,6 +151,7 @@ pub async fn execute_call(
proc_create_tag_from_timestamp(ctx, catalog, catalog_name,
&args).await
}
"create_global_index" => proc_create_global_index(ctx, catalog,
catalog_name, &args).await,
+ "drop_global_index" => proc_drop_global_index(ctx, catalog,
catalog_name, &args).await,
"create_lumina_index" => proc_create_lumina_index(ctx, catalog,
catalog_name, &args).await,
_ => Err(DataFusionError::Plan(format!(
"Unknown procedure: {proc_name}"
@@ -558,6 +560,40 @@ async fn proc_create_global_index(
ok_result(ctx)
}
+async fn proc_drop_global_index(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let table = get_table(catalog, catalog_name, args).await?;
+ let index_column = require_arg(args, "index_column")?;
+ let index_type = args
+ .get("index_type")
+ .map(String::as_str)
+ .unwrap_or("btree");
+ if !index_type.eq_ignore_ascii_case("btree") {
+ return Err(DataFusionError::NotImplemented(format!(
+ "drop_global_index only supports index_type => 'btree', got
'{index_type}'"
+ )));
+ }
+ if args.contains_key("partitions") {
+ return Err(DataFusionError::NotImplemented(
+ "drop_global_index partitions are not supported for btree
yet".to_string(),
+ ));
+ }
+ if args.contains_key("dry_run") {
+ return Err(DataFusionError::NotImplemented(
+ "drop_global_index dry_run is not supported for btree
yet".to_string(),
+ ));
+ }
+
+ let mut builder = table.new_btree_global_index_drop_builder();
+ builder.with_index_column(index_column);
+ builder.execute().await.map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
fn parse_key_value_options(options: &str) -> DFResult<HashMap<String, String>>
{
let mut parsed = HashMap::new();
for entry in options.split(',').map(str::trim).filter(|s| !s.is_empty()) {
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index ec2d486..9e6a256 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -208,6 +208,101 @@ async fn
test_create_global_index_builds_btree_and_filter_reads() {
assert_eq!(rows, vec![(2, "bob".to_string())]);
}
+#[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;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.btree_drop (id, name) VALUES (1, 'alice'),
(2, 'bob')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table => 'test_db.btree_drop',
index_column => 'id')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.btree_drop (id, name) VALUES (3, 'carol')",
+ )
+ .await;
+
+ let fast_rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.btree_drop WHERE id >= 2",
+ )
+ .await;
+ assert_eq!(fast_rows, vec![(2, "bob".to_string())]);
+
+ exec(
+ &sql_context,
+ "CALL sys.drop_global_index(table => 'test_db.btree_drop',
index_column => 'id', index_type => 'btree')",
+ )
+ .await;
+
+ let index_count = row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.`btree_drop$table_indexes` \
+ WHERE index_type = 'btree' AND index_field_name = 'id'",
+ )
+ .await;
+ assert_eq!(index_count, 0);
+
+ let rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.btree_drop WHERE id >= 2",
+ )
+ .await;
+ assert_eq!(rows, vec![(2, "bob".to_string()), (3, "carol".to_string())]);
+}
+
+#[tokio::test]
+async fn test_drop_global_index_requires_index_column() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop_missing_col").await;
+
+ assert_sql_error(
+ &sql_context,
+ "CALL sys.drop_global_index(table =>
'test_db.btree_drop_missing_col')",
+ "Missing required argument: 'index_column'",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_drop_global_index_rejects_non_btree() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop_bad_type").await;
+
+ assert_sql_error(
+ &sql_context,
+ "CALL sys.drop_global_index(table => 'test_db.btree_drop_bad_type',
index_column => 'id', index_type => 'bitmap')",
+ "only supports index_type => 'btree'",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_drop_global_index_without_match_succeeds() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop_no_match").await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.btree_drop_no_match (id, name) VALUES (1,
'alice')",
+ )
+ .await;
+
+ exec(
+ &sql_context,
+ "CALL sys.drop_global_index(table => 'test_db.btree_drop_no_match',
index_column => 'id')",
+ )
+ .await;
+
+ let rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.btree_drop_no_match WHERE id = 1",
+ )
+ .await;
+ assert_eq!(rows, vec![(1, "alice".to_string())]);
+}
+
#[tokio::test]
async fn test_create_global_index_fast_full_detail_after_append() {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_coverage").await;
diff --git a/crates/paimon/src/table/btree_global_index_drop_builder.rs
b/crates/paimon/src/table/btree_global_index_drop_builder.rs
new file mode 100644
index 0000000..50f0e30
--- /dev/null
+++ b/crates/paimon/src/table/btree_global_index_drop_builder.rs
@@ -0,0 +1,357 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::spec::{DataField, FileKind, IndexFileMeta, IndexManifest};
+use crate::table::{CommitMessage, SnapshotManager, Table, TableCommit};
+use crate::{Error, Result};
+use std::collections::HashMap;
+
+const BTREE_INDEX_TYPE: &str = "btree";
+
+pub struct BTreeGlobalIndexDropBuilder<'a> {
+ table: &'a Table,
+ index_column: Option<String>,
+}
+
+impl<'a> BTreeGlobalIndexDropBuilder<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ Self {
+ table,
+ index_column: None,
+ }
+ }
+
+ pub fn with_index_column(&mut self, column: &str) -> &mut Self {
+ self.index_column = Some(column.to_string());
+ self
+ }
+
+ pub async fn execute(&self) -> Result<usize> {
+ let index_column = self
+ .index_column
+ .as_deref()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "BTree global index column is required".to_string(),
+ source: None,
+ })?;
+ let index_field = find_index_field(self.table, index_column)?;
+
+ let snapshot_manager = SnapshotManager::new(
+ self.table.file_io().clone(),
+ self.table.location().to_string(),
+ );
+ let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
+ return Ok(0);
+ };
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Ok(0);
+ };
+
+ let index_entries = IndexManifest::read(
+ self.table.file_io(),
+ &snapshot_manager.manifest_path(index_manifest_name),
+ )
+ .await?;
+ let mut deletions_by_partition_bucket: HashMap<(Vec<u8>, i32),
Vec<IndexFileMeta>> =
+ HashMap::new();
+ let mut dropped = 0;
+ for entry in index_entries {
+ if entry.kind != FileKind::Add || entry.index_file.index_type !=
BTREE_INDEX_TYPE {
+ continue;
+ }
+ let Some(global_meta) =
entry.index_file.global_index_meta.as_ref() else {
+ continue;
+ };
+ if global_meta.index_field_id != index_field.id() {
+ continue;
+ }
+ dropped += 1;
+ deletions_by_partition_bucket
+ .entry((entry.partition, entry.bucket))
+ .or_default()
+ .push(entry.index_file);
+ }
+ if dropped == 0 {
+ return Ok(0);
+ }
+
+ let mut groups = deletions_by_partition_bucket
+ .into_iter()
+ .collect::<Vec<_>>();
+ groups.sort_by(
+ |((left_partition, left_bucket), _), ((right_partition,
right_bucket), _)| {
+ left_partition
+ .cmp(right_partition)
+ .then(left_bucket.cmp(right_bucket))
+ },
+ );
+ let messages = groups
+ .into_iter()
+ .map(|((partition, bucket), deleted_index_files)| {
+ let mut message = CommitMessage::new(partition, bucket,
vec![]);
+ message.deleted_index_files = deleted_index_files;
+ message
+ })
+ .collect::<Vec<_>>();
+
+ TableCommit::new(
+ self.table.clone(),
+ format!(
+ "global-index-{}-drop-{}",
+ BTREE_INDEX_TYPE,
+ uuid::Uuid::new_v4()
+ ),
+ )
+ .commit_if_latest_snapshot(messages, snapshot.id())
+ .await?;
+
+ Ok(dropped)
+ }
+}
+
+fn find_index_field<'a>(table: &'a Table, column: &str) -> Result<&'a
DataField> {
+ table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == column)
+ .ok_or_else(|| Error::ColumnNotExist {
+ full_name: table.identifier().full_name(),
+ column: column.to_string(),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::Identifier;
+ use crate::io::FileIOBuilder;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{
+ BinaryRow, DataFileMeta, DataType, DeletionVectorMeta,
GlobalIndexMeta, IndexManifestEntry,
+ IntType, Schema, TableSchema, VarCharType,
+ };
+ use crate::table::TableCommit;
+ use chrono::{DateTime, Utc};
+ use indexmap::IndexMap;
+
+ fn test_table(table_path: &str) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .build()
+ .unwrap();
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "test_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+ }
+
+ async fn setup_dirs(table: &Table) {
+ table
+ .file_io()
+ .mkdirs(&format!("{}/snapshot/", table.location()))
+ .await
+ .unwrap();
+ table
+ .file_io()
+ .mkdirs(&format!("{}/manifest/", table.location()))
+ .await
+ .unwrap();
+ }
+
+ fn global_index_file(
+ index_type: &str,
+ name: &str,
+ index_field_id: i32,
+ row_range_start: i64,
+ row_range_end: i64,
+ ) -> IndexFileMeta {
+ IndexFileMeta {
+ index_type: index_type.to_string(),
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count: (row_range_end - row_range_start + 1) as i32,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start,
+ row_range_end,
+ index_field_id,
+ extra_field_ids: None,
+ index_meta: None,
+ }),
+ }
+ }
+
+ fn hash_index_file(name: &str) -> IndexFileMeta {
+ IndexFileMeta {
+ index_type: "HASH".to_string(),
+ file_name: name.to_string(),
+ file_size: 64,
+ row_count: 1,
+ deletion_vectors_ranges: None,
+ global_index_meta: None,
+ }
+ }
+
+ fn deletion_vector_index_file(name: &str) -> IndexFileMeta {
+ IndexFileMeta {
+ index_type: "DELETION_VECTORS".to_string(),
+ file_name: name.to_string(),
+ file_size: 64,
+ row_count: 1,
+ deletion_vectors_ranges: Some(IndexMap::from([(
+ "data-0.parquet".to_string(),
+ DeletionVectorMeta {
+ offset: 1,
+ length: 8,
+ cardinality: Some(1),
+ },
+ )])),
+ global_index_meta: None,
+ }
+ }
+
+ fn data_file(name: &str) -> DataFileMeta {
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count: 1,
+ min_key: vec![],
+ max_key: vec![],
+ key_stats: BinaryTableStats::empty(),
+ value_stats: BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level: 0,
+ extra_files: vec![],
+ creation_time: Some(
+ "2024-09-06T07:45:55.039+00:00"
+ .parse::<DateTime<Utc>>()
+ .unwrap(),
+ ),
+ delete_row_count: None,
+ embedded_index: None,
+ first_row_id: None,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ }
+ }
+
+ async fn latest_index_entries(table: &Table) -> Vec<IndexManifestEntry> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Vec::new();
+ };
+ IndexManifest::read(
+ table.file_io(),
+ &snapshot_manager.manifest_path(index_manifest_name),
+ )
+ .await
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_drop_btree_global_index_only_removes_target_entries() {
+ let table = test_table("memory:/test_drop_btree_global_index");
+ 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_INDEX_TYPE, "btree-id.index", 0, 0, 9),
+ global_index_file(BTREE_INDEX_TYPE, "btree-name.index", 1, 0, 9),
+ global_index_file("full-text", "fulltext-id.index", 0, 100, 109),
+ hash_index_file("hash.index"),
+ deletion_vector_index_file("dv.index"),
+ ];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+
+ let dropped = table
+ .new_btree_global_index_drop_builder()
+ .with_index_column("id")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(dropped, 1);
+
+ let mut remaining = latest_index_entries(&table)
+ .await
+ .into_iter()
+ .map(|entry| entry.index_file.file_name)
+ .collect::<Vec<_>>();
+ remaining.sort();
+ assert_eq!(
+ remaining,
+ vec![
+ "btree-name.index".to_string(),
+ "dv.index".to_string(),
+ "fulltext-id.index".to_string(),
+ "hash.index".to_string(),
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_drop_btree_global_index_is_idempotent_without_match() {
+ let table =
test_table("memory:/test_drop_btree_global_index_idempotent");
+ 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_INDEX_TYPE,
+ "btree-name.index",
+ 1,
+ 0,
+ 9,
+ )];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+
+ let dropped = table
+ .new_btree_global_index_drop_builder()
+ .with_index_column("id")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(dropped, 0);
+
+ let remaining = latest_index_entries(&table).await;
+ assert_eq!(remaining.len(), 1);
+ assert_eq!(remaining[0].index_file.file_name, "btree-name.index");
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 930f9ef..edc3719 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -22,6 +22,7 @@ pub(crate) mod bin_pack;
mod blob_file_writer;
mod branch_manager;
mod btree_global_index_build_builder;
+mod btree_global_index_drop_builder;
mod bucket_assigner;
mod bucket_assigner_constant;
mod bucket_assigner_cross;
@@ -71,6 +72,7 @@ use crate::Result;
use arrow_array::RecordBatch;
pub use branch_manager::BranchManager;
pub use btree_global_index_build_builder::BTreeGlobalIndexBuildBuilder;
+pub use btree_global_index_drop_builder::BTreeGlobalIndexDropBuilder;
pub use commit_message::CommitMessage;
pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo};
pub use data_evolution_writer::{DataEvolutionDeleteWriter,
DataEvolutionWriter};
@@ -212,6 +214,10 @@ impl Table {
BTreeGlobalIndexBuildBuilder::new(self)
}
+ pub fn new_btree_global_index_drop_builder(&self) ->
BTreeGlobalIndexDropBuilder<'_> {
+ BTreeGlobalIndexDropBuilder::new(self)
+ }
+
/// Create a write builder for write/commit.
///
/// Reference: [pypaimon
FileStoreTable.new_write_builder](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/table/file_store_table.py).
