diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index 72b13e903b5..2e107a8ff2e 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -738,16 +738,17 @@ impl Dataset { let message_len = LittleEndian::read_u32(&last_block[offset_in_block..offset_in_block + 4]) as usize; let message_data = &last_block[offset_in_block + 4..offset_in_block + 4 + message_len]; - let transaction: Transaction = - lance_table::format::pb::Transaction::decode(message_data)?.try_into()?; - - let metadata_cache = session.metadata_cache.for_dataset(uri); - let metadata_key = TransactionKey { - version: manifest_location.version, - }; - metadata_cache - .insert_with_key(&metadata_key, Arc::new(transaction)) - .await; + if let Some(transaction) = + decode_inline_transaction(message_data, manifest_location.version) + { + let metadata_cache = session.metadata_cache.for_dataset(uri); + let metadata_key = TransactionKey { + version: manifest_location.version, + }; + metadata_cache + .insert_with_key(&metadata_key, Arc::new(transaction)) + .await; + } } if manifest.should_use_legacy_format() { @@ -3662,6 +3663,31 @@ impl ManifestWriteConfig { } } +/// Decode an inline transaction section for opportunistic caching. +/// +/// Returns `None` instead of failing when the transaction cannot be decoded: +/// the section may have been written by a newer version of Lance with an +/// operation type this version does not know, and that must not prevent +/// opening the dataset. Paths that need the transaction contents surface the +/// error at their call sites instead. +fn decode_inline_transaction(message_data: &[u8], version: u64) -> Option { + match lance_table::format::pb::Transaction::decode(message_data) + .map_err(Error::from) + .and_then(Transaction::try_from) + { + Ok(transaction) => Some(transaction), + Err(err) => { + log::warn!( + "Failed to decode the inline transaction of version {}; \ + it may have been written by a newer version of Lance: {}", + version, + err + ); + None + } + } +} + /// Commit a manifest file and create a copy at the latest manifest path. #[allow(clippy::too_many_arguments)] pub(crate) async fn write_manifest_file( diff --git a/rust/lance/src/dataset/tests/dataset_transactions.rs b/rust/lance/src/dataset/tests/dataset_transactions.rs index 3e2a4caa3b3..548c6985cb3 100644 --- a/rust/lance/src/dataset/tests/dataset_transactions.rs +++ b/rust/lance/src/dataset/tests/dataset_transactions.rs @@ -208,6 +208,35 @@ async fn test_session_store_registry() { assert_eq!(registry.active_stores().len(), 0); } +#[test] +fn test_decode_inline_transaction_tolerates_unknown_operations() { + use crate::dataset::decode_inline_transaction; + use lance_table::format::pb; + use prost::Message; + + // A transaction written by a newer version of Lance may carry an operation + // this version cannot decode; prost surfaces it as a missing oneof. This + // must not fail (it would prevent opening the dataset), only skip caching. + let unknown_operation = pb::Transaction { + read_version: 1, + uuid: "test".to_string(), + ..Default::default() + }; + assert!(decode_inline_transaction(&unknown_operation.encode_to_vec(), 42).is_none()); + + // Corrupt bytes are likewise tolerated. + assert!(decode_inline_transaction(&[0xff, 0xff, 0xff], 42).is_none()); + + // A decodable transaction is returned. + let known = pb::Transaction::from(&Transaction::new( + 1, + Operation::Append { fragments: vec![] }, + None, + )); + let decoded = decode_inline_transaction(&known.encode_to_vec(), 42).unwrap(); + assert!(matches!(decoded.operation, Operation::Append { .. })); +} + #[tokio::test] async fn test_migrate_v2_manifest_paths() { let test_uri = TempStrDir::default();