From 539395c286a99605e3dafa4ef9c2d5477b5ae4d8 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 18 Jul 2026 13:22:42 +0800 Subject: [PATCH 1/4] fix: make FTS metadata loading retry-safe --- rust/lance-core/src/error.rs | 23 ++++++- rust/lance-index/src/scalar/inverted/index.rs | 41 ++++++++++++- rust/lance/src/dataset.rs | 1 + rust/lance/src/index.rs | 60 +++++++++++++++---- rust/lance/src/io/commit.rs | 1 + rust/lance/src/session/index_caches.rs | 33 ++++++++-- 6 files changed, 141 insertions(+), 18 deletions(-) diff --git a/rust/lance-core/src/error.rs b/rust/lance-core/src/error.rs index a85f08cb741..5676e92c6a5 100644 --- a/rust/lance-core/src/error.rs +++ b/rust/lance-core/src/error.rs @@ -905,14 +905,18 @@ pub fn get_caller_location() -> &'static std::panic::Location<'static> { /// Wrap an error in a new error type that implements Clone /// /// This is useful when two threads/streams share a common fallible source -/// The base error will always have the full error. Any cloned results will -/// only have Error::Cloned with the to_string of the base error. +/// Definite not-found errors preserve their variant so compatibility fallbacks +/// can distinguish them from transient I/O failures. Other cloned results use +/// Error::Cloned with the string representation of the base error. pub struct CloneableError(pub Error); impl Clone for CloneableError { #[track_caller] fn clone(&self) -> Self { - Self(Error::cloned(self.0.to_string())) + match &self.0 { + Error::NotFound { uri, .. } => Self(Error::not_found(uri.clone())), + error => Self(Error::cloned(error.to_string())), + } } } @@ -930,6 +934,19 @@ mod test { use super::*; use std::fmt; + #[test] + fn cloneable_error_preserves_not_found_variant() { + let original = CloneableError(Error::not_found("metadata.lance")); + let cloned = original.clone(); + assert!(matches!(original.0, Error::NotFound { .. })); + assert!(matches!(cloned.0, Error::NotFound { .. })); + + let original = CloneableError(Error::timeout("metadata read timed out")); + let cloned = original.clone(); + assert!(matches!(original.0, Error::Timeout { .. })); + assert!(matches!(cloned.0, Error::Cloned { .. })); + } + #[test] fn test_caller_location_capture() { let current_fn = get_caller_location(); diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs index c1740545226..23732fd3e89 100644 --- a/rust/lance-index/src/scalar/inverted/index.rs +++ b/rust/lance-index/src/scalar/inverted/index.rs @@ -48,6 +48,7 @@ use lance_core::utils::tokio::{get_num_compute_intensive_cpus, spawn_cpu}; use lance_core::utils::tracing::{IO_TYPE_LOAD_SCALAR_PART, TRACE_IO_EVENTS}; use lance_core::{Error, ROW_ID, ROW_ID_FIELD, Result}; use lance_select::{RowAddrMask, RowAddrTreeMap}; +use object_store::Error as ObjectStoreError; use roaring::RoaringBitmap; use std::sync::LazyLock; use tokio::{sync::OnceCell, task::spawn_blocking}; @@ -127,6 +128,27 @@ pub const POSITIONS_CODEC_VARINT_DOC_DELTA_V2: &str = "varint_doc_delta_v2"; pub const POSITIONS_CODEC_PACKED_DELTA_V1: &str = "packed_delta_v1"; pub const DELETED_FRAGMENTS_COL: &str = "deleted_fragments"; +fn is_missing_index_file_error(error: &Error) -> bool { + match error { + Error::NotFound { .. } => true, + Error::IO { source, .. } | Error::Wrapped { error: source, .. } => { + is_missing_index_file_source(source.as_ref()) + } + _ => false, + } +} + +fn is_missing_index_file_source(source: &(dyn std::error::Error + 'static)) -> bool { + if let Some(error) = source.downcast_ref::() { + return is_missing_index_file_error(error); + } + if let Some(error) = source.downcast_ref::() { + return matches!(error, ObjectStoreError::NotFound { .. }) + || std::error::Error::source(error).is_some_and(is_missing_index_file_source); + } + source.source().is_some_and(is_missing_index_file_source) +} + // Just a heuristic when we need to pre-allocate memory for tokens pub const ESTIMATED_MAX_TOKENS_PER_ROW: usize = 4 * 1024; @@ -1218,7 +1240,7 @@ impl InvertedIndex { .ok_or(Error::index("params not found in metadata".to_owned()))?; Ok(serde_json::from_str::(params)?) } - Err(_) => { + Err(error) if is_missing_index_file_error(&error) => { // Legacy format: params live in the tokens file (see // `load_legacy_index`). let reader = store.open_index_file(TOKENS_FILE).await?; @@ -1230,6 +1252,7 @@ impl InvertedIndex { .transpose()? .unwrap_or_default()) } + Err(error) => Err(error), } } @@ -7164,6 +7187,22 @@ mod tests { use super::*; + #[test] + fn params_legacy_fallback_only_accepts_missing_metadata() { + assert!(is_missing_index_file_error(&Error::not_found( + METADATA_FILE + ))); + assert!(is_missing_index_file_error(&Error::wrapped(Box::new( + Error::not_found(METADATA_FILE), + )))); + assert!(!is_missing_index_file_error(&Error::timeout( + "metadata read timed out" + ))); + assert!(!is_missing_index_file_error(&Error::io( + "metadata read was denied" + ))); + } + async fn write_single_partition_index( store: Arc, params: InvertedIndexParams, diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index 70e96e323ab..cdc92ec91e4 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -741,6 +741,7 @@ impl Dataset { let ds_index_cache = session.index_cache.for_dataset(uri); let metadata_key = crate::session::index_caches::IndexMetadataKey { version: manifest_location.version, + store_identity: &object_store.store_prefix, }; ds_index_cache .insert_with_key(&metadata_key, Arc::new(indices)) diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index a1d38f0e4b1..a68f848c7e8 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -1314,10 +1314,11 @@ impl DatasetIndexExt for Dataset { async fn load_indices(&self) -> Result>> { let metadata_key = IndexMetadataKey { version: self.version().version, + store_identity: &self.object_store.store_prefix, }; - let mut indices = match self.index_cache.get_with_key(&metadata_key).await { - Some(indices) => indices, - None => { + let mut indices = self + .index_cache + .get_or_insert_with_key(metadata_key, || async { let mut loaded_indices = read_manifest_indexes( &self.object_store, &self.manifest_location, @@ -1325,13 +1326,9 @@ impl DatasetIndexExt for Dataset { ) .await?; retain_supported_indices(&mut loaded_indices); - let loaded_indices = Arc::new(loaded_indices); - self.index_cache - .insert_with_key(&metadata_key, loaded_indices.clone()) - .await; - loaded_indices - } - }; + Ok(loaded_indices) + }) + .await?; // Infer details for legacy vector indices (once per index name, concurrently). // This may run on indices that were opportunistically cached during Dataset::open @@ -4350,6 +4347,48 @@ mod tests { assert_eq!(indices2.len(), 1); } + #[tokio::test] + async fn test_load_indices_singleflights_concurrent_cache_misses() { + let session = Arc::new(Session::default()); + let write_params = WriteParams { + session: Some(session.clone()), + ..Default::default() + }; + let test_dir = TempStrDir::default(); + let schema = Arc::new(Schema::new(vec![Field::new("tag", DataType::Utf8, false)])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![Arc::new(StringArray::from(vec!["a", "b", "c"]))], + ) + .unwrap(); + let mut dataset = Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema), + &test_dir, + Some(write_params), + ) + .await + .unwrap(); + dataset + .create_index( + &["tag"], + IndexType::Bitmap, + None, + &ScalarIndexParams::default(), + false, + ) + .await + .unwrap(); + + session.index_cache.clear().await; + let before = session.index_cache_stats().await; + let results = futures::future::join_all((0..32).map(|_| dataset.load_indices())).await; + assert!(results.iter().all(|result| result.is_ok())); + let after = session.index_cache_stats().await; + + assert_eq!(after.misses - before.misses, 1); + assert_eq!(after.hits - before.hits, 31); + } + #[tokio::test] async fn test_remap_empty() { let data = gen_batch() @@ -4736,6 +4775,7 @@ mod tests { // We commit by doing a delete("false") after replacing the cached indices. let metadata_key = crate::session::index_caches::IndexMetadataKey { version: dataset.version().version, + store_identity: &dataset.object_store.store_prefix, }; dataset .index_cache diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index 35cd275b8ee..54ad8355cb6 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -1073,6 +1073,7 @@ pub(crate) async fn commit_transaction( if !indices.is_empty() { let key = IndexMetadataKey { version: target_version, + store_identity: &dataset.object_store.store_prefix, }; dataset .index_cache diff --git a/rust/lance/src/session/index_caches.rs b/rust/lance/src/session/index_caches.rs index e93b7208b09..60e52cf55be 100644 --- a/rust/lance/src/session/index_caches.rs +++ b/rust/lance/src/session/index_caches.rs @@ -94,16 +94,22 @@ impl CacheKey for FragReuseIndexKey<'_> { } } -#[derive(Debug)] -pub struct IndexMetadataKey { +#[derive(Clone, Copy, Debug)] +pub struct IndexMetadataKey<'a> { pub version: u64, + pub store_identity: &'a str, } -impl CacheKey for IndexMetadataKey { +impl CacheKey for IndexMetadataKey<'_> { type ValueType = Vec; fn key(&self) -> Cow<'_, str> { - Cow::Owned(self.version.to_string()) + Cow::Owned(format!( + "{}:{}/{}", + self.store_identity.len(), + self.store_identity, + self.version + )) } fn type_name() -> &'static str { @@ -145,3 +151,22 @@ impl CacheKey for ScalarIndexDetailsKey<'_> { "ScalarIndexDetails" } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn index_metadata_key_isolates_object_store_identity() { + let first = IndexMetadataKey { + version: 7, + store_identity: "s3$first-options", + }; + let second = IndexMetadataKey { + version: 7, + store_identity: "s3$second-options", + }; + + assert_ne!(first.key(), second.key()); + } +} From ce64c6d0da36688d3fea26930f128c8d15704c50 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 18 Jul 2026 13:44:11 +0800 Subject: [PATCH 2/4] fix: expose typed not-found detection --- rust/lance-core/src/error.rs | 22 ++++++++++ rust/lance-index/src/scalar/inverted/index.rs | 40 +++---------------- rust/lance/src/dataset/fragment.rs | 12 +++--- 3 files changed, 33 insertions(+), 41 deletions(-) diff --git a/rust/lance-core/src/error.rs b/rust/lance-core/src/error.rs index 5676e92c6a5..0ee3bd895fa 100644 --- a/rust/lance-core/src/error.rs +++ b/rust/lance-core/src/error.rs @@ -519,6 +519,17 @@ impl Error { NotFoundSnafu { uri: uri.into() }.build() } + /// Return whether this error or one of its typed sources is a missing object. + pub fn is_not_found(&self) -> bool { + match self { + Self::NotFound { .. } => true, + Self::IO { source, .. } | Self::Wrapped { error: source, .. } => { + error_source_is_not_found(source.as_ref()) + } + _ => false, + } + } + #[track_caller] pub fn wrapped(error: BoxedError) -> Self { WrappedSnafu { error }.build() @@ -700,6 +711,17 @@ impl Error { } } +fn error_source_is_not_found(source: &(dyn std::error::Error + 'static)) -> bool { + if let Some(error) = source.downcast_ref::() { + return error.is_not_found(); + } + if let Some(error) = source.downcast_ref::() { + return matches!(error, object_store::Error::NotFound { .. }) + || std::error::Error::source(error).is_some_and(error_source_is_not_found); + } + source.source().is_some_and(error_source_is_not_found) +} + pub trait LanceOptionExt { /// Unwraps an option, returning an internal error if the option is None. /// diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs index 23732fd3e89..ec57a55c2a5 100644 --- a/rust/lance-index/src/scalar/inverted/index.rs +++ b/rust/lance-index/src/scalar/inverted/index.rs @@ -48,7 +48,6 @@ use lance_core::utils::tokio::{get_num_compute_intensive_cpus, spawn_cpu}; use lance_core::utils::tracing::{IO_TYPE_LOAD_SCALAR_PART, TRACE_IO_EVENTS}; use lance_core::{Error, ROW_ID, ROW_ID_FIELD, Result}; use lance_select::{RowAddrMask, RowAddrTreeMap}; -use object_store::Error as ObjectStoreError; use roaring::RoaringBitmap; use std::sync::LazyLock; use tokio::{sync::OnceCell, task::spawn_blocking}; @@ -128,27 +127,6 @@ pub const POSITIONS_CODEC_VARINT_DOC_DELTA_V2: &str = "varint_doc_delta_v2"; pub const POSITIONS_CODEC_PACKED_DELTA_V1: &str = "packed_delta_v1"; pub const DELETED_FRAGMENTS_COL: &str = "deleted_fragments"; -fn is_missing_index_file_error(error: &Error) -> bool { - match error { - Error::NotFound { .. } => true, - Error::IO { source, .. } | Error::Wrapped { error: source, .. } => { - is_missing_index_file_source(source.as_ref()) - } - _ => false, - } -} - -fn is_missing_index_file_source(source: &(dyn std::error::Error + 'static)) -> bool { - if let Some(error) = source.downcast_ref::() { - return is_missing_index_file_error(error); - } - if let Some(error) = source.downcast_ref::() { - return matches!(error, ObjectStoreError::NotFound { .. }) - || std::error::Error::source(error).is_some_and(is_missing_index_file_source); - } - source.source().is_some_and(is_missing_index_file_source) -} - // Just a heuristic when we need to pre-allocate memory for tokens pub const ESTIMATED_MAX_TOKENS_PER_ROW: usize = 4 * 1024; @@ -1240,7 +1218,7 @@ impl InvertedIndex { .ok_or(Error::index("params not found in metadata".to_owned()))?; Ok(serde_json::from_str::(params)?) } - Err(error) if is_missing_index_file_error(&error) => { + Err(error) if error.is_not_found() => { // Legacy format: params live in the tokens file (see // `load_legacy_index`). let reader = store.open_index_file(TOKENS_FILE).await?; @@ -7189,18 +7167,10 @@ mod tests { #[test] fn params_legacy_fallback_only_accepts_missing_metadata() { - assert!(is_missing_index_file_error(&Error::not_found( - METADATA_FILE - ))); - assert!(is_missing_index_file_error(&Error::wrapped(Box::new( - Error::not_found(METADATA_FILE), - )))); - assert!(!is_missing_index_file_error(&Error::timeout( - "metadata read timed out" - ))); - assert!(!is_missing_index_file_error(&Error::io( - "metadata read was denied" - ))); + assert!(Error::not_found(METADATA_FILE).is_not_found()); + assert!(Error::wrapped(Box::new(Error::not_found(METADATA_FILE))).is_not_found()); + assert!(!Error::timeout("metadata read timed out").is_not_found()); + assert!(!Error::io("metadata read was denied").is_not_found()); } async fn write_single_partition_index( diff --git a/rust/lance/src/dataset/fragment.rs b/rust/lance/src/dataset/fragment.rs index b365dcb4d26..74073c46d2b 100644 --- a/rust/lance/src/dataset/fragment.rs +++ b/rust/lance/src/dataset/fragment.rs @@ -3748,11 +3748,11 @@ mod tests { // A take that hits the coverage does need the file, so it now fails with // a not-found error naming the missing overlay file. let err = frag.take(&[5], &val_only).await.unwrap_err(); - let err = format!("{err:?}"); + let message = format!("{err:?}"); assert!( - err.contains("miss.lance") && err.to_lowercase().contains("not found"), + err.is_not_found() && message.contains("miss.lance"), "take hitting the overlay's coverage should fail with a not-found error \ - for its missing file, got: {err}", + for its missing file, got: {message}", ); } @@ -4192,11 +4192,11 @@ mod tests { // Projecting the overlaid `val` column does need the file, so it fails // with a not-found error naming the missing overlay file. let err = frag.take(&[0], &val_only).await.unwrap_err(); - let err = format!("{err:?}"); + let message = format!("{err:?}"); assert!( - err.contains("valov.lance") && err.to_lowercase().contains("not found"), + err.is_not_found() && message.contains("valov.lance"), "projecting the overlaid column should fail with a not-found error \ - for its missing file, got: {err}", + for its missing file, got: {message}", ); } From df0e099f303a0a01fb09a9cf8aaf491a2aceacb8 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sun, 19 Jul 2026 03:11:51 +0800 Subject: [PATCH 3/4] fix: preserve index loader error compatibility --- rust/lance-core/src/cache/mod.rs | 33 +++++ rust/lance-core/src/cache/moka.rs | 27 ++-- rust/lance-core/src/error.rs | 68 +++++++-- rust/lance-index/src/scalar/inverted/index.rs | 131 ++++++++++++++++-- 4 files changed, 222 insertions(+), 37 deletions(-) diff --git a/rust/lance-core/src/cache/mod.rs b/rust/lance-core/src/cache/mod.rs index 4f93f261bea..cd7476011bf 100644 --- a/rust/lance-core/src/cache/mod.rs +++ b/rust/lance-core/src/cache/mod.rs @@ -595,8 +595,10 @@ impl CacheStats { #[cfg(test)] mod tests { use super::*; + use crate::Error; use std::collections::{BTreeSet, HashMap}; use std::marker::PhantomData; + use std::sync::atomic::AtomicUsize; struct TestKey { key: String, @@ -919,6 +921,37 @@ mod tests { assert_eq!(cache.stats().await.hits, 1); } + #[tokio::test] + async fn test_cache_coalesces_concurrent_loader_errors() { + let cache = LanceCache::with_capacity(1000); + let barrier = Arc::new(tokio::sync::Barrier::new(5)); + let loader_calls = Arc::new(AtomicUsize::new(0)); + + let results = futures::future::join_all((0..5).map(|_| { + let cache = cache.clone(); + let barrier = barrier.clone(); + let loader_calls = loader_calls.clone(); + async move { + barrier.wait().await; + cache + .get_or_insert_with_key(TestKey::>::new("error"), || async move { + loader_calls.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + Err(Error::timeout("metadata read timed out")) + }) + .await + } + })) + .await; + + assert_eq!(loader_calls.load(Ordering::Relaxed), 1); + assert!( + results + .iter() + .all(|result| matches!(result, Err(Error::Timeout { .. }))) + ); + } + #[tokio::test] async fn test_custom_backend() { use async_trait::async_trait; diff --git a/rust/lance-core/src/cache/moka.rs b/rust/lance-core/src/cache/moka.rs index fd86b064f6b..024d11cdce8 100644 --- a/rust/lance-core/src/cache/moka.rs +++ b/rust/lance-core/src/cache/moka.rs @@ -9,6 +9,7 @@ use async_trait::async_trait; use futures::Future; use crate::Result; +use crate::error::CloneableError; use super::CacheCodec; use super::backend::{CacheBackend, CacheEntry, CacheKeyIterator, InternalCacheKey}; @@ -88,37 +89,25 @@ impl CacheBackend for MokaCacheBackend { loader: Pin> + Send + 'a>>, _codec: Option, ) -> Result<(CacheEntry, bool)> { - // Use moka's built-in dedup: optionally_get_with runs the init future - // at most once per key, even under concurrent access. - let (error_tx, error_rx) = tokio::sync::oneshot::channel(); - // Track whether the loader actually ran (= cache miss). let was_miss = Arc::new(AtomicBool::new(false)); let was_miss_clone = was_miss.clone(); let init = async move { was_miss_clone.store(true, Ordering::Relaxed); - match loader.await { - Ok((entry, size_bytes)) => Some(MokaCacheEntry { entry, size_bytes }), - Err(e) => { - let _ = error_tx.send(e); - None - } - } + loader + .await + .map(|(entry, size_bytes)| MokaCacheEntry { entry, size_bytes }) + .map_err(CloneableError) }; let owned_key = key.clone(); - match self.cache.optionally_get_with(owned_key, init).await { - Some(record) => { + match self.cache.try_get_with(owned_key, init).await { + Ok(record) => { let was_cached = !was_miss.load(Ordering::Relaxed); Ok((record.entry, was_cached)) } - None => match error_rx.await { - Ok(err) => Err(err), - Err(_) => Err(crate::Error::internal( - "Failed to retrieve error from cache loader", - )), - }, + Err(error) => Err(Arc::unwrap_or_clone(error).0), } } diff --git a/rust/lance-core/src/error.rs b/rust/lance-core/src/error.rs index 0ee3bd895fa..7bbd155f9a1 100644 --- a/rust/lance-core/src/error.rs +++ b/rust/lance-core/src/error.rs @@ -292,6 +292,7 @@ pub enum Error { Stop, #[snafu(display("Wrapped error: {error}, {location}"))] Wrapped { + #[snafu(source)] error: BoxedError, #[snafu(implicit)] location: Location, @@ -532,7 +533,7 @@ impl Error { #[track_caller] pub fn wrapped(error: BoxedError) -> Self { - WrappedSnafu { error }.build() + WrappedSnafu.into_error(error) } #[track_caller] @@ -927,16 +928,41 @@ pub fn get_caller_location() -> &'static std::panic::Location<'static> { /// Wrap an error in a new error type that implements Clone /// /// This is useful when two threads/streams share a common fallible source -/// Definite not-found errors preserve their variant so compatibility fallbacks -/// can distinguish them from transient I/O failures. Other cloned results use -/// Error::Cloned with the string representation of the base error. +/// Definite not-found errors preserve typed source-chain detection and their +/// human-readable representation. Timeout and I/O errors preserve their error +/// categories. Other cloned results use Error::Cloned with the string +/// representation of the base error. pub struct CloneableError(pub Error); +struct DisplayError(Error); + +impl fmt::Debug for DisplayError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, f) + } +} + +impl fmt::Display for DisplayError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.0, f) + } +} + +impl std::error::Error for DisplayError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.0) + } +} + impl Clone for CloneableError { #[track_caller] fn clone(&self) -> Self { match &self.0 { - Error::NotFound { uri, .. } => Self(Error::not_found(uri.clone())), + Error::NotFound { uri, .. } => Self(Error::wrapped(Box::new(DisplayError( + Error::not_found(uri.clone()), + )))), + Error::Timeout { message, .. } => Self(Error::timeout(message.clone())), + Error::IO { source, .. } => Self(Error::io(source.to_string())), error => Self(Error::cloned(error.to_string())), } } @@ -954,19 +980,45 @@ impl From> for CloneableResult { #[cfg(test)] mod test { use super::*; + use std::error::Error as _; use std::fmt; #[test] - fn cloneable_error_preserves_not_found_variant() { + fn cloneable_error_preserves_not_found_contract() { let original = CloneableError(Error::not_found("metadata.lance")); let cloned = original.clone(); assert!(matches!(original.0, Error::NotFound { .. })); - assert!(matches!(cloned.0, Error::NotFound { .. })); + assert!(cloned.0.is_not_found()); + assert!(cloned.0.to_string().to_lowercase().contains("not found")); + assert!( + format!("{:?}", cloned.0) + .to_lowercase() + .contains("not found") + ); + assert!(cloned.0.source().is_some_and(|source| source.is::() + || source.source().is_some_and(|source| source.is::()))); + let downstream_error = Error::wrapped(Box::new(Error::io_source(Box::new( + object_store::Error::Generic { + store: "N/A", + source: Box::new(cloned.0), + }, + )))); + assert!(downstream_error.is_not_found()); + assert!( + format!("{downstream_error:?}") + .to_lowercase() + .contains("not found") + ); let original = CloneableError(Error::timeout("metadata read timed out")); let cloned = original.clone(); assert!(matches!(original.0, Error::Timeout { .. })); - assert!(matches!(cloned.0, Error::Cloned { .. })); + assert!(matches!(cloned.0, Error::Timeout { .. })); + + let original = CloneableError(Error::io("metadata read was denied")); + let cloned = original.clone(); + assert!(matches!(original.0, Error::IO { .. })); + assert!(matches!(cloned.0, Error::IO { .. })); } #[test] diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs index ec57a55c2a5..f165582b144 100644 --- a/rust/lance-index/src/scalar/inverted/index.rs +++ b/rust/lance-index/src/scalar/inverted/index.rs @@ -1218,10 +1218,14 @@ impl InvertedIndex { .ok_or(Error::index("params not found in metadata".to_owned()))?; Ok(serde_json::from_str::(params)?) } - Err(error) if error.is_not_found() => { + Err(metadata_error) => { // Legacy format: params live in the tokens file (see - // `load_legacy_index`). - let reader = store.open_index_file(TOKENS_FILE).await?; + // `load_legacy_index`). Some S3 configurations return 403 for + // a missing object, so the readable legacy file is the + // authoritative format probe. + let Ok(reader) = store.open_index_file(TOKENS_FILE).await else { + return Err(metadata_error); + }; Ok(reader .schema() .metadata @@ -1230,7 +1234,6 @@ impl InvertedIndex { .transpose()? .unwrap_or_default()) } - Err(error) => Err(error), } } @@ -7165,12 +7168,120 @@ mod tests { use super::*; - #[test] - fn params_legacy_fallback_only_accepts_missing_metadata() { - assert!(Error::not_found(METADATA_FILE).is_not_found()); - assert!(Error::wrapped(Box::new(Error::not_found(METADATA_FILE))).is_not_found()); - assert!(!Error::timeout("metadata read timed out").is_not_found()); - assert!(!Error::io("metadata read was denied").is_not_found()); + #[derive(Debug)] + struct MetadataAccessDeniedStore { + inner: Arc, + } + + impl DeepSizeOf for MetadataAccessDeniedStore { + fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize { + self.inner.deep_size_of_children(context) + } + } + + #[async_trait] + impl IndexStore for MetadataAccessDeniedStore { + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn clone_arc(&self) -> Arc { + Arc::new(Self { + inner: self.inner.clone(), + }) + } + + fn io_parallelism(&self) -> usize { + self.inner.io_parallelism() + } + + async fn new_index_file( + &self, + name: &str, + schema: Arc, + ) -> Result> { + self.inner.new_index_file(name, schema).await + } + + async fn open_index_file(&self, name: &str) -> Result> { + if name == METADATA_FILE { + Err(Error::io("metadata access denied")) + } else { + self.inner.open_index_file(name).await + } + } + + fn with_io_priority(&self, io_priority: u64) -> Arc { + Arc::new(Self { + inner: self.inner.with_io_priority(io_priority), + }) + } + + async fn copy_index_file( + &self, + name: &str, + dest_store: &dyn IndexStore, + ) -> Result { + self.inner.copy_index_file(name, dest_store).await + } + + async fn rename_index_file( + &self, + name: &str, + new_name: &str, + ) -> Result { + self.inner.rename_index_file(name, new_name).await + } + + async fn delete_index_file(&self, name: &str) -> Result<()> { + self.inner.delete_index_file(name).await + } + + async fn list_files_with_sizes(&self) -> Result> { + self.inner.list_files_with_sizes().await + } + } + + #[tokio::test] + async fn params_legacy_fallback_probes_tokens_after_metadata_access_denied() { + let tmpdir = TempObjDir::default(); + let inner: Arc = Arc::new(LanceIndexStore::new( + ObjectStore::local().into(), + tmpdir.clone(), + Arc::new(LanceCache::no_cache()), + )); + let expected = InvertedIndexParams::default(); + let metadata = HashMap::from([( + "tokenizer".to_owned(), + serde_json::to_string(&expected).unwrap(), + )]); + let mut writer = inner + .new_index_file(TOKENS_FILE, Arc::new(Schema::empty())) + .await + .unwrap(); + writer.finish_with_metadata(metadata).await.unwrap(); + let store = MetadataAccessDeniedStore { inner }; + + let actual = InvertedIndex::load_params(&store).await.unwrap(); + assert_eq!( + serde_json::to_value(actual).unwrap(), + serde_json::to_value(expected).unwrap() + ); + } + + #[tokio::test] + async fn params_legacy_probe_preserves_metadata_error_when_tokens_are_missing() { + let tmpdir = TempObjDir::default(); + let inner: Arc = Arc::new(LanceIndexStore::new( + ObjectStore::local().into(), + tmpdir.clone(), + Arc::new(LanceCache::no_cache()), + )); + let store = MetadataAccessDeniedStore { inner }; + + let error = InvertedIndex::load_params(&store).await.unwrap_err(); + assert!(matches!(error, Error::IO { .. })); + assert!(error.to_string().contains("metadata access denied")); } async fn write_single_partition_index( From 95de72f12df4d810a51b30db7a4cf40de6c2967f Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sun, 19 Jul 2026 04:08:48 +0800 Subject: [PATCH 4/4] fix: preserve not-found across repeated clones --- rust/lance-core/src/error.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/rust/lance-core/src/error.rs b/rust/lance-core/src/error.rs index 7bbd155f9a1..5d92b69457e 100644 --- a/rust/lance-core/src/error.rs +++ b/rust/lance-core/src/error.rs @@ -961,6 +961,9 @@ impl Clone for CloneableError { Error::NotFound { uri, .. } => Self(Error::wrapped(Box::new(DisplayError( Error::not_found(uri.clone()), )))), + error if error.is_not_found() => Self(Error::wrapped(Box::new(DisplayError( + Error::not_found(error.to_string()), + )))), Error::Timeout { message, .. } => Self(Error::timeout(message.clone())), Error::IO { source, .. } => Self(Error::io(source.to_string())), error => Self(Error::cloned(error.to_string())), @@ -987,9 +990,18 @@ mod test { fn cloneable_error_preserves_not_found_contract() { let original = CloneableError(Error::not_found("metadata.lance")); let cloned = original.clone(); + let cloned_again = cloned.clone(); assert!(matches!(original.0, Error::NotFound { .. })); assert!(cloned.0.is_not_found()); + assert!(cloned_again.0.is_not_found()); assert!(cloned.0.to_string().to_lowercase().contains("not found")); + assert!( + cloned_again + .0 + .to_string() + .to_lowercase() + .contains("not found") + ); assert!( format!("{:?}", cloned.0) .to_lowercase()