fix: preserve Windows local store contracts

rust/lancedb/src/database/listing.rs

@@ -8,8 +8,10 @@ use std::fs::create_dir_all;
 use std::path::Path;
 use std::{collections::HashMap, sync::Arc};
 
-use lance::dataset::refs::Ref;
-use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
+use lance::dataset::transaction::{Operation, Transaction};
+use lance::dataset::{
+    CommitBuilder, ReadParams, WriteDestination, WriteMode, builder::DatasetBuilder,
+};
 use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_encoding::version::LanceFileVersion;
@@ -23,9 +25,7 @@ use crate::connection::ConnectRequest;
 use crate::database::ReadConsistency;
 use crate::database::namespace::LanceNamespaceDatabase;
 use crate::error::{CreateDirSnafu, Error, Result};
-use crate::io::object_store::MirroringObjectStoreWrapper;
-#[cfg(any(windows, test))]
-use crate::io::object_store::register_windows_file_store;
+use crate::io::object_store::{MirroringObjectStoreWrapper, new_default_session};
 use crate::table::NativeTable;
 use crate::utils::validate_table_name;
 
@@ -47,6 +47,15 @@ pub const OPT_NEW_TABLE_STORAGE_VERSION: &str = "new_table_data_storage_version"
 pub const OPT_NEW_TABLE_V2_MANIFEST_PATHS: &str = "new_table_enable_v2_manifest_paths";
 pub const OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS: &str = "new_table_enable_stable_row_ids";
 
