fix: forward commit outcome capabilities

# Conflicts:
#	rust/lancedb/src/database/listing.rs

python/python/lancedb/db.py

@@ -716,20 +716,9 @@ class LanceDBConnection(DBConnection):
         if not isinstance(uri, Path):
             scheme = get_uri_scheme(uri)
         is_local = isinstance(uri, Path) or scheme == "file"
-        if is_local:
+        is_file_uri = isinstance(uri, str) and uri.lower().startswith("file:")
+        if is_local and not is_file_uri:
             if isinstance(uri, str):
-                # Strip file:// or file:/ scheme if present
-                # file:///path becomes file:/path after URL normalization
-                if uri.startswith("file://"):
-                    uri = uri[7:]  # Remove "file://"
-                elif uri.startswith("file:/"):
-                    uri = uri[5:]  # Remove "file:"
-
-                if sys.platform == "win32":
-                    # On Windows, a path like /C:/path should become C:/path
-                    if len(uri) >= 3 and uri[0] == "/" and uri[2] == ":":
-                        uri = uri[1:]
-
                 uri = Path(uri)
             uri = uri.expanduser().absolute()
             Path(uri).mkdir(parents=True, exist_ok=True)

python/python/lancedb/namespace.py

@@ -919,17 +919,7 @@ class LanceNamespaceDBConnection(DBConnection):
             The namespace client for this connection.
         """
         if self._namespace_client is None:
-            if (
-                self._namespace_client_impl is None
-                or self._namespace_client_properties is None
-            ):
-                raise ValueError(
-                    "Cannot construct a Python namespace client without "
-                    "namespace implementation properties"
-                )
-            self._namespace_client = namespace_connect(
-                self._namespace_client_impl, self._namespace_client_properties
-            )
+            self._namespace_client = LOOP.run(self._inner.namespace_client())
         return self._namespace_client
 
 
@@ -1370,17 +1360,7 @@ class AsyncLanceNamespaceDBConnection:
             The namespace client for this connection.
         """
         if self._namespace_client is None:
-            if (
-                self._namespace_client_impl is None
-                or self._namespace_client_properties is None
-            ):
-                raise ValueError(
-                    "Cannot construct a Python namespace client without "
-                    "namespace implementation properties"
-                )
-            self._namespace_client = namespace_connect(
-                self._namespace_client_impl, self._namespace_client_properties
-            )
+            self._namespace_client = await self._inner.namespace_client()
         return self._namespace_client
 
 

python/python/tests/test_db.py

@@ -89,6 +89,32 @@ def test_sync_debugger_inspection_does_not_use_background_loop(tmp_path, monkeyp
     assert repr(table) == f"LanceTable(name='test', _conn={db!r})"
 
 
+def test_connect_preserves_file_uri_authority(monkeypatch):
+    uri = "file://server/share/database"
+    received = []
+
+    async def fake_connect(passed_uri, *_args):
+        received.append(passed_uri)
+        return SimpleNamespace(uri=passed_uri)
+
+    monkeypatch.setattr("lancedb.db.lancedb_connect", fake_connect)
+    db = lancedb.connect(uri)
+
+    assert received == [uri]
+    assert db.uri == uri
+
+
+def test_connect_file_uri_lifecycle(tmp_path):
+    uri = (tmp_path / "sync").as_uri()
+    db = lancedb.connect(uri)
+
+    db.create_table("test", data=[{"id": 1}])
+    assert db.table_names() == ["test"]
+    assert db.open_table("test").count_rows() == 1
+    db.drop_table("test")
+    assert db.table_names() == []
+
+
 def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch):
     from lancedb.background_loop import LOOP
     from lancedb.db import LanceDBConnection
@@ -405,6 +431,35 @@ async def test_connect(tmp_path):
     assert str(db) == f"ListingDatabase(uri={tmp_path}, read_consistency_interval=5s)"
 
 
+@pytest.mark.asyncio
+async def test_connect_async_preserves_file_uri_authority(monkeypatch):
+    uri = "file://server/share/database"
+    received = []
+
+    async def fake_connect(passed_uri, *_args):
+        received.append(passed_uri)
+        return SimpleNamespace(uri=passed_uri)
+
+    monkeypatch.setattr(lancedb, "lancedb_connect", fake_connect)
+    db = await lancedb.connect_async(uri)
+
+    assert received == [uri]
+    assert db.uri == uri
+
+
+@pytest.mark.asyncio
+async def test_connect_async_file_uri_lifecycle(tmp_path):
+    uri = (tmp_path / "async").as_uri()
+    db = await lancedb.connect_async(uri)
+
+    await db.create_table("test", data=[{"id": 1}])
+    assert await db.table_names() == ["test"]
+    table = await db.open_table("test")
+    assert await table.count_rows() == 1
+    await db.drop_table("test")
+    assert await db.table_names() == []
+
+
 @pytest.mark.asyncio
 async def test_close(mem_db_async: lancedb.AsyncConnection):
     assert mem_db_async.is_open()
@@ -1180,6 +1235,40 @@ def test_clone_table_deep_clone_fails(tmp_path):
         db.clone_table("cloned", source_uri, is_shallow=False)
 
 
+class _UnsupportedNamespaceConfig:
+    async def namespace_client_config(self):
+        raise RuntimeError("UNC namespace client export is not supported")
+
+
+def test_sync_namespace_client_propagates_export_guard(monkeypatch):
+    from lancedb.db import AsyncConnection, LanceDBConnection
+
+    monkeypatch.setattr(
+        "lancedb.db.namespace_connect",
+        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+    )
+    db = LanceDBConnection.__new__(LanceDBConnection)
+    db._conn = AsyncConnection(_UnsupportedNamespaceConfig())
+    db._cached_namespace_client = None
+
+    with pytest.raises(RuntimeError, match="UNC namespace client export"):
+        db.namespace_client()
+
+
+@pytest.mark.asyncio
+async def test_async_namespace_client_propagates_export_guard(monkeypatch):
+    from lancedb.db import AsyncConnection
+
+    monkeypatch.setattr(
+        "lancedb.db.namespace_connect",
+        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+    )
+    db = AsyncConnection(_UnsupportedNamespaceConfig())
+
+    with pytest.raises(RuntimeError, match="UNC namespace client export"):
+        await db.namespace_client()
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="Namespace client issues")
 def test_namespace_client_native_storage(tmp_path):
     """Test namespace_client() returns DirectoryNamespace for native storage."""

python/python/tests/test_namespace.py

@@ -60,6 +60,11 @@ class _NamespaceClient:
         return _ipc_file()
 
 
+class _UnsupportedNamespaceConfig:
+    async def namespace_client_config(self):
+        raise RuntimeError("UNC namespace client export is not supported")
+
+
 def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
     table = LanceTable.__new__(LanceTable)
     table._table = _FailingSyncInner()
@@ -138,6 +143,24 @@ class TestNamespaceConnection:
         db.drop_namespace(["test_ns"])
         assert "test_ns" not in db.list_namespaces().namespaces
 
