fix(windows): reuse object store on table open

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<AtomicUsize>);
+
+    impl WrappingObjectStore for PassthroughStoreWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            target: Arc<dyn object_store::ObjectStore>,
+        ) -> Arc<dyn object_store::ObjectStore> {
+            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.

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<bool>,
     ) -> Result<Self> {
         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);