From 5fa98b9af80a9c14e39a34b61e98bcc76c2722d3 Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Thu, 6 Aug 2026 09:13:59 +0000 Subject: [PATCH] fix: address Windows file store review feedback --- python/python/lancedb/db.py | 15 +- python/python/tests/test_db.py | 55 ++++ rust/lancedb/src/database/listing.rs | 51 ++++ rust/lancedb/src/io/object_store.rs | 383 +++++++++++++++++++++++++-- 4 files changed, 470 insertions(+), 34 deletions(-) diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index b48e84cc5..14f31f21a 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -713,20 +713,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) diff --git a/python/python/tests/test_db.py b/python/python/tests/test_db.py index 86b70b75d..96fde4427 100644 --- a/python/python/tests/test_db.py +++ b/python/python/tests/test_db.py @@ -77,6 +77,32 @@ def test_sync_repr_does_not_use_background_loop(tmp_path, monkeypatch): 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_ingest_pd(tmp_path): db = lancedb.connect(tmp_path) @@ -376,6 +402,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() diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs index 7a68b6b58..9e284b17a 100644 --- a/rust/lancedb/src/database/listing.rs +++ b/rust/lancedb/src/database/listing.rs @@ -704,6 +704,27 @@ 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(), + }; + match std::fs::remove_dir_all(filesystem_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } + } + /// Get the URI of a table in the database. fn table_uri(&self, name: &str) -> Result { validate_table_name(name)?; @@ -774,6 +795,17 @@ 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| Error::Runtime { + message: format!("Failed to remove table directory '{table_uri}': {error}"), + })?; + } } Ok(()) } @@ -1397,6 +1429,25 @@ mod tests { .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()); + #[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 { .. }))); } #[tokio::test] diff --git a/rust/lancedb/src/io/object_store.rs b/rust/lancedb/src/io/object_store.rs index d12c08542..16cec7736 100644 --- a/rust/lancedb/src/io/object_store.rs +++ b/rust/lancedb/src/io/object_store.rs @@ -38,12 +38,16 @@ 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 `file-object-store` scheme. +/// 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. #[cfg(any(windows, test))] #[derive(Debug, Default)] -struct PrefixedFileStoreProvider; +struct PrefixedFileStoreProvider { + base_paths: std::sync::Mutex< + std::collections::HashMap>>>, + >, +} #[cfg(any(windows, test))] impl PrefixedFileStoreProvider { @@ -52,14 +56,23 @@ impl PrefixedFileStoreProvider { LanceError::invalid_input(format!("Unable to convert URL '{url}' to a local path")) })?; - let root = filesystem_path.ancestors().last().ok_or_else(|| { - LanceError::invalid_input(format!( + 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(|_| { + let relative = filesystem_path.strip_prefix(&root).map_err(|_| { LanceError::invalid_input(format!( "Local path '{}' is not beneath store root '{}'", filesystem_path.display(), @@ -83,7 +96,190 @@ impl PrefixedFileStoreProvider { .collect::>>()? .join("/"); - Ok((root.to_path_buf(), Path::parse(relative)?)) + Ok((root, Path::parse(relative)?)) + } + + fn base_paths_for( + &self, + root: &std::path::Path, + base_path: &Path, + ) -> Arc>> { + let mut roots = self.base_paths.lock().expect("base-path lock poisoned"); + let base_paths = roots + .entry(root.to_path_buf()) + .or_insert_with(|| Arc::new(std::sync::RwLock::new(Vec::new()))) + .clone(); + let mut paths = base_paths.write().expect("base-path lock poisoned"); + if !paths.iter().any(|path| base_path.prefix_matches(path)) { + paths.retain(|path| !path.prefix_matches(base_path)); + paths.push(base_path.clone()); + } + drop(paths); + base_paths + } +} + +/// 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, + root: std::path::PathBuf, + base_paths: Arc>>, + absolute_alias: Path, +} + +#[cfg(any(windows, test))] +impl RootedLocalFileSystem { + fn new( + root: std::path::PathBuf, + base_paths: Arc>>, + ) -> LanceResult { + let absolute_alias = Path::from_absolute_path(&root)?; + Ok(Self { + inner: Arc::new(LocalFileSystem::new_with_prefix(&root)?), + root, + base_paths, + absolute_alias, + }) + } + + fn path_from_parts<'a>(parts: impl Iterator>) -> 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(); + }; + let suffix = Self::path_from_parts(suffix); + + // A UNC absolute path loses its server when converted to an object-store + // path, leaving `share/`. Only strip that ambiguous share segment + // when the remainder is inside a base that requested this shared store. + let base_paths = self.base_paths.read().expect("base-path lock poisoned"); + if base_paths + .iter() + .any(|base| base.as_ref().is_empty() || suffix.prefix_matches(base)) + { + suffix + } else { + path.clone() + } + } + + 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 { + self.inner + .put_opts(&self.normalize(location), payload, options) + .await + } + + async fn put_multipart_opts( + &self, + location: &Path, + options: PutMultipartOptions, + ) -> Result> { + self.inner + .put_multipart_opts(&self.normalize(location), options) + .await + } + + async fn get_opts(&self, location: &Path, options: GetOptions) -> Result { + 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>, + ) -> BoxStream<'static, Result> { + 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> { + 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 { + 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 } } @@ -95,27 +291,39 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider { base_path: Url, params: &ObjectStoreParams, ) -> LanceResult { - let (root, _) = Self::root_and_relative_path(&base_path)?; - let store = Arc::new(LocalFileSystem::new_with_prefix(root)?); - let location = Url::parse("file-object-store:///").expect("static URL must be valid"); + let (root, relative_path) = Self::root_and_relative_path(&base_path)?; + let base_paths = self.base_paths_for(&root, &relative_path); + let raw_store: Arc = + Arc::new(RootedLocalFileSystem::new(root, base_paths)?); + let location = Url::parse("lancedb-file:///").expect("static URL must be valid"); let storage_options = StorageOptions::new(params.storage_options().cloned().unwrap_or_default()); - Ok(lance::io::ObjectStore::new( - store, + // 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, - params.block_size, - params.object_store_wrapper.clone(), + 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 { - Self::root_and_relative_path(url).map(|(_, path)| path) + Self::root_and_relative_path(url).map(|(root, path)| { + self.base_paths_for(&root, &path); + path + }) } fn calculate_object_store_prefix( @@ -123,7 +331,8 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider { url: &Url, _storage_options: Option<&std::collections::HashMap>, ) -> LanceResult { - let (root, _) = Self::root_and_relative_path(url)?; + let (root, path) = Self::root_and_relative_path(url)?; + self.base_paths_for(&root, &path); let root = root.canonicalize()?; Ok(format!("file${}", root.display())) } @@ -133,12 +342,29 @@ impl ObjectStoreProvider for PrefixedFileStoreProvider { /// roots by using `LocalFileSystem::new_with_prefix`. #[cfg(any(windows, test))] pub(crate) fn register_windows_file_store(registry: &Arc) { - registry.insert("file", Arc::new(PrefixedFileStoreProvider)); + registry.insert("file", Arc::new(PrefixedFileStoreProvider::default())); } #[cfg(test)] mod prefixed_file_store_test { use super::*; + 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, + ) -> Arc { + self.calls.fetch_add(1, Ordering::Relaxed); + original + } + } #[tokio::test] async fn anchors_new_and_existing_directories_at_a_filesystem_prefix() { @@ -149,7 +375,7 @@ mod prefixed_file_store_test { let table_url = Url::from_directory_path(&table_path).unwrap(); let registry = Arc::new(ObjectStoreRegistry::default()); - registry.insert("file", Arc::new(PrefixedFileStoreProvider)); + registry.insert("file", Arc::new(PrefixedFileStoreProvider::default())); // A new table is relative to its filesystem root. The non-`file` // scheme proves Lance will not bypass this prefixed store. @@ -160,7 +386,9 @@ mod prefixed_file_store_test { ) .await .unwrap(); - assert_eq!(store.scheme(), "file-object-store"); + assert_eq!(store.scheme(), "lancedb-file"); + 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(); @@ -193,6 +421,119 @@ mod prefixed_file_store_test { .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::default())); + 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(), ¶ms) + .await + .unwrap(); + assert_eq!(wrapper.calls.load(Ordering::Relaxed), 1); + + 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); + } + + #[test] + fn normalizes_absolute_drive_and_unc_aliases() { + let tempdir = tempfile::tempdir().unwrap(); + let drive_bases = Arc::new(std::sync::RwLock::new(vec![Path::from("Users/db")])); + let mut drive_store = + RootedLocalFileSystem::new(tempdir.path().to_path_buf(), drive_bases).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 unc_bases = Arc::new(std::sync::RwLock::new(vec![Path::from("share/db")])); + let mut unc_store = + RootedLocalFileSystem::new(tempdir.path().to_path_buf(), unc_bases).unwrap(); + unc_store.absolute_alias = Path::from("share"); + assert_eq!( + unc_store.normalize(&Path::from("share/share/db/table.lance")), + Path::from("share/db/table.lance") + ); + assert_eq!( + unc_store.normalize(&Path::from("share/db/table.lance")), + Path::from("share/db/table.lance") + ); + } + + #[tokio::test] + async fn routes_unc_alias_lifecycle_through_the_prefix() { + let tempdir = tempfile::tempdir().unwrap(); + let base_paths = Arc::new(std::sync::RwLock::new(vec![Path::from("share/db")])); + let mut store = + RootedLocalFileSystem::new(tempdir.path().to_path_buf(), base_paths).unwrap(); + store.absolute_alias = Path::from("share"); + let table = Path::from("share/share/db/test.lance"); + 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::>().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")); + } } #[derive(Debug)]