+fn session_or_default(
+    session: Option<Arc<lance::session::Session>>,
+) -> (Arc<lance::session::Session>, bool) {
+    match session {
+        Some(session) => (session, false),
+        None => (new_default_session(), true),
+    }
+}
+
 /// Controls how new tables should be created
 #[derive(Clone, Debug, Default)]
 pub struct NewTableConfig {
@@ -361,19 +370,20 @@ impl ListingDatabase {
         uri: &str,
         storage_options: &HashMap<String, String>,
         session: Arc<lance::session::Session>,
+        prepare_native_directory: bool,
     ) -> Result<String> {
         match url::Url::parse(uri) {
             Ok(url) if url.scheme().len() == 1 && cfg!(windows) => {
-                Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
-                #[cfg(windows)]
-                register_windows_file_store(&session.store_registry());
+                if prepare_native_directory {
+                    Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
+                }
                 let (object_store, _) = ObjectStore::from_uri_and_params(
                     session.store_registry(),
                     uri,
                     &ObjectStoreParams::default(),
                 )
                 .await?;
-                if object_store.is_local() {
+                if object_store.is_local() && prepare_native_directory {
                     Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
                 }
                 Ok(uri.to_string())
@@ -405,11 +415,10 @@ impl ListingDatabase {
                 let plain_uri = url.to_string();
 
                 #[cfg(windows)]
-                if url.scheme() == "file" {
+                if url.scheme() == "file" && prepare_native_directory {
                     Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
                         path: plain_uri.clone(),
                     })?;
-                    register_windows_file_store(&session.store_registry());
                 }
 
                 let os_params = ObjectStoreParams {
@@ -428,7 +437,7 @@ impl ListingDatabase {
                     &os_params,
                 )
                 .await?;
-                if object_store.is_local() {
+                if object_store.is_local() && prepare_native_directory {
                     Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
                         path: plain_uri.clone(),
                     })?;
@@ -437,16 +446,16 @@ impl ListingDatabase {
                 Ok(plain_uri)
             }
             Err(_) => {
-                Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
-                #[cfg(windows)]
-                register_windows_file_store(&session.store_registry());
+                if prepare_native_directory {
+                    Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
+                }
                 let (object_store, _) = ObjectStore::from_uri_and_params(
                     session.store_registry(),
                     uri,
                     &ObjectStoreParams::default(),
                 )
                 .await?;
-                if object_store.is_local() {
+                if object_store.is_local() && prepare_native_directory {
                     Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
                 }
                 Ok(uri.to_string())
@@ -458,13 +467,14 @@ impl ListingDatabase {
         request: &ConnectRequest,
     ) -> Result<LanceNamespaceDatabase> {
         let options = ListingDatabaseOptions::parse_from_map(&request.options)?;
-        let session = request
-            .session
-            .clone()
-            .unwrap_or_else(|| Arc::new(lance::session::Session::default()));
-        let namespace_root =
-            Self::prepare_namespace_root(&request.uri, &options.storage_options, session.clone())
-                .await?;
+        let (session, owns_session) = session_or_default(request.session.clone());
+        let namespace_root = Self::prepare_namespace_root(
+            &request.uri,
+            &options.storage_options,
+            session.clone(),
+            owns_session || !cfg!(windows),
+        )
+        .await?;
         let ns_properties = Self::build_manifest_enabled_namespace_client_properties(
             &namespace_root,
             &options.storage_options,
@@ -563,16 +573,13 @@ impl ListingDatabase {
                     url.to_string()
                 };
 
-                let session = request
-                    .session
-                    .clone()
-                    .unwrap_or_else(|| Arc::new(lance::session::Session::default()));
+                let (session, owns_session) = session_or_default(request.session.clone());
+                let prepare_native_directory = owns_session || !cfg!(windows);
                 #[cfg(windows)]
-                if url.scheme() == "file" {
+                if url.scheme() == "file" && prepare_native_directory {
                     Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
                         path: storage_base_uri.clone(),
                     })?;
-                    register_windows_file_store(&session.store_registry());
                 }
                 let os_params = ObjectStoreParams {
                     storage_options_accessor: if options.storage_options.is_empty() {
@@ -590,7 +597,7 @@ impl ListingDatabase {
                     &os_params,
                 )
                 .await?;
-                if object_store.is_local() {
+                if object_store.is_local() && prepare_native_directory {
                     Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
                         path: storage_base_uri.clone(),
                     })?;
@@ -648,17 +655,18 @@ impl ListingDatabase {
         namespace_client_properties: HashMap<String, String>,
         session: Option<Arc<lance::session::Session>>,
     ) -> Result<Self> {
-        let session = session.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
-        Self::try_create_dir(path).context(CreateDirSnafu { path })?;
-        #[cfg(windows)]
-        register_windows_file_store(&session.store_registry());
+        let (session, owns_session) = session_or_default(session);
+        let prepare_native_directory = owns_session || !cfg!(windows);
+        if prepare_native_directory {
+            Self::try_create_dir(path).context(CreateDirSnafu { path })?;
+        }
         let (object_store, base_path) = ObjectStore::from_uri_and_params(
             session.store_registry(),
             path,
             &ObjectStoreParams::default(),
         )
         .await?;
-        if object_store.is_local() {
+        if object_store.is_local() && prepare_native_directory {
             Self::try_create_dir(path).context(CreateDirSnafu { path })?;
         }
 
@@ -704,23 +712,6 @@ impl ListingDatabase {
         Ok(())
     }
 
-    #[cfg(any(windows, test))]
-    fn try_remove_dir_all(path: &str) -> core::result::Result<(), std::io::Error> {
-        let filesystem_path = match url::Url::parse(path) {
-            Ok(mut url) if url.scheme() == "file" => {
-                url.set_query(None);
-                url.to_file_path().map_err(|_| {
-                    std::io::Error::new(
-                        std::io::ErrorKind::InvalidInput,
-                        format!("Unable to convert URL '{url}' to a local path"),
-                    )
-                })?
-            }
-            _ => Path::new(path).to_path_buf(),
-        };
-        std::fs::remove_dir_all(filesystem_path)
-    }
-
     /// Get the URI of a table in the database.
     fn table_uri(&self, name: &str) -> Result<String> {
         validate_table_name(name)?;
@@ -791,28 +782,6 @@ impl ListingDatabase {
                     },
                     _ => Error::from(err),
                 })?;
-
-            // The prefixed Windows store deliberately uses a custom scheme so
-            // Lance never follows object deletion with cwd-relative native
-            // cleanup. Remove the now-empty directory through its full file URI.
-            #[cfg(any(windows, test))]
-            if self.object_store.scheme() == "lancedb-file" {
-                let table_uri = self.table_uri(&name)?;
-                Self::try_remove_dir_all(&table_uri).map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        Error::TableNotFound {
-                            name: name.clone(),
-                            source: Box::new(error),
-                        }
-                    } else {
-                        Error::Runtime {
-                            message: format!(
-                                "Failed to remove table directory '{table_uri}': {error}"
-                            ),
-                        }
-                    }
-                })?;
-            }
         }
         Ok(())
     }
@@ -1179,24 +1148,60 @@ impl Database for ListingDatabase {
             ..Default::default()
         };
 
-        let mut source_dataset = DatasetBuilder::from_uri(&request.source_uri)
+        let source_dataset = DatasetBuilder::from_uri(&request.source_uri)
             .with_read_params(read_params.clone())
             .load()
             .await
             .map_err(|e| -> Error { e.into() })?;
 
-        let version_ref = match (request.source_version, request.source_tag) {
-            (Some(v), None) => Ok(Ref::Version(None, Some(v))),
-            (None, Some(tag)) => Ok(Ref::Tag(tag)),
-            (None, None) => Ok(Ref::Version(None, Some(source_dataset.version().version))),
+        let (ref_name, version_number) = match (request.source_version, request.source_tag) {
+            (Some(version), None) => Ok((None, version)),
+            (None, Some(tag)) => {
+                let tag = source_dataset.tags().get(&tag).await?;
+                Ok((tag.branch, tag.version))
+            }
+            (None, None) => Ok((None, source_dataset.version().version)),
             _ => Err(Error::InvalidInput {
                 message: "Cannot specify both source_version and source_tag".to_string(),
             }),
         }?;
 
         let target_uri = self.table_uri(&request.target_table_name)?;
-        source_dataset
-            .shallow_clone(&target_uri, version_ref, Some(storage_params))
+        let source_location = source_dataset
+            .branch_location()
+            .find_branch(ref_name.as_deref())?;
+        let (source_store, _) = ObjectStore::from_uri_and_params(
+            self.session.store_registry(),
+            &source_location.uri,
+            &storage_params,
+        )
+        .await?;
+        let (target_store, _) = ObjectStore::from_uri_and_params(
+            self.session.store_registry(),
+            &target_uri,
+            &storage_params,
+        )
+        .await?;
+        let clone_op = Operation::Clone {
+            is_shallow: true,
+            ref_name,
+            ref_version: version_number,
+            ref_path: source_location.uri,
+            branch_name: None,
+        };
+        let transaction = Transaction::new(version_number, clone_op, None);
+        CommitBuilder::new(WriteDestination::Uri(&target_uri))
+            .with_store_params(storage_params)
+            .with_object_store(target_store)
+            .with_source_store(source_store)
+            .with_session(self.session.clone())
+            .with_storage_format(
+                source_dataset
+                    .manifest
+                    .data_storage_format
+                    .lance_file_version()?,
+            )
+            .execute(transaction)
             .await
             .map_err(|e| -> Error { e.into() })?;
 
@@ -1205,7 +1210,7 @@ impl Database for ListingDatabase {
             &request.target_table_name,
             request.target_namespace_path,
             self.store_wrapper.clone(),
-            None,
+            Some(read_params),
             self.read_consistency_interval,
             request.namespace_client,
             HashSet::new(), // listing database doesn't support server-side queries
@@ -1383,12 +1388,27 @@ mod tests {
         (tempdir, db)
     }
 
+    #[test]
+    fn preserves_caller_owned_session_provider() {
+        let registry = Arc::new(lance_io::object_store::ObjectStoreRegistry::default());
+        let provider = registry.get_provider("file").unwrap();
+        let session = Arc::new(lance::session::Session::new(16, 16, registry.clone()));
+
+        let (selected, owns_session) = session_or_default(Some(session.clone()));
+
+        assert!(!owns_session);
+        assert!(Arc::ptr_eq(&selected, &session));
+        assert!(Arc::ptr_eq(
+            &registry.get_provider("file").unwrap(),
+            &provider
+        ));
+    }
+
     #[tokio::test]
     async fn test_listing_database_with_prefixed_file_store() {
         let tempdir = tempdir().unwrap();
         let uri = tempdir.path().to_str().unwrap();
-        let session = Arc::new(lance::session::Session::default());
-        register_windows_file_store(&session.store_registry());
+        let session = crate::io::object_store::new_prefixed_file_session();
 
         let request = ConnectRequest {
             uri: uri.to_string(),
@@ -1685,6 +1705,65 @@ mod tests {
         assert_eq!(source_count, 3);
     }
 
+    #[tokio::test]
+    async fn test_clone_table_across_databases_uses_target_store() {
+        let source_dir = tempdir().unwrap();
+        let target_dir = tempdir().unwrap();
+        let source_request = ConnectRequest {
+            uri: source_dir.path().to_string_lossy().into_owned(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: None,
+        };
+        let target_request = ConnectRequest {
+            uri: target_dir.path().to_string_lossy().into_owned(),
+            ..source_request.clone()
+        };
+        let source_db = ListingDatabase::connect_with_options(&source_request)
+            .await
+            .unwrap();
+        let target_db = ListingDatabase::connect_with_options(&target_request)
+            .await
+            .unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        source_db
+            .create_table(CreateTableRequest {
+                name: "source".to_string(),
+                namespace_path: vec![],
+                data: Box::new(
+                    RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2, 3]))])
+                        .unwrap(),
+                ),
+                mode: CreateTableMode::Create,
+                write_options: Default::default(),
+                location: None,
+                namespace_client: None,
+            })
+            .await
+            .unwrap();
+
+        let target = target_db
+            .clone_table(CloneTableRequest {
+                target_table_name: "target".to_string(),
+                target_namespace_path: vec![],
+                source_uri: source_db.table_uri("source").unwrap(),
+                source_version: None,
+                source_tag: None,
+                is_shallow: true,
+                namespace_client: None,
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(target.count_rows(None).await.unwrap(), 3);
+        assert!(target_dir.path().join("target.lance").exists());
+        assert!(!source_dir.path().join("target.lance").exists());
+    }
+
     #[tokio::test]
     async fn test_clone_table_with_storage_options() {
         let tempdir = tempdir().unwrap();

rust/lancedb/src/io/object_store.rs

@@ -38,9 +38,9 @@ pub mod io_tracking;
 /// UNC-share root keeps the UNC authority in the filesystem prefix and exposes
 /// only paths relative to that prefix to `object_store`.
 ///
-/// The returned Lance store deliberately uses the `lancedb-file` scheme.
-/// The regular `file` scheme enables optimized readers and writers that bypass
-/// the configured object store and would reintroduce the broken UNC conversion.
+/// Extracted paths retain the native drive or UNC-share root as a structural
+/// first component. This keeps Lance's local classification and optimized I/O
+/// safe without recovering absolute path provenance from ambiguous path text.
 #[cfg(any(windows, test))]
 #[derive(Debug)]
 struct PrefixedFileStoreProvider;
@@ -95,18 +95,21 @@ impl PrefixedFileStoreProvider {
         Ok((root, Path::parse(relative)?))
     }
 
-    /// Select the immutable path scope shared by a database and its tables.
+    /// Preserve the native filesystem root as a structural path component.
     ///
-    /// LanceDB table URIs end in `.lance`. Their parent is the listing database
-    /// path, so using it as the store scope lets opens and shallow clones reuse
-    /// the connection store while databases elsewhere on the same drive or UNC
-    /// share retain distinct cache identities.
-    fn store_scope_path(path: &Path) -> Path {
-        if path.extension() == Some("lance") {
-            path.parent().unwrap_or_default()
-        } else {
-            path.clone()
+    /// `object_store::Path::from_absolute_path` converts a UNC path to its URL
+    /// path and loses the server. Keeping `C:` or `\\server\share` as the first
+    /// component distinguishes an absolute path from every relative path while
+    /// remaining directly usable by Windows filesystem APIs.
+    fn rooted_path(root: &std::path::Path, relative: &Path) -> LanceResult<Path> {
+        let root = root.to_string_lossy();
+        let root = root.trim_end_matches(['/', '\\']);
+        if root.is_empty() {
+            return Ok(relative.clone());
         }
+        Ok(relative
+            .parts()
+            .fold(Path::parse(root)?, |path, part| path.join(part)))
     }
 }
 
@@ -123,18 +126,16 @@ impl PrefixedFileStoreProvider {
 struct RootedLocalFileSystem {
     inner: Arc<LocalFileSystem>,
     root: std::path::PathBuf,
-    base_path: Path,
     absolute_alias: Path,
 }
 
 #[cfg(any(windows, test))]
 impl RootedLocalFileSystem {
-    fn new(root: std::path::PathBuf, base_path: Path) -> LanceResult<Self> {
-        let absolute_alias = Path::from_absolute_path(&root)?;
+    fn new(root: std::path::PathBuf) -> LanceResult<Self> {
+        let absolute_alias = PrefixedFileStoreProvider::rooted_path(&root, &Path::default())?;
         Ok(Self {
             inner: Arc::new(LocalFileSystem::new_with_prefix(&root)?),
             root,
-            base_path,
             absolute_alias,
         })
     }
@@ -150,17 +151,7 @@ impl RootedLocalFileSystem {
         let Some(suffix) = path.prefix_match(&self.absolute_alias) else {
             return path.clone();
         };
-        let suffix = Self::path_from_parts(suffix);
-
-        // A UNC absolute path loses its server when converted to an object-store
-        // path, leaving `share/<path>`. Each store is cached for exactly one base
-        // path, so only strip that ambiguous share segment when the remainder is
-        // inside this store's immutable base.
-        if self.base_path.as_ref().is_empty() || suffix.prefix_matches(&self.base_path) {
-            suffix
-        } else {
-            path.clone()
-        }
+        Self::path_from_parts(suffix)
     }
 
     fn restore_prefix(&self, path: Path, requested: &Path, normalized: &Path) -> Path {
@@ -276,11 +267,17 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider {
         base_path: Url,
         params: &ObjectStoreParams,
     ) -> LanceResult<lance::io::ObjectStore> {
-        let (root, relative_path) = Self::root_and_relative_path(&base_path)?;
-        let store_scope = Self::store_scope_path(&relative_path);
-        let raw_store: Arc<dyn ObjectStore> =
-            Arc::new(RootedLocalFileSystem::new(root, store_scope)?);
-        let location = Url::parse("lancedb-file:///").expect("static URL must be valid");
+        let (root, _) = Self::root_and_relative_path(&base_path)?;
+        let raw_store: Arc<dyn ObjectStore> = Arc::new(RootedLocalFileSystem::new(root)?);
+        // `file` keeps local planning and manifest behavior. With a custom
+        // wrapper, `file+uring` is also classified as local but (on Windows)
+        // routes readers and writers through the wrapped object store instead
+        // of the native fast path.
+        let location = if params.object_store_wrapper.is_some() {
+            Url::parse("file+uring:///").expect("static URL must be valid")
+        } else {
+            Url::parse("file:///").expect("static URL must be valid")
+        };
         let storage_options =
             StorageOptions::new(params.storage_options().cloned().unwrap_or_default());
 
@@ -305,7 +302,8 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider {
     }
 
     fn extract_path(&self, url: &Url) -> LanceResult<Path> {
-        Self::root_and_relative_path(url).map(|(_, path)| path)
+        let (root, relative) = Self::root_and_relative_path(url)?;
+        Self::rooted_path(&root, &relative)
     }
 
     fn calculate_object_store_prefix(
@@ -313,23 +311,34 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider {
         url: &Url,
         _storage_options: Option<&std::collections::HashMap<String, String>>,
     ) -> LanceResult<String> {
-        let (root, path) = Self::root_and_relative_path(url)?;
+        let (root, _) = Self::root_and_relative_path(url)?;
         let root = root.canonicalize()?;
-        let store_scope = Self::store_scope_path(&path);
-        // Scope the registry cache to one database path. A root-wide store
-        // cannot distinguish a legitimate relative path beginning with a UNC
-        // share name from the absolute alias produced by the default file
-        // provider, while a table-wide store cannot service sibling targets
-        // during shallow clone.
-        Ok(format!("file${}${store_scope}", root.display()))
+        // One store per drive or UNC share keeps registry and metrics
+        // cardinality bounded. Absolute-vs-relative provenance is carried by
+        // the extracted path instead of the cache key.
+        Ok(format!("file${}", root.display()))
     }
 }
 
-/// Replace Lance's default Windows file provider with one that preserves UNC
-/// roots by using `LocalFileSystem::new_with_prefix`.
-#[cfg(any(windows, test))]
-pub(crate) fn register_windows_file_store(registry: &Arc<ObjectStoreRegistry>) {
-    registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+/// Build LanceDB's default session with the Windows file fallback installed.
+///
+/// Callers that provide a Session retain its registry unchanged.
+pub(crate) fn new_default_session() -> Arc<lance::session::Session> {
+    let session = Arc::new(lance::session::Session::default());
+    #[cfg(windows)]
+    session
+        .store_registry()
+        .insert("file", Arc::new(PrefixedFileStoreProvider));
+    session
+}
+
+#[cfg(test)]
+pub(crate) fn new_prefixed_file_session() -> Arc<lance::session::Session> {
+    let session = Arc::new(lance::session::Session::default());
+    session
+        .store_registry()
+        .insert("file", Arc::new(PrefixedFileStoreProvider));
+    session
 }
 
 #[cfg(test)]
@@ -364,8 +373,8 @@ mod prefixed_file_store_test {
         let registry = Arc::new(ObjectStoreRegistry::default());
         registry.insert("file", Arc::new(PrefixedFileStoreProvider));
 
-        // A new table is relative to its filesystem root. The non-`file`
-        // scheme proves Lance will not bypass this prefixed store.
+        // The extracted path remains absolute for Lance's local fast paths,
+        // while the inner store strips the structural root before delegation.
         let (store, base_path) = lance::io::ObjectStore::from_uri_and_params(
             registry.clone(),
             table_url.as_str(),
@@ -373,7 +382,9 @@ mod prefixed_file_store_test {
         )
         .await
         .unwrap();
-        assert_eq!(store.scheme(), "lancedb-file");
+        assert_eq!(store.scheme(), "file");
+        assert!(store.is_local());
+        assert!(!store.is_cloud());
         assert_eq!(store.block_size(), 4 * 1024);
         assert_eq!(store.io_parallelism(), DEFAULT_LOCAL_IO_PARALLELISM);
         assert_eq!(base_path.filename(), Some("test.lance"));
@@ -426,6 +437,9 @@ mod prefixed_file_store_test {
                 .await
                 .unwrap();
         assert_eq!(wrapper.calls.load(Ordering::Relaxed), 1);
+        assert_eq!(store.scheme(), "file+uring");
+        assert!(store.is_local());
+        assert!(!store.is_cloud());
 
         store
             .inner
@@ -441,7 +455,7 @@ mod prefixed_file_store_test {
     }
 
     #[tokio::test]
-    async fn scopes_store_cache_to_requested_base_path() {
+    async fn reuses_store_cache_across_database_paths() {
         let tempdir = tempfile::tempdir().unwrap();
         let first_url = Url::from_directory_path(tempdir.path().join("database")).unwrap();
         let second_url = Url::from_directory_path(tempdir.path().join("share/database")).unwrap();
@@ -463,8 +477,9 @@ mod prefixed_file_store_test {
         .await
         .unwrap();
 
-        assert!(!Arc::ptr_eq(&first, &second));
-        assert_eq!(registry.stats().misses, 2);
+        assert!(Arc::ptr_eq(&first, &second));
+        assert_eq!(registry.stats().misses, 1);
+        assert_eq!(registry.stats().hits, 1);
     }
 
     #[tokio::test]
@@ -509,21 +524,17 @@ mod prefixed_file_store_test {
     #[test]
     fn normalizes_absolute_drive_and_unc_aliases() {
         let tempdir = tempfile::tempdir().unwrap();
-        let mut drive_store =
-            RootedLocalFileSystem::new(tempdir.path().to_path_buf(), Path::from("Users/db"))
-                .unwrap();
+        let mut drive_store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
         drive_store.absolute_alias = Path::from("C:");
         assert_eq!(
             drive_store.normalize(&Path::from("C:/Users/db/table.lance")),
             Path::from("Users/db/table.lance")
         );
 
-        let mut unc_store =
-            RootedLocalFileSystem::new(tempdir.path().to_path_buf(), Path::from("share/db"))
-                .unwrap();
-        unc_store.absolute_alias = Path::from("share");
+        let mut unc_store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
+        unc_store.absolute_alias = Path::parse(r"\\server\share").unwrap();
         assert_eq!(
-            unc_store.normalize(&Path::from("share/share/db/table.lance")),
+            unc_store.normalize(&Path::parse(r"\\server\share/share/db/table.lance").unwrap()),
             Path::from("share/db/table.lance")
         );
         assert_eq!(
@@ -535,15 +546,15 @@ mod prefixed_file_store_test {
     #[test]
     fn relative_unc_alias_does_not_cross_database_roots() {
         let tempdir = tempfile::tempdir().unwrap();
-        let mut store =
-            RootedLocalFileSystem::new(tempdir.path().to_path_buf(), Path::from("share/database"))
-                .unwrap();
-        store.absolute_alias = Path::from("share");
+        let mut store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
+        store.absolute_alias = Path::parse(r"\\server\share").unwrap();
 
         let relative = Path::from("share/database/table.lance/marker");
         assert_eq!(store.normalize(&relative), relative);
         assert_eq!(
-            store.normalize(&Path::from("share/share/database/table.lance/marker")),
+            store.normalize(
+                &Path::parse(r"\\server\share/share/database/table.lance/marker").unwrap()
+            ),
             relative
         );
     }
@@ -551,11 +562,9 @@ mod prefixed_file_store_test {
     #[tokio::test]
     async fn routes_unc_alias_lifecycle_through_the_prefix() {
         let tempdir = tempfile::tempdir().unwrap();
-        let mut store =
-            RootedLocalFileSystem::new(tempdir.path().to_path_buf(), Path::from("share/db"))
-                .unwrap();
-        store.absolute_alias = Path::from("share");
-        let table = Path::from("share/share/db/test.lance");
+        let mut store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
+        store.absolute_alias = Path::parse(r"\\server\share").unwrap();
+        let table = Path::parse(r"\\server\share/share/db/test.lance").unwrap();
         let marker = table.clone().join("marker");
 
         store
@@ -603,6 +612,26 @@ mod prefixed_file_store_test {
         assert_eq!(root, std::path::PathBuf::from(r"\\server\share\"));
         assert_eq!(path, Path::from("database"));
     }
+
+    #[test]
+    fn preserves_drive_and_unc_roots_in_extracted_paths() {
+        assert_eq!(
+            PrefixedFileStoreProvider::rooted_path(
+                std::path::Path::new(r"C:\"),
+                &Path::from("database/table.lance")
+            )
+            .unwrap(),
+            Path::from("C:/database/table.lance")
+        );
+        assert_eq!(
+            PrefixedFileStoreProvider::rooted_path(
+                std::path::Path::new(r"\\server\share\"),
+                &Path::from("database/table.lance")
+            )
+            .unwrap(),
+            Path::parse(r"\\server\share/database/table.lance").unwrap()
+        );
+    }
 }
 
 #[derive(Debug)]