+    def test_sync_namespace_client_propagates_export_guard(self, monkeypatch):
+        from lancedb.db import AsyncConnection
+
+        monkeypatch.setattr(
+            "lancedb.namespace.namespace_connect",
+            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+        )
+        db = lancedb.LanceNamespaceDBConnection.__new__(
+            lancedb.LanceNamespaceDBConnection
+        )
+        db._namespace_client = None
+        db._namespace_client_impl = "dir"
+        db._namespace_client_properties = {"root": "file://server/share/database"}
+        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
+
+        with pytest.raises(RuntimeError, match="UNC namespace client export"):
+            db.namespace_client()
+
     def test_create_table_through_namespace(self):
         """Test creating a table through namespace."""
         db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
@@ -639,6 +662,24 @@ class TestAsyncNamespaceConnection:
         await db.drop_namespace(["test_ns"])
         assert "test_ns" not in (await db.list_namespaces()).namespaces
 
+    async def test_async_namespace_client_propagates_export_guard(self, monkeypatch):
+        from lancedb.db import AsyncConnection
+
+        monkeypatch.setattr(
+            "lancedb.namespace.namespace_connect",
+            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+        )
+        db = lancedb.AsyncLanceNamespaceDBConnection.__new__(
+            lancedb.AsyncLanceNamespaceDBConnection
+        )
+        db._namespace_client = None
+        db._namespace_client_impl = "dir"
+        db._namespace_client_properties = {"root": "file://server/share/database"}
+        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
+
+        with pytest.raises(RuntimeError, match="UNC namespace client export"):
+            await db.namespace_client()
+
     async def test_async_namespace_client_is_lazy(self):
         """namespace_client() should still return the backing client on demand."""
         pytest.importorskip("lance")

rust/lancedb/src/database/listing.rs

@@ -8,14 +8,23 @@ 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_file::version::LanceFileVersion;
 use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
-use lance_table::io::commit::commit_handler_from_url;
+use lance_table::{
+    format::{IndexMetadata, Manifest, Transaction as LanceTableTransaction},
+    io::commit::{
+        CommitError, CommitHandler, ManifestLocation, ManifestNamingScheme, ManifestWriter,
+        commit_handler_from_url,
+    },
+};
 use object_store::local::LocalFileSystem;
+use object_store::path::Path as ObjectPath;
 use snafu::ResultExt;
 
 use crate::blob::{ensure_blob_storage_version, has_blob_columns};
@@ -23,7 +32,9 @@ 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(windows)]
+use crate::io::object_store::rooted_file_commit_handler;
+use crate::io::object_store::{MirroringObjectStoreWrapper, new_default_session};
 use crate::table::NativeTable;
 use crate::utils::validate_table_name;
 
@@ -45,6 +56,101 @@ 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),
+    }
+}
+
+fn clone_target_read_params(
+    store_options: ObjectStoreParams,
+    session: Arc<lance::session::Session>,
+    commit_handler: Arc<dyn CommitHandler>,
+) -> ReadParams {
+    ReadParams {
+        store_options: Some(store_options),
+        session: Some(session),
+        commit_handler: Some(commit_handler),
+        ..Default::default()
+    }
+}
+
+/// Routes clone source metadata through the source commit handler while all
+/// destination operations continue to use the target handler.
+#[derive(Debug)]
+struct CloneCommitHandler {
+    source_base: ObjectPath,
+    source: Arc<dyn CommitHandler>,
+    target: Arc<dyn CommitHandler>,
+}
+
+#[async_trait::async_trait]
+impl CommitHandler for CloneCommitHandler {
+    fn is_version_not_found_definitive(&self) -> bool {
+        self.target.is_version_not_found_definitive()
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        self.target.propagate_commit_error_after_success()
+    }
+
+    async fn resolve_latest_location(
+        &self,
+        base_path: &ObjectPath,
+        object_store: &ObjectStore,
+    ) -> lance_core::Result<ManifestLocation> {
+        self.target
+            .resolve_latest_location(base_path, object_store)
+            .await
+    }
+
+    async fn resolve_version_location(
+        &self,
+        base_path: &ObjectPath,
+        version: u64,
+        object_store: &dyn object_store::ObjectStore,
+    ) -> lance_core::Result<ManifestLocation> {
+        let handler = if base_path == &self.source_base {
+            &self.source
+        } else {
+            &self.target
+        };
+        handler
+            .resolve_version_location(base_path, version, object_store)
+            .await
+    }
+
+    async fn commit(
+        &self,
+        manifest: &mut Manifest,
+        indices: Option<Vec<IndexMetadata>>,
+        base_path: &ObjectPath,
+        object_store: &ObjectStore,
+        manifest_writer: ManifestWriter,
+        naming_scheme: ManifestNamingScheme,
+        transaction: Option<LanceTableTransaction>,
+    ) -> std::result::Result<ManifestLocation, CommitError> {
+        self.target
+            .commit(
+                manifest,
+                indices,
+                base_path,
+                object_store,
+                manifest_writer,
+                naming_scheme,
+                transaction,
+            )
+            .await
+    }
+
+    async fn delete(&self, base_path: &ObjectPath) -> lance_core::Result<()> {
+        self.target.delete(base_path).await
+    }
+}
+
 /// Controls how new tables should be created
 #[derive(Clone, Debug, Default)]
 pub struct NewTableConfig {
@@ -355,20 +461,45 @@ impl ListingDatabase {
         url.to_string()
     }
 
+    fn rooted_commit_handler_for_uri(uri: &str) -> Option<Arc<dyn CommitHandler>> {
+        #[cfg(windows)]
+        {
+            let is_file = match url::Url::parse(uri) {
+                Ok(url) => url.scheme() == "file" || url.scheme().len() == 1,
+                Err(_) => true,
+            };
+            return is_file.then(rooted_file_commit_handler);
+        }
+
+        #[cfg(not(windows))]
+        {
+            let _ = uri;
+            None
+        }
+    }
+
+    fn rooted_commit_handler(&self) -> Option<Arc<dyn CommitHandler>> {
+        Self::rooted_commit_handler_for_uri(&self.uri)
+    }
+
     async fn prepare_namespace_root(
         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) => {
+                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())
@@ -399,6 +530,13 @@ impl ListingDatabase {
                 url.set_query(None);
                 let plain_uri = url.to_string();
 
+                #[cfg(windows)]
+                if url.scheme() == "file" && prepare_native_directory {
+                    Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
+                        path: plain_uri.clone(),
+                    })?;
+                }
+
                 let os_params = ObjectStoreParams {
                     storage_options_accessor: if storage_options.is_empty() {
                         None
@@ -415,7 +553,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(),
                     })?;
@@ -424,13 +562,16 @@ impl ListingDatabase {
                 Ok(plain_uri)
             }
             Err(_) => {
+                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())
@@ -442,13 +583,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,
@@ -547,10 +689,14 @@ 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" && prepare_native_directory {
+                    Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
+                        path: storage_base_uri.clone(),
+                    })?;
+                }
                 let os_params = ObjectStoreParams {
                     storage_options_accessor: if options.storage_options.is_empty() {
                         None
@@ -567,7 +713,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(),
                     })?;
@@ -625,14 +771,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()));
+        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 })?;
         }
 
