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 efe6f4e5 feat(datafusion): support the delete_branch procedure (#941)
efe6f4e5 is described below
commit efe6f4e5c0281e5a80e173d3527f7200434f3f2b
Author: jackylee <[email protected]>
AuthorDate: Fri Oct 16 12:12:18 2026 +0800
feat(datafusion): support the delete_branch procedure (#941)
---
crates/integrations/datafusion/src/procedures.rs | 54 ++++
crates/integrations/datafusion/tests/procedures.rs | 283 +++++++++++++++++++++
.../datafusion/tests/rest_management_procedures.rs | 26 ++
crates/paimon/src/table/branch_manager.rs | 95 ++++++-
4 files changed, 453 insertions(+), 5 deletions(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index ce150229..0161536c 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -178,6 +178,7 @@ fn declared_parameters(proc_name: &str) -> Option<&'static
[&'static str]> {
"rollback_to_timestamp" => &["table", "timestamp"],
"create_tag_from_timestamp" => &["table", "tag", "timestamp"],
"rename_branch" => &["table", "from_branch", "to_branch"],
+ "delete_branch" => &["table", "branch"],
"create_global_index" => &["table", "index_column", "index_type",
"options"],
"drop_global_index" => &[
"table",
@@ -291,6 +292,7 @@ pub async fn execute_call(
proc_create_tag_from_timestamp(ctx, catalog, catalog_name,
&args).await
}
"rename_branch" => proc_rename_branch(ctx, catalog, catalog_name,
&args).await,
+ "delete_branch" => proc_delete_branch(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,
@@ -491,6 +493,58 @@ async fn proc_rename_branch(
ok_result(ctx)
}
+async fn proc_delete_branch(
+ 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 branch_str = require_arg(args, "branch")?;
+
+ // A REST catalog owns branch metadata and the Rust REST catalog has no
+ // branch-drop endpoint yet (Java `RESTCatalog.dropBranch` DELETEs through
the
+ // catalog). Dropping the physical branch directory here would leave the
+ // catalog still listing the branch while its data is gone. Refuse before
+ // deleting anything, until a catalog-aware path exists.
+ if table.rest_env().is_some() {
+ return Err(DataFusionError::NotImplemented(
+ "delete_branch is not yet supported for tables managed by a REST
catalog".to_string(),
+ ));
+ }
+
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ let options = table.schema().options();
+ // Java `Table.deleteBranches` splits on comma and passes each token
through
+ // unchanged. Branch identities keep leading/trailing spaces (the shared
+ // validator and readers treat `" b1"` and `"b1"` as distinct), so the
+ // delete_tag trim convention must not apply here — trimming could delete a
+ // different branch than the one requested.
+ for branch_name in branch_str.split(',') {
+ if branch_name.is_empty() {
+ continue;
+ }
+ // Validate the logical name before any existence check or deletion: a
+ // separator-bearing name like `prod/schema` would otherwise slip past
the
+ // configured-branch guard (it is not literally `prod`) and recursively
+ // delete the inner directory of a protected branch.
+
BranchManager::validate_branch_name(branch_name).map_err(to_datafusion_error)?;
+ BranchManager::ensure_branch_deletable(options, branch_name)
+ .map_err(to_datafusion_error)?;
+ if !bm
+ .branch_exists(branch_name)
+ .await
+ .map_err(to_datafusion_error)?
+ {
+ continue;
+ }
+ bm.drop_branch(branch_name)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ ok_result(ctx)
+}
+
async fn proc_delete_tag(
ctx: &SessionContext,
catalog: &Arc<dyn Catalog>,
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index 21bfc73e..f4463685 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -412,6 +412,289 @@ async fn
test_rename_branch_rejects_unopenable_source_and_target() {
table.copy_with_branch("b1").await.unwrap();
}
+#[tokio::test]
+async fn test_delete_branch() {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.t1 (id INT, name VARCHAR(100), PRIMARY
KEY (id))",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t1 VALUES (1, 'alice')",
+ )
+ .await;
+
+ // Seed a branch through the core manager (create_branch is a separate PR).
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch("b1").await.unwrap();
+ assert!(bm.branch_exists("b1").await.unwrap(), "seed branch exists");
+
+ exec(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch => 'b1')",
+ )
+ .await;
+
+ assert!(
+ !bm.branch_exists("b1").await.unwrap(),
+ "delete_branch should remove the branch"
+ );
+ let count = row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.`t1$branches` WHERE branch_name = 'b1'",
+ )
+ .await;
+ assert_eq!(count, 0);
+
+ // Deleting a branch that no longer exists is a no-op, not an error.
+ exec(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch => 'b1')",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_delete_branch_preserves_leading_space_identity() {
+ // A branch name keeps its leading/trailing spaces: the shared validator
+ // permits `" b1"` (only all-whitespace is rejected) and the reader treats
it
+ // as distinct from `"b1"`. Java `Table.deleteBranches` passes each comma
+ // token through unchanged, so `delete_branch(branch => ' b1')` must drop
+ // `" b1"` and leave `"b1"` intact — trimming the token would delete the
+ // wrong branch.
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.t1 (id INT, name VARCHAR(100), PRIMARY
KEY (id))",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t1 VALUES (1, 'alice')",
+ )
+ .await;
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch("b1").await.unwrap();
+ bm.create_branch(" b1").await.unwrap();
+
+ exec(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch => ' b1')",
+ )
+ .await;
+
+ assert!(
+ !bm.branch_exists(" b1").await.unwrap(),
+ "the leading-space branch is the one requested and must be deleted"
+ );
+ assert!(
+ bm.branch_exists("b1").await.unwrap(),
+ "the un-spaced branch must survive — trimming the token would delete
it"
+ );
+}
+
+#[tokio::test]
+async fn test_delete_branch_preserves_trailing_space_identity() {
+ // `"prod "` passes every guard as a branch distinct from the protected
+ // `prod`, so deleting it must not resolve to `prod`'s directory.
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.t1 (id INT, name VARCHAR(100), PRIMARY
KEY (id)) WITH ('scan.primary-branch' = 'prod')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t1 VALUES (1, 'alice')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "CALL sys.create_tag(table => 'test_db.t1', tag => 'v1')",
+ )
+ .await;
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch_from_tag("prod", "v1").await.unwrap();
+ bm.create_branch_from_tag("prod ", "v1").await.unwrap();
+
+ exec(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch => 'prod ')",
+ )
+ .await;
+
+ assert!(!bm.branch_exists("prod ").await.unwrap());
+ assert_eq!(
+ collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.`t1$branch_prod`"
+ )
+ .await,
+ vec![(1, "alice".to_string())],
+ "the protected branch must stay readable"
+ );
+}
+
+#[tokio::test]
+async fn test_delete_branch_preserves_scan_configured_branch() {
+ // A branch named by `scan.primary-branch` / `scan.fallback-branch` is on a
+ // reader's path, so `delete_branch` must refuse it (mirrors Java
+ // `AbstractFileStoreTable.deleteBranch`) while still allowing unrelated
+ // branches to be dropped.
+ for (table_name, option_key) in [
+ ("tp", "scan.primary-branch"),
+ ("tf", "scan.fallback-branch"),
+ ] {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.{table_name} (id INT, name
VARCHAR(100), PRIMARY KEY (id)) WITH ('{option_key}' = 'prod')"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
+ &format!("INSERT INTO paimon.test_db.{table_name} VALUES (1,
'alice')"),
+ )
+ .await;
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", table_name))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch("prod").await.unwrap();
+ bm.create_branch("tmp").await.unwrap();
+
+ // The configured branch cannot be deleted, and the error names the
option.
+ assert_sql_error(
+ &sql_context,
+ &format!("CALL sys.delete_branch(table => 'test_db.{table_name}',
branch => 'prod')"),
+ option_key,
+ )
+ .await;
+ assert!(
+ bm.branch_exists("prod").await.unwrap(),
+ "{option_key}: protected branch must remain"
+ );
+
+ // An unrelated branch is still deletable, so the guard is not
over-broad.
+ exec(
+ &sql_context,
+ &format!("CALL sys.delete_branch(table => 'test_db.{table_name}',
branch => 'tmp')"),
+ )
+ .await;
+ assert!(
+ !bm.branch_exists("tmp").await.unwrap(),
+ "{option_key}: unrelated branch should be deletable"
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_delete_branch_batch_stops_at_protected_branch() {
+ // A comma-separated request must still catch a protected branch even when
+ // it is not listed first. Matches Java `Table.deleteBranches`, which loops
+ // `deleteBranch` per name; the protected branch is never dropped.
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.t1 (id INT, name VARCHAR(100), PRIMARY
KEY (id)) WITH ('scan.primary-branch' = 'prod')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t1 VALUES (1, 'alice')",
+ )
+ .await;
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch("prod").await.unwrap();
+ bm.create_branch("keep").await.unwrap();
+
+ // 'prod' is protected and appears second; the call must fail and leave it.
+ assert_sql_error(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch => 'keep,prod')",
+ "scan.primary-branch",
+ )
+ .await;
+ assert!(
+ bm.branch_exists("prod").await.unwrap(),
+ "protected branch must survive the batch"
+ );
+}
+
+#[tokio::test]
+async fn test_delete_branch_rejects_path_separated_name() {
+ // A separator-bearing name slips past the configured-branch guard (it is
not
+ // literally `prod`), so without name validation `branch_exists` finds the
real
+ // `branch-prod/schema` directory and `drop_branch` deletes it, destroying
the
+ // protected branch's schema. The name must be rejected before any
deletion.
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ exec(&sql_context, "CREATE SCHEMA paimon.test_db").await;
+ exec(
+ &sql_context,
+ "CREATE TABLE paimon.test_db.t1 (id INT, name VARCHAR(100), PRIMARY
KEY (id)) WITH ('scan.primary-branch' = 'prod')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t1 VALUES (1, 'alice')",
+ )
+ .await;
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t1"))
+ .await
+ .unwrap();
+ let bm = BranchManager::new(table.file_io().clone(),
table.location().to_string());
+ bm.create_branch("prod").await.unwrap();
+
+ assert_sql_error(
+ &sql_context,
+ "CALL sys.delete_branch(table => 'test_db.t1', branch =>
'prod/schema')",
+ "path separator",
+ )
+ .await;
+
+ // The protected branch and its schema survive and stay openable.
+ assert!(bm.branch_exists("prod").await.unwrap(), "prod must remain");
+ table.copy_with_branch("prod").await.unwrap();
+}
+
#[tokio::test]
async fn test_create_lumina_index_requires_index_column() {
let (_tmp, sql_context) = setup_table_with_snapshots().await;
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
index bc6a6b85..d38efa9d 100644
--- a/crates/integrations/datafusion/tests/rest_management_procedures.rs
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -680,6 +680,32 @@ async fn test_rename_branch_rejected_on_rest_catalog() {
.await;
}
+#[tokio::test]
+async fn test_delete_branch_rejected_on_rest_catalog() {
+ // A REST catalog owns branch metadata and the Rust REST catalog has no
+ // branch-drop endpoint yet, so `delete_branch` must refuse rather than
delete
+ // a physical branch directory while the catalog still lists it. Without
the
+ // guard the filesystem `BranchManager` runs and reports a different
error, so
+ // this assertion is non-vacuous.
+ let (tmp, server, context) = setup().await;
+
+ // `get_table` needs a schema-bearing response; register one so resolution
+ // reaches the REST guard rather than failing earlier on a missing schema.
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .build()
+ .unwrap();
+ let path = format!("file://{}/{DATABASE}.db/{TABLE}",
tmp.path().display());
+ server.add_table_with_schema(DATABASE, TABLE, schema, &path);
+
+ common::assert_sql_error(
+ &context,
+ &format!("CALL sys.delete_branch(table => '{DATABASE}.{TABLE}', branch
=> 'b1')"),
+ "REST catalog",
+ )
+ .await;
+}
+
#[tokio::test]
async fn test_argument_validation_follows_java() {
let (_tmp, server, context) = setup().await;
diff --git a/crates/paimon/src/table/branch_manager.rs
b/crates/paimon/src/table/branch_manager.rs
index ca35ec4c..4439137b 100644
--- a/crates/paimon/src/table/branch_manager.rs
+++ b/crates/paimon/src/table/branch_manager.rs
@@ -24,6 +24,12 @@ use crate::catalog::DEFAULT_MAIN_BRANCH;
use crate::io::FileIO;
use crate::table::{SchemaManager, SnapshotManager, TagManager};
use opendal::raw::get_basename;
+use std::collections::HashMap;
+
+/// Table option a reader falls back to when the primary branch has no data.
+const SCAN_FALLBACK_BRANCH: &str = "scan.fallback-branch";
+/// Table option naming the branch a reader treats as primary.
+const SCAN_PRIMARY_BRANCH: &str = "scan.primary-branch";
const BRANCH_DIR: &str = "branch";
const BRANCH_PREFIX: &str = "branch-";
@@ -62,13 +68,22 @@ impl BranchManager {
)
}
+ /// [`branch_path`](Self::branch_path) in directory form
(`branch/branch-my_branch/`).
+ ///
+ /// OpenDAL trims whitespace from both ends of a path, and a recursive
delete
+ /// the backend cannot do natively lists by key prefix. The trailing `/`
keeps
+ /// `prod ` from resolving to `prod`, and `b1` from also matching `b10/`.
+ fn branch_dir(&self, branch_name: &str) -> String {
+ format!("{}/", self.branch_path(branch_name))
+ }
+
/// Validate branch name format.
///
/// Rules:
/// - Cannot be "main"
/// - Cannot be blank or whitespace only
/// - Cannot be a pure numeric string
- fn validate_branch_name(branch_name: &str) -> crate::Result<()> {
+ pub fn validate_branch_name(branch_name: &str) -> crate::Result<()> {
// Enforce the same name contract the catalog/table reader applies
// (`copy_with_branch` and `$branch_...` resolution): reject blank,
// `.`/`..`, path separators and control characters. A name accepted
here
@@ -110,8 +125,7 @@ impl BranchManager {
/// Check if a branch exists.
pub async fn branch_exists(&self, branch_name: &str) ->
crate::Result<bool> {
- let path = format!("{}/", self.branch_path(branch_name));
- self.file_io.exists(&path).await
+ self.file_io.exists(&self.branch_dir(branch_name)).await
}
/// List all branch names sorted ascending. Returns an empty vector when
the
@@ -198,8 +212,37 @@ impl BranchManager {
source: None,
});
}
- let path = self.branch_path(branch_name);
- self.file_io.delete_dir(&path).await?;
+ // Delete the same directory form that `branch_exists` checked.
+ self.file_io
+ .delete_dir(&self.branch_dir(branch_name))
+ .await?;
+ Ok(())
+ }
+
+ /// Reject deleting a branch a reader is configured to consult.
+ ///
+ /// A branch named by `scan.primary-branch` or `scan.fallback-branch` is
part
+ /// of a table's read path; dropping it would break reads that fall back to
+ /// it. Mirrors Java `AbstractFileStoreTable.deleteBranch`, which refuses
the
+ /// deletion and asks the caller to unset the option first. Callers pass
the
+ /// table options because these keys live on the schema, not the manager.
+ pub fn ensure_branch_deletable(
+ options: &HashMap<String, String>,
+ branch_name: &str,
+ ) -> crate::Result<()> {
+ for key in [SCAN_PRIMARY_BRANCH, SCAN_FALLBACK_BRANCH] {
+ if options
+ .get(key)
+ .is_some_and(|configured| configured == branch_name)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Cannot delete branch '{branch_name}' because it is
configured as '{key}'. Unset '{key}' first."
+ ),
+ source: None,
+ });
+ }
+ }
Ok(())
}
@@ -519,6 +562,24 @@ mod tests {
assert!(!bm.branch_exists("to_drop").await.unwrap());
}
+ #[tokio::test]
+ async fn test_drop_branch_deletes_only_the_named_directory() {
+ // The memory backend deletes recursively by key prefix, as S3 and OSS
do.
+ let file_io = test_file_io();
+ let table_path = "memory:/test_drop_branch_exact".to_string();
+ let schema_manager = SchemaManager::new(file_io.clone(),
table_path.clone());
+ write_schema(&file_io, &schema_manager, &test_schema()).await;
+
+ let bm = BranchManager::new(file_io, table_path);
+ for name in ["b1", "b10", "prod", "prod "] {
+ bm.create_branch(name).await.unwrap();
+ }
+ bm.drop_branch("b1").await.unwrap();
+ bm.drop_branch("prod ").await.unwrap();
+
+ assert_eq!(bm.list_all().await.unwrap(), vec!["b10", "prod"]);
+ }
+
#[tokio::test]
async fn test_drop_nonexistent_branch() {
let file_io = test_file_io();
@@ -529,6 +590,30 @@ mod tests {
assert!(msg.contains("doesn't exist"));
}
+ #[test]
+ fn test_ensure_branch_deletable_rejects_scan_branches() {
+ for key in ["scan.primary-branch", "scan.fallback-branch"] {
+ let mut options = HashMap::new();
+ options.insert(key.to_string(), "prod".to_string());
+
+ // The branch the option points at is part of a read path.
+ let err = BranchManager::ensure_branch_deletable(&options,
"prod").unwrap_err();
+ let msg = format!("{err}");
+ assert!(msg.contains("prod"), "{msg}");
+ assert!(msg.contains(key), "{msg}");
+
+ // A different branch is unaffected even while the option is set,
+ // so the guard keys on the configured value, not its mere
presence.
+ BranchManager::ensure_branch_deletable(&options,
"feature").unwrap();
+ }
+ }
+
+ #[test]
+ fn test_ensure_branch_deletable_without_scan_options() {
+ let options = HashMap::new();
+ BranchManager::ensure_branch_deletable(&options, "any").unwrap();
+ }
+
#[tokio::test]
async fn test_rename_branch() {
let file_io = test_file_io();
