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 819d5a58 fix: support REST branches with distinct identities (#1109)
819d5a58 is described below
commit 819d5a5865b3169ebdec5aafa4ad6e273d0d8cd6
Author: shyjsarah <[email protected]>
AuthorDate: Mon Oct 12 23:30:37 2026 +0800
fix: support REST branches with distinct identities (#1109)
---
crates/paimon/src/table/mod.rs | 12 ++++++------
crates/paimon/src/table/rest_env.rs | 24 ++++++++++++++++++++++--
crates/paimon/tests/rest_catalog_test.rs | 25 +++++++++++++------------
3 files changed, 41 insertions(+), 20 deletions(-)
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index d1bb0057..ff1db75b 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -903,12 +903,12 @@ impl Table {
// 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())
+ let (current, mut table) = tokio::try_join!(
+ env.current_table_identity_checked(),
+ env.get_table(&identifier)
+ )?;
+ if current.path.as_deref() != Some(self.location())
+ || table.location() != self.location()
{
return Err(crate::Error::DataInvalid {
message: format!(
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index b8cdc314..dfb4b3c7 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -155,7 +155,7 @@ impl RESTEnv {
schema_id: i64,
fields: &[crate::spec::DataField],
) -> Result<GetTableResponse> {
- let response = self.api.get_table(&self.identifier).await?;
+ let response = self.current_table_identity_checked().await?;
let name = self.identifier.full_name();
let same = |what: &str, loaded: String, now: Option<String>| match now
{
Some(now) if now == loaded => Ok(()),
@@ -168,7 +168,6 @@ impl RESTEnv {
source: None,
}),
};
- same("uuid", self.uuid.clone(), response.id.clone())?;
same(
"schema",
schema_id.to_string(),
@@ -194,6 +193,27 @@ impl RESTEnv {
Ok(response)
}
+ /// Refused unless the identifier still resolves to this catalog entity.
+ pub(crate) async fn current_table_identity_checked(&self) ->
Result<GetTableResponse> {
+ let response = self.api.get_table(&self.identifier).await?;
+ let name = self.identifier.full_name();
+ match response.id.as_deref() {
+ Some(uuid) if uuid == self.uuid => {}
+ uuid => {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "table '{name}' now resolves to uuid {}, not the {}
this handle was loaded \
+ with; re-load the table before reading it",
+ uuid.unwrap_or("nothing the server reports"),
+ self.uuid
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(response)
+ }
+
/// Get the table identifier.
pub fn identifier(&self) -> &Identifier {
&self.identifier
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index d63a66c4..3d599167 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -5437,7 +5437,7 @@ async fn test_rest_catalog_manages_policies_end_to_end() {
}
#[tokio::test]
-async fn test_branch_write_copy_loads_catalog_schema_without_schema_files() {
+async fn test_branch_write_copy_loads_catalog_schema_with_distinct_identity() {
let ctx = setup_catalog(vec!["default"]).await;
let path = "memory:/branch-write";
ctx.server
@@ -5452,7 +5452,8 @@ async fn
test_branch_write_copy_loads_catalog_schema_without_schema_files() {
.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");
+ ctx.server
+ .set_table_uuid("default", "t$branch_dev", "branch-t");
let main = ctx
.catalog
.get_table(&Identifier::new("default", "t"))
@@ -5490,27 +5491,26 @@ async fn
test_branch_write_copy_loads_catalog_schema_without_schema_files() {
}
#[tokio::test]
-async fn test_branch_write_copy_rejects_replaced_identity_or_location() {
+async fn test_branch_write_copy_rejects_replaced_source_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");
+ ctx.server
+ .set_table_uuid("default", "t$branch_dev", "branch-t");
let main = ctx
.catalog
.get_table(&Identifier::new("default", "t"))
.await
.unwrap();
- main.copy_with_branch("dev").await.unwrap();
+ let branch = 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();
+ .set_table_uuid("default", "t$branch_dev", "recreated-branch");
+ let error = branch.copy_with_branch("main").await.unwrap_err();
assert!(
- error
- .to_string()
- .contains("no longer belongs to the loaded table"),
+ error.to_string().contains("now resolves to uuid"),
"{error}"
);
ctx.server.add_table_with_schema(
@@ -5519,8 +5519,9 @@ async fn
test_branch_write_copy_rejects_replaced_identity_or_location() {
test_schema(),
"memory:/replacement",
);
- ctx.server.set_table_uuid("default", "t$branch_dev", "t");
- let error = main.copy_with_branch("dev").await.unwrap_err();
+ ctx.server
+ .set_table_uuid("default", "t$branch_dev", "branch-t");
+ let error = branch.copy_with_branch("main").await.unwrap_err();
assert!(
error
.to_string()
