rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-18 12:08:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve UNC roots for local object stores

Browse Source
This commit is contained in:
Gatefixer
2026-08-05 20:44:50 +00:00
parent c7ea91f3ea
commit b525cbbe6a
2 changed files with 273 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/lancedb/src/database/listing.rs
+95 -22
View File
@@ -24,6 +24,8 @@ use crate::database::ReadConsistency;
use crate::database::namespace::LanceNamespaceDatabase;
use crate::error::{CreateDirSnafu, Error, Result};
use crate::io::object_store::MirroringObjectStoreWrapper;
#[cfg(any(windows, test))]
use crate::io::object_store::register_windows_file_store;
use crate::table::NativeTable;
use crate::utils::validate_table_name;
@@ -362,6 +364,9 @@ impl ListingDatabase {
) -> Result<String> {
match url::Url::parse(uri) {
Ok(url) if url.scheme().len() == 1 && cfg!(windows) => {
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
#[cfg(windows)]
register_windows_file_store(&session.store_registry());
let (object_store, _) = ObjectStore::from_uri_and_params(
session.store_registry(),
uri,
@@ -399,6 +404,14 @@ impl ListingDatabase {
url.set_query(None);
let plain_uri = url.to_string();
#[cfg(windows)]
if url.scheme() == "file" {
Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
path: plain_uri.clone(),
})?;
register_windows_file_store(&session.store_registry());
}
let os_params = ObjectStoreParams {
storage_options_accessor: if storage_options.is_empty() {
None
@@ -424,6 +437,9 @@ impl ListingDatabase {
Ok(plain_uri)
}
Err(_) => {
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
#[cfg(windows)]
register_windows_file_store(&session.store_registry());
let (object_store, _) = ObjectStore::from_uri_and_params(
session.store_registry(),
uri,
@@ -551,6 +567,13 @@ impl ListingDatabase {
.session
.clone()
.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
#[cfg(windows)]
if url.scheme() == "file" {
Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
path: storage_base_uri.clone(),
})?;
register_windows_file_store(&session.store_registry());
}
let os_params = ObjectStoreParams {
storage_options_accessor: if options.storage_options.is_empty() {
None
@@ -626,6 +649,9 @@ impl ListingDatabase {
session: Option<Arc<lance::session::Session>>,
) -> Result<Self> {
let session = session.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
Self::try_create_dir(path).context(CreateDirSnafu { path })?;
#[cfg(windows)]
register_windows_file_store(&session.store_registry());
let (object_store, base_path) = ObjectStore::from_uri_and_params(
session.store_registry(),
path,
@@ -662,20 +688,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)?;
}
@@ -1322,6 +1344,61 @@ mod tests {
(tempdir, db)
}
#[tokio::test]
async fn test_listing_database_with_prefixed_file_store() {
let tempdir = tempdir().unwrap();
let uri = tempdir.path().to_str().unwrap();
let session = Arc::new(lance::session::Session::default());
register_windows_file_store(&session.store_registry());
let 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);
}
#[tokio::test]
async fn test_listing_database_root_ops_do_not_create_manifest() {
let tempdir = tempdir().unwrap();
@@ -2376,18 +2453,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/io/object_store.rs
+178 -1
View File
@@ -1,7 +1,7 @@
// 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};
@@ -15,9 +15,186 @@ use object_store::{
use async_trait::async_trait;
#[cfg(any(windows, test))]
use lance_core::{Error as LanceError, Result as LanceResult};
#[cfg(any(windows, test))]
use lance_io::object_store::{
DEFAULT_LOCAL_IO_PARALLELISM, ObjectStoreParams, ObjectStoreProvider, ObjectStoreRegistry,
StorageOptions,
};
#[cfg(any(windows, test))]
use object_store::local::LocalFileSystem;
#[cfg(any(windows, test))]
use url::Url;
#[cfg(test)]
pub mod io_tracking;
/// 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`.
///
/// The returned Lance store deliberately uses the `file-object-store` 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;
#[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 root = filesystem_path.ancestors().last().ok_or_else(|| {
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.to_path_buf(), Path::parse(relative)?))
}
}
#[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 store = Arc::new(LocalFileSystem::new_with_prefix(root)?);
let location = Url::parse("file-object-store:///").expect("static URL must be valid");
let storage_options =
StorageOptions::new(params.storage_options().cloned().unwrap_or_default());
Ok(lance::io::ObjectStore::new(
store,
location,
params.block_size,
params.object_store_wrapper.clone(),
false,
false,
DEFAULT_LOCAL_IO_PARALLELISM,
storage_options.download_retry_count(),
params.storage_options(),
))
}
fn extract_path(&self, url: &Url) -> LanceResult<Path> {
Self::root_and_relative_path(url).map(|(_, path)| path)
}
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()?;
Ok(format!("file${}", root.display()))
}
}
/// Replace Lance's default Windows file provider with one that preserves UNC
/// roots by using `LocalFileSystem::new_with_prefix`.
#[cfg(any(windows, test))]
pub(crate) fn register_windows_file_store(registry: &Arc<ObjectStoreRegistry>) {
registry.insert("file", Arc::new(PrefixedFileStoreProvider));
}
#[cfg(test)]
mod prefixed_file_store_test {
use super::*;
#[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));
// A new table is relative to its filesystem root. The non-`file`
// scheme proves Lance will not bypass this prefixed store.
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-object-store");
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");
}
}
#[derive(Debug)]
struct MirroringObjectStore {
primary: Arc<dyn ObjectStore>,