From 6ae90441ca92bc9edd46022879aa9a559e77628d Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Thu, 29 Oct 2026 10:32:49 +0800 Subject: [PATCH] feat(datafusion): add create_branch SQL procedure Expose Core branch creation through SQL for non-REST main tables. Support empty and tag-based branches with ignore_if_exists, including retained tags whose live snapshots have expired. Reject invalid names and unsupported sources before writing. Add SQL end-to-end tests and document the supported scope. Signed-off-by: QuakeWang --- .../integrations/datafusion/src/procedures.rs | 53 ++ .../datafusion/tests/create_branch.rs | 565 ++++++++++++++++++ .../datafusion/tests/create_branch/support.rs | 129 ++++ .../tests/rest_management_procedures.rs | 39 ++ docs/src/sql.md | 53 ++ 5 files changed, 839 insertions(+) create mode 100644 crates/integrations/datafusion/tests/create_branch.rs create mode 100644 crates/integrations/datafusion/tests/create_branch/support.rs diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 160fd34b1..19df4563c 100644 --- a/crates/integrations/datafusion/src/procedures.rs +++ b/crates/integrations/datafusion/src/procedures.rs @@ -18,6 +18,7 @@ //! CALL procedure support for Paimon tables. //! //! Supported procedures: +//! - `CALL sys.create_branch(table => '...', branch => '...'[, tag => '...', ignore_if_exists => true])` //! - `CALL sys.create_tag(table => '...', tag => '...', snapshot_id => ..., time_retained => '...')` //! - `CALL sys.replace_tag(table => '...', tag => '...', snapshot_id => ..., time_retained => '...')` //! - `CALL sys.delete_tag(table => '...', tag => '...')` @@ -202,6 +203,7 @@ async fn earlier_or_equal_from_all( /// a misspelled `index_type` on `create_global_index` would build the default index. fn declared_parameters(proc_name: &str) -> Option<&'static [&'static str]> { Some(match proc_name { + "create_branch" => &["table", "branch", "tag", "ignore_if_exists"], "create_tag" => &["table", "tag", "snapshot_id", "time_retained"], "replace_tag" => &["table", "tag", "snapshot_id", "time_retained"], "delete_tag" => &["table", "tag"], @@ -324,6 +326,7 @@ pub async fn execute_procedure( reject_unknown_args(proc_name, &args)?; match proc_name { + "create_branch" => proc_create_branch(ctx, catalog, catalog_name, &args).await, "create_tag" => proc_create_tag(ctx, catalog, catalog_name, &args).await, "replace_tag" => proc_replace_tag(ctx, catalog, catalog_name, &args).await, "delete_tag" => proc_delete_tag(ctx, catalog, catalog_name, &args).await, @@ -663,6 +666,56 @@ async fn proc_replace_tag( ok_result(ctx) } +async fn proc_create_branch( + ctx: &SessionContext, + catalog: &Arc, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + let branch = require_arg(args, "branch")?; + let ignore_if_exists = bool_arg(args, "ignore_if_exists")?; + // Validate before the existence check: an invalid name must not become a + // successful no-op just because it resolves to an existing directory. + BranchManager::validate_branch_name(branch).map_err(to_datafusion_error)?; + if branch.contains('$') { + return Err(DataFusionError::Plan(format!( + "Branch name '{branch}' cannot contain the SQL suffix delimiter '$'" + ))); + } + let table = get_table(catalog, catalog_name, args).await?; + if table.rest_env().is_some() { + return Err(DataFusionError::NotImplemented( + "create_branch is not supported for tables managed by a REST catalog".to_string(), + )); + } + if table.schema().core_options().is_format_table() { + return Err(DataFusionError::NotImplemented( + "create_branch requires a FileStoreTable; format-table is not supported".to_string(), + )); + } + if !table.is_main_branch() { + return Err(DataFusionError::NotImplemented( + "create_branch requires a main-branch source table".to_string(), + )); + } + + let manager = BranchManager::new(table.file_io().clone(), table.location().to_string()); + if ignore_if_exists + && manager + .branch_exists(branch) + .await + .map_err(to_datafusion_error)? + { + return ok_result(ctx); + } + match args.get("tag") { + Some(tag) => manager.create_branch_from_tag(branch, tag).await, + None => manager.create_branch(branch).await, + } + .map_err(to_datafusion_error)?; + ok_result(ctx) +} + async fn proc_rename_branch( ctx: &SessionContext, catalog: &Arc, diff --git a/crates/integrations/datafusion/tests/create_branch.rs b/crates/integrations/datafusion/tests/create_branch.rs new file mode 100644 index 000000000..ca5954c4e --- /dev/null +++ b/crates/integrations/datafusion/tests/create_branch.rs @@ -0,0 +1,565 @@ +// 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. + +mod common; +#[path = "create_branch/support.rs"] +mod support; + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use common::{ + assert_sql_error, collect_id_name, create_sql_context, create_test_env, exec, row_count, + string_value, +}; +use paimon::catalog::Identifier; +use paimon::io::{FileIOBuilder, FileIOProvider}; +use paimon::spec::{DataType, IntType, Schema}; +use paimon::table::{BranchManager, Table}; +use paimon::{Catalog, FileSystemCatalog}; +use paimon_datafusion::SQLContext; + +async fn setup() -> (tempfile::TempDir, Arc, SQLContext, Table) { + let (tmp, catalog) = create_test_env(); + let context = create_sql_context(catalog.clone()).await; + exec(&context, "CREATE SCHEMA paimon.test_db").await; + exec( + &context, + "CREATE TABLE paimon.test_db.events (id INT, name STRING)", + ) + .await; + exec( + &context, + "INSERT INTO paimon.test_db.events VALUES (1, 'alice')", + ) + .await; + exec( + &context, + "CALL sys.create_tag(table => 'test_db.events', tag => 'release', time_retained => '1d')", + ) + .await; + exec( + &context, + "ALTER TABLE paimon.test_db.events ADD COLUMN extra INT", + ) + .await; + exec( + &context, + "INSERT INTO paimon.test_db.events VALUES (2, 'bob', 20)", + ) + .await; + let table = catalog + .get_table(&Identifier::new("test_db", "events")) + .await + .unwrap(); + (tmp, catalog, context, table) +} + +async fn assert_ok(context: &SQLContext, args: &str) { + let batches = context + .sql(&format!("CALL paimon.sys.create_branch({args})")) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 1); + assert_eq!(batches[0].num_columns(), 1); + assert_eq!(batches[0].schema().field(0).name(), "result"); + assert_eq!(string_value(batches[0].column(0).as_ref(), 0), "OK"); +} + +fn table_files(root: &Path) -> BTreeMap> { + 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 { + files.insert(path.clone(), std::fs::read(path).unwrap()); + } + } + files +} + +fn main_files(root: &Path) -> BTreeMap> { + table_files(root) + .into_iter() + .filter(|(path, _)| !path.strip_prefix(root).unwrap().starts_with("branch")) + .collect() +} + +#[tokio::test] +async fn empty_branch_uses_latest_schema_and_does_not_inherit_data_or_tags() { + let (tmp, _, context, table) = setup().await; + let root = tmp.path().join("test_db.db/events"); + let before = main_files(&root); + assert_ok( + &context, + "table => 'paimon.test_db.events', branch => 'empty'", + ) + .await; + let branch = table.copy_with_branch("empty").await.unwrap(); + assert_eq!(branch.schema().id(), table.schema().id()); + assert_eq!(branch.schema().fields(), table.schema().fields()); + assert!(branch + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .is_none()); + assert!(!root.join("branch/branch-empty/snapshot").exists()); + assert!(!root.join("branch/branch-empty/tag").exists()); + assert_eq!( + row_count( + &context, + "SELECT * FROM paimon.test_db.`events$branch_empty`" + ) + .await, + 0 + ); + assert_eq!( + row_count( + &context, + "SELECT * FROM paimon.test_db.`events$branches` WHERE branch_name = 'empty'" + ) + .await, + 1 + ); + assert_eq!(main_files(&root), before); + assert_eq!( + collect_id_name(&context, "SELECT id, name FROM paimon.test_db.events").await, + vec![(1, "alice".into()), (2, "bob".into())] + ); +} + +#[tokio::test] +async fn empty_branch_can_be_created_before_the_first_commit() { + let (tmp, catalog) = create_test_env(); + let context = create_sql_context(catalog).await; + exec(&context, "CREATE SCHEMA paimon.test_db").await; + exec(&context, "CREATE TABLE paimon.test_db.fresh (id INT)").await; + let root = tmp.path().join("test_db.db/fresh"); + let before = main_files(&root); + assert_ok( + &context, + "table => 'test_db.fresh', branch => 'empty', ignore_if_exists => true", + ) + .await; + assert_eq!( + row_count( + &context, + "SELECT * FROM paimon.test_db.`fresh$branch_empty`" + ) + .await, + 0 + ); + assert_eq!( + row_count( + &context, + "SELECT * FROM paimon.test_db.`fresh$branches` WHERE branch_name = 'empty'" + ) + .await, + 1 + ); + assert!(root.join("branch/branch-empty/schema/schema-0").is_file()); + assert_eq!(main_files(&root), before); +} + +#[tokio::test] +async fn tag_branch_reads_historical_schema_and_rows_with_live_or_expired_snapshot() { + let (tmp, _, context, table) = setup().await; + let root = tmp.path().join("test_db.db/events"); + let tag = table.tag_manager().get("release").await.unwrap().unwrap(); + assert!(tag.schema_id() < table.schema().id()); + for (name, expire) in [("live", false), ("expired", true)] { + if expire { + table + .file_io() + .delete_file(&table.snapshot_manager().snapshot_path(tag.id())) + .await + .unwrap(); + assert!(!table + .file_io() + .exists(&table.snapshot_manager().snapshot_path(tag.id())) + .await + .unwrap()); + } + let before = main_files(&root); + assert_ok(&context, &format!("table => 'test_db.events', branch => '{name}', tag => 'release', ignore_if_exists => false")).await; + let branch = table.copy_with_branch(name).await.unwrap(); + assert_eq!(branch.schema().id(), tag.schema_id()); + assert_eq!(branch.schema().fields().len(), 2); + assert_eq!( + collect_id_name( + &context, + &format!("SELECT id, name FROM paimon.test_db.`events$branch_{name}`") + ) + .await, + vec![(1, "alice".into())] + ); + assert_eq!( + row_count( + &context, + &format!( + "SELECT * FROM paimon.test_db.`events$branches` WHERE branch_name = '{name}'" + ) + ) + .await, + 1 + ); + assert_eq!( + std::fs::read(root.join(format!("branch/branch-{name}/tag/tag-release"))).unwrap(), + before[&root.join("tag/tag-release")] + ); + assert_eq!(main_files(&root), before); + assert_eq!( + collect_id_name(&context, "SELECT id, name FROM paimon.test_db.events").await, + vec![(1, "alice".into()), (2, "bob".into())] + ); + assert_eq!( + collect_id_name( + &context, + "SELECT id, name FROM paimon.test_db.events VERSION AS OF 'release'" + ) + .await, + vec![(1, "alice".into())] + ); + } +} + +#[tokio::test] +async fn ignore_existing_preserves_empty_and_tagged_branches_without_resolving_tag() { + let (tmp, _, context, _) = setup().await; + for (branch, tag) in [("empty", ""), ("tagged", ", tag => 'release'")] { + assert_ok( + &context, + &format!( + "table => 'test_db.events', branch => '{branch}'{tag}, ignore_if_exists => 'TrUe'" + ), + ) + .await; + } + let root = tmp.path().join("test_db.db/events"); + let before = table_files(&root); + for branch in ["empty", "tagged"] { + for tag in ["", ", tag => 'release'", ", tag => 'missing'"] { + for ignore in ["", ", ignore_if_exists => false"] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events', branch => '{branch}'{tag}{ignore})"), + "already exists").await; + assert_eq!(table_files(&root), before); + } + assert_ok(&context, &format!("table => 'test_db.events', branch => '{branch}'{tag}, ignore_if_exists => true")).await; + assert_eq!(table_files(&root), before); + } + } + assert_eq!( + row_count( + &context, + "SELECT * FROM paimon.test_db.`events$branch_empty`" + ) + .await, + 0 + ); + assert_eq!( + collect_id_name( + &context, + "SELECT id, name FROM paimon.test_db.`events$branch_tagged`" + ) + .await, + vec![(1, "alice".into())] + ); +} + +#[tokio::test] +async fn invalid_names_are_rejected_before_ignore_existing_checks() { + let (tmp, _, context, _) = setup().await; + assert_ok(&context, "table => 'test_db.events', branch => 'keep'").await; + let root = tmp.path().join("test_db.db/events"); + for name in ["main", "123"] { + let path = root.join(format!("branch/branch-{name}")); + std::fs::create_dir(&path).unwrap(); + std::fs::write(path.join("marker"), "keep").unwrap(); + } + let before = table_files(&root); + for (name, message) in [ + ("main", "default branch"), + ("", "empty"), + (" ", "whitespace"), + ("123", "numeric"), + (".", "cannot be"), + ("..", "cannot be"), + ("keep/schema", "path separators"), + ("a\\b", "path separators"), + ("a\n", "control characters"), + ] { + for ignore in [false, true] { + for tag in ["", ", tag => 'release'"] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events', branch => '{name}'{tag}, ignore_if_exists => {ignore})"), message).await; + assert_eq!(table_files(&root), before); + } + } + } +} + +#[tokio::test] +async fn sql_delimiter_in_branch_names_is_rejected_before_writing_or_ignoring() { + let (tmp, _, context, table) = setup().await; + let root = tmp.path().join("test_db.db/events"); + BranchManager::new(table.file_io().clone(), table.location().to_string()) + .create_branch_from_tag("legacy$blue", "release") + .await + .unwrap(); + let before = table_files(&root); + + for branch in ["dev$blue", "$dev", "dev$", "legacy$blue"] { + for tag in [", tag => 'release'", ""] { + for ignore in [ + "", + ", ignore_if_exists => false", + ", ignore_if_exists => true", + ] { + assert_sql_error( + &context, + &format!( + "CALL sys.create_branch(table => 'test_db.events', branch => '{branch}'{tag}{ignore})" + ), + "SQL suffix delimiter '$'", + ) + .await; + assert_eq!(table_files(&root), before); + if branch != "legacy$blue" { + assert!(!root.join(format!("branch/branch-{branch}")).exists()); + } + } + } + } +} + +#[tokio::test] +async fn branch_names_preserve_spaces_and_tag_names_follow_core_validation() { + let (tmp, _, context, table) = setup().await; + let root = tmp.path().join("test_db.db/events"); + let before = main_files(&root); + assert_ok( + &context, + "table => 'test_db.events', branch => 'dev', tag => 'release'", + ) + .await; + for name in [" dev", "dev "] { + assert_ok( + &context, + &format!("table => 'test_db.events', branch => '{name}', ignore_if_exists => true"), + ) + .await; + assert_eq!( + row_count( + &context, + &format!("SELECT * FROM paimon.test_db.`events$branch_{name}`") + ) + .await, + 0 + ); + } + assert_eq!( + row_count(&context, "SELECT * FROM paimon.test_db.`events$branches`").await, + 3 + ); + assert_eq!( + collect_id_name( + &context, + "SELECT id, name FROM paimon.test_db.`events$branch_dev`" + ) + .await, + vec![(1, "alice".into())] + ); + assert_eq!(main_files(&root), before); + exec( + &context, + "CALL sys.create_tag(table => 'test_db.events', tag => ' release')", + ) + .await; + let before = main_files(&root); + assert_ok( + &context, + "table => 'test_db.events', branch => 'spaced_tag', tag => ' release'", + ) + .await; + assert_eq!( + table + .copy_with_branch("spaced_tag") + .await + .unwrap() + .schema() + .id(), + table.schema().id() + ); + assert_eq!( + collect_id_name( + &context, + "SELECT id, name FROM paimon.test_db.`events$branch_spaced_tag`" + ) + .await, + vec![(1, "alice".into()), (2, "bob".into())] + ); + assert_eq!(main_files(&root), before); +} + +#[tokio::test] +async fn invalid_arguments_and_missing_tag_leave_metadata_unchanged() { + let (tmp, _, context, _) = setup().await; + assert_ok(&context, "table => 'test_db.events', branch => 'existing'").await; + let root = tmp.path().join("test_db.db/events"); + let before = table_files(&root); + for (arguments, message) in [ + ("branch => 'new'", "Missing required argument: 'table'"), + ("table => 'test_db.events'", "Missing required argument: 'branch'"), + ("table => 'test_db.events', branch => 'new', tag => 'missing'", "Tag 'missing' does not exist"), + ("table => 'test_db.events', branch => 'new', tag => ''", "tag name"), + ("table => 'test_db.events', branch => 'new', tag => 'release '", "Invalid tag name"), + ("table => 'test_db.events', branch => 'new', ignore_if_exists => true, tag => 'missing'", "Tag 'missing' does not exist"), + ("table => 'test_db.events', branch => 'existing', ignore_if_exists => 'yes'", "Invalid ignore_if_exists"), + ("table => 'test_db.events', branch => 'existing', ignore_if_exists => 1", "Invalid ignore_if_exists"), + ("table => 'test_db.events', branch => 'new', ignore_if_exists => NULL", "Unsupported argument value"), + ("table => 'test_db.events', branch => 'existing', ignoreIfExists => true", "Argument ignoreifexists is unknown"), + ("table => 'test_db.events', branch => 'existing', tags => 'release', ignore_if_exists => true", "Argument tags is unknown"), + ("table => 'test_db.events', branch => 'new', branch => 'other'", "argument branch is duplicated"), + ("'test_db.events', 'new'", "require named arguments"), + ("table => 'paimon.other.events', branch => 'new'", "does not exist"), + ("table => 'other.test_db.events', branch => 'new'", "Catalog name mismatch"), + ] { + assert_sql_error(&context, &format!("CALL sys.create_branch({arguments})"), message).await; + assert_eq!(table_files(&root), before); + } + assert!(!root.join("branch/branch-new").exists()); +} + +#[tokio::test] +async fn unsupported_table_types_are_rejected_even_when_target_exists() { + for table_type in [ + "format-table", + "object-table", + "iceberg-table", + "lance-table", + ] { + let (tmp, catalog) = create_test_env(); + catalog + .create_database("test_db", false, Default::default()) + .await + .unwrap(); + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("type", table_type) + .option("file.format", "parquet") + .build() + .unwrap(); + catalog + .create_table(&Identifier::new("test_db", "events"), schema, false) + .await + .unwrap(); + let root = tmp.path().join("test_db.db/events"); + std::fs::create_dir_all(root.join("branch/branch-existing")).unwrap(); + std::fs::write(root.join("branch/branch-existing/marker"), "keep").unwrap(); + let before = table_files(&root); + let context = create_sql_context(catalog).await; + let message = if table_type == "format-table" { + "requires a FileStoreTable" + } else { + "cannot be read as a Paimon table" + }; + for branch in ["new", "existing"] { + for tag in ["", ", tag => 'missing'"] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events', branch => '{branch}'{tag}, ignore_if_exists => true)"), message).await; + assert_eq!(table_files(&root), before); + } + } + assert!(!root.join("branch/branch-new").exists()); + } +} + +#[tokio::test] +async fn branch_suffixes_are_not_resolved_and_non_main_tables_are_rejected() { + let (tmp, catalog, context, table) = setup().await; + assert_ok( + &context, + "table => 'test_db.events', branch => 'dev', tag => 'release'", + ) + .await; + let root = tmp.path().join("test_db.db/events"); + let before = table_files(&root); + for suffix in ["$branch_dev", "$branch_main", "$branches"] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events{suffix}', branch => 'new', ignore_if_exists => true)"), + "does not exist").await; + assert_eq!(table_files(&root), before); + } + let branch = table.copy_with_branch("dev").await.unwrap(); + let context = support::context_with_table(catalog, branch).await; + for branch in ["new", "dev"] { + for tag in ["", ", tag => 'release'"] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events', branch => '{branch}'{tag}, ignore_if_exists => true)"), + "requires a main-branch source table").await; + assert_eq!(table_files(&root), before); + } + } + assert!(!root.join("branch/branch-new").exists()); +} + +#[derive(Debug, Default)] +struct FailingProvider { + calls: AtomicUsize, +} + +#[async_trait::async_trait] +impl FileIOProvider for FailingProvider { + async fn create(&self, _path: &str) -> paimon::Result<(opendal::Operator, String)> { + self.calls.fetch_add(1, Ordering::SeqCst); + Err(opendal::Error::new( + opendal::ErrorKind::PermissionDenied, + "injected branch storage failure", + ) + .into()) + } +} + +#[tokio::test] +async fn ignore_existing_does_not_swallow_storage_errors() { + let (tmp, catalog, _, table) = setup().await; + let root = tmp.path().join("test_db.db/events"); + let before = table_files(&root); + let provider = Arc::new(FailingProvider::default()); + let io = FileIOBuilder::new("file") + .with_provider(provider.clone()) + .build() + .unwrap(); + let table = Table::new( + io, + table.identifier().clone(), + table.location().into(), + table.schema().clone(), + None, + ); + let context = support::context_with_table(catalog, table).await; + for ignore in [false, true] { + assert_sql_error(&context, &format!("CALL sys.create_branch(table => 'test_db.events', branch => 'new', ignore_if_exists => {ignore})"), + "injected branch storage failure").await; + assert_eq!(table_files(&root), before); + } + assert_eq!(provider.calls.load(Ordering::SeqCst), 2); +} diff --git a/crates/integrations/datafusion/tests/create_branch/support.rs b/crates/integrations/datafusion/tests/create_branch/support.rs new file mode 100644 index 000000000..f2e176390 --- /dev/null +++ b/crates/integrations/datafusion/tests/create_branch/support.rs @@ -0,0 +1,129 @@ +// 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 std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use paimon::catalog::{Database, Identifier}; +use paimon::spec::{Schema, SchemaChange}; +use paimon::table::Table; +use paimon::{Catalog, FileSystemCatalog}; +use paimon_datafusion::SQLContext; + +pub async fn context_with_table(inner: Arc, table: Table) -> SQLContext { + let mut context = SQLContext::new(); + context + .register_catalog_with_default_db( + "paimon", + Arc::new(ResolvedTableCatalog { inner, table }), + None, + ) + .await + .unwrap(); + context +} + +struct ResolvedTableCatalog { + inner: Arc, + table: Table, +} + +#[async_trait] +impl Catalog for ResolvedTableCatalog { + async fn list_databases(&self) -> paimon::Result> { + self.inner.list_databases().await + } + + async fn create_database( + &self, + name: &str, + ignore_if_exists: bool, + properties: HashMap, + ) -> paimon::Result<()> { + self.inner + .create_database(name, ignore_if_exists, properties) + .await + } + + async fn get_database(&self, name: &str) -> paimon::Result { + self.inner.get_database(name).await + } + + async fn drop_database( + &self, + name: &str, + ignore_if_not_exists: bool, + cascade: bool, + ) -> paimon::Result<()> { + self.inner + .drop_database(name, ignore_if_not_exists, cascade) + .await + } + + async fn get_table(&self, identifier: &Identifier) -> paimon::Result { + assert_eq!(identifier, self.table.identifier()); + Ok(self.table.clone()) + } + + async fn list_tables(&self, database_name: &str) -> paimon::Result> { + self.inner.list_tables(database_name).await + } + + async fn create_table( + &self, + identifier: &Identifier, + creation: Schema, + ignore_if_exists: bool, + ) -> paimon::Result<()> { + self.inner + .create_table(identifier, creation, ignore_if_exists) + .await + } + + async fn drop_table( + &self, + identifier: &Identifier, + ignore_if_not_exists: bool, + ) -> paimon::Result<()> { + self.inner + .drop_table(identifier, ignore_if_not_exists) + .await + } + + async fn rename_table( + &self, + from: &Identifier, + to: &Identifier, + ignore_if_not_exists: bool, + ) -> paimon::Result<()> { + self.inner + .rename_table(from, to, ignore_if_not_exists) + .await + } + + async fn alter_table( + &self, + identifier: &Identifier, + changes: Vec, + ignore_if_not_exists: bool, + ) -> paimon::Result<()> { + self.inner + .alter_table(identifier, changes, ignore_if_not_exists) + .await + } +} diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs b/crates/integrations/datafusion/tests/rest_management_procedures.rs index 486e4c8b6..e1a50b31c 100644 --- a/crates/integrations/datafusion/tests/rest_management_procedures.rs +++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs @@ -714,6 +714,45 @@ async fn test_replace_tag_rejected_on_rest_catalog_without_changing_metadata() { } } +#[tokio::test] +async fn test_create_branch_rejects_rest_tables_even_when_target_exists() { + let (tmp, server, context) = setup().await; + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .build() + .unwrap(); + let root = tmp.path().join(format!("{DATABASE}.db/{TABLE}")); + let schema_path = root.join("schema/schema-0"); + let branch_path = root.join("branch/branch-existing/schema/schema-0"); + let schema_bytes = serde_json::to_vec(&paimon::spec::TableSchema::new(0, &schema)).unwrap(); + for path in [&schema_path, &branch_path] { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, &schema_bytes).unwrap(); + } + server.add_table_with_schema( + DATABASE, + TABLE, + schema, + &format!("file://{}", root.display()), + ); + + for branch in ["new", "existing"] { + for tag in ["", ", tag => 'missing'"] { + common::assert_sql_error( + &context, + &format!( + "CALL sys.create_branch(table => '{DATABASE}.{TABLE}', branch => '{branch}'{tag}, ignore_if_exists => true)" + ), + "create_branch is not supported for tables managed by a REST catalog", + ) + .await; + assert_eq!(std::fs::read(&schema_path).unwrap(), schema_bytes); + assert_eq!(std::fs::read(&branch_path).unwrap(), schema_bytes); + assert_eq!(std::fs::read_dir(root.join("branch")).unwrap().count(), 1); + } + } +} + #[tokio::test] async fn test_rename_branch_rejected_on_rest_catalog() { // A REST catalog is authoritative for branch metadata and has no rename diff --git a/docs/src/sql.md b/docs/src/sql.md index 8fff70bdb..f5b677b66 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1153,6 +1153,59 @@ consumers before cleanup: running scans retain their loaded positions, and check completion can recreate deleted records. A fresh scan without an explicit checkpoint follows its existing initial scan behavior after its consumer record is deleted. +### create_branch + +Create a branch from the main source table: + +```sql +-- Copy the latest schema without inheriting data, snapshots, or tags. +CALL sys.create_branch(table => 'paimon.my_db.my_table', branch => 'empty_branch'); + +-- Start from a tag's historical schema and data. +CALL sys.create_branch( + table => 'paimon.my_db.my_table', + branch => 'release_branch', + tag => 'release', + ignore_if_exists => true +); +``` + +`table` and `branch` are required. `table` accepts `database.table` or +`catalog.database.table`. Omitting `tag` creates a branch containing schema +metadata only, even when the source table has data. A specified tag must exist; +its retained snapshot can be used even after the live snapshot file has expired, +provided the referenced schemas, manifests, and data files remain available. +The source table is not modified. Query the new branch through `$branch_` +and list it through [`$branches`](#branches). + +`ignore_if_exists` defaults to `false`: an existing branch causes an error. +With `true`, an existing branch is left unchanged and the tag is not resolved, +even if the supplied tag does not exist. Both creation and an ignored request +return one row with `result = 'OK'`. Boolean literals and case-insensitive +`'true'` / `'false'` strings are accepted; other values are rejected. This +parameter uses the Rust SQL name `ignore_if_exists`, rather than Java Spark's +`ignoreIfExists`. + +Branch names must pass the Core name validator: `main`, empty or whitespace-only +names, pure ASCII numeric names, `.`, `..`, path separators, and control +characters are rejected. SQL also rejects `$` in target branch names because it +separates branch and system-table suffixes in object names. Branch names preserve +leading and trailing spaces. +When creating a branch, tag names are passed unchanged to Core; trailing +whitespace is invalid. +Branch-name and source-table validation also apply when `ignore_if_exists` is +`true`. + +Only non-REST FileStore tables resolved on `main` are supported. REST tables, +format tables, external engine tables, and already resolved non-main source +tables are rejected. The procedure does not resolve `$branch_` or system +table suffixes in its `table` argument and does not create from another branch. + +Creation reuses the Core metadata-copy path. Existence checks and file copies +are not transactional; storage errors are returned and can leave partial branch +metadata. `ignore_if_exists` checks name existence and does not repair a partial +creation. + ### create_tag Create a named tag from a snapshot: