From e619eb0942edcd8e29d0b34992dce1a04d7b3dfc Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Sat, 8 Aug 2026 20:32:50 +0000 Subject: [PATCH] fix(windows): reuse object store on table open --- rust/lancedb/src/database/listing.rs | 23 ++++++++++++++++++++++- rust/lancedb/src/table.rs | 28 ++++++++++++++++------------ 2 files changed, 38 insertions(+), 13 deletions(-) diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs index a610b5560..ed217e632 100644 --- a/rust/lancedb/src/database/listing.rs +++ b/rust/lancedb/src/database/listing.rs @@ -1321,8 +1321,23 @@ mod tests { use arrow_schema::{DataType, Field, Schema}; use futures::TryStreamExt; use std::path::PathBuf; + use std::sync::atomic::{AtomicUsize, Ordering}; use tempfile::tempdir; + #[derive(Debug)] + struct PassthroughStoreWrapper(Arc); + + impl WrappingObjectStore for PassthroughStoreWrapper { + fn wrap( + &self, + _store_prefix: &str, + target: Arc, + ) -> Arc { + self.0.fetch_add(1, Ordering::Relaxed); + target + } + } + async fn setup_database() -> (tempfile::TempDir, ListingDatabase) { let tempdir = tempdir().unwrap(); let uri = tempdir.path().to_str().unwrap(); @@ -1422,9 +1437,13 @@ mod tests { read_consistency_interval: None, session: Some(session), }; - let db = ListingDatabase::connect_with_options(&request) + let mut db = ListingDatabase::connect_with_options(&request) .await .unwrap(); + // A connection-level write wrapper must not prevent table opens from + // reusing the connection's registered object store. + let wrapper_calls = Arc::new(AtomicUsize::new(0)); + db.store_wrapper = Some(Arc::new(PassthroughStoreWrapper(wrapper_calls.clone()))); let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])); db.create_table(CreateTableRequest { @@ -1440,6 +1459,7 @@ mod tests { .unwrap(); let before_open = registry.stats(); + let wrapper_calls_before_open = wrapper_calls.load(Ordering::Relaxed); for _ in 0..3 { let table = db .open_table(OpenTableRequest { @@ -1459,6 +1479,7 @@ mod tests { let after_open = registry.stats(); assert_eq!(after_open.misses, before_open.misses); assert!(after_open.hits >= before_open.hits + 3); + assert!(wrapper_calls.load(Ordering::Relaxed) >= wrapper_calls_before_open + 3); } /// Regression test for https://github.com/lancedb/lancedb/issues/3197. diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 5120c48b7..3e0137314 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -60,7 +60,7 @@ use crate::index::{IndexConfig, IndexStatisticsImpl, IndexType}; use crate::job::Job; use crate::query::{IntoQueryVector, Query, QueryExecutionOptions, TakeQuery, VectorQuery}; use crate::table::datafusion::insert::InsertExec; -use crate::utils::{PatchReadParam, PatchWriteParam, resolve_arrow_field_path}; +use crate::utils::{PatchWriteParam, resolve_arrow_field_path}; use self::dataset::DatasetConsistencyWrapper; use self::merge::MergeInsertBuilder; @@ -2385,11 +2385,6 @@ impl NativeTable { managed_versioning: Option, ) -> Result { let params = params.unwrap_or_default(); - // patch the params if we have a write store wrapper - let params = match write_store_wrapper.clone() { - Some(wrapper) => params.patch_with_store_wrapper(wrapper)?, - None => params, - }; // Build table_id from namespace + name let mut table_id = namespace.clone(); @@ -2444,6 +2439,14 @@ impl NativeTable { } Err(e) => return Err(e.into()), }; + // Resolve the store from the session registry before applying a + // connection-level write wrapper. Wrapper identity is part of the + // registry key, so including it in ReadParams prevents reuse when the + // opened table (and its wrapped store) is short-lived. + let dataset = match write_store_wrapper { + Some(wrapper) => dataset.with_object_store_wrappers([wrapper]), + None => dataset, + }; let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval); let id = Self::build_id(&namespace, name); @@ -2546,12 +2549,6 @@ impl NativeTable { params.session(sess); } - // patch the params if we have a write store wrapper - let params = match write_store_wrapper.clone() { - Some(wrapper) => params.patch_with_store_wrapper(wrapper)?, - None => params, - }; - // Build table_id from namespace + name let mut table_id = namespace.clone(); table_id.push(name.to_string()); @@ -2573,6 +2570,13 @@ impl NativeTable { }, e => e.into(), })?; + // Apply the write wrapper after the session registry has resolved the + // shared store. The cloned dataset retains the wrapper for subsequent + // reads, manifest commits, and any additional base stores. + let dataset = match write_store_wrapper { + Some(wrapper) => dataset.with_object_store_wrappers([wrapper]), + None => dataset, + }; let uri = dataset.uri().to_string(); let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval);