Merge origin/main into gatekeeper/fix-472-1

Cargo.lock

@@ -5495,6 +5495,7 @@ dependencies = [
  "urlencoding",
  "uuid",
  "walkdir",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]

rust/lancedb/Cargo.toml

@@ -115,6 +115,11 @@ serial_test = "3"
 [target.'cfg(unix)'.dev-dependencies]
 pprof = { version = "0.14", features = ["flamegraph"] }
 
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.61", features = [
+    "Win32_Foundation",
+    "Win32_Storage_FileSystem",
+] }
 
 [features]
 default = []

rust/lancedb/src/database/listing.rs

@@ -25,7 +25,7 @@ use crate::database::namespace::LanceNamespaceDatabase;
 use crate::error::{CreateDirSnafu, Error, Result};
 use crate::io::object_store::MirroringObjectStoreWrapper;
 use crate::table::NativeTable;
-use crate::utils::validate_table_name;
+use crate::utils::{PatchStoreParam, validate_table_name};
 
 use lance_namespace::models::{
     CreateNamespaceRequest, CreateNamespaceResponse, DescribeNamespaceRequest,
@@ -355,6 +355,14 @@ impl ListingDatabase {
         url.to_string()
     }
 
+    #[cfg(any(windows, test))]
+    fn uses_local_file_provider(object_store: &ObjectStore) -> bool {
+        matches!(
+            object_store.scheme(),
+            "file" | "file-object-store" | "file+uring"
+        )
+    }
+
     async fn prepare_namespace_root(
         uri: &str,
         storage_options: &HashMap<String, String>,
@@ -581,6 +589,17 @@ impl ListingDatabase {
                     }
                     None => None,
                 };
+                #[cfg(windows)]
+                let write_store_wrapper = if Self::uses_local_file_provider(&object_store) {
+                    // Local manifest commits need create-only rename semantics,
+                    // including on filesystems that do not support hard links.
+                    Some(
+                        Arc::new(crate::io::object_store::windows::WindowsLocalFileSystemWrapper)
+                            as Arc<dyn WrappingObjectStore>,
+                    )
+                } else {
+                    write_store_wrapper
+                };
 
                 let namespace_database = Self::connect_namespace_database(
                     &storage_base_uri,
@@ -645,12 +664,22 @@ impl ListingDatabase {
         )
         .await?;
 
+        #[cfg(windows)]
+        let write_store_wrapper = Self::uses_local_file_provider(&object_store).then(|| {
+            // Local manifest commits need create-only rename semantics,
+            // including on filesystems that do not support hard links.
+            Arc::new(crate::io::object_store::windows::WindowsLocalFileSystemWrapper)
+                as Arc<dyn WrappingObjectStore>
+        });
+        #[cfg(not(windows))]
+        let write_store_wrapper = None;
+
         Ok(Self {
             uri: path.to_string(),
             query_string: None,
             base_path,
             object_store,
-            store_wrapper: None,
+            store_wrapper: write_store_wrapper,
             read_consistency_interval,
             storage_options: HashMap::new(),
             storage_options_provider: None,
@@ -1112,6 +1141,12 @@ impl Database for ListingDatabase {
             },
             ..Default::default()
         };
+        let storage_params = match self.store_wrapper.clone() {
+            Some(wrapper) => Some(storage_params)
+                .patch_with_store_wrapper(wrapper)?
+                .expect("patching store params always returns parameters"),
+            None => storage_params,
+        };
         let read_params = ReadParams {
             store_options: Some(storage_params.clone()),
             session: Some(self.session.clone()),
@@ -1295,6 +1330,7 @@ mod tests {
     use crate::connection::ConnectRequest;
     use crate::data::scannable::Scannable;
     use crate::database::{CreateTableMode, CreateTableRequest};
+    use crate::io::object_store::io_tracking::IoStatsHolder;
     use crate::query::QueryRequest;
     use crate::table::{AnyQuery, WriteOptions};
     use arrow_array::{Int32Array, RecordBatch, StringArray};
@@ -1302,11 +1338,26 @@ mod tests {
     use futures::{TryStreamExt, stream::once};
     use std::path::PathBuf;
     use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::time::Duration;
     use tempfile::tempdir;
     use tokio::sync::Barrier;
     use tokio::time::timeout;
 
+    #[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();
@@ -1491,6 +1542,25 @@ mod tests {
         assert!(!tempdir.path().join("__manifest").exists());
     }
 
+    #[tokio::test]
+    async fn test_file_object_store_uses_local_file_provider() {
+        let tempdir = tempdir().unwrap();
+        let path = tempdir.path().to_string_lossy().replace('\\', "/");
+        let uri = if path.starts_with('/') {
+            format!("file-object-store://{path}")
+        } else {
+            format!("file-object-store:///{path}")
+        };
+        let registry = Arc::new(lance_io::object_store::ObjectStoreRegistry::default());
+        let (store, _) =
+            ObjectStore::from_uri_and_params(registry, &uri, &ObjectStoreParams::default())
+                .await
+                .unwrap();
+
+        assert_eq!(store.scheme(), "file-object-store");
+        assert!(ListingDatabase::uses_local_file_provider(&store));
+    }
+
     /// Regression test for https://github.com/lancedb/lancedb/issues/1600.
     ///
     /// Opening a table used to create a separate object-store client instead of
@@ -1514,9 +1584,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 {
@@ -1532,6 +1606,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 {
@@ -1551,6 +1626,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.
@@ -1694,6 +1770,46 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_clone_table_uses_connection_store_wrapper() {
+        let (_tempdir, mut db) = setup_database().await;
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        db.create_table(CreateTableRequest {
+            name: "source_table".to_string(),
+            namespace_path: vec![],
+            data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        let source_uri = db.table_uri("source_table").unwrap();
+        let tracker = IoStatsHolder::default();
+        db.store_wrapper = Some(Arc::new(tracker.clone()));
+        let _ = tracker.incremental_stats();
+
+        db.clone_table(CloneTableRequest {
+            target_table_name: "cloned_table".to_string(),
+            target_namespace_path: vec![],
+            source_uri,
+            source_version: None,
+            source_tag: None,
+            is_shallow: true,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        let stats = tracker.incremental_stats();
+        assert!(
+            stats.write_iops > 0,
+            "clone bypassed the wrapper: {stats:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_clone_table_with_data() {
         let (_tempdir, db) = setup_database().await;

rust/lancedb/src/io/object_store.rs

@@ -18,6 +18,9 @@ use async_trait::async_trait;
 #[cfg(test)]
 pub mod io_tracking;
 
+#[cfg(windows)]
+pub(crate) mod windows;
+
 #[derive(Debug)]
 struct MirroringObjectStore {
     primary: Arc<dyn ObjectStore>,

rust/lancedb/src/io/object_store/windows.rs

@@ -0,0 +1,223 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Windows local filesystem compatibility for atomic manifest commits.
+
+use std::ffi::OsStr;
+use std::fmt::{Display, Formatter};
+use std::os::windows::ffi::OsStrExt;
+use std::path::{Path as StdPath, PathBuf};
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures::stream::BoxStream;
+use lance::io::WrappingObjectStore;
+use object_store::{
+    CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
+    ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, RenameOptions,
+    RenameTargetMode, Result, UploadPart, path::Path,
+};
+use windows_sys::Win32::Foundation::{ERROR_ALREADY_EXISTS, ERROR_FILE_EXISTS};
+use windows_sys::Win32::Storage::FileSystem::MoveFileExW;
+
+const STORE_NAME: &str = "WindowsLocalFileSystem";
+
+/// Uses the Windows move primitive for create-only renames on local stores.
+///
+/// `object_store` implements create-only local renames with a hard link followed
+/// by a delete. Some Windows filesystems do not support hard links, but
+/// `MoveFileExW` without `MOVEFILE_REPLACE_EXISTING` provides the same atomic
+/// create-only rename semantics without requiring them.
+#[derive(Debug, Default)]
+pub struct WindowsLocalFileSystemWrapper;
+
+impl WrappingObjectStore for WindowsLocalFileSystemWrapper {
+    fn wrap(&self, _store_prefix: &str, target: Arc<dyn ObjectStore>) -> Arc<dyn ObjectStore> {
+        Arc::new(WindowsLocalFileSystem { target })
+    }
+}
+
+#[derive(Debug)]
+struct WindowsLocalFileSystem {
+    target: Arc<dyn ObjectStore>,
+}
+
+impl Display for WindowsLocalFileSystem {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{STORE_NAME}({})", self.target)
+    }
+}
+
+#[async_trait::async_trait]
+#[deny(clippy::missing_trait_methods)]
+impl ObjectStore for WindowsLocalFileSystem {
+    async fn put_opts(
+        &self,
+        location: &Path,
+        bytes: PutPayload,
+        opts: PutOptions,
+    ) -> Result<PutResult> {
+        self.target.put_opts(location, bytes, opts).await
+    }
+
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOptions,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        self.target.put_multipart_opts(location, opts).await
+    }
+
+    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
+        self.target.get_opts(location, options).await
+    }
+
+    async fn get_ranges(
+        &self,
+        location: &Path,
+        ranges: &[std::ops::Range<u64>],
+    ) -> Result<Vec<Bytes>> {
+        self.target.get_ranges(location, ranges).await
+    }
+
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, Result<Path>>,
+    ) -> BoxStream<'static, Result<Path>> {
+        self.target.delete_stream(locations)
+    }
+
+    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
+        self.target.list(prefix)
+    }
+
+    fn list_with_offset(
+        &self,
+        prefix: Option<&Path>,
+        offset: &Path,
+    ) -> BoxStream<'static, Result<ObjectMeta>> {
+        self.target.list_with_offset(prefix, offset)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        self.target.list_with_delimiter(prefix).await
+    }
+
+    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
+        self.target.copy_opts(from, to, options).await
+    }
+
+    async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
+        if options.target_mode != RenameTargetMode::Create {
+            return self.target.rename_opts(from, to, options).await;
+        }
+
+        let from = PathBuf::from(from.as_ref());
+        let to = PathBuf::from(to.as_ref());
+        tokio::task::spawn_blocking(move || move_file_if_not_exists(&from, &to))
+            .await
+            .map_err(|source| Error::Generic {
+                store: STORE_NAME,
+                source: Box::new(source),
+            })?
+    }
+}
+
+fn move_file_if_not_exists(from: &StdPath, to: &StdPath) -> Result<()> {
+    let from_wide = null_terminated_wide(from.as_os_str());
+    let to_wide = null_terminated_wide(to.as_os_str());
+
+    // SAFETY: both pointers reference null-terminated UTF-16 buffers that remain
+    // alive for the duration of this call. A zero flag value deliberately omits
+    // MOVEFILE_REPLACE_EXISTING, giving this operation create-only semantics.
+    if unsafe { MoveFileExW(from_wide.as_ptr(), to_wide.as_ptr(), 0) } != 0 {
+        return Ok(());
+    }
+
+    let source = std::io::Error::last_os_error();
+    let path = to.to_string_lossy().into_owned();
+    match source.raw_os_error().map(|code| code as u32) {
+        Some(ERROR_ALREADY_EXISTS | ERROR_FILE_EXISTS) => Err(Error::AlreadyExists {
+            path,
+            source: Box::new(source),
+        }),
+        _ if source.kind() == std::io::ErrorKind::NotFound => Err(Error::NotFound {
+            path,
+            source: Box::new(source),
+        }),
+        _ => Err(Error::Generic {
+            store: STORE_NAME,
+            source: Box::new(source),
+        }),
+    }
+}
+
+fn null_terminated_wide(value: &OsStr) -> Vec<u16> {
+    value.encode_wide().chain(Some(0)).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use object_store::memory::InMemory;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn create_only_rename_does_not_use_hard_links() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let source_path = tempdir.path().join("staged.manifest");
+        let destination_path = tempdir.path().join("1.manifest");
+        std::fs::write(&source_path, b"manifest").unwrap();
+
+        let source = Path::from_absolute_path(&source_path).unwrap();
+        let destination = Path::from_absolute_path(&destination_path).unwrap();
+        let store = WindowsLocalFileSystem {
+            // The source does not exist in this inner store. Delegating the
+            // rename would fail, proving the wrapper uses the native move path.
+            target: Arc::new(InMemory::new()),
+        };
+
+        store
+            .rename_opts(
+                &source,
+                &destination,
+                RenameOptions::new().with_target_mode(RenameTargetMode::Create),
+            )
+            .await
+            .unwrap();
+
+        assert!(!source_path.exists());
+        assert_eq!(std::fs::read(destination_path).unwrap(), b"manifest");
+    }
+
+    #[tokio::test]
+    async fn create_only_rename_preserves_existing_destination() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let source_path = tempdir.path().join("staged.manifest");
+        let destination_path = tempdir.path().join("1.manifest");
+        std::fs::write(&source_path, b"new manifest").unwrap();
+        std::fs::write(&destination_path, b"existing manifest").unwrap();
+
+        let source = Path::from_absolute_path(&source_path).unwrap();
+        let destination = Path::from_absolute_path(&destination_path).unwrap();
+        let store = WindowsLocalFileSystem {
+            target: Arc::new(InMemory::new()),
+        };
+
+        let error = store
+            .rename_opts(
+                &source,
+                &destination,
+                RenameOptions::new().with_target_mode(RenameTargetMode::Create),
+            )
+            .await
+            .unwrap_err();
+
+        assert!(matches!(error, Error::AlreadyExists { .. }));
+        assert_eq!(std::fs::read(source_path).unwrap(), b"new manifest");
+        assert_eq!(
+            std::fs::read(destination_path).unwrap(),
+            b"existing manifest"
+        );
+    }
+}

rust/lancedb/src/table.rs

@@ -2335,10 +2335,19 @@ 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,
+        let has_caller_store_wrapper = params
+            .store_options
+            .as_ref()
+            .and_then(|options| options.object_store_wrapper.as_ref())
+            .is_some();
+        // A caller wrapper must remain outside connection-level compatibility
+        // behavior. When there is no caller wrapper, apply the compatibility
+        // layer after loading so the session's registered store can be reused.
+        let (params, wrapper_after_load) = match write_store_wrapper {
+            Some(wrapper) if has_caller_store_wrapper => {
+                (params.patch_with_store_wrapper(wrapper)?, None)
+            }
+            wrapper => (params, wrapper),
         };
 
         // Build table_id from namespace + name
@@ -2397,6 +2406,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 wrapper_after_load {
+            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);
@@ -2498,11 +2515,16 @@ impl NativeTable {
         if let Some(sess) = session {
             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,
+        let has_caller_store_wrapper = params
+            .store_options
+            .as_ref()
+            .and_then(|options| options.object_store_wrapper.as_ref())
+            .is_some();
+        let (params, wrapper_after_load) = match write_store_wrapper {
+            Some(wrapper) if has_caller_store_wrapper => {
+                (params.patch_with_store_wrapper(wrapper)?, None)
+            }
+            wrapper => (params, wrapper),
         };
 
         // Build table_id from namespace + name
@@ -2526,6 +2548,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 wrapper_after_load {
+            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);
@@ -3660,8 +3689,8 @@ pub struct FragmentSummaryStats {
 #[cfg(test)]
 #[allow(deprecated)]
 mod tests {
-    use std::sync::Arc;
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+    use std::sync::{Arc, Mutex};
     use std::time::Duration;
 
     use arrow_array::{
@@ -4033,6 +4062,66 @@ mod tests {
         }
     }
 
+    #[derive(Debug)]
+    struct OrderedStoreWrapper {
+        name: &'static str,
+        order: Arc<Mutex<Vec<&'static str>>>,
+    }
+
+    impl WrappingObjectStore for OrderedStoreWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            original: Arc<dyn object_store::ObjectStore>,
+        ) -> Arc<dyn object_store::ObjectStore> {
+            self.order.lock().unwrap().push(self.name);
+            original
+        }
+    }
+
+    #[tokio::test]
+    async fn test_open_with_params_keeps_caller_store_wrapper_outermost() {
+        let tmp_dir = tempdir().unwrap();
+        let dataset_path = tmp_dir.path().join("test.lance");
+        let uri = dataset_path.to_str().unwrap();
+        let batch = make_test_batches();
+        let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema());
+        Dataset::write(reader, uri, None).await.unwrap();
+
+        let order = Arc::new(Mutex::new(Vec::new()));
+        let caller_wrapper = Arc::new(OrderedStoreWrapper {
+            name: "caller",
+            order: order.clone(),
+        });
+        let compatibility_wrapper = Arc::new(OrderedStoreWrapper {
+            name: "compatibility",
+            order: order.clone(),
+        });
+        let params = ReadParams {
+            store_options: Some(ObjectStoreParams {
+                object_store_wrapper: Some(caller_wrapper),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+
+        NativeTable::open_with_params(
+            uri,
+            "test",
+            vec![],
+            Some(compatibility_wrapper),
+            Some(params),
+            None,
+            None,
+            HashSet::new(),
+            None,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(*order.lock().unwrap(), vec!["compatibility", "caller"]);
+    }
+
     #[tokio::test]
     async fn test_open_table_options() {
         let tmp_dir = tempdir().unwrap();

rust/lancedb/src/utils/mod.rs

@@ -14,6 +14,7 @@ use lance::arrow::json::JsonDataType;
 use lance::dataset::{ReadParams, WriteParams};
 use lance::index::vector::utils::infer_vector_dim;
 use lance::io::{ObjectStoreParams, WrappingObjectStore};
+use lance_io::object_store::ChainedWrappingObjectStore;
 use std::pin::Pin;
 
 use crate::error::{Error, Result};
@@ -37,13 +38,13 @@ impl PatchStoreParam for Option<ObjectStoreParams> {
         wrapper: Arc<dyn WrappingObjectStore>,
     ) -> Result<Option<ObjectStoreParams>> {
         let mut params = self.unwrap_or_default();
-        if params.object_store_wrapper.is_some() {
-            return Err(Error::Other {
-                message: "can not patch param because object store is already set".into(),
-                source: None,
-            });
-        }
-        params.object_store_wrapper = Some(wrapper);
+        params.object_store_wrapper = Some(match params.object_store_wrapper.take() {
+            // The wrapper being patched in is connection-level compatibility
+            // behavior. Keep it closest to the target store so an existing
+            // caller wrapper remains outermost and can observe every operation.
+            Some(existing) => Arc::new(ChainedWrappingObjectStore::new(vec![wrapper, existing])),
+            None => wrapper,
+        });
 
         Ok(Some(params))
     }
@@ -472,14 +473,60 @@ impl Stream for MaxBatchLengthStream {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Mutex;
+
     use arrow_array::Int32Array;
     use arrow_schema::Field;
     use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
     use futures::{StreamExt, stream};
+    use object_store::{ObjectStore, memory::InMemory};
     use tokio::time::sleep;
 
     use super::*;
 
+    #[derive(Debug)]
+    struct OrderedStoreWrapper {
+        name: &'static str,
+        order: Arc<Mutex<Vec<&'static str>>>,
+    }
+
+    impl WrappingObjectStore for OrderedStoreWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            original: Arc<dyn ObjectStore>,
+        ) -> Arc<dyn ObjectStore> {
+            self.order.lock().unwrap().push(self.name);
+            original
+        }
+    }
+
+    #[test]
+    fn test_patch_store_param_keeps_caller_wrapper_outermost() {
+        let order = Arc::new(Mutex::new(Vec::new()));
+        let params = Some(ObjectStoreParams {
+            object_store_wrapper: Some(Arc::new(OrderedStoreWrapper {
+                name: "caller",
+                order: order.clone(),
+            })),
+            ..Default::default()
+        });
+
+        let params = params
+            .patch_with_store_wrapper(Arc::new(OrderedStoreWrapper {
+                name: "compatibility",
+                order: order.clone(),
+            }))
+            .unwrap()
+            .unwrap();
+        params
+            .object_store_wrapper
+            .unwrap()
+            .wrap("memory", Arc::new(InMemory::new()) as Arc<dyn ObjectStore>);
+
+        assert_eq!(*order.lock().unwrap(), vec!["compatibility", "caller"]);
+    }
+
     #[test]
     fn test_guess_default_column() {
         let schema_no_vector = Schema::new(vec![