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 e2872baf Support Java-aligned Paimon branch writes (#1033)
e2872baf is described below
commit e2872baf3b408a209f40315478c4567486c07987
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 5 09:14:26 2026 +0800
Support Java-aligned Paimon branch writes (#1033)
---
bindings/python/tests/test_resolved_table.py | 16 +-
bindings/python/tests/test_table.py | 34 +
.../datafusion/tests/sql_context_tests.rs | 18 +-
crates/paimon/src/catalog/mod.rs | 12 +
crates/paimon/src/table/branch_write_tests.rs | 728 +++++++++++++++++++++
crates/paimon/src/table/bucket_assigner_dynamic.rs | 7 +-
crates/paimon/src/table/mod.rs | 29 +-
crates/paimon/src/table/postpone_bucket_plan.rs | 6 +-
.../table/postpone_fixed_bucket_write_builder.rs | 2 +-
crates/paimon/src/table/rest_env.rs | 10 +-
crates/paimon/src/table/snapshot_commit.rs | 26 +-
crates/paimon/src/table/table_commit.rs | 47 +-
.../paimon/src/table/table_commit/parity_tests.rs | 8 +-
.../src/table/table_commit/recovery_tests.rs | 8 +-
.../src/table/table_commit/row_tracking_tests.rs | 1 +
crates/paimon/src/table/table_write.rs | 36 +-
crates/paimon/src/table/write_builder.rs | 34 +-
crates/paimon/tests/rest_catalog_test.rs | 93 +++
18 files changed, 1025 insertions(+), 90 deletions(-)
diff --git a/bindings/python/tests/test_resolved_table.py
b/bindings/python/tests/test_resolved_table.py
index 566885c7..36cfe8b9 100644
--- a/bindings/python/tests/test_resolved_table.py
+++ b/bindings/python/tests/test_resolved_table.py
@@ -255,16 +255,20 @@ def
test_resolved_rest_response_keeps_snapshot_and_token_refresh(
if '$branch_' in object_name else 'main'
)
assert table.branch() == expected_branch
- if '$branch_' in object_name:
- with pytest.raises(NotImplementedError, match='Writing to Paimon
branch'):
- table.new_batch_write_builder().new_write()
+ builder = table.new_batch_write_builder()
+ builder.new_write().close()
+ builder.new_commit().close()
assert len(token_requests) == (0 if external else 1)
assert all(path.endswith('/token') for path in requests)
assert _read(table) == (1, [{'id': 1, 'name': 'a'}])
assert all(path.endswith(('/token', '/snapshot')) for path in requests)
- encoded_name = object_name.replace('$', '%24')
- assert any(path.endswith(
- f'/databases/db/tables/{encoded_name}/snapshot') for path in
requests)
+ # Java switchToBranch uses the base identifier for main, including an
+ # explicitly supplied $branch_main alias.
+ encoded_name = 't' if expected_branch == 'main' else
object_name.replace('$', '%24')
+ snapshot_paths = [path for path in requests if
path.endswith('/snapshot')]
+ assert snapshot_paths
+ assert all(path.endswith(
+ f'/databases/db/tables/{encoded_name}/snapshot') for path in
snapshot_paths)
assert len(token_requests) == (0 if external else 2)
finally:
server.shutdown()
diff --git a/bindings/python/tests/test_table.py
b/bindings/python/tests/test_table.py
index cd7d11af..2a5b10f6 100644
--- a/bindings/python/tests/test_table.py
+++ b/bindings/python/tests/test_table.py
@@ -74,6 +74,40 @@ def test_branch_snapshots_are_isolated(branch_tables):
assert empty.list_snapshots() == []
[email protected]("branch", ["blue", "empty"])
+def test_branch_write_commit_keeps_main_history(branch_tables, branch):
+ main, blue, empty = branch_tables
+ table = {"blue": blue, "empty": empty}[branch]
+ root = Path(main.location())
+ main_snapshots = {
+ path.name: path.read_bytes() for path in (root / "snapshot").iterdir()
+ }
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.record_batch(
+ [[3], ["new"]], schema=pa.schema([("id", pa.int32()), ("dt",
pa.string())])
+ ))
+ messages = writer.prepare_commit()
+ assert messages
+ committer = builder.new_commit()
+ committer.commit(messages)
+ writer.close()
+ committer.close()
+
+ assert table.latest_snapshot().id() == (2 if branch == "blue" else 1)
+ for target, expected in ((table, [1, 3] if branch == "blue" else [3]),
(main, [1, 2])):
+ reader = target.new_read_builder()
+ batches = reader.new_read().read(reader.new_scan().plan().splits())
+ assert sorted(pa.Table.from_batches(batches).column("id").to_pylist())
== expected
+ assert {
+ path.name: path.read_bytes() for path in (root / "snapshot").iterdir()
+ } == main_snapshots
+ if branch == "blue":
+ assert empty.latest_snapshot() is None
+ else:
+ assert blue.latest_snapshot().id() == 1
+
+
def test_branch_tags_are_isolated(branch_tables):
main, blue, empty = branch_tables
assert [(t.name(), t.snapshot_id()) for t in main.list_tags()] ==
[("main", 2)]
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index 16a66f29..8e0eccea 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -528,12 +528,22 @@ async fn test_select_branch_table_reads_branch_snapshot()
{
let branch_table = table.copy_with_branch("b1").await.unwrap();
let write_builder = branch_table.new_write_builder();
- assert!(write_builder.new_write().is_err());
+ // Core branch writes are supported. The SQL layer below still validates
+ // branch-qualified DML targets separately.
+ assert!(write_builder.new_write().is_ok());
+ assert!(write_builder.try_new_commit().is_ok());
assert!(write_builder
.new_data_evolution_writer(vec!["name".to_string()])
- .is_err());
- assert!(write_builder.new_delete().is_err());
- assert!(write_builder.try_new_commit().is_err());
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("'data-evolution.enabled' = 'true'"));
+ assert!(write_builder
+ .new_delete()
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("'data-evolution.enabled' = 'true'"));
assert_sql_error_contains(
&sql_context,
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 3ffdedd7..20a13da0 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -165,6 +165,18 @@ impl Identifier {
.unwrap_or_else(|| DEFAULT_MAIN_BRANCH.to_string()))
}
+ /// Java's Identifier(database, tableName, branch) for branch-scoped
metadata.
+ pub(crate) fn with_branch(&self, branch: &str) -> Result<Self> {
+ validate_branch_name(branch)?;
+ let table = self.parsed_object_name()?.table;
+ let object = if branch == DEFAULT_MAIN_BRANCH {
+ table
+ } else {
+ format!("{table}$branch_{branch}")
+ };
+ Ok(Self::new(self.database(), object))
+ }
+
pub fn system_table_name(&self) -> Result<Option<String>> {
Ok(self.parsed_object_name()?.system_table)
}
diff --git a/crates/paimon/src/table/branch_write_tests.rs
b/crates/paimon/src/table/branch_write_tests.rs
new file mode 100644
index 00000000..283705dd
--- /dev/null
+++ b/crates/paimon/src/table/branch_write_tests.rs
@@ -0,0 +1,728 @@
+// 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 super::*;
+use crate::io::FileIOBuilder;
+use crate::spec::{DataType, IntType, Schema};
+use arrow_array::{Int32Array, RecordBatch};
+use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
+use std::sync::Arc;
+
+async fn table(primary_key: bool, options: &[(&str, &str)]) ->
(tempfile::TempDir, Table) {
+ let dir = tempfile::tempdir().unwrap();
+ // Scope the real filesystem operator to this temporary directory.
OpenDAL's
+ // default root "/" cannot list an absolute path on another Windows drive.
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some(dir.path().to_str().unwrap().to_string());
+ let file_io = FileIOBuilder::new("file")
+ .with_fs_operator(opendal::Operator::from_config(config).unwrap())
+ .build()
+ .unwrap();
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()));
+ if primary_key {
+ builder = builder.primary_key(vec!["id"]).option("bucket", "1");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("db", "t"),
+ "file:/db.db/t".into(),
+ TableSchema::new(0, &builder.build().unwrap()),
+ None,
+ );
+ file_io
+ .new_output(&table.schema_manager().schema_path(0))
+ .unwrap()
+ .write(bytes::Bytes::from(
+ serde_json::to_vec(table.schema()).unwrap(),
+ ))
+ .await
+ .unwrap();
+ (dir, table)
+}
+
+fn batch(ids: Vec<i32>, values: Vec<i32>) -> RecordBatch {
+ RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("id", ArrowType::Int32, true),
+ Field::new("value", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+}
+
+async fn append(table: &Table, user: &str, data: RecordBatch) ->
Vec<CommitMessage> {
+ let builder = table.new_write_builder().with_commit_user(user).unwrap();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ builder
+ .try_new_commit()
+ .unwrap()
+ .commit(messages.clone())
+ .await
+ .unwrap();
+ messages
+}
+
+async fn tagged_branch(table: &Table) -> Table {
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ table.tag_manager().create("base", &snapshot).await.unwrap();
+ BranchManager::new(table.file_io().clone(), table.location().into())
+ .create_branch_from_tag("dev", "base")
+ .await
+ .unwrap();
+ table.copy_with_branch("dev").await.unwrap()
+}
+
+#[tokio::test]
+async fn branch_writes_keep_append_and_primary_key_histories_isolated() {
+ for primary_key in [false, true] {
+ let (_dir, main) = table(primary_key, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ append(&main, "writer", batch(vec![1], vec![100])).await;
+ append(&branch, "writer", batch(vec![1], vec![20])).await;
+ let main_rows = table_write::tests::read_id_value_rows(&main).await;
+ let branch_rows =
table_write::tests::read_id_value_rows(&branch).await;
+ assert_eq!(
+ main_rows,
+ if primary_key {
+ vec![(1, 100)]
+ } else {
+ vec![(1, 10), (1, 100)]
+ }
+ );
+ assert_eq!(
+ branch_rows,
+ if primary_key {
+ vec![(1, 20)]
+ } else {
+ vec![(1, 10), (1, 20)]
+ }
+ );
+ let main_snapshot = main
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let branch_snapshot = branch
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(main_snapshot.id(), 2);
+ assert_eq!(branch_snapshot.id(), 2);
+ assert_ne!(main_snapshot.uuid(), branch_snapshot.uuid());
+ }
+}
+
+async fn stage(
+ table: &Table,
+ user: &str,
+ data: RecordBatch,
+ overwrite: bool,
+) -> Vec<CommitMessage> {
+ let builder = table.new_write_builder().with_commit_user(user).unwrap();
+ let builder = if overwrite {
+ builder.with_overwrite()
+ } else {
+ builder
+ };
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ writer.prepare_commit().await.unwrap()
+}
+
+async fn latest(table: &Table) -> crate::spec::Snapshot {
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+}
+
+#[tokio::test]
+async fn branch_write_checkpoint_recovery_is_scoped_to_the_branch() {
+ let (_dir, main) = table(true, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ for (target, value) in [(&main, 100), (&branch, 20)] {
+ let messages = stage(target, "same-writer", batch(vec![1],
vec![value]), false).await;
+ let commit = target
+ .new_write_builder()
+ .with_commit_user("same-writer")
+ .unwrap()
+ .new_commit();
+ commit
+ .filter_and_commit_with_identifier(messages.clone(), 7)
+ .await
+ .unwrap();
+ assert_eq!(latest(target).await.id(), 2);
+ assert_eq!(latest(target).await.commit_identifier(), 7);
+ commit
+ .filter_and_commit_with_identifier(messages, 7)
+ .await
+ .unwrap();
+ assert_eq!(latest(target).await.id(), 2);
+ }
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 100)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 20)]
+ );
+}
+
+#[tokio::test]
+async fn branch_write_overwrite_and_truncate_preserve_main_and_shared_files() {
+ for primary_key in [false, true] {
+ let (_dir, main) = table(primary_key, &[]).await;
+ let original = append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ let before = latest(&main).await;
+ let messages = stage(&branch, "overwrite", batch(vec![2], vec![20]),
true).await;
+ let commit = branch
+ .new_write_builder()
+ .with_commit_user("overwrite")
+ .unwrap()
+ .new_commit();
+ commit.overwrite(messages, None).await.unwrap();
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(2, 20)]
+ );
+ assert_eq!(latest(&main).await, before);
+ commit.truncate_table().await.unwrap();
+ assert_eq!(latest(&branch).await.id(), 3);
+ assert!(table_write::tests::read_id_value_rows(&branch)
+ .await
+ .is_empty());
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 10)]
+ );
+ let scan = main.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(
+ scan.splits()
+ .iter()
+ .map(|split| split.row_count())
+ .sum::<i64>(),
+ 1
+ );
+ assert!(!original.is_empty());
+ }
+}
+
+#[tokio::test]
+async fn branch_write_dynamic_buckets_load_only_the_branch_hash_index() {
+ let (_dir, main) = table(
+ true,
+ &[("bucket", "-1"), ("dynamic-bucket.target-row-num", "1")],
+ )
+ .await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ append(&main, "main", batch(vec![2], vec![20])).await;
+ append(&main, "main", batch(vec![3], vec![30])).await;
+ let messages = append(&branch, "dev", batch(vec![4], vec![40])).await;
+ assert_eq!(
+ messages
+ .iter()
+ .map(|message| message.bucket)
+ .collect::<Vec<_>>(),
+ vec![1]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 10), (4, 40)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 10), (2, 20), (3, 30)]
+ );
+}
+
+fn update_batch(row_id: i64, value: i32) -> RecordBatch {
+ RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("_ROW_ID", ArrowType::Int64, false),
+ Field::new("value", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(arrow_array::Int64Array::from(vec![row_id])),
+ Arc::new(Int32Array::from(vec![value])),
+ ],
+ )
+ .unwrap()
+}
+
+#[tokio::test]
+async fn branch_write_row_ids_and_conflicts_use_branch_snapshot_history() {
+ let (_dir, main) = table(
+ false,
+ &[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ],
+ )
+ .await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ let stale_main = main
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update_batch(0, 100)])
+ .await
+ .unwrap();
+ let update = branch
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update_batch(0, 20)])
+ .await
+ .unwrap();
+ branch
+ .new_write_builder()
+ .new_commit()
+ .commit(update)
+ .await
+ .unwrap();
+ // A change to the same column in another branch is not a conflict.
+ main.new_write_builder()
+ .new_commit()
+ .commit(stale_main)
+ .await
+ .unwrap();
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 20)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 100)]
+ );
+ let stale = branch
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update_batch(0, 21)])
+ .await
+ .unwrap();
+ let fresh = branch
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update_batch(0, 22)])
+ .await
+ .unwrap();
+ branch
+ .new_write_builder()
+ .new_commit()
+ .commit(fresh)
+ .await
+ .unwrap();
+ let error = branch
+ .new_write_builder()
+ .new_commit()
+ .commit(stale)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("multiple MERGE INTO"), "{error}");
+ assert_eq!(latest(&branch).await.id(), 3);
+ assert_eq!(latest(&branch).await.next_row_id(), Some(1));
+ assert_eq!(latest(&main).await.id(), 2);
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 22)]
+ );
+}
+
+#[tokio::test]
+async fn branch_write_empty_branch_starts_its_own_row_id_space() {
+ let (_dir, main) = table(
+ false,
+ &[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ],
+ )
+ .await;
+ append(&main, "seed", batch(vec![1, 2], vec![10, 20])).await;
+ BranchManager::new(main.file_io().clone(), main.location().into())
+ .create_branch("empty")
+ .await
+ .unwrap();
+ let branch = main.copy_with_branch("empty").await.unwrap();
+ append(&branch, "dev", batch(vec![3], vec![30])).await;
+ assert_eq!(latest(&branch).await.id(), 1);
+ assert_eq!(latest(&branch).await.next_row_id(), Some(1));
+ assert_eq!(latest(&main).await.next_row_id(), Some(2));
+ let update = branch
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update_batch(0, 31)])
+ .await
+ .unwrap();
+ branch
+ .new_write_builder()
+ .new_commit()
+ .commit(update)
+ .await
+ .unwrap();
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(3, 31)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 10), (2, 20)]
+ );
+}
+
+#[tokio::test]
+async fn branch_write_missing_schema_never_creates_a_branch() {
+ let (_dir, main) = table(false, &[]).await;
+ let branch = main
+ .copy_with_resolved_schema(main.schema().clone(), "absent")
+ .unwrap();
+ let mut writer = branch.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(vec![1], vec![10]))
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("does not contain schema"),
+ "{error}"
+ );
+ let messages = stage(&main, "writer", batch(vec![1], vec![10]),
false).await;
+ let error = branch
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("does not exist"), "{error}");
+ assert!(branch
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ assert!(main
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ BranchManager::new(main.file_io().clone(), main.location().into())
+ .create_branch("absent")
+ .await
+ .unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(vec![1], vec![10]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("cannot be reused"), "{error}");
+ assert!(writer.prepare_commit().await.is_err());
+}
+
+#[tokio::test]
+async fn branch_write_historical_handles_remain_read_only() {
+ let (_dir, main) = table(false, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ let snapshot = latest(&branch).await;
+ let mut historical = branch.copy_with_pinned_snapshot(&snapshot);
+ // The pinned snapshot remains read-only even with no selector in its
options.
+ historical.schema = historical
+ .schema
+ .copy_with_replaced_options(branch.schema().options().clone());
+ assert!(historical.new_write_builder().new_write().is_err());
+ assert!(historical.new_write_builder().try_new_commit().is_err());
+ assert!(historical
+ .new_write_builder()
+ .new_commit()
+ .commit(vec![])
+ .await
+ .is_err());
+ assert_eq!(latest(&branch).await, snapshot);
+}
+
+#[tokio::test]
+async fn
branch_write_renaming_publisher_routes_snapshot_and_hint_by_argument() {
+ use super::snapshot_commit::{RenamingSnapshotCommit, SnapshotCommit};
+ let (_dir, main) = table(false, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let original = latest(&main).await;
+ BranchManager::new(main.file_io().clone(), main.location().into())
+ .create_branch("dev")
+ .await
+ .unwrap();
+ let publisher = RenamingSnapshotCommit::new(main.snapshot_manager());
+ assert!(publisher.commit(None, &original, "dev", &[]).await.unwrap());
+ let branch = main.copy_with_branch("dev").await.unwrap();
+ assert_eq!(latest(&branch).await, original);
+ assert!(!publisher.commit(None, &original, "dev", &[]).await.unwrap());
+ assert_eq!(latest(&main).await, original);
+ let branch_hint = format!("{}/LATEST",
branch.snapshot_manager().snapshot_dir());
+ assert_eq!(
+ main.file_io()
+ .new_input(&branch_hint)
+ .unwrap()
+ .read()
+ .await
+ .unwrap()
+ .as_ref(),
+ b"1"
+ );
+ assert!(publisher
+ .commit(None, &original, "../invalid", &[])
+ .await
+ .is_err());
+}
+
+#[tokio::test]
+async fn branch_write_schema_id_is_resolved_in_its_own_namespace() {
+ let (_dir, main) = table(false, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ let evolved = main
+ .schema()
+ .apply_changes(vec![crate::spec::SchemaChange::SetOption {
+ key: "target-file-row-num".into(),
+ value: "10".into(),
+ }])
+ .unwrap();
+ main.file_io()
+ .new_output(&main.schema_manager().schema_path(evolved.id()))
+ .unwrap()
+ .write(bytes::Bytes::from(serde_json::to_vec(&evolved).unwrap()))
+ .await
+ .unwrap();
+ // A stale writer may publish under the latest schema ID, but that ID must
+ // come from its branch, not from an independently evolving main branch.
+ append(&main, "main", batch(vec![2], vec![20])).await;
+ append(&branch, "dev", batch(vec![3], vec![30])).await;
+ assert_eq!(latest(&main).await.schema_id(), 1);
+ assert_eq!(latest(&branch).await.schema_id(), 0);
+ assert_eq!(main.copy_with_branch("dev").await.unwrap().schema().id(), 0);
+ let evolved = evolved
+ .apply_changes(vec![crate::spec::SchemaChange::SetOption {
+ key: "target-file-row-num".into(),
+ value: "20".into(),
+ }])
+ .unwrap();
+ branch
+ .file_io()
+ .new_output(&branch.schema_manager().schema_path(evolved.id()))
+ .unwrap()
+ .write(bytes::Bytes::from(serde_json::to_vec(&evolved).unwrap()))
+ .await
+ .unwrap();
+ append(&branch, "dev", batch(vec![4], vec![40])).await;
+ assert_eq!(latest(&branch).await.schema_id(), 2);
+ assert_eq!(latest(&main).await.schema_id(), 1);
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 10), (3, 30), (4, 40)]
+ );
+}
+
+#[tokio::test]
+async fn
branch_write_rest_publisher_routes_by_argument_and_preserves_payload() {
+ use crate::api::rest_api::RESTApi;
+ use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
+ use axum::{extract::Path, routing::post, Json, Router};
+ use std::sync::Mutex;
+
+ let requests = Arc::new(Mutex::new(Vec::new()));
+ let recorded = requests.clone();
+ let app = Router::new().route(
+ "/v1/databases/:db/tables/:table/commit",
+ post(
+ move |Path((db, table)): Path<(String, String)>,
+ Json(body): Json<serde_json::Value>| {
+ let recorded = recorded.clone();
+ async move {
+ recorded.lock().unwrap().push((db, table, body));
+ Json(serde_json::json!({"success": true}))
+ }
+ },
+ ),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let address = listener.local_addr().unwrap();
+ let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
+ let mut options = crate::Options::new();
+ options.set("uri", format!("http://{address}";));
+ options.set("warehouse", "test");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ let api = Arc::new(RESTApi::new(options, false).await.unwrap());
+ // Selecting main must strip the old branch from the publisher's
identifier.
+ let publisher = RESTSnapshotCommit::new(
+ api,
+ Identifier::new("db", "t$branch_old"),
+ "table-uuid".into(),
+ );
+ let (_dir, main) = table(false, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let snapshot = latest(&main).await;
+ for branch in ["main", "dev", "audit"] {
+ assert!(publisher
+ .commit(Some("base-uuid"), &snapshot, branch, &[])
+ .await
+ .unwrap());
+ }
+ assert!(publisher
+ .commit(None, &snapshot, "../bad", &[])
+ .await
+ .is_err());
+ let requests = requests.lock().unwrap();
+ assert_eq!(requests.len(), 3);
+ for (request, name) in requests.iter().zip(["t", "t$branch_dev",
"t$branch_audit"]) {
+ assert_eq!(request.0, "db");
+ assert_eq!(request.1, name);
+ assert_eq!(request.2["tableId"], "table-uuid");
+ assert_eq!(request.2["baseSnapshotUuid"], "base-uuid");
+ assert_eq!(
+ request.2["snapshot"],
+ serde_json::to_value(&snapshot).unwrap()
+ );
+ assert_eq!(request.2["statistics"], serde_json::json!([]));
+ }
+ server.abort();
+}
+
+#[tokio::test]
+async fn branch_write_sequence_numbers_ignore_main_advancement() {
+ let (_dir, main) = table(true, &[]).await;
+ append(&main, "seed", batch(vec![1], vec![10])).await;
+ let branch = tagged_branch(&main).await;
+ append(&main, "main", batch(vec![1], vec![100])).await;
+ append(&main, "main", batch(vec![1], vec![200])).await;
+ let messages = stage(&branch, "dev", batch(vec![1], vec![20]),
false).await;
+ assert_eq!(messages[0].new_files[0].min_sequence_number, 1);
+ assert_eq!(messages[0].new_files[0].max_sequence_number, 1);
+ branch
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(latest(&main).await.id(), 3);
+ assert_eq!(latest(&branch).await.id(), 2);
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 200)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 20)]
+ );
+}
+
+async fn fixed_bucket_write(table: &Table, value: i32, buckets: i32,
overwrite: bool) {
+ let plan_batch = RecordBatch::try_from_iter([(
+ "total_buckets",
+ Arc::new(Int32Array::from(vec![buckets])) as arrow_array::ArrayRef,
+ )])
+ .unwrap();
+ let plan = PostponeBucketPlan::from_arrow(table, &plan_batch).unwrap();
+ let builder = table
+ .new_postpone_fixed_bucket_write_builder()
+ .unwrap()
+ .with_bucket_plan(plan);
+ let builder = if overwrite {
+ builder.with_overwrite()
+ } else {
+ builder
+ };
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(vec![1], vec![value]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ builder
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn branch_write_postpone_restores_branch_bucket_layout_and_sequence() {
+ let (_dir, main) = table(
+ true,
+ &[("bucket", "-2"), ("postpone.default-bucket-num", "2")],
+ )
+ .await;
+ fixed_bucket_write(&main, 10, 2, false).await;
+ let branch = tagged_branch(&main).await;
+ // Main changes the partition's layout and advances beyond dev's baseline.
+ fixed_bucket_write(&main, 100, 3, true).await;
+ fixed_bucket_write(&main, 200, 3, false).await;
+ let builder = branch.new_postpone_fixed_bucket_write_builder().unwrap();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(vec![1], vec![20]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages[0].total_buckets, Some(2));
+ assert_eq!(messages[0].check_from_snapshot, Some(1));
+ assert_eq!(messages[0].new_files[0].min_sequence_number, 1);
+ builder
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(latest(&main).await.id(), 3);
+ assert_eq!(latest(&branch).await.id(), 2);
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 200)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 20)]
+ );
+}
diff --git a/crates/paimon/src/table/bucket_assigner_dynamic.rs
b/crates/paimon/src/table/bucket_assigner_dynamic.rs
index 6dbc50aa..ee30e5ef 100644
--- a/crates/paimon/src/table/bucket_assigner_dynamic.rs
+++ b/crates/paimon/src/table/bucket_assigner_dynamic.rs
@@ -31,7 +31,7 @@ use crate::table::bucket_assigner::{BatchAssignOutput,
BucketAssigner, Partition
use crate::table::data_file_path_factory::{DataFilePath, DataFilePathFactory};
use crate::table::index_file_path::IndexFileLocation;
use crate::table::partition_filter::PartitionFilter;
-use crate::table::{Snapshot, SnapshotManager, Table, TableScan};
+use crate::table::{Snapshot, Table, TableScan};
use crate::Result;
use arrow_array::RecordBatch;
use rand::seq::SliceRandom;
@@ -540,10 +540,7 @@ impl DynamicBucketAssigner {
self.cached_index_entries = Some(Vec::new());
return Ok(());
}
- let snapshot_manager = SnapshotManager::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- );
+ let snapshot_manager = self.table.snapshot_manager();
let latest_snapshot = snapshot_manager.get_latest_snapshot().await?;
let entries = if let Some(snapshot) = &latest_snapshot {
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 67b40c72..82039211 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -26,6 +26,8 @@ mod bitmap_global_index_reader;
mod bitmap_global_index_writer;
mod blob_resolver;
mod branch_manager;
+#[cfg(test)]
+mod branch_write_tests;
mod bucket_assigner;
mod bucket_assigner_constant;
mod bucket_assigner_cross;
@@ -284,7 +286,8 @@ impl Table {
/// invariants are validated — primary-key/partition columns must exist and
/// field names/ids must be unique — so a malformed external schema is
/// rejected here instead of panicking or reading the wrong column later.
The
- /// branch only selects the branch-scoped managers used by subsequent
reads.
+ /// branch selects the metadata managers for subsequent reads and writes.
+ /// Filesystem writes require the supplied schema to exist in that branch.
pub fn from_resolved_schema(
file_io: FileIO,
identifier: Identifier,
@@ -805,6 +808,30 @@ impl Table {
validate_branch_name(branch_name)?;
branch_name.to_string()
};
+ if let Some(env) = &self.rest_env {
+ // Catalog branches may have no schema directory in the data store.
+ // Load their schema, identity and FileIO through the same catalog.
+ let identifier = env.identifier().with_branch(&branch)?;
+ let mut table = env.get_table(&identifier).await?;
+ if table.location() != self.location()
+ || table
+ .rest_env
+ .as_ref()
+ .is_none_or(|branch_env| branch_env.uuid() != env.uuid())
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Branch '{branch}' no longer belongs to the loaded
table '{}'; reload the table before switching branches",
+ self.identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ let mut options = table.schema.options().clone();
+ options.insert("branch".to_string(), branch);
+ table.schema = table.schema.copy_with_replaced_options(options);
+ return Ok(table);
+ }
let schema_manager = if branch == DEFAULT_MAIN_BRANCH {
SchemaManager::new(self.file_io.clone(), self.location.clone())
} else {
diff --git a/crates/paimon/src/table/postpone_bucket_plan.rs
b/crates/paimon/src/table/postpone_bucket_plan.rs
index 613ba6b4..f17c6db6 100644
--- a/crates/paimon/src/table/postpone_bucket_plan.rs
+++ b/crates/paimon/src/table/postpone_bucket_plan.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::spec::batch_to_serialized_bytes;
-use crate::table::{SnapshotManager, Table, TableScan};
+use crate::table::{Table, TableScan};
use crate::Result;
use arrow_array::{Array, Int32Array, RecordBatch};
use std::collections::HashMap;
@@ -135,9 +135,7 @@ impl PostponeBucketPlan {
if snapshot_id == 0 {
return Ok(());
}
- let snapshot = SnapshotManager::new(table.file_io().clone(),
table.location().to_string())
- .get_snapshot(snapshot_id)
- .await?;
+ let snapshot =
table.snapshot_manager().get_snapshot(snapshot_id).await?;
let entries = TableScan::new(table, None, vec![], None, None, None)
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
diff --git a/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
b/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
index bc1537cf..6f7e3d3c 100644
--- a/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
+++ b/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
@@ -77,7 +77,7 @@ impl<'a> PostponeFixedBucketWriteBuilder<'a> {
}
pub fn try_new_commit(&self) -> Result<PostponeFixedBucketTableCommit> {
- self.table.ensure_not_branch_reference_for_write()?;
+ ensure_table_write_allowed(self.table)?;
Ok(self.new_commit())
}
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index cf3ad2a6..1b3375bb 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -433,15 +433,7 @@ impl RESTEnv {
&self,
branch: &str,
) -> Result<Option<crate::spec::Snapshot>> {
- let parsed_identifier = self.identifier.parsed_object_name()?;
- let object = if parsed_identifier.branch() == Some(branch) {
- format!("{}$branch_{branch}", parsed_identifier.table())
- } else if branch == crate::catalog::DEFAULT_MAIN_BRANCH {
- parsed_identifier.table().to_string()
- } else {
- format!("{}$branch_{branch}", parsed_identifier.table())
- };
- let identifier = Identifier::new(self.identifier.database(), object);
+ let identifier = self.identifier.with_branch(branch)?;
match self.api.load_snapshot(&identifier).await {
Ok(snapshot) => Ok(snapshot.map(|snapshot| snapshot.snapshot)),
Err(Error::RestApi {
diff --git a/crates/paimon/src/table/snapshot_commit.rs
b/crates/paimon/src/table/snapshot_commit.rs
index 942e6d77..7f646090 100644
--- a/crates/paimon/src/table/snapshot_commit.rs
+++ b/crates/paimon/src/table/snapshot_commit.rs
@@ -17,10 +17,10 @@
//! SnapshotCommit abstraction for atomic snapshot commits.
//!
-//! Reference: [pypaimon
snapshot_commit.py](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/snapshot/snapshot_commit.py)
+//! Reference: [Java
SnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java)
use crate::api::rest_api::RESTApi;
-use crate::catalog::Identifier;
+use crate::catalog::{validate_branch_name, Identifier};
use crate::spec::{PartitionStatistics, Snapshot};
use crate::table::SnapshotManager;
use crate::Result;
@@ -35,18 +35,20 @@ use std::sync::Arc;
#[async_trait]
pub trait SnapshotCommit: Send + Sync {
/// Commit the given snapshot. Returns true if successful, false if
- /// another writer won the race.
+ /// another writer won the race. `branch` selects the snapshot namespace;
+ /// data files and manifests remain shared under the table location.
async fn commit(
&self,
base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
+ branch: &str,
statistics: &[PartitionStatistics],
) -> Result<bool>;
}
/// A SnapshotCommit using file renaming to commit.
///
-/// Reference: [pypaimon
RenamingSnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/snapshot/renaming_snapshot_commit.py)
+/// Reference: [Java
RenamingSnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java)
pub struct RenamingSnapshotCommit {
snapshot_manager: SnapshotManager,
}
@@ -63,16 +65,22 @@ impl SnapshotCommit for RenamingSnapshotCommit {
&self,
_base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
+ branch: &str,
_statistics: &[PartitionStatistics],
) -> Result<bool> {
- // statistics are not used in file system mode (same as Python)
- self.snapshot_manager.commit_snapshot(snapshot).await
+ // Like Java RenamingSnapshotCommit, the argument selects the metadata
+ // namespace even when this publisher was constructed for another
branch.
+ validate_branch_name(branch)?;
+ self.snapshot_manager
+ .with_branch(branch)
+ .commit_snapshot(snapshot)
+ .await
}
}
/// A SnapshotCommit using REST API to commit (e.g. REST Catalog).
///
-/// Reference: [pypaimon
RESTSnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/snapshot/catalog_snapshot_commit.py)
+/// Reference: [Java
CatalogSnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java)
pub struct RESTSnapshotCommit {
api: Arc<RESTApi>,
identifier: Identifier,
@@ -95,11 +103,13 @@ impl SnapshotCommit for RESTSnapshotCommit {
&self,
base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
+ branch: &str,
statistics: &[PartitionStatistics],
) -> Result<bool> {
+ let identifier = self.identifier.with_branch(branch)?;
self.api
.commit_snapshot(
- &self.identifier,
+ &identifier,
&self.uuid,
base_snapshot_uuid,
snapshot,
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 78f17710..17cc19ba 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -262,7 +262,7 @@ impl TableCommit {
mut commits: Vec<(i64, Vec<CommitMessage>)>,
) -> Result<usize> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
commits.sort_by_key(|(id, _)| *id);
for pair in commits.windows(2) {
if pair[0].0 == pair[1].0 {
@@ -360,7 +360,7 @@ impl TableCommit {
// A refusal here must not clean up: a retry with an identifier that
// already committed names files a snapshot references.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, false)?;
validate_bucket_ownership(&commit_messages)?;
@@ -408,7 +408,7 @@ impl TableCommit {
) -> Result<()> {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, false)?;
validate_bucket_ownership(&commit_messages)?;
@@ -504,7 +504,7 @@ impl TableCommit {
) -> Result<()> {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, true)?;
validate_bucket_ownership(&commit_messages)?;
@@ -765,7 +765,7 @@ impl TableCommit {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.ensure_not_format_table()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
if partitions.is_empty() {
return Ok(());
@@ -813,7 +813,7 @@ impl TableCommit {
partitions: Vec<HashMap<String, Option<Datum>>>,
commit_identifier: i64,
) -> Result<()> {
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
if partitions.is_empty() {
return Err(crate::Error::DataInvalid {
@@ -847,7 +847,7 @@ impl TableCommit {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.ensure_not_format_table()?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
self.try_commit(
CommitEntriesPlan::Overwrite {
@@ -895,7 +895,7 @@ impl TableCommit {
}
CoreOptions::new(self.table.schema().options())
.ensure_type_paimon_served(&self.table.identifier().full_name())?;
- self.table.ensure_not_branch_reference_for_write()?;
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
let table_path = self.table.location().trim_end_matches('/');
let index_file_in_data_file_dir =
@@ -1175,7 +1175,12 @@ impl TableCommit {
let base_snapshot_uuid =
latest_snapshot.as_ref().and_then(Snapshot::uuid);
let publication_error = match self
.snapshot_commit
- .commit(base_snapshot_uuid, &snapshot, &statistics)
+ .commit(
+ base_snapshot_uuid,
+ &snapshot,
+ self.table.branch(),
+ &statistics,
+ )
.await
{
Ok(true) => return Ok(CommitAttemptResult::Success),
@@ -1372,7 +1377,7 @@ impl TableCommit {
if let Some(env) = &self.table.rest_env {
return env
.api()
- .get_table(env.identifier())
+ .get_table(&env.identifier().with_branch(self.table.branch())?)
.await?
.schema_id
.ok_or_else(|| crate::Error::DataInvalid {
@@ -1380,14 +1385,20 @@ impl TableCommit {
source: None,
});
}
- // Tables constructed directly by callers need not have schema files.
- Ok(self
- .table
- .schema_manager()
- .latest()
- .await?
- .map(|schema| schema.id())
- .unwrap_or(self.table.schema().id()))
+ let latest = self.table.schema_manager().latest().await?;
+ if let Some(schema) = latest {
+ return Ok(schema.id());
+ }
+ // A resolved schema may open a main table without on-disk metadata.
+ // Selecting a branch must never implicitly create it during
publication.
+ if self.table.is_main_branch() {
+ Ok(self.table.schema().id())
+ } else {
+ Err(crate::Error::DataInvalid {
+ message: format!("Branch '{}' does not exist.",
self.table.branch()),
+ source: None,
+ })
+ }
}
async fn inherited_statistics(
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
index e4aa339c..bd272425 100644
--- a/crates/paimon/src/table/table_commit/parity_tests.rs
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -307,7 +307,13 @@ struct LostResponseCommit {
#[async_trait::async_trait]
impl SnapshotCommit for LostResponseCommit {
- async fn commit(&self, _: Option<&str>, snapshot: &Snapshot, _:
&[PartitionStatistics]) -> Result<bool> {
+ async fn commit(
+ &self,
+ _: Option<&str>,
+ snapshot: &Snapshot,
+ _: &str,
+ _: &[PartitionStatistics],
+ ) -> Result<bool> {
let attempt = self.calls.fetch_add(1,
std::sync::atomic::Ordering::SeqCst);
if attempt == 0 {
if self.publish_first {
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
index 32b5d3ec..fbe1d67f 100644
--- a/crates/paimon/src/table/table_commit/recovery_tests.rs
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -399,7 +399,13 @@ struct ConcurrentDvCommit {
#[async_trait::async_trait]
impl SnapshotCommit for ConcurrentDvCommit {
- async fn commit(&self, _: Option<&str>, snapshot: &Snapshot, _:
&[PartitionStatistics]) -> Result<bool> {
+ async fn commit(
+ &self,
+ _: Option<&str>,
+ snapshot: &Snapshot,
+ _: &str,
+ _: &[PartitionStatistics],
+ ) -> Result<bool> {
if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
let other = TableCommit::new(self.table.clone(),
"concurrent".into());
match self.change {
diff --git a/crates/paimon/src/table/table_commit/row_tracking_tests.rs
b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
index 611d023d..598ee627 100644
--- a/crates/paimon/src/table/table_commit/row_tracking_tests.rs
+++ b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
@@ -244,6 +244,7 @@ impl SnapshotCommit for ConcurrentRowTrackingAppend {
&self,
_: Option<&str>,
snapshot: &Snapshot,
+ _: &str,
_: &[PartitionStatistics],
) -> Result<bool> {
let first = {
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 3154b23a..30384d9f 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -46,7 +46,7 @@ use crate::table::postpone_file_writer::{PostponeFileWriter,
PostponeWriteConfig
use crate::table::prepared_files::PreparedFiles;
use crate::table::row_kind_generator::RowKindGenerator;
use crate::table::write_batch_normalize::normalize_write_array;
-use crate::table::{Snapshot, SnapshotManager, Table, TableScan};
+use crate::table::{Snapshot, Table, TableScan};
use crate::Result;
use arrow_array::{ArrayRef, RecordBatch};
use std::collections::{HashMap, HashSet};
@@ -131,6 +131,7 @@ pub struct TableWrite {
// Keep the Format Table state off ordinary Paimon write futures' stacks.
format_writer: Option<Box<FormatTableWriter>>,
table: Table,
+ branch_schema_checked: bool,
write_schema: Arc<arrow_schema::Schema>,
partition_writers: HashMap<PartitionBucketKey, FileWriter>,
partition_computer: PartitionComputer,
@@ -195,6 +196,7 @@ impl TableWrite {
Ok(Self {
format_writer: Some(Box::new(format_writer)),
table: table.clone(),
+ branch_schema_checked: table.is_main_branch() ||
table.rest_env.is_some(),
write_schema: build_target_arrow_schema(schema.fields())?,
partition_writers: HashMap::new(),
partition_computer: PartitionComputer::new(
@@ -473,6 +475,7 @@ impl TableWrite {
Ok(Self {
format_writer: None,
table: table.clone(),
+ branch_schema_checked: table.is_main_branch() ||
table.rest_env.is_some(),
write_schema,
partition_writers: HashMap::new(),
partition_computer,
@@ -524,8 +527,7 @@ impl TableWrite {
let latest_snapshot = match sequence_snapshot {
Some(snapshot) => snapshot,
None => {
- let snapshot_manager =
- SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot_manager = table.snapshot_manager();
snapshot_manager.get_latest_snapshot().await?
}
};
@@ -548,10 +550,7 @@ impl TableWrite {
}
pub(super) async fn pin_sequence_snapshot(&mut self) -> Result<i64> {
- let snapshot_manager = SnapshotManager::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- );
+ let snapshot_manager = self.table.snapshot_manager();
let snapshot = snapshot_manager.get_latest_snapshot().await?;
let snapshot_id = snapshot.as_ref().map_or(0, Snapshot::id);
self.sequence_snapshot = Some(snapshot);
@@ -1000,6 +999,8 @@ impl TableWrite {
) -> Result<()> {
self.ensure_active()?;
let result = async {
+ self.ensure_branch_schema_exists().await?;
+
let key = (partition_bytes, bucket);
if !self.partition_writers.contains_key(&key) {
self.create_writer(key.0.clone(), key.1).await?;
@@ -1017,6 +1018,27 @@ impl TableWrite {
result
}
+ /// Resolved filesystem handles must name an existing branch schema before
+ /// creating output. Main and REST tables retain their established
metadata path.
+ async fn ensure_branch_schema_exists(&mut self) -> Result<()> {
+ if self.branch_schema_checked {
+ return Ok(());
+ }
+ let schema_path =
self.table.schema_manager().schema_path(self.schema_id);
+ if !self.table.file_io().exists(&schema_path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Branch '{}' does not contain schema {}.",
+ self.table.branch(),
+ self.schema_id
+ ),
+ source: None,
+ });
+ }
+ self.branch_schema_checked = true;
+ Ok(())
+ }
+
/// Routing can update persistent bucket-index state before it fails, just
/// as file writes can produce output. Neither failure permits a later
commit.
async fn fail_write(&mut self) {
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 59513838..41712038 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -199,7 +199,7 @@ impl<'a> PaimonWriteBuilder<'a> {
/// Try to create a new TableCommit for committing write results.
pub fn try_new_commit(&self) -> crate::Result<TableCommit> {
- self.ensure_main_branch_write()?;
+ ensure_table_write_allowed(self.table)?;
Ok(TableCommit::new(
self.table.clone(),
self.commit_user.clone(),
@@ -239,19 +239,15 @@ impl<'a> PaimonWriteBuilder<'a> {
&self,
update_columns: Vec<String>,
) -> crate::Result<DataEvolutionWriter> {
- self.ensure_main_branch_write()?;
+ ensure_table_write_allowed(self.table)?;
DataEvolutionWriter::new(self.table, update_columns)
}
/// Create a new writer for data-evolution row-id deletes.
pub fn new_delete(&self) -> crate::Result<DataEvolutionDeleteWriter> {
- self.ensure_main_branch_write()?;
+ ensure_table_write_allowed(self.table)?;
DataEvolutionDeleteWriter::new(self.table)
}
-
- fn ensure_main_branch_write(&self) -> crate::Result<()> {
- self.table.ensure_not_branch_reference_for_write()
- }
}
pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> {
@@ -265,11 +261,10 @@ pub(super) fn ensure_table_write_allowed(table: &Table)
-> crate::Result<()> {
),
});
}
- table.ensure_not_branch_reference_for_write()?;
// A time-travel table may carry a historical schema.
let selector =
crate::spec::CoreOptions::new(table.schema().options()).try_time_travel_selector();
- if !matches!(selector, Ok(None)) {
+ if table.is_time_traveled() || !matches!(selector, Ok(None)) {
return Err(crate::Error::Unsupported {
message: "Cannot write to a table with a time-travel option set \
(scan.version / scan.timestamp-millis / scan.timestamp /
scan.watermark / scan.snapshot-id / scan.tag-name)"
@@ -488,31 +483,20 @@ mod tests {
}
#[tokio::test]
- async fn test_branch_reference_rejects_write_and_index_builders() {
+ async fn test_branch_reference_write_and_index_capabilities() {
let table = as_main_branch_reference(test_postpone_pk_table(
&test_file_io(),
"memory:/test_branch_reference_writes",
));
- let write_err =
table.new_write_builder().try_new_commit().err().unwrap();
- assert!(
- matches!(write_err, crate::Error::Unsupported { ref message }
- if message == "Writing to Paimon branch 'main' is not
supported"),
- "Expected branch write rejection, got: {write_err:?}"
- );
-
- let commit_err = table
+ table.new_write_builder().new_write().unwrap();
+ table
.new_write_builder()
- .new_commit()
+ .try_new_commit()
+ .unwrap()
.commit(Vec::new())
.await
- .err()
.unwrap();
- assert!(
- matches!(commit_err, crate::Error::Unsupported { ref message }
- if message == "Writing to Paimon branch 'main' is not
supported"),
- "Expected branch commit rejection, got: {commit_err:?}"
- );
let index_err = table
.new_btree_global_index_build_builder()
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 24f217c3..c0d0c38f 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -3138,3 +3138,96 @@ async fn test_rest_catalog_manages_policies_end_to_end()
{
.unwrap();
assert!(page.elements.is_empty());
}
+
+#[tokio::test]
+async fn test_branch_write_copy_loads_catalog_schema_without_schema_files() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let path = "memory:/branch-write";
+ ctx.server
+ .add_table_with_schema("default", "t", test_schema(), path);
+ let branch_schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("target-file-row-num", "2")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "t$branch_dev",
branch_schema.clone(), path);
+ ctx.server
+ .set_table_schema_id("default", "t$branch_dev", branch_schema, 7);
+ ctx.server.set_table_uuid("default", "t$branch_dev", "t");
+ let main = ctx
+ .catalog
+ .get_table(&Identifier::new("default", "t"))
+ .await
+ .unwrap();
+ let branch = main.copy_with_branch("dev").await.unwrap();
+ assert_eq!(branch.branch(), "dev");
+ assert_eq!(branch.schema().id(), 7);
+ assert_eq!(branch.schema().fields()[1].name(), "value");
+ assert_eq!(branch.location(), main.location());
+ assert!(branch.schema_manager().latest().await.unwrap().is_none());
+ let mut writer = branch.new_write_builder().new_write().unwrap();
+ let data = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10])) as arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files[0].schema_id, 7);
+ // Branch resolution must retain the REST environment so switching back
+ // loads the catalog's main schema instead of looking for a schema file.
+ let main_again = branch.copy_with_branch("main").await.unwrap();
+ assert_eq!(main_again.branch(), "main");
+ assert_eq!(main_again.schema().id(), 0);
+ assert_eq!(main_again.schema().fields()[1].name(), "name");
+ assert!(main.copy_with_branch("missing").await.is_err());
+}
+
+#[tokio::test]
+async fn test_branch_write_copy_rejects_replaced_identity_or_location() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let path = "memory:/branch-identity";
+ ctx.server
+ .add_table_with_schema("default", "t", test_schema(), path);
+ ctx.server
+ .add_table_with_schema("default", "t$branch_dev", test_schema(), path);
+ ctx.server.set_table_uuid("default", "t$branch_dev", "t");
+ let main = ctx
+ .catalog
+ .get_table(&Identifier::new("default", "t"))
+ .await
+ .unwrap();
+ main.copy_with_branch("dev").await.unwrap();
+ ctx.server
+ .set_table_uuid("default", "t$branch_dev", "recreated-table");
+ let error = main.copy_with_branch("dev").await.unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains("no longer belongs to the loaded table"),
+ "{error}"
+ );
+ ctx.server.add_table_with_schema(
+ "default",
+ "t$branch_dev",
+ test_schema(),
+ "memory:/replacement",
+ );
+ ctx.server.set_table_uuid("default", "t$branch_dev", "t");
+ let error = main.copy_with_branch("dev").await.unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains("no longer belongs to the loaded table"),
+ "{error}"
+ );
+}