@@ -662,20 +812,16 @@ impl ListingDatabase {
 
     /// Try to create a local directory to store the lancedb dataset
     fn try_create_dir(path: &str) -> core::result::Result<(), std::io::Error> {
-        // Strip file:// or file:/ scheme if present to get the actual filesystem path
-        // Note: file:///path becomes file:/path after url.to_string(), so we need to handle both
-        let fs_path = if let Some(stripped) = path.strip_prefix("file://") {
-            // file:///path or file://host/path format
-            stripped
-        } else if let Some(stripped) = path.strip_prefix("file:") {
-            // file:/path format (from url.to_string() on file:///path)
-            // The path after "file:" should already start with "/" for absolute paths
-            stripped
-        } else {
-            path
+        let filesystem_path = match url::Url::parse(path) {
+            Ok(url) if url.scheme() == "file" => 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(),
         };
-
-        let path = Path::new(fs_path);
+        let path = filesystem_path.as_path();
         if !path.try_exists()? {
             create_dir_all(path)?;
         }
@@ -733,7 +879,10 @@ impl ListingDatabase {
         if let Some(query_string) = &self.query_string {
             uri.push_str(&format!("?{}", query_string));
         }
-        let commit_handler = commit_handler_from_url(&uri, &Some(object_store_params)).await?;
+        let commit_handler = match self.rooted_commit_handler() {
+            Some(handler) => handler,
+            None => commit_handler_from_url(&uri, &Some(object_store_params)).await?,
+        };
         for name in names {
             let dir_name = format!("{}.{}", name, LANCE_EXTENSION);
             let full_path = self.base_path.clone().join(dir_name.clone());
@@ -866,6 +1015,9 @@ impl ListingDatabase {
             write_params.mode = WriteMode::Overwrite;
         }
 
+        if write_params.commit_handler.is_none() {
+            write_params.commit_handler = self.rooted_commit_handler();
+        }
         write_params.session = Some(self.session.clone());
 
         write_params
@@ -1112,30 +1264,90 @@ impl Database for ListingDatabase {
             },
             ..Default::default()
         };
+        let source_commit_handler = Self::rooted_commit_handler_for_uri(&request.source_uri);
         let read_params = ReadParams {
             store_options: Some(storage_params.clone()),
             session: Some(self.session.clone()),
+            commit_handler: source_commit_handler.clone(),
             ..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, source_base) = 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 source_commit_handler = match source_commit_handler {
+            Some(handler) => handler,
+            None => {
+                commit_handler_from_url(&source_location.uri, &Some(storage_params.clone())).await?
+            }
+        };
+        let target_commit_handler = match self.rooted_commit_handler() {
+            Some(handler) => handler,
+            None => commit_handler_from_url(&target_uri, &Some(storage_params.clone())).await?,
+        };
+        let target_read_params = clone_target_read_params(
+            storage_params.clone(),
+            self.session.clone(),
+            target_commit_handler.clone(),
+        );
+        let clone_commit_handler = Arc::new(CloneCommitHandler {
+            source_base,
+            source: source_commit_handler,
+            target: target_commit_handler,
+        });
+        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_commit_handler(clone_commit_handler)
+            .with_session(self.session.clone())
+            .with_storage_format(
+                source_dataset
+                    .manifest
+                    .data_storage_format
+                    .lance_file_format()
+                    .to_selector(),
+            )
+            .execute(transaction)
             .await
             .map_err(|e| -> Error { e.into() })?;
 
@@ -1144,7 +1356,7 @@ impl Database for ListingDatabase {
             &request.target_table_name,
             request.target_namespace_path,
             self.store_wrapper.clone(),
-            None,
+            Some(target_read_params),
             self.read_consistency_interval,
             request.namespace_client,
             HashSet::new(), // listing database doesn't support server-side queries
@@ -1210,6 +1422,9 @@ impl Database for ListingDatabase {
             }
             default_params
         });
+        if read_params.commit_handler.is_none() {
+            read_params.commit_handler = self.rooted_commit_handler();
+        }
         read_params.session(self.session.clone());
 
         let native_table = Arc::new(
@@ -1307,6 +1522,60 @@ mod tests {
     use tokio::sync::Barrier;
     use tokio::time::timeout;
 
+    #[derive(Debug)]
+    struct TargetOnlyCommitHandler;
+
+    #[async_trait::async_trait]
+    impl CommitHandler for TargetOnlyCommitHandler {
+        async fn resolve_version_location(
+            &self,
+            _base_path: &ObjectPath,
+            _version: u64,
+            _object_store: &dyn object_store::ObjectStore,
+        ) -> lance_core::Result<ManifestLocation> {
+            Err(lance_core::Error::invalid_input(
+                "target commit handler was used for the source manifest",
+            ))
+        }
+
+        async fn commit(
+            &self,
+            manifest: &mut Manifest,
+            indices: Option<Vec<IndexMetadata>>,
+            base_path: &ObjectPath,
+            object_store: &ObjectStore,
+            manifest_writer: ManifestWriter,
+            naming_scheme: ManifestNamingScheme,
+            transaction: Option<LanceTableTransaction>,
+        ) -> std::result::Result<ManifestLocation, CommitError> {
+            lance_table::io::commit::ConditionalPutCommitHandler
+                .commit(
+                    manifest,
+                    indices,
+                    base_path,
+                    object_store,
+                    manifest_writer,
+                    naming_scheme,
+                    transaction,
+                )
+                .await
+        }
+    }
+
+    #[test]
+    fn commit_handler_forwards_commit_outcome_capabilities() {
+        let target: Arc<dyn CommitHandler> =
+            Arc::new(lance_table::io::commit::ConditionalPutCommitHandler);
+        let handler = CloneCommitHandler {
+            source_base: ObjectPath::from("source"),
+            source: target.clone(),
+            target,
+        };
+
+        assert!(handler.is_version_not_found_definitive());
+        assert!(!handler.propagate_commit_error_after_success());
+    }
+
     async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
         let tempdir = tempdir().unwrap();
         let uri = tempdir.path().to_str().unwrap();
@@ -1329,6 +1598,199 @@ 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
+        ));
+    }
+
+    #[test]
+    fn clone_target_read_params_use_the_target_commit_handler() {
+        let session = Arc::new(lance::session::Session::default());
+        let target_handler: Arc<dyn CommitHandler> = Arc::new(TargetOnlyCommitHandler);
+
+        let params = clone_target_read_params(
+            ObjectStoreParams::default(),
+            session.clone(),
+            target_handler.clone(),
+        );
+
+        assert!(Arc::ptr_eq(params.session.as_ref().unwrap(), &session));
+        assert!(Arc::ptr_eq(
+            params.commit_handler.as_ref().unwrap(),
+            &target_handler
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_listing_database_with_prefixed_file_store() {
+        let tempdir = tempdir().unwrap();
+        let uri = tempdir.path().to_str().unwrap();
+        let session = crate::io::object_store::new_prefixed_file_session();
+
+        let request = ConnectRequest {
+            uri: uri.to_string(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: Some(session),
+        };
+        let db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2, 3]))]).unwrap();
+        db.create_table(CreateTableRequest {
+            name: "test".to_string(),
+            namespace_path: vec![],
+            data: Box::new(batch),
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        #[allow(deprecated)]
+        let table_names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert_eq!(table_names, vec!["test"]);
+
+        let table = db
+            .open_table(OpenTableRequest {
+                name: "test".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+            .await
+            .unwrap();
+        assert_eq!(table.count_rows(None).await.unwrap(), 3);
+        drop(table);
+
+        db.drop_table("test", &[]).await.unwrap();
+        assert!(!tempdir.path().join("test.lance").exists());
+        assert!(matches!(
+            db.drop_table("test", &[]).await,
+            Err(Error::TableNotFound { .. })
+        ));
+        #[allow(deprecated)]
+        let table_names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert!(table_names.is_empty());
+        let reopened = db
+            .open_table(OpenTableRequest {
+                name: "test".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+            .await;
+        assert!(matches!(reopened, Err(Error::TableNotFound { .. })));
+    }
+
+    #[cfg(windows)]
+    #[tokio::test]
+    async fn reopens_a_table_through_a_unc_authority() {
+        use std::path::{Component, Prefix};
+
+        let tempdir = tempdir().unwrap();
+        let drive = match tempdir.path().components().next() {
+            Some(Component::Prefix(prefix)) => match prefix.kind() {
+                Prefix::Disk(drive) | Prefix::VerbatimDisk(drive) => drive,
+                other => panic!("expected a drive-backed temporary directory, got {other:?}"),
+            },
+            other => panic!("expected an absolute Windows temporary directory, got {other:?}"),
+        };
+        let drive_root = PathBuf::from(format!("{}:\\", drive as char));
+        let relative = tempdir.path().strip_prefix(&drive_root).unwrap();
+        // `url` treats `localhost` as an empty file-URL authority, turning
+        // `file://localhost/C$/...` into the invalid drive-relative
+        // `file:///C$/...`. Use the machine's actual network name so this
+        // remains a genuine UNC URL throughout the connection lifecycle.
+        let Some(computer_name) = std::env::var_os("COMPUTERNAME") else {
+            eprintln!("skipping UNC lifecycle test because COMPUTERNAME is unavailable");
+            return;
+        };
+        let unc_path = PathBuf::from(format!(
+            r"\\{}\{}$",
+            computer_name.to_string_lossy(),
+            drive as char
+        ))
+        .join(relative);
+        if !unc_path.try_exists().unwrap_or(false) {
+            eprintln!("skipping UNC lifecycle test because the local admin share is unavailable");
+            return;
+        }
+        let uri = url::Url::from_directory_path(&unc_path)
+            .unwrap()
+            .to_string();
+        let request = ConnectRequest {
+            uri,
+            #[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 db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let table = db
+            .create_table(CreateTableRequest {
+                name: "unc_reopen".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();
+        drop(table);
+
+        let reopened = db
+            .open_table(OpenTableRequest {
+                name: "unc_reopen".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+            .await
+            .unwrap();
+        assert_eq!(reopened.count_rows(None).await.unwrap(), 3);
+    }
+
     struct BarrierScannable {
         batch: RecordBatch,
         barrier: Arc<Barrier>,
@@ -1749,6 +2211,173 @@ 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 clone_routes_source_manifest_through_source_commit_handler() {
+        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 source_uri = source_db.table_uri("source").unwrap();
+        let target_uri = target_db.table_uri("target").unwrap();
+        let storage_params = ObjectStoreParams::default();
+        let source_dataset = DatasetBuilder::from_uri(&source_uri)
+            .with_read_params(ReadParams {
+                session: Some(target_db.session.clone()),
+                ..Default::default()
+            })
+            .load()
+            .await
+            .unwrap();
+        let source_location = source_dataset.branch_location().find_branch(None).unwrap();
+        let (source_store, source_base) = ObjectStore::from_uri_and_params(
+            target_db.session.store_registry(),
+            &source_location.uri,
+            &storage_params,
+        )
+        .await
+        .unwrap();
+        let (target_store, _) = ObjectStore::from_uri_and_params(
+            target_db.session.store_registry(),
+            &target_uri,
+            &storage_params,
+        )
+        .await
+        .unwrap();
+        let source_handler = commit_handler_from_url(&source_location.uri, &None)
+            .await
+            .unwrap();
+        let clone_handler = Arc::new(CloneCommitHandler {
+            source_base,
+            source: source_handler,
+            target: Arc::new(TargetOnlyCommitHandler),
+        });
+        let version = source_dataset.version().version;
+        let transaction = Transaction::new(
+            version,
+            Operation::Clone {
+                is_shallow: true,
+                ref_name: None,
+                ref_version: version,
+                ref_path: source_location.uri,
+                branch_name: None,
+            },
+            None,
+        );
+
+        CommitBuilder::new(WriteDestination::Uri(&target_uri))
+            .with_store_params(storage_params)
+            .with_object_store(target_store)
+            .with_source_store(source_store)
+            .with_commit_handler(clone_handler)
+            .with_session(target_db.session.clone())
+            .with_storage_format(
+                source_dataset
+                    .manifest
+                    .data_storage_format
+                    .lance_file_format()
+                    .to_selector(),
+            )
+            .execute(transaction)
+            .await
+            .unwrap();
+
+        assert!(target_dir.path().join("target.lance").exists());
+    }
+
     #[tokio::test]
     async fn test_clone_table_with_storage_options() {
         let tempdir = tempdir().unwrap();
@@ -2653,18 +3282,14 @@ mod tests {
     /// as `path=<base table>` + `query=/_mem_wal/...`, causing
     /// `Dataset::write` to find the base table dataset and falsely
     /// report `Dataset already exists`.
-    ///
-    /// Skipped on Windows: `try_create_dir` does not understand
-    /// `file:///C:/…` paths so `connect_with_options` fails before
-    /// even reaching the URL-mutation logic. The pure URL-mutation
-    /// invariant is covered by
-    /// `test_capture_query_treats_empty_as_none` above, which runs
-    /// on all platforms.
-    #[cfg(not(windows))]
     #[tokio::test]
     async fn test_table_uri_url_path_has_no_trailing_question_mark() {
         let tempdir = tempdir().unwrap();
-        let uri = format!("file://{}", tempdir.path().to_str().unwrap());
+        let uri = url::Url::from_directory_path(tempdir.path())
+            .unwrap()
+            .to_string()
+            .trim_end_matches('/')
+            .to_string();
 
         let request = ConnectRequest {
             uri: uri.clone(),

rust/lancedb/src/database/namespace.rs

@@ -59,6 +59,46 @@ fn is_table_already_exists_namespace_error(err: &lance::Error) -> bool {
 /// via the `delimiter` property.
 const DEFAULT_NAMESPACE_DELIMITER: &str = "$";
 
+#[cfg(any(windows, test))]
+fn is_unc_root(root: &str) -> bool {
+    let normalized = root.replace('\\', "/");
+    let lowercase = normalized.to_ascii_lowercase();
+
+    if lowercase.starts_with("//?/unc/") {
+        return true;
+    }
+    if lowercase.starts_with("//?/") || lowercase.starts_with("//./") {
+        return false;
+    }
+    if let Some(authority) = normalized.strip_prefix("//") {
+        return !authority.is_empty() && !authority.starts_with('/');
+    }
+
+    normalized.split_once("://").is_some_and(|(scheme, path)| {
+        scheme.eq_ignore_ascii_case("file") && !path.is_empty() && !path.starts_with('/')
+    })
+}
+
+#[cfg(any(windows, test))]
+fn is_directory_unc_root(ns_impl: &str, ns_properties: &HashMap<String, String>) -> bool {
+    ns_impl.eq_ignore_ascii_case("dir")
+        && ns_properties
+            .get("root")
+            .is_some_and(|root| is_unc_root(root))
+}
+
+#[cfg(any(windows, test))]
+#[derive(Debug)]
+struct UnavailableUncDirectoryNamespace;
+
+#[cfg(any(windows, test))]
+#[async_trait]
+impl LanceNamespace for UnavailableUncDirectoryNamespace {
+    fn namespace_id(&self) -> String {
+        "unavailable-unc-directory".to_string()
+    }
+}
+
 /// A database implementation that uses lance-namespace for table management
 pub struct LanceNamespaceDatabase {
     namespace: Arc<dyn LanceNamespace>,
@@ -92,6 +132,17 @@ fn resolve_delimiter(ns_properties: &HashMap<String, String>) -> String {
 }
 
 impl LanceNamespaceDatabase {
+    fn ensure_storage_supported(&self) -> Result<()> {
+        #[cfg(any(windows, test))]
+        if is_directory_unc_root(&self.ns_impl, &self.ns_properties) {
+            return Err(Error::NotSupported {
+                message: "Directory namespace operations are not supported for UNC roots because the namespace manifest resolver does not preserve file URI authorities; use flat root tables or a non-UNC namespace root"
+                    .to_string(),
+            });
+        }
+        Ok(())
+    }
+
     pub fn from_namespace_client(
         namespace_client: Arc<dyn LanceNamespace>,
         namespace_client_impl: String,
@@ -153,6 +204,25 @@ impl LanceNamespaceDatabase {
         pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
         new_table_config: NewTableConfig,
     ) -> Result<Self> {
+        #[cfg(any(windows, test))]
+        if is_directory_unc_root(ns_impl, &ns_properties) {
+            let freshness_baselines: FreshnessBaselines = Arc::new(Mutex::new(HashMap::new()));
+            let delimiter = resolve_delimiter(&ns_properties);
+            return Ok(Self {
+                namespace: Arc::new(UnavailableUncDirectoryNamespace),
+                storage_options,
+                read_consistency_interval,
+                session,
+                uri: format!("namespace://{}", ns_impl),
+                pushdown_operations,
+                ns_impl: ns_impl.to_string(),
+                ns_properties,
+                new_table_config,
+                freshness_baselines,
+                delimiter,
+            });
+        }
+
         let mut builder = ConnectBuilder::new(ns_impl);
         for (key, value) in ns_properties.clone() {
             builder = builder.property(key, value);
@@ -310,6 +380,7 @@ impl Database for LanceNamespaceDatabase {
         &self,
         request: ListNamespacesRequest,
     ) -> Result<ListNamespacesResponse> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.list_namespaces(request).await?)
     }
 
@@ -317,10 +388,12 @@ impl Database for LanceNamespaceDatabase {
         &self,
         request: CreateNamespaceRequest,
     ) -> Result<CreateNamespaceResponse> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.create_namespace(request).await?)
     }
 
     async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.drop_namespace(request).await?)
     }
 
@@ -328,10 +401,12 @@ impl Database for LanceNamespaceDatabase {
         &self,
         request: DescribeNamespaceRequest,
     ) -> Result<DescribeNamespaceResponse> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.describe_namespace(request).await?)
     }
 
     async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
+        self.ensure_storage_supported()?;
         let ns_request = ListTablesRequest {
             id: Some(request.namespace_path),
             page_token: request.start_after,
@@ -345,10 +420,12 @@ impl Database for LanceNamespaceDatabase {
     }
 
     async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.list_tables(request).await?)
     }
 
     async fn create_table(&self, request: DbCreateTableRequest) -> Result<Arc<dyn BaseTable>> {
+        self.ensure_storage_supported()?;
         let mut table_id = request.namespace_path.clone();
         table_id.push(request.name.clone());
         let mut existing_table = None;
@@ -518,6 +595,7 @@ impl Database for LanceNamespaceDatabase {
     }
 
     async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
+        self.ensure_storage_supported()?;
         let native_table = NativeTable::open_from_namespace(
             self.namespace.clone(),
             &request.name,
@@ -547,6 +625,7 @@ impl Database for LanceNamespaceDatabase {
         cur_namespace_path: &[String],
         new_namespace_path: &[String],
     ) -> Result<()> {
+        self.ensure_storage_supported()?;
         let mut cur_table_id = cur_namespace_path.to_vec();
         cur_table_id.push(cur_name.to_string());
 
@@ -573,6 +652,7 @@ impl Database for LanceNamespaceDatabase {
     }
 
     async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> {
+        self.ensure_storage_supported()?;
         let mut table_id = namespace_path.to_vec();
         table_id.push(name.to_string());
 
@@ -612,10 +692,12 @@ impl Database for LanceNamespaceDatabase {
     }
 
     async fn namespace_client(&self) -> Result<Arc<dyn LanceNamespace>> {
+        self.ensure_storage_supported()?;
         Ok(self.namespace.clone())
     }
 
     async fn namespace_client_config(&self) -> Result<(String, HashMap<String, String>)> {
+        self.ensure_storage_supported()?;
         Ok((self.ns_impl.clone(), self.ns_properties.clone()))
     }
 }
@@ -644,6 +726,56 @@ mod tests {
         RecordBatch::try_new(schema, vec![Arc::new(id_array), Arc::new(name_array)]).unwrap()
     }
 
+    #[test]
+    fn detects_unc_namespace_roots_without_misclassifying_drives() {
+        for root in [
+            "file://server/share/database",
+            "FILE://server/share/database",
+            r"\\server\share\database",
+            r"\\?\UNC\server\share\database",
+        ] {
+            assert!(is_unc_root(root), "expected an UNC root: {root}");
+        }
+        for root in [
+            "file:///C:/database",
+            r"C:\database",
+            r"\\?\C:\database",
+            "/var/lib/database",
+        ] {
+            assert!(!is_unc_root(root), "expected a non-UNC root: {root}");
+        }
+    }
+
+    #[tokio::test]
+    async fn directory_namespace_fails_closed_before_accessing_an_unc_root() {
+        let mut properties = HashMap::new();
+        properties.insert(
+            "root".to_string(),
+            "file://server/share/database".to_string(),
+        );
+        let db = LanceNamespaceDatabase::connect(
+            "dir",
+            properties,
+            HashMap::new(),
+            None,
+            None,
+            HashSet::new(),
+        )
+        .await
+        .unwrap();
+
+        let error = db
+            .list_tables(ListTablesRequest::default())
+            .await
+            .unwrap_err();
+        assert!(matches!(error, Error::NotSupported { .. }));
+        assert!(error.to_string().contains("UNC roots"));
+
+        let error = db.namespace_client_config().await.unwrap_err();
+        assert!(matches!(error, Error::NotSupported { .. }));
+        assert!(error.to_string().contains("UNC roots"));
+    }
+
     #[tokio::test]
     async fn test_namespace_connection_simple() {
         // Test that namespace connections work with simple connect_namespace(impl_type, properties)

rust/lancedb/src/io/object_store.rs

@@ -1,12 +1,22 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-//! A mirroring object store that mirror writes to a secondary object store
+//! Object-store providers and adapters used by LanceDB.
 
 use std::{fmt::Formatter, sync::Arc};
 
+#[cfg(any(windows, test))]
+use futures::TryStreamExt;
 use futures::{StreamExt, TryFutureExt, stream::BoxStream};
 use lance::io::WrappingObjectStore;
+#[cfg(any(windows, test))]
+use lance_table::{
+    format::{IndexMetadata, Manifest, Transaction},
+    io::commit::{
+        CommitError, CommitHandler, ManifestLocation, ManifestNamingScheme, ManifestWriter,
+        RenameCommitHandler,
+    },
+};
 use object_store::{
     CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
     ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
@@ -15,9 +25,797 @@ use object_store::{
 
 use async_trait::async_trait;
 
+#[cfg(any(windows, test))]
+use lance_core::{Error as LanceError, Result as LanceResult};
+#[cfg(test)]
+use lance_io::object_store::ObjectStoreRegistry;
+#[cfg(any(windows, test))]
+use lance_io::object_store::{
+    DEFAULT_LOCAL_IO_PARALLELISM, ObjectStoreParams, ObjectStoreProvider, StorageOptions,
+};
+#[cfg(any(windows, test))]
+use object_store::local::LocalFileSystem;
+#[cfg(any(windows, test))]
+use url::Url;
+
 #[cfg(test)]
 pub mod io_tracking;
 
+/// A local commit handler that resolves the latest manifest through the object store.
+///
+/// Lance's native local shortcut reconstructs the selected manifest from a
+/// filesystem path. On Windows that conversion drops the authority from a UNC
+/// path. Listing through the already-rooted object store preserves the structural
+/// server/share prefix while retaining the normal atomic-rename commit behavior.
+#[derive(Debug)]
+#[cfg(any(windows, test))]
+struct RootedFileCommitHandler;
+
+#[cfg(any(windows, test))]
+#[async_trait]
+impl CommitHandler for RootedFileCommitHandler {
+    fn is_version_not_found_definitive(&self) -> bool {
+        RenameCommitHandler.is_version_not_found_definitive()
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        RenameCommitHandler.propagate_commit_error_after_success()
+    }
+
+    async fn resolve_latest_location(
+        &self,
+        base_path: &Path,
+        object_store: &lance::io::ObjectStore,
+    ) -> LanceResult<ManifestLocation> {
+        self.list_manifest_locations(base_path, object_store, true)
+            .try_next()
+            .await?
+            .ok_or_else(|| LanceError::not_found(base_path.to_string()))
+    }
+
+    async fn commit(
+        &self,
+        manifest: &mut Manifest,
+        indices: Option<Vec<IndexMetadata>>,
+        base_path: &Path,
+        object_store: &lance::io::ObjectStore,
+        manifest_writer: ManifestWriter,
+        naming_scheme: ManifestNamingScheme,
+        transaction: Option<Transaction>,
+    ) -> std::result::Result<ManifestLocation, CommitError> {
+        RenameCommitHandler
+            .commit(
+                manifest,
+                indices,
+                base_path,
+                object_store,
+                manifest_writer,
+                naming_scheme,
+                transaction,
+            )
+            .await
+    }
+}
+
+#[cfg(any(windows, test))]
+pub(crate) fn rooted_file_commit_handler() -> Arc<dyn CommitHandler> {
+    Arc::new(RootedFileCommitHandler)
+}
+
+/// A file-store provider that anchors each request at its filesystem root.
+///
+/// On Windows, an unprefixed [`LocalFileSystem`] cannot service UNC paths. Its
+/// conversion to an object-store [`Path`] drops the UNC host, so subsequent I/O
+/// is directed at a different local path. Anchoring the store at the drive or
+/// UNC-share root keeps the UNC authority in the filesystem prefix and exposes
+/// only paths relative to that prefix to `object_store`.
+///
+/// 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;
+
+#[cfg(any(windows, test))]
+impl PrefixedFileStoreProvider {
+    fn root_and_relative_path(url: &Url) -> LanceResult<(std::path::PathBuf, Path)> {
+        let filesystem_path = url.to_file_path().map_err(|_| {
+            LanceError::invalid_input(format!("Unable to convert URL '{url}' to a local path"))
+        })?;
+
+        let mut root = std::path::PathBuf::new();
+        for component in filesystem_path.components() {
+            match component {
+                std::path::Component::Prefix(_) | std::path::Component::RootDir => {
+                    root.push(component.as_os_str());
+                }
+                _ => break,
+            }
+        }
+        if root.as_os_str().is_empty() {
+            return Err(LanceError::invalid_input(format!(
+                "Local path '{}' has no filesystem root",
+                filesystem_path.display()
+            )));
+        }
+
+        let relative = filesystem_path.strip_prefix(&root).map_err(|_| {
+            LanceError::invalid_input(format!(
+                "Local path '{}' is not beneath store root '{}'",
+                filesystem_path.display(),
+                root.display()
+            ))
+        })?;
+        let relative = relative
+            .components()
+            .filter_map(|component| match component {
+                std::path::Component::Normal(part) => Some(part),
+                _ => None,
+            })
+            .map(|part| {
+                part.to_str().ok_or_else(|| {
+                    LanceError::invalid_input(format!(
+                        "Local path '{}' is not valid UTF-8",
+                        filesystem_path.display()
+                    ))
+                })
+            })
+            .collect::<LanceResult<Vec<_>>>()?
+            .join("/");
+
+        Ok((root, Path::parse(relative)?))
+    }
+
+    /// Preserve the native filesystem root as a structural path component.
+    ///
+    /// `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)))
+    }
+}
+
+/// A local store rooted at a Windows drive or UNC share.
+///
+/// Most calls use paths returned by [`PrefixedFileStoreProvider`], which are
+/// relative to `root`. Some Lance operations retain an existing object store
+/// while independently re-extracting a Windows file URI with the default file
+/// provider. Those paths include the drive (`C:/...`) or UNC share
+/// (`share/...`) again. Normalize that absolute alias before delegating so the
+/// filesystem prefix is never applied twice.
+#[cfg(any(windows, test))]
+#[derive(Debug, Clone)]
+struct RootedLocalFileSystem {
+    inner: Arc<LocalFileSystem>,
+    root: std::path::PathBuf,
+    absolute_alias: Path,
+}
+
+#[cfg(any(windows, test))]
+impl RootedLocalFileSystem {
+    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,
+            absolute_alias,
+        })
+    }
+
+    fn path_from_parts<'a>(parts: impl Iterator<Item = object_store::path::PathPart<'a>>) -> Path {
+        parts.fold(Path::default(), |path, part| path.join(part))
+    }
+
+    fn normalize(&self, path: &Path) -> Path {
+        if self.absolute_alias.as_ref().is_empty() {
+            return path.clone();
+        }
+        let Some(suffix) = path.prefix_match(&self.absolute_alias) else {
+            return path.clone();
+        };
+        Self::path_from_parts(suffix)
+    }
+
+    fn restore_prefix(&self, path: Path, requested: &Path, normalized: &Path) -> Path {
+        if requested == normalized {
+            return path;
+        }
+        path.prefix_match(normalized)
+            .map(|suffix| suffix.fold(requested.clone(), |path, part| path.join(part)))
+            .unwrap_or(path)
+    }
+}
+
+#[cfg(any(windows, test))]
+impl std::fmt::Display for RootedLocalFileSystem {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "RootedLocalFileSystem({})", self.root.display())
+    }
+}
+
+#[cfg(any(windows, test))]
+#[async_trait]
+impl ObjectStore for RootedLocalFileSystem {
+    async fn put_opts(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+        options: PutOptions,
+    ) -> Result<PutResult> {
+        self.inner
+            .put_opts(&self.normalize(location), payload, options)
+            .await
+    }
+
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        options: PutMultipartOptions,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        self.inner
+            .put_multipart_opts(&self.normalize(location), options)
+            .await
+    }
+
+    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
+        let normalized = self.normalize(location);
+        let mut result = self.inner.get_opts(&normalized, options).await?;
+        result.meta.location = location.clone();
+        Ok(result)
+    }
+
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, Result<Path>>,
+    ) -> BoxStream<'static, Result<Path>> {
+        let store = self.clone();
+        locations
+            .map(move |location| {
+                let store = store.clone();
+                async move {
+                    let location = location?;
+                    let normalized = store.normalize(&location);
+                    store.inner.delete(&normalized).await?;
+                    Ok(location)
+                }
+            })
+            .buffered(10)
+            .boxed()
+    }
+
+    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
+        let requested = prefix.cloned().unwrap_or_default();
+        let normalized = self.normalize(&requested);
+        let store = self.clone();
+        self.inner
+            .list(prefix.map(|_| &normalized))
+            .map(move |result| {
+                result.map(|mut meta| {
+                    meta.location = store.restore_prefix(meta.location, &requested, &normalized);
+                    meta
+                })
+            })
+            .boxed()
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let requested = prefix.cloned().unwrap_or_default();
+        let normalized = self.normalize(&requested);
+        let mut result = self
+            .inner
+            .list_with_delimiter(prefix.map(|_| &normalized))
+            .await?;
+        for meta in &mut result.objects {
+            meta.location = self.restore_prefix(meta.location.clone(), &requested, &normalized);
+        }
+        for path in &mut result.common_prefixes {
+            *path = self.restore_prefix(path.clone(), &requested, &normalized);
+        }
+        Ok(result)
+    }
+
+    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
+        self.inner
+            .copy_opts(&self.normalize(from), &self.normalize(to), options)
+            .await
+    }
+}
+
+#[cfg(any(windows, test))]
+#[async_trait]
+impl ObjectStoreProvider for PrefixedFileStoreProvider {
+    async fn new_store(
+        &self,
+        base_path: Url,
+        params: &ObjectStoreParams,
+    ) -> LanceResult<lance::io::ObjectStore> {
+        let (root, _) = Self::root_and_relative_path(&base_path)?;
+        let raw_store: Arc<dyn ObjectStore> = Arc::new(RootedLocalFileSystem::new(root)?);
+        // Native local shortcuts are safe only when the rooted filesystem is
+        // the final store. An arbitrary wrapper can redirect I/O, so use
+        // Lance's non-cloud object-store route in that case. This keeps local
+        // scan planning without enabling native copy/delete or the io_uring
+        // scheduler, all of which would bypass the wrapper.
+        let location = if params.object_store_wrapper.is_some() {
+            Url::parse("memory:///").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());
+
+        // ObjectStore::new initializes the private local-store fields. The
+        // registry owns tracing, custom wrapper, and I/O tracker installation,
+        // so restore the raw store before returning to avoid applying them twice.
+        let mut store = lance::io::ObjectStore::new(
+            raw_store.clone(),
+            location,
+            Some(params.block_size.unwrap_or(4 * 1024)),
+            None,
+            false,
+            false,
+            DEFAULT_LOCAL_IO_PARALLELISM,
+            storage_options.download_retry_count(),
+            params.storage_options(),
+        );
+        store.inner = raw_store;
+        store.store_prefix =
+            self.calculate_object_store_prefix(&base_path, params.storage_options())?;
+        Ok(store)
+    }
+
+    fn extract_path(&self, url: &Url) -> LanceResult<Path> {
+        let (root, relative) = Self::root_and_relative_path(url)?;
+        Self::rooted_path(&root, &relative)
+    }
+
+    fn calculate_object_store_prefix(
+        &self,
+        url: &Url,
+        _storage_options: Option<&std::collections::HashMap<String, String>>,
+    ) -> LanceResult<String> {
+        let (root, _) = Self::root_and_relative_path(url)?;
+        let root = root.canonicalize()?;
+        // 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()))
+    }
+}
+
+/// 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)]
+mod prefixed_file_store_test {
+    use super::*;
+    use object_store::memory::InMemory;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[derive(Debug, Default)]
+    struct CountingWrapper {
+        calls: AtomicUsize,
+    }
+
+    impl WrappingObjectStore for CountingWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            original: Arc<dyn ObjectStore>,
+        ) -> Arc<dyn ObjectStore> {
+            self.calls.fetch_add(1, Ordering::Relaxed);
+            original
+        }
+    }
+
+    #[derive(Debug, Default)]
+    struct MemoryRedirectWrapper {
+        store: Arc<InMemory>,
+    }
+
+    impl WrappingObjectStore for MemoryRedirectWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            _original: Arc<dyn ObjectStore>,
+        ) -> Arc<dyn ObjectStore> {
+            self.store.clone()
+        }
+    }
+
+    #[test]
+    fn commit_handler_forwards_commit_outcome_capabilities() {
+        let handler = RootedFileCommitHandler;
+
+        assert!(handler.is_version_not_found_definitive());
+        assert!(!handler.propagate_commit_error_after_success());
+    }
+
+    #[tokio::test]
+    async fn anchors_new_and_existing_directories_at_a_filesystem_prefix() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let database_path = tempdir.path().join("database");
+        std::fs::create_dir(&database_path).unwrap();
+        let table_path = database_path.join("test.lance");
+        let table_url = Url::from_directory_path(&table_path).unwrap();
+
+        let registry = Arc::new(ObjectStoreRegistry::default());
+        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+
+        // 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(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+        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"));
+        let initial_base_path = base_path.clone();
+
+        let marker = base_path.join("marker");
+        store
+            .inner
+            .put(&marker, bytes::Bytes::from_static(b"new").into())
+            .await
+            .unwrap();
+        assert_eq!(std::fs::read(table_path.join("marker")).unwrap(), b"new");
+
+        // Once the table directory exists, a fresh store uses the same stable
+        // root and object-store path.
+        drop(store);
+        let (store, base_path) = lance::io::ObjectStore::from_uri_and_params(
+            registry,
+            table_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+        assert_eq!(base_path, initial_base_path);
+        let contents = store
+            .inner
+            .get(&base_path.join("marker"))
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(contents.as_ref(), b"new");
+    }
+
+    #[tokio::test]
+    async fn applies_wrapper_and_io_tracking_once() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let table_url = Url::from_directory_path(tempdir.path().join("test.lance")).unwrap();
+        let registry = Arc::new(ObjectStoreRegistry::default());
+        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+        let wrapper = Arc::new(CountingWrapper::default());
+        let params = ObjectStoreParams {
+            object_store_wrapper: Some(wrapper.clone()),
+            ..Default::default()
+        };
+
+        let (store, base_path) =
+            lance::io::ObjectStore::from_uri_and_params(registry, table_url.as_str(), &params)
+                .await
+                .unwrap();
+        assert_eq!(wrapper.calls.load(Ordering::Relaxed), 1);
+        assert_eq!(store.scheme(), "memory");
+        assert!(!store.is_local());
+        assert!(!store.is_cloud());
+        assert!(!store.prefers_lite_scheduler());
+
+        store
+            .inner
+            .put(
+                &base_path.join("marker"),
+                bytes::Bytes::from_static(b"tracked").into(),
+            )
+            .await
+            .unwrap();
+        let stats = store.io_tracker().stats();
+        assert_eq!(stats.write_iops, 1);
+        assert_eq!(stats.written_bytes, 7);
+    }
+
+    #[tokio::test]
+    async fn wrapped_store_cleanup_does_not_bypass_the_wrapper() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let table_path = tempdir.path().join("test.lance");
+        std::fs::create_dir(&table_path).unwrap();
+        std::fs::write(table_path.join("native-marker"), b"native").unwrap();
+
+        let table_url = Url::from_directory_path(&table_path).unwrap();
+        let registry = Arc::new(ObjectStoreRegistry::default());
+        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+        let wrapper = Arc::new(MemoryRedirectWrapper::default());
+        let params = ObjectStoreParams {
+            object_store_wrapper: Some(wrapper.clone()),
+            ..Default::default()
+        };
+        let (store, base_path) =
+            lance::io::ObjectStore::from_uri_and_params(registry, table_url.as_str(), &params)
+                .await
+                .unwrap();
+        let wrapped_marker = base_path.clone().join("wrapped-marker");
+        store
+            .inner
+            .put(
+                &wrapped_marker,
+                bytes::Bytes::from_static(b"wrapped").into(),
+            )
+            .await
+            .unwrap();
+
+        store.remove_dir_all(base_path).await.unwrap();
+
+        assert!(table_path.join("native-marker").exists());
+        assert!(matches!(
+            wrapper.store.head(&wrapped_marker).await,
+            Err(object_store::Error::NotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn rooted_commit_handler_uses_store_paths_for_latest_manifest() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let versions = tempdir.path().join("_versions");
+        std::fs::create_dir(&versions).unwrap();
+        std::fs::write(versions.join("2.manifest"), b"native").unwrap();
+
+        let base_path = Path::from_absolute_path(tempdir.path()).unwrap();
+        let raw_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        raw_store
+            .put(
+                &base_path.clone().join("_versions").join("1.manifest"),
+                bytes::Bytes::from_static(b"wrapped").into(),
+            )
+            .await
+            .unwrap();
+        let mut store = lance::io::ObjectStore::new(
+            raw_store.clone(),
+            Url::parse("file:///").unwrap(),
+            Some(4 * 1024),
+            None,
+            false,
+            false,
+            DEFAULT_LOCAL_IO_PARALLELISM,
+            0,
+            None,
+        );
+        store.inner = raw_store;
+
+        let native = RenameCommitHandler
+            .resolve_latest_location(&base_path, &store)
+            .await
+            .unwrap();
+        assert_eq!(native.version, 2);
+
+        let rooted = rooted_file_commit_handler()
+            .resolve_latest_location(&base_path, &store)
+            .await
+            .unwrap();
+        assert_eq!(rooted.version, 1);
+        assert_eq!(
+            rooted.path,
+            base_path.clone().join("_versions").join("1.manifest")
+        );
+    }
+
+    #[tokio::test]
+    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();
+        let registry = Arc::new(ObjectStoreRegistry::default());
+        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+
+        let (first, _) = lance::io::ObjectStore::from_uri_and_params(
+            registry.clone(),
+            first_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+        let (second, _) = lance::io::ObjectStore::from_uri_and_params(
+            registry.clone(),
+            second_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+
+        assert!(Arc::ptr_eq(&first, &second));
+        assert_eq!(registry.stats().misses, 1);
+        assert_eq!(registry.stats().hits, 1);
+    }
+
+    #[tokio::test]
+    async fn reuses_database_store_for_table_and_clone_targets() {
+        let tempdir = tempfile::tempdir().unwrap();
+        let database_url = Url::from_directory_path(tempdir.path().join("database")).unwrap();
+        let source_url =
+            Url::from_directory_path(tempdir.path().join("database/source.lance")).unwrap();
+        let target_url =
+            Url::from_directory_path(tempdir.path().join("database/target.lance")).unwrap();
+        let registry = Arc::new(ObjectStoreRegistry::default());
+        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
+
+        let (database, _) = lance::io::ObjectStore::from_uri_and_params(
+            registry.clone(),
+            database_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+        let (source, _) = lance::io::ObjectStore::from_uri_and_params(
+            registry.clone(),
+            source_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+        let (target, _) = lance::io::ObjectStore::from_uri_and_params(
+            registry.clone(),
+            target_url.as_str(),
+            &ObjectStoreParams::default(),
+        )
+        .await
+        .unwrap();
+
+        assert!(Arc::ptr_eq(&database, &source));
+        assert!(Arc::ptr_eq(&database, &target));
+        assert_eq!(registry.stats().misses, 1);
+        assert_eq!(registry.stats().hits, 2);
+    }
+
+    #[test]
+    fn normalizes_absolute_drive_and_unc_aliases() {
+        let tempdir = tempfile::tempdir().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()).unwrap();
+        unc_store.absolute_alias = Path::parse(r"\\server\share").unwrap();
+        assert_eq!(
+            unc_store.normalize(&Path::parse(r"\\server\share/share/db/table.lance").unwrap()),
+            Path::from("share/db/table.lance")
+        );
+        assert_eq!(
+            unc_store.normalize(&Path::from("share/db/table.lance")),
+            Path::from("share/db/table.lance")
+        );
+    }
+
+    #[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()).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::parse(r"\\server\share/share/database/table.lance/marker").unwrap()
+            ),
+            relative
+        );
+    }
+
+    #[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()).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
+            .put(&marker, bytes::Bytes::from_static(b"unc").into())
+            .await
+            .unwrap();
+        assert_eq!(
+            std::fs::read(tempdir.path().join("share/db/test.lance/marker")).unwrap(),
+            b"unc"
+        );
+
+        let listed = store.list(Some(&table)).collect::<Vec<_>>().await;
+        assert_eq!(listed.len(), 1);
+        assert_eq!(listed[0].as_ref().unwrap().location, marker);
+        assert_eq!(
+            store
+                .get(&marker)
+                .await
+                .unwrap()
+                .bytes()
+                .await
+                .unwrap()
+                .as_ref(),
+            b"unc"
+        );
+
+        store.delete(&marker).await.unwrap();
+        assert!(!tempdir.path().join("share/db/test.lance/marker").exists());
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn extracts_drive_and_unc_share_roots() {
+        let (root, path) = PrefixedFileStoreProvider::root_and_relative_path(
+            &Url::parse("file:///C:/database").unwrap(),
+        )
+        .unwrap();
+        assert_eq!(root, std::path::PathBuf::from(r"C:\"));
+        assert_eq!(path, Path::from("database"));
+
+        let (root, path) = PrefixedFileStoreProvider::root_and_relative_path(
+            &Url::parse("file://server/share/database").unwrap(),
+        )
+        .unwrap();
+        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)]
 struct MirroringObjectStore {
     primary: Arc<dyn ObjectStore>,