From 363dc69447f91ca4bdd8bc9f35aa3ed8ae6aac7a Mon Sep 17 00:00:00 2001 From: forwardxu Date: Thu, 28 May 2026 10:23:02 +0800 Subject: [PATCH 1/6] feat(namespace-dir): implement alter_transaction for DirectoryNamespace Implement the alter_transaction method for DirectoryNamespace, which allows altering a transaction's status and properties. This completes one of the remaining unimplemented methods in the LanceNamespace trait. The implementation supports: - SetStatusAction: change transaction status (Queued/Running/Succeeded/Failed/Canceled) - SetPropertyAction: set transaction properties with Overwrite/Fail/Skip modes - UnsetPropertyAction: remove transaction properties with Skip/Fail modes All actions in a request are applied atomically - either all succeed or the operation fails with an appropriate error. Tests added: - test_alter_transaction_set_status - test_alter_transaction_set_property - test_alter_transaction_set_property_fail_mode - test_alter_transaction_unset_property - test_alter_transaction_invalid_status - test_alter_transaction_not_found - test_alter_transaction_missing_id --- rust/lance-namespace-impls/src/dir.rs | 423 +++++++++++++++++++++++++- 1 file changed, 416 insertions(+), 7 deletions(-) diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index bd5fa70f89d..c92dc4320f5 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -48,13 +48,14 @@ use crate::context::DynamicContextProvider; use lance_namespace::models::{ AlterTableAddColumnsRequest, AlterTableAddColumnsResponse, AlterTableAlterColumnsRequest, AlterTableAlterColumnsResponse, AlterTableDropColumnsRequest, AlterTableDropColumnsResponse, - AnalyzeTableQueryPlanRequest, BatchDeleteTableVersionsRequest, - BatchDeleteTableVersionsResponse, BranchContents as ModelBranchContents, CountTableRowsRequest, - CreateNamespaceRequest, CreateNamespaceResponse, CreateTableBranchRequest, - CreateTableBranchResponse, CreateTableIndexRequest, CreateTableIndexResponse, - CreateTableRequest, CreateTableResponse, CreateTableScalarIndexResponse, CreateTableTagRequest, - CreateTableTagResponse, CreateTableVersionRequest, CreateTableVersionResponse, - DeclareTableRequest, DeclareTableResponse, DeleteFromTableRequest, DeleteFromTableResponse, + AlterTransactionRequest, AlterTransactionResponse, AnalyzeTableQueryPlanRequest, + BatchDeleteTableVersionsRequest, BatchDeleteTableVersionsResponse, + BranchContents as ModelBranchContents, CountTableRowsRequest, CreateNamespaceRequest, + CreateNamespaceResponse, CreateTableBranchRequest, CreateTableBranchResponse, + CreateTableIndexRequest, CreateTableIndexResponse, CreateTableRequest, CreateTableResponse, + CreateTableScalarIndexResponse, CreateTableTagRequest, CreateTableTagResponse, + CreateTableVersionRequest, CreateTableVersionResponse, DeclareTableRequest, + DeclareTableResponse, DeleteFromTableRequest, DeleteFromTableResponse, DeleteTableBranchRequest, DeleteTableBranchResponse, DeleteTableTagRequest, DeleteTableTagResponse, DescribeNamespaceRequest, DescribeNamespaceResponse, DescribeTableIndexStatsRequest, DescribeTableIndexStatsResponse, DescribeTableRequest, @@ -3764,6 +3765,165 @@ impl LanceNamespace for DirectoryNamespace { Ok(Self::transaction_response(version, &transaction)) } + async fn alter_transaction( + &self, + request: AlterTransactionRequest, + ) -> Result { + self.record_op("alter_transaction"); + + // Parse the request ID: must include table id and transaction identifier + let mut request_id = request.id.ok_or_else(|| { + lance_core::Error::from(NamespaceError::InvalidInput { + message: "Transaction id must include table id and transaction identifier" + .to_string(), + }) + })?; + if request_id.len() < 2 { + return Err(NamespaceError::InvalidInput { + message: format!( + "Transaction request id must include table id and transaction identifier, got {:?}", + request_id + ), + } + .into()); + } + + let txn_id = request_id.pop().expect("request_id len checked above"); + let table_id = Some(request_id); + let table_uri = self.resolve_table_location(&table_id).await?; + let dataset = self + .load_dataset(&table_uri, None, "alter_transaction") + .await?; + let (version, transaction) = self.find_transaction(&dataset, &txn_id).await?; + + // Collect current transaction properties + let mut properties = transaction + .transaction_properties + .as_ref() + .map(|props| (**props).clone()) + .unwrap_or_default(); + properties.insert("uuid".to_string(), transaction.uuid.clone()); + properties.insert("version".to_string(), version.to_string()); + properties.insert( + "read_version".to_string(), + transaction.read_version.to_string(), + ); + properties.insert( + "operation".to_string(), + Self::transaction_operation_name(&transaction), + ); + if let Some(tag) = &transaction.tag { + properties.insert("tag".to_string(), tag.clone()); + } + + // Process each action in the request + let mut final_status = "SUCCEEDED".to_string(); + for action in &request.actions { + if let Some(ref set_status) = action.set_status_action + && let Some(ref status) = set_status.status + { + // Validate the status value (case-insensitive) + let normalized = status.to_lowercase().replace('_', ""); + match normalized.as_str() { + "queued" | "running" | "succeeded" | "failed" | "canceled" => { + final_status = status.clone(); + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid transaction status '{}'. Valid values are: Queued, Running, Succeeded, Failed, Canceled", + status + ), + } + .into()); + } + } + } + + if let Some(ref set_property) = action.set_property_action + && let (Some(key), Some(value)) = (&set_property.key, &set_property.value) + { + let mode = set_property + .mode + .as_deref() + .unwrap_or("Overwrite") + .to_lowercase(); + match mode.as_str() { + "overwrite" => { + properties.insert(key.clone(), value.clone()); + } + "fail" => { + if properties.contains_key(key) { + return Err(NamespaceError::ConcurrentModification { + message: format!( + "Property '{}' already exists and mode is 'Fail'", + key + ), + } + .into()); + } + properties.insert(key.clone(), value.clone()); + } + "skip" => { + if !properties.contains_key(key) { + properties.insert(key.clone(), value.clone()); + } + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid set_property mode '{}'. Valid values are: Overwrite, Fail, Skip", + mode + ), + } + .into()); + } + } + } + + if let Some(ref unset_property) = action.unset_property_action + && let Some(ref key) = unset_property.key + { + let mode = unset_property + .mode + .as_deref() + .unwrap_or("Skip") + .to_lowercase(); + match mode.as_str() { + "skip" => { + properties.remove(key); + } + "fail" => { + if !properties.contains_key(key) { + return Err(NamespaceError::InvalidInput { + message: format!( + "Property '{}' does not exist and mode is 'Fail'", + key + ), + } + .into()); + } + properties.remove(key); + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid unset_property mode '{}'. Valid values are: Skip, Fail", + mode + ), + } + .into()); + } + } + } + } + + Ok(AlterTransactionResponse { + status: final_status, + properties: Some(properties), + }) + } + async fn create_table_scalar_index( &self, request: CreateTableIndexRequest, @@ -12968,4 +13128,253 @@ mod tests { .expect("reopen branch failed"); assert_eq!(scan_id_column(&reopened).await, vec![1, 2]); } + + #[tokio::test] + async fn test_alter_transaction_set_status() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + DescribeTransactionRequest, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + if let Some(txn_id) = transaction_id { + // First verify the transaction exists + let describe_resp = namespace + .describe_transaction(DescribeTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(describe_resp.status, "SUCCEEDED"); + + // Alter the transaction status + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "Canceled"); + assert!(response.properties.is_some()); + let props = response.properties.unwrap(); + assert_eq!(props.get("uuid"), Some(&txn_id)); + assert_eq!(props.get("operation"), Some(&"CreateIndex".to_string())); + } + } + + #[tokio::test] + async fn test_alter_transaction_set_property() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + if let Some(txn_id) = transaction_id { + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("custom_value".to_string()), + mode: None, + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert_eq!(props.get("custom_key"), Some(&"custom_value".to_string())); + } + } + + #[tokio::test] + async fn test_alter_transaction_set_property_fail_mode() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + if let Some(txn_id) = transaction_id { + // Try to set a property that already exists (uuid) with Fail mode + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("uuid".to_string()), + value: Some("new_value".to_string()), + mode: Some("Fail".to_string()), + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + } + + #[tokio::test] + async fn test_alter_transaction_unset_property() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + AlterTransactionUnsetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + if let Some(txn_id) = transaction_id { + // First set a custom property, then unset it + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![ + AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("temp_key".to_string()), + value: Some("temp_value".to_string()), + mode: None, + })), + unset_property_action: None, + }, + AlterTransactionAction { + set_status_action: None, + set_property_action: None, + unset_property_action: Some(Box::new(AlterTransactionUnsetProperty { + key: Some("temp_key".to_string()), + mode: None, + })), + }, + ], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert!(!props.contains_key("temp_key")); + } + } + + #[tokio::test] + async fn test_alter_transaction_invalid_status() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + if let Some(txn_id) = transaction_id { + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("InvalidStatus".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + } + + #[tokio::test] + async fn test_alter_transaction_not_found() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + + // Try to alter a non-existent transaction + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), "non_existent_txn".to_string()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_alter_transaction_missing_id() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + + // Try with missing id + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: None, + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + + // Try with insufficient id parts + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } } From 8ad21c9c7c8aad3bf3f7199eb47079be2d8f1d08 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Wed, 1 Jul 2026 09:59:33 +0800 Subject: [PATCH 2/6] fix(namespace-impls): persist alter_transaction changes via sidecar Previously, alter_transaction only mutated properties in memory and returned them, so subsequent describe_transaction / alter_transaction calls could not observe the modifications. Since Lance embeds the Transaction into the manifest and treats it as immutable, we cannot rewrite the transaction file in place. Introduce a namespace-owned sidecar file at {table}/_alter_transactions/{uuid}.json that records the accumulated alterations (status, added/overwritten properties, tombstoned properties). describe_transaction and follow-up alter_transaction calls now merge the sidecar on top of the immutable transaction so the altered state survives across calls. Also treat uuid/version/read_version/operation/tag as reserved keys that cannot be modified, since they are derived from the immutable Transaction metadata. --- rust/lance-namespace-impls/Cargo.toml | 10 +- rust/lance-namespace-impls/src/dir.rs | 305 +++++++++++++++++++++++--- 2 files changed, 278 insertions(+), 37 deletions(-) diff --git a/rust/lance-namespace-impls/Cargo.toml b/rust/lance-namespace-impls/Cargo.toml index 27b9a4bc0e2..64e281efdf1 100644 --- a/rust/lance-namespace-impls/Cargo.toml +++ b/rust/lance-namespace-impls/Cargo.toml @@ -13,8 +13,8 @@ rust-version.workspace = true [features] default = ["dir-aws", "dir-azure", "dir-gcp", "dir-oss", "dir-huggingface"] -rest = ["dep:reqwest", "dep:serde"] -rest-adapter = ["dep:axum", "dep:tower", "dep:tower-http", "dep:serde"] +rest = ["dep:reqwest"] +rest-adapter = ["dep:axum", "dep:tower", "dep:tower-http"] # Cloud storage features for directory implementation - align with lance-io dir-gcp = ["lance-io/gcp", "lance/gcp"] dir-aws = ["lance-io/aws", "lance/aws"] @@ -24,8 +24,8 @@ dir-huggingface = ["lance-io/huggingface", "lance/huggingface"] dir-goosefs = ["lance-io/goosefs", "lance/goosefs"] # Credential vending features credential-vendor-aws = ["dep:aws-sdk-sts", "dep:aws-config", "dep:sha2", "dep:base64"] -credential-vendor-gcp = ["dep:reqwest", "dep:serde", "dep:sha2", "dep:base64", "dep:ring", "dep:rustls-pki-types"] -credential-vendor-azure = ["dep:reqwest", "dep:serde", "dep:sha2", "dep:base64", "dep:chrono", "dep:hmac", "dep:quick-xml"] +credential-vendor-gcp = ["dep:reqwest", "dep:sha2", "dep:base64", "dep:ring", "dep:rustls-pki-types"] +credential-vendor-azure = ["dep:reqwest", "dep:sha2", "dep:base64", "dep:chrono", "dep:hmac", "dep:quick-xml"] [dependencies] lance-namespace.workspace = true @@ -58,12 +58,12 @@ datafusion-physical-plan = { workspace = true } axum = { workspace = true, optional = true } tower = { workspace = true, optional = true } tower-http = { workspace = true, optional = true, features = ["trace", "cors", "normalize-path"] } -serde = { workspace = true, optional = true } # Common dependencies async-trait.workspace = true bytes.workspace = true tokio = { workspace = true, features = ["sync", "full"] } +serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } futures.workspace = true log.workspace = true diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index c92dc4320f5..73f6b17b0fe 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -914,6 +914,29 @@ struct TableDeleteEntry { ranges: Vec<(i64, i64)>, } +/// Persistent record of `alter_transaction` outcomes for a single transaction. +/// +/// Lance's transaction file is immutable once written, so we record any +/// modifications (status transitions, extra properties, tombstoned properties) +/// in a namespace-owned sidecar file. The sidecar is then merged into the +/// response of subsequent `describe_transaction` / `alter_transaction` calls. +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +struct TransactionAlteration { + /// The most recently applied status, if any. + #[serde(default, skip_serializing_if = "Option::is_none")] + status: Option, + /// User-defined properties layered on top of the immutable transaction + /// properties. Values here take precedence over the transaction's own + /// properties when both are present. + #[serde(default)] + properties: HashMap, + /// Names of transaction properties that have been tombstoned via + /// `unset_property_action`. A tombstoned key is hidden from the response + /// even when the immutable transaction still carries it. + #[serde(default)] + removed_properties: std::collections::HashSet, +} + impl DirectoryNamespace { /// Apply pagination to a list of table names /// @@ -2065,12 +2088,29 @@ impl DirectoryNamespace { fn transaction_response( version: u64, transaction: &Transaction, + alteration: Option, ) -> DescribeTransactionResponse { let mut properties = transaction .transaction_properties .as_ref() .map(|properties| (**properties).clone()) .unwrap_or_default(); + + // Apply persisted alterations on top of the immutable transaction + // properties so callers see the current effective state. + let mut effective_status = "SUCCEEDED".to_string(); + if let Some(alteration) = alteration { + for key in &alteration.removed_properties { + properties.remove(key); + } + for (key, value) in alteration.properties { + properties.insert(key, value); + } + if let Some(status) = alteration.status { + effective_status = status; + } + } + properties.insert("uuid".to_string(), transaction.uuid.clone()); properties.insert("version".to_string(), version.to_string()); properties.insert( @@ -2086,7 +2126,7 @@ impl DirectoryNamespace { } DescribeTransactionResponse { - status: "SUCCEEDED".to_string(), + status: effective_status, properties: Some(properties), } } @@ -2175,6 +2215,86 @@ impl DirectoryNamespace { .into()) } + /// Relative directory (under a table's Lance root) used to persist + /// alter_transaction outcomes. The Lance transaction file itself is + /// immutable, so we keep alterations in a namespace-owned sidecar. + const TRANSACTION_ALTERATIONS_DIR: &'static str = "_alter_transactions"; + + fn transaction_alteration_path(&self, table_uri: &str, txn_uuid: &str) -> Result { + let table_path = self.object_store_path_from_uri(table_uri)?; + Ok(table_path + .join(Self::TRANSACTION_ALTERATIONS_DIR) + .join(format!("{}.json", txn_uuid).as_str())) + } + + async fn load_transaction_alteration( + &self, + table_uri: &str, + txn_uuid: &str, + ) -> Result> { + let path = self.transaction_alteration_path(table_uri, txn_uuid)?; + match self.object_store.inner.get(&path).await { + Ok(get_result) => { + let bytes = get_result.bytes().await.map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to read alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + let alteration: TransactionAlteration = + serde_json::from_slice(&bytes).map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to parse alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + Ok(Some(alteration)) + } + Err(ObjectStoreError::NotFound { .. }) => Ok(None), + Err(e) => Err(lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to load alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + .into()), + } + } + + async fn save_transaction_alteration( + &self, + table_uri: &str, + txn_uuid: &str, + alteration: &TransactionAlteration, + ) -> Result<()> { + let path = self.transaction_alteration_path(table_uri, txn_uuid)?; + let bytes = serde_json::to_vec(alteration).map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to serialize alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + self.object_store + .inner + .put(&path, bytes.into()) + .await + .map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to persist alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + Ok(()) + } + fn table_full_uri(&self, table_name: &str) -> String { format!("{}/{}.lance", &self.root, table_name) } @@ -3762,7 +3882,13 @@ impl LanceNamespace for DirectoryNamespace { .await?; let (version, transaction) = self.find_transaction(&dataset, &id).await?; - Ok(Self::transaction_response(version, &transaction)) + // Merge any persisted alter_transaction changes stored in the sidecar + // so that describe_transaction reflects the latest altered state. + let sidecar = self + .load_transaction_alteration(&table_uri, &transaction.uuid) + .await?; + + Ok(Self::transaction_response(version, &transaction, sidecar)) } async fn alter_transaction( @@ -3796,28 +3922,19 @@ impl LanceNamespace for DirectoryNamespace { .await?; let (version, transaction) = self.find_transaction(&dataset, &txn_id).await?; - // Collect current transaction properties - let mut properties = transaction - .transaction_properties - .as_ref() - .map(|props| (**props).clone()) + // Reserved keys are derived from the immutable Transaction metadata and + // must not be modified via alter_transaction. They are only surfaced in + // the response for the caller's convenience. + const RESERVED_KEYS: &[&str] = &["uuid", "version", "read_version", "operation", "tag"]; + let is_reserved = |key: &str| RESERVED_KEYS.iter().any(|k| *k == key); + + // Load the existing sidecar (if any) so alterations accumulate across + // successive alter_transaction calls. + let mut sidecar = self + .load_transaction_alteration(&table_uri, &transaction.uuid) + .await? .unwrap_or_default(); - properties.insert("uuid".to_string(), transaction.uuid.clone()); - properties.insert("version".to_string(), version.to_string()); - properties.insert( - "read_version".to_string(), - transaction.read_version.to_string(), - ); - properties.insert( - "operation".to_string(), - Self::transaction_operation_name(&transaction), - ); - if let Some(tag) = &transaction.tag { - properties.insert("tag".to_string(), tag.clone()); - } - // Process each action in the request - let mut final_status = "SUCCEEDED".to_string(); for action in &request.actions { if let Some(ref set_status) = action.set_status_action && let Some(ref status) = set_status.status @@ -3826,7 +3943,7 @@ impl LanceNamespace for DirectoryNamespace { let normalized = status.to_lowercase().replace('_', ""); match normalized.as_str() { "queued" | "running" | "succeeded" | "failed" | "canceled" => { - final_status = status.clone(); + sidecar.status = Some(status.clone()); } _ => { return Err(NamespaceError::InvalidInput { @@ -3843,6 +3960,15 @@ impl LanceNamespace for DirectoryNamespace { if let Some(ref set_property) = action.set_property_action && let (Some(key), Some(value)) = (&set_property.key, &set_property.value) { + if is_reserved(key) { + return Err(NamespaceError::InvalidInput { + message: format!( + "Property '{}' is reserved and cannot be modified", + key + ), + } + .into()); + } let mode = set_property .mode .as_deref() @@ -3850,10 +3976,17 @@ impl LanceNamespace for DirectoryNamespace { .to_lowercase(); match mode.as_str() { "overwrite" => { - properties.insert(key.clone(), value.clone()); + sidecar.properties.insert(key.clone(), value.clone()); } "fail" => { - if properties.contains_key(key) { + // Consider both the immutable transaction properties + // and any values previously written to the sidecar. + let exists = sidecar.properties.contains_key(key) + || transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); + if exists { return Err(NamespaceError::ConcurrentModification { message: format!( "Property '{}' already exists and mode is 'Fail'", @@ -3862,11 +3995,16 @@ impl LanceNamespace for DirectoryNamespace { } .into()); } - properties.insert(key.clone(), value.clone()); + sidecar.properties.insert(key.clone(), value.clone()); } "skip" => { - if !properties.contains_key(key) { - properties.insert(key.clone(), value.clone()); + let exists = sidecar.properties.contains_key(key) + || transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); + if !exists { + sidecar.properties.insert(key.clone(), value.clone()); } } _ => { @@ -3884,17 +4022,35 @@ impl LanceNamespace for DirectoryNamespace { if let Some(ref unset_property) = action.unset_property_action && let Some(ref key) = unset_property.key { + if is_reserved(key) { + return Err(NamespaceError::InvalidInput { + message: format!( + "Property '{}' is reserved and cannot be modified", + key + ), + } + .into()); + } let mode = unset_property .mode .as_deref() .unwrap_or("Skip") .to_lowercase(); + let exists_in_transaction = transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); match mode.as_str() { "skip" => { - properties.remove(key); + sidecar.properties.remove(key); + if exists_in_transaction { + // Track a tombstone so describe_transaction can + // hide the immutable property from the response. + sidecar.removed_properties.insert(key.clone()); + } } "fail" => { - if !properties.contains_key(key) { + if !sidecar.properties.contains_key(key) && !exists_in_transaction { return Err(NamespaceError::InvalidInput { message: format!( "Property '{}' does not exist and mode is 'Fail'", @@ -3903,7 +4059,10 @@ impl LanceNamespace for DirectoryNamespace { } .into()); } - properties.remove(key); + sidecar.properties.remove(key); + if exists_in_transaction { + sidecar.removed_properties.insert(key.clone()); + } } _ => { return Err(NamespaceError::InvalidInput { @@ -3918,9 +4077,22 @@ impl LanceNamespace for DirectoryNamespace { } } + // Persist the accumulated alterations so subsequent calls observe + // them. The transaction file itself is immutable in Lance, so we + // record alter_transaction outcomes in a namespace-owned sidecar. + self.save_transaction_alteration(&table_uri, &transaction.uuid, &sidecar) + .await?; + + // Assemble the response by merging the immutable transaction metadata + // with the persisted alterations. + let final_status = sidecar + .status + .clone() + .unwrap_or_else(|| "SUCCEEDED".to_string()); + let response = Self::transaction_response(version, &transaction, Some(sidecar)); Ok(AlterTransactionResponse { status: final_status, - properties: Some(properties), + properties: response.properties, }) } @@ -13377,4 +13549,73 @@ mod tests { .await; assert!(result.is_err()); } + + #[tokio::test] + async fn test_alter_transaction_persists_changes() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + AlterTransactionSetStatus, DescribeTransactionRequest, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + let txn_id = transaction_id.expect("scalar index should produce a transaction id"); + + // Alter status and set a custom property. + namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![ + AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }, + AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("owner".to_string()), + value: Some("alice".to_string()), + mode: None, + })), + unset_property_action: None, + }, + ], + ..Default::default() + }) + .await + .unwrap(); + + // The changes must survive across a fresh describe_transaction call, + // proving the alteration was persisted to the transaction file. + let describe_resp = namespace + .describe_transaction(DescribeTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + ..Default::default() + }) + .await + .unwrap(); + let props = describe_resp.properties.expect("properties should be set"); + assert_eq!(props.get("owner"), Some(&"alice".to_string())); + // The internal `_status` marker should not leak into the response but + // must be present on disk so subsequent alter_transaction calls can + // observe the previously set status. + assert!(!props.contains_key("_status")); + + let follow_up = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(follow_up.status, "Canceled"); + let follow_up_props = follow_up.properties.unwrap(); + assert_eq!(follow_up_props.get("owner"), Some(&"alice".to_string())); + } } From 240d47214af2267eedef481c9e8536b5da172089 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Wed, 1 Jul 2026 10:06:15 +0800 Subject: [PATCH 3/6] style(namespace-impls): fix rustfmt issues in alter_transaction reserved key checks --- rust/lance-namespace-impls/src/dir.rs | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index 73f6b17b0fe..3a549bed129 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -3962,10 +3962,7 @@ impl LanceNamespace for DirectoryNamespace { { if is_reserved(key) { return Err(NamespaceError::InvalidInput { - message: format!( - "Property '{}' is reserved and cannot be modified", - key - ), + message: format!("Property '{}' is reserved and cannot be modified", key), } .into()); } @@ -4024,10 +4021,7 @@ impl LanceNamespace for DirectoryNamespace { { if is_reserved(key) { return Err(NamespaceError::InvalidInput { - message: format!( - "Property '{}' is reserved and cannot be modified", - key - ), + message: format!("Property '{}' is reserved and cannot be modified", key), } .into()); } From f140451ea56f432b32d18ffd9721605ee58b03e3 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Wed, 1 Jul 2026 10:20:07 +0800 Subject: [PATCH 4/6] fix: address clippy warnings in dir.rs - Remove useless conversion .into() on lance_core::Error - Use RESERVED_KEYS.contains(&key) instead of iter().any() --- rust/lance-namespace-impls/src/dir.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index 3a549bed129..8b1864830d7 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -2260,8 +2260,7 @@ impl DirectoryNamespace { "Failed to load alter_transaction sidecar for '{}': {}", txn_uuid, e ), - }) - .into()), + })), } } @@ -3926,7 +3925,7 @@ impl LanceNamespace for DirectoryNamespace { // must not be modified via alter_transaction. They are only surfaced in // the response for the caller's convenience. const RESERVED_KEYS: &[&str] = &["uuid", "version", "read_version", "operation", "tag"]; - let is_reserved = |key: &str| RESERVED_KEYS.iter().any(|k| *k == key); + let is_reserved = |key: &str| RESERVED_KEYS.contains(&key); // Load the existing sidecar (if any) so alterations accumulate across // successive alter_transaction calls. From 4a1a1eeac9acba4e08d1869b2f8648a71945bea9 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Wed, 1 Jul 2026 13:04:11 +0800 Subject: [PATCH 5/6] fix(namespace-impls): address review comments on alter_transaction - Cargo.toml: revert to original by dropping serde derive; sidecar now serializes/deserializes via serde_json::Value manually, so no changes to Cargo.toml are needed. - Replace 'if let Some(txn_id) = transaction_id' with expect(...) in five alter_transaction tests to avoid silent false-positive passes. - test_alter_transaction_set_property_fail_mode: use a non-reserved 'custom_key' (set once, then set again with mode='Fail') so the test actually exercises the mode='Fail' branch instead of tripping the reserved-key guard first. --- rust/lance-namespace-impls/Cargo.toml | 10 +- rust/lance-namespace-impls/src/dir.rs | 420 +++++++++++++++++--------- 2 files changed, 287 insertions(+), 143 deletions(-) diff --git a/rust/lance-namespace-impls/Cargo.toml b/rust/lance-namespace-impls/Cargo.toml index 64e281efdf1..27b9a4bc0e2 100644 --- a/rust/lance-namespace-impls/Cargo.toml +++ b/rust/lance-namespace-impls/Cargo.toml @@ -13,8 +13,8 @@ rust-version.workspace = true [features] default = ["dir-aws", "dir-azure", "dir-gcp", "dir-oss", "dir-huggingface"] -rest = ["dep:reqwest"] -rest-adapter = ["dep:axum", "dep:tower", "dep:tower-http"] +rest = ["dep:reqwest", "dep:serde"] +rest-adapter = ["dep:axum", "dep:tower", "dep:tower-http", "dep:serde"] # Cloud storage features for directory implementation - align with lance-io dir-gcp = ["lance-io/gcp", "lance/gcp"] dir-aws = ["lance-io/aws", "lance/aws"] @@ -24,8 +24,8 @@ dir-huggingface = ["lance-io/huggingface", "lance/huggingface"] dir-goosefs = ["lance-io/goosefs", "lance/goosefs"] # Credential vending features credential-vendor-aws = ["dep:aws-sdk-sts", "dep:aws-config", "dep:sha2", "dep:base64"] -credential-vendor-gcp = ["dep:reqwest", "dep:sha2", "dep:base64", "dep:ring", "dep:rustls-pki-types"] -credential-vendor-azure = ["dep:reqwest", "dep:sha2", "dep:base64", "dep:chrono", "dep:hmac", "dep:quick-xml"] +credential-vendor-gcp = ["dep:reqwest", "dep:serde", "dep:sha2", "dep:base64", "dep:ring", "dep:rustls-pki-types"] +credential-vendor-azure = ["dep:reqwest", "dep:serde", "dep:sha2", "dep:base64", "dep:chrono", "dep:hmac", "dep:quick-xml"] [dependencies] lance-namespace.workspace = true @@ -58,12 +58,12 @@ datafusion-physical-plan = { workspace = true } axum = { workspace = true, optional = true } tower = { workspace = true, optional = true } tower-http = { workspace = true, optional = true, features = ["trace", "cors", "normalize-path"] } +serde = { workspace = true, optional = true } # Common dependencies async-trait.workspace = true bytes.workspace = true tokio = { workspace = true, features = ["sync", "full"] } -serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } futures.workspace = true log.workspace = true diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index 8b1864830d7..c7a18217ff9 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -920,23 +920,141 @@ struct TableDeleteEntry { /// modifications (status transitions, extra properties, tombstoned properties) /// in a namespace-owned sidecar file. The sidecar is then merged into the /// response of subsequent `describe_transaction` / `alter_transaction` calls. -#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +/// +/// Serialization is implemented manually via `serde_json::Value` to avoid +/// pulling in `serde`'s `derive` feature for this crate. +#[derive(Debug, Clone, Default)] struct TransactionAlteration { /// The most recently applied status, if any. - #[serde(default, skip_serializing_if = "Option::is_none")] status: Option, /// User-defined properties layered on top of the immutable transaction /// properties. Values here take precedence over the transaction's own /// properties when both are present. - #[serde(default)] properties: HashMap, /// Names of transaction properties that have been tombstoned via /// `unset_property_action`. A tombstoned key is hidden from the response /// even when the immutable transaction still carries it. - #[serde(default)] removed_properties: std::collections::HashSet, } +impl TransactionAlteration { + /// JSON field names used for the sidecar on-disk representation. + const F_STATUS: &'static str = "status"; + const F_PROPERTIES: &'static str = "properties"; + const F_REMOVED_PROPERTIES: &'static str = "removed_properties"; + + fn to_json(&self) -> serde_json::Value { + let mut obj = serde_json::Map::new(); + if let Some(ref status) = self.status { + obj.insert( + Self::F_STATUS.to_string(), + serde_json::Value::String(status.clone()), + ); + } + let props = self + .properties + .iter() + .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone()))) + .collect::>(); + obj.insert( + Self::F_PROPERTIES.to_string(), + serde_json::Value::Object(props), + ); + let removed = self + .removed_properties + .iter() + .map(|k| serde_json::Value::String(k.clone())) + .collect::>(); + obj.insert( + Self::F_REMOVED_PROPERTIES.to_string(), + serde_json::Value::Array(removed), + ); + serde_json::Value::Object(obj) + } + + fn from_json(value: serde_json::Value) -> std::result::Result { + let obj = match value { + serde_json::Value::Object(m) => m, + other => { + return Err(format!( + "expected JSON object for TransactionAlteration, got {}", + other + )); + } + }; + + let status = match obj.get(Self::F_STATUS) { + None | Some(serde_json::Value::Null) => None, + Some(serde_json::Value::String(s)) => Some(s.clone()), + Some(other) => { + return Err(format!("field 'status' must be a string, got {}", other)); + } + }; + + let mut properties = HashMap::new(); + if let Some(props_val) = obj.get(Self::F_PROPERTIES) { + match props_val { + serde_json::Value::Null => {} + serde_json::Value::Object(m) => { + for (k, v) in m { + match v { + serde_json::Value::String(s) => { + properties.insert(k.clone(), s.clone()); + } + other => { + return Err(format!( + "property '{}' must be a string, got {}", + k, other + )); + } + } + } + } + other => { + return Err(format!( + "field 'properties' must be an object, got {}", + other + )); + } + } + } + + let mut removed_properties = std::collections::HashSet::new(); + if let Some(removed_val) = obj.get(Self::F_REMOVED_PROPERTIES) { + match removed_val { + serde_json::Value::Null => {} + serde_json::Value::Array(arr) => { + for item in arr { + match item { + serde_json::Value::String(s) => { + removed_properties.insert(s.clone()); + } + other => { + return Err(format!( + "field 'removed_properties' must contain strings, got {}", + other + )); + } + } + } + } + other => { + return Err(format!( + "field 'removed_properties' must be an array, got {}", + other + )); + } + } + } + + Ok(Self { + status, + properties, + removed_properties, + }) + } +} + impl DirectoryNamespace { /// Apply pagination to a list of table names /// @@ -2243,15 +2361,22 @@ impl DirectoryNamespace { ), }) })?; - let alteration: TransactionAlteration = - serde_json::from_slice(&bytes).map_err(|e| { - lance_core::Error::from(NamespaceError::Internal { - message: format!( - "Failed to parse alter_transaction sidecar for '{}': {}", - txn_uuid, e - ), - }) - })?; + let value: serde_json::Value = serde_json::from_slice(&bytes).map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to parse alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + let alteration = TransactionAlteration::from_json(value).map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to parse alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; Ok(Some(alteration)) } Err(ObjectStoreError::NotFound { .. }) => Ok(None), @@ -2271,7 +2396,7 @@ impl DirectoryNamespace { alteration: &TransactionAlteration, ) -> Result<()> { let path = self.transaction_alteration_path(table_uri, txn_uuid)?; - let bytes = serde_json::to_vec(alteration).map_err(|e| { + let bytes = serde_json::to_vec(&alteration.to_json()).map_err(|e| { lance_core::Error::from(NamespaceError::Internal { message: format!( "Failed to serialize alter_transaction sidecar for '{}': {}", @@ -13303,40 +13428,40 @@ mod tests { let (namespace, _temp_dir) = create_test_namespace().await; create_scalar_table(&namespace, "users").await; - let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); - if let Some(txn_id) = transaction_id { - // First verify the transaction exists - let describe_resp = namespace - .describe_transaction(DescribeTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - ..Default::default() - }) - .await - .unwrap(); - assert_eq!(describe_resp.status, "SUCCEEDED"); + // First verify the transaction exists + let describe_resp = namespace + .describe_transaction(DescribeTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(describe_resp.status, "SUCCEEDED"); - // Alter the transaction status - let response = namespace - .alter_transaction(AlterTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - actions: vec![AlterTransactionAction { - set_status_action: Some(Box::new(AlterTransactionSetStatus { - status: Some("Canceled".to_string()), - })), - set_property_action: None, - unset_property_action: None, - }], - ..Default::default() - }) - .await - .unwrap(); - assert_eq!(response.status, "Canceled"); - assert!(response.properties.is_some()); - let props = response.properties.unwrap(); - assert_eq!(props.get("uuid"), Some(&txn_id)); - assert_eq!(props.get("operation"), Some(&"CreateIndex".to_string())); - } + // Alter the transaction status + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "Canceled"); + assert!(response.properties.is_some()); + let props = response.properties.unwrap(); + assert_eq!(props.get("uuid"), Some(&txn_id)); + assert_eq!(props.get("operation"), Some(&"CreateIndex".to_string())); } #[tokio::test] @@ -13347,29 +13472,29 @@ mod tests { let (namespace, _temp_dir) = create_test_namespace().await; create_scalar_table(&namespace, "users").await; - let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); - if let Some(txn_id) = transaction_id { - let response = namespace - .alter_transaction(AlterTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - actions: vec![AlterTransactionAction { - set_status_action: None, - set_property_action: Some(Box::new(AlterTransactionSetProperty { - key: Some("custom_key".to_string()), - value: Some("custom_value".to_string()), - mode: None, - })), - unset_property_action: None, - }], - ..Default::default() - }) - .await - .unwrap(); - assert_eq!(response.status, "SUCCEEDED"); - let props = response.properties.unwrap(); - assert_eq!(props.get("custom_key"), Some(&"custom_value".to_string())); - } + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("custom_value".to_string()), + mode: None, + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert_eq!(props.get("custom_key"), Some(&"custom_value".to_string())); } #[tokio::test] @@ -13380,27 +13505,46 @@ mod tests { let (namespace, _temp_dir) = create_test_namespace().await; create_scalar_table(&namespace, "users").await; - let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); - if let Some(txn_id) = transaction_id { - // Try to set a property that already exists (uuid) with Fail mode - let result = namespace - .alter_transaction(AlterTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - actions: vec![AlterTransactionAction { - set_status_action: None, - set_property_action: Some(Box::new(AlterTransactionSetProperty { - key: Some("uuid".to_string()), - value: Some("new_value".to_string()), - mode: Some("Fail".to_string()), - })), - unset_property_action: None, - }], - ..Default::default() - }) - .await; - assert!(result.is_err()); - } + // First, set a non-reserved property so it exists in the sidecar. + namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("initial_value".to_string()), + mode: None, + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + + // Now try to set the same property again with Fail mode, which must + // exercise the mode='Fail' branch (not the reserved-key guard). + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("new_value".to_string()), + mode: Some("Fail".to_string()), + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); } #[tokio::test] @@ -13412,40 +13556,40 @@ mod tests { let (namespace, _temp_dir) = create_test_namespace().await; create_scalar_table(&namespace, "users").await; - let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); - if let Some(txn_id) = transaction_id { - // First set a custom property, then unset it - let response = namespace - .alter_transaction(AlterTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - actions: vec![ - AlterTransactionAction { - set_status_action: None, - set_property_action: Some(Box::new(AlterTransactionSetProperty { - key: Some("temp_key".to_string()), - value: Some("temp_value".to_string()), - mode: None, - })), - unset_property_action: None, - }, - AlterTransactionAction { - set_status_action: None, - set_property_action: None, - unset_property_action: Some(Box::new(AlterTransactionUnsetProperty { - key: Some("temp_key".to_string()), - mode: None, - })), - }, - ], - ..Default::default() - }) - .await - .unwrap(); - assert_eq!(response.status, "SUCCEEDED"); - let props = response.properties.unwrap(); - assert!(!props.contains_key("temp_key")); - } + // First set a custom property, then unset it + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![ + AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("temp_key".to_string()), + value: Some("temp_value".to_string()), + mode: None, + })), + unset_property_action: None, + }, + AlterTransactionAction { + set_status_action: None, + set_property_action: None, + unset_property_action: Some(Box::new(AlterTransactionUnsetProperty { + key: Some("temp_key".to_string()), + mode: None, + })), + }, + ], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert!(!props.contains_key("temp_key")); } #[tokio::test] @@ -13456,24 +13600,24 @@ mod tests { let (namespace, _temp_dir) = create_test_namespace().await; create_scalar_table(&namespace, "users").await; - let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); - if let Some(txn_id) = transaction_id { - let result = namespace - .alter_transaction(AlterTransactionRequest { - id: Some(vec!["users".to_string(), txn_id.clone()]), - actions: vec![AlterTransactionAction { - set_status_action: Some(Box::new(AlterTransactionSetStatus { - status: Some("InvalidStatus".to_string()), - })), - set_property_action: None, - unset_property_action: None, - }], - ..Default::default() - }) - .await; - assert!(result.is_err()); - } + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("InvalidStatus".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); } #[tokio::test] From 6f1d6f2e67a86193395591aef754b1268185a5cc Mon Sep 17 00:00:00 2001 From: forwardxu Date: Wed, 1 Jul 2026 14:42:46 +0800 Subject: [PATCH 6/6] refactor(namespace-impls): simplify sidecar (de)serialization Adopt the same pattern used in dir/manifest.rs, which relies on the built-in Serialize/Deserialize impls for Option, HashMap and HashSet transitively provided by serde_json. - Replace hand-rolled serde_json::Value match with a single serde_json::from_slice::> plus per-field from_value calls. - Fold the two-step to_json + serde_json::to_vec into a single to_json_bytes helper. - No Cargo.toml changes; still compiles under --no-default-features --features dir-aws where serde is optional. Net effect: -81 lines in dir.rs, all 8 alter_transaction tests still pass, cargo fmt/clippy clean. --- rust/lance-namespace-impls/src/dir.rs | 149 ++++++-------------------- 1 file changed, 34 insertions(+), 115 deletions(-) diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index c7a18217ff9..dc2d83cf278 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -943,114 +943,41 @@ impl TransactionAlteration { const F_PROPERTIES: &'static str = "properties"; const F_REMOVED_PROPERTIES: &'static str = "removed_properties"; - fn to_json(&self) -> serde_json::Value { - let mut obj = serde_json::Map::new(); - if let Some(ref status) = self.status { - obj.insert( - Self::F_STATUS.to_string(), - serde_json::Value::String(status.clone()), - ); - } - let props = self - .properties - .iter() - .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone()))) - .collect::>(); - obj.insert( - Self::F_PROPERTIES.to_string(), - serde_json::Value::Object(props), - ); - let removed = self - .removed_properties - .iter() - .map(|k| serde_json::Value::String(k.clone())) - .collect::>(); - obj.insert( - Self::F_REMOVED_PROPERTIES.to_string(), - serde_json::Value::Array(removed), - ); - serde_json::Value::Object(obj) + /// Serialize this alteration to a JSON byte vector. + /// + /// Uses the same pattern as `dir/manifest.rs`: rely on the built-in + /// `Serialize` impls for `Option`, `HashMap` and + /// `HashSet` provided by the `serde` crate (transitively pulled in + /// by `serde_json`), so no `serde` derive nor extra dependency is needed. + fn to_json_bytes(&self) -> serde_json::Result> { + serde_json::to_vec(&serde_json::json!({ + Self::F_STATUS: self.status, + Self::F_PROPERTIES: self.properties, + Self::F_REMOVED_PROPERTIES: self.removed_properties, + })) } - fn from_json(value: serde_json::Value) -> std::result::Result { - let obj = match value { - serde_json::Value::Object(m) => m, - other => { - return Err(format!( - "expected JSON object for TransactionAlteration, got {}", - other - )); - } - }; - - let status = match obj.get(Self::F_STATUS) { - None | Some(serde_json::Value::Null) => None, - Some(serde_json::Value::String(s)) => Some(s.clone()), - Some(other) => { - return Err(format!("field 'status' must be a string, got {}", other)); - } - }; - - let mut properties = HashMap::new(); - if let Some(props_val) = obj.get(Self::F_PROPERTIES) { - match props_val { - serde_json::Value::Null => {} - serde_json::Value::Object(m) => { - for (k, v) in m { - match v { - serde_json::Value::String(s) => { - properties.insert(k.clone(), s.clone()); - } - other => { - return Err(format!( - "property '{}' must be a string, got {}", - k, other - )); - } - } - } - } - other => { - return Err(format!( - "field 'properties' must be an object, got {}", - other - )); - } - } - } - - let mut removed_properties = std::collections::HashSet::new(); - if let Some(removed_val) = obj.get(Self::F_REMOVED_PROPERTIES) { - match removed_val { - serde_json::Value::Null => {} - serde_json::Value::Array(arr) => { - for item in arr { - match item { - serde_json::Value::String(s) => { - removed_properties.insert(s.clone()); - } - other => { - return Err(format!( - "field 'removed_properties' must contain strings, got {}", - other - )); - } - } - } - } - other => { - return Err(format!( - "field 'removed_properties' must be an array, got {}", - other - )); - } - } - } - + /// Deserialize an alteration from JSON bytes, mirroring the + /// `serde_json::from_slice::>(...)` idiom already + /// used in `dir/manifest.rs`. Missing / null fields fall back to defaults + /// so that the sidecar format stays forward-compatible. + fn from_json_slice(bytes: &[u8]) -> serde_json::Result { + let mut obj: serde_json::Map = serde_json::from_slice(bytes)?; Ok(Self { - status, - properties, - removed_properties, + status: serde_json::from_value( + obj.remove(Self::F_STATUS) + .unwrap_or(serde_json::Value::Null), + )?, + properties: serde_json::from_value( + obj.remove(Self::F_PROPERTIES) + .unwrap_or(serde_json::Value::Null), + ) + .unwrap_or_default(), + removed_properties: serde_json::from_value( + obj.remove(Self::F_REMOVED_PROPERTIES) + .unwrap_or(serde_json::Value::Null), + ) + .unwrap_or_default(), }) } } @@ -2361,15 +2288,7 @@ impl DirectoryNamespace { ), }) })?; - let value: serde_json::Value = serde_json::from_slice(&bytes).map_err(|e| { - lance_core::Error::from(NamespaceError::Internal { - message: format!( - "Failed to parse alter_transaction sidecar for '{}': {}", - txn_uuid, e - ), - }) - })?; - let alteration = TransactionAlteration::from_json(value).map_err(|e| { + let alteration = TransactionAlteration::from_json_slice(&bytes).map_err(|e| { lance_core::Error::from(NamespaceError::Internal { message: format!( "Failed to parse alter_transaction sidecar for '{}': {}", @@ -2396,7 +2315,7 @@ impl DirectoryNamespace { alteration: &TransactionAlteration, ) -> Result<()> { let path = self.transaction_alteration_path(table_uri, txn_uuid)?; - let bytes = serde_json::to_vec(&alteration.to_json()).map_err(|e| { + let bytes = alteration.to_json_bytes().map_err(|e| { lance_core::Error::from(NamespaceError::Internal { message: format!( "Failed to serialize alter_transaction sidecar for '{}': {}",