From 7d6690a10a007670625c3a8a4d908178d7ffbd13 Mon Sep 17 00:00:00 2001 From: shyjsarah <44659226+shyjsarah@users.noreply.github.com> Date: Mon, 12 Oct 2026 02:10:50 -0700 Subject: [PATCH] fix: support REST branches with distinct identities --- 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 d1bb00575..ff1db75b0 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 48ad5e19c..8076f54e5 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 { - 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| 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 { + 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 c95b91a47..565037902 100644 --- a/crates/paimon/tests/rest_catalog_test.rs +++ b/crates/paimon/tests/rest_catalog_test.rs @@ -5211,7 +5211,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 @@ -5226,7 +5226,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")) @@ -5264,27 +5265,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( @@ -5293,8 +5293,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()