mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-12 00:02:21 +00:00
Compare commits
12
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9ba03810e7 | ||
|
|
b4053059bf | ||
|
|
7a7b7a3941 | ||
|
|
d52940cdab | ||
|
|
fafc297675 | ||
|
|
82ebddbc10 | ||
|
|
f114bba752 | ||
|
|
78024a30ce | ||
|
|
72500192e6 | ||
|
|
8165857a50 | ||
|
|
5fa98b9af8 | ||
|
|
b525cbbe6a |
@@ -716,20 +716,9 @@ class LanceDBConnection(DBConnection):
|
|||||||
if not isinstance(uri, Path):
|
if not isinstance(uri, Path):
|
||||||
scheme = get_uri_scheme(uri)
|
scheme = get_uri_scheme(uri)
|
||||||
is_local = isinstance(uri, Path) or scheme == "file"
|
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):
|
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 = Path(uri)
|
||||||
uri = uri.expanduser().absolute()
|
uri = uri.expanduser().absolute()
|
||||||
Path(uri).mkdir(parents=True, exist_ok=True)
|
Path(uri).mkdir(parents=True, exist_ok=True)
|
||||||
|
|||||||
@@ -919,17 +919,7 @@ class LanceNamespaceDBConnection(DBConnection):
|
|||||||
The namespace client for this connection.
|
The namespace client for this connection.
|
||||||
"""
|
"""
|
||||||
if self._namespace_client is None:
|
if self._namespace_client is None:
|
||||||
if (
|
self._namespace_client = LOOP.run(self._inner.namespace_client())
|
||||||
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
|
|
||||||
)
|
|
||||||
return self._namespace_client
|
return self._namespace_client
|
||||||
|
|
||||||
|
|
||||||
@@ -1370,17 +1360,7 @@ class AsyncLanceNamespaceDBConnection:
|
|||||||
The namespace client for this connection.
|
The namespace client for this connection.
|
||||||
"""
|
"""
|
||||||
if self._namespace_client is None:
|
if self._namespace_client is None:
|
||||||
if (
|
self._namespace_client = await self._inner.namespace_client()
|
||||||
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
|
|
||||||
)
|
|
||||||
return self._namespace_client
|
return self._namespace_client
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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})"
|
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):
|
def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch):
|
||||||
from lancedb.background_loop import LOOP
|
from lancedb.background_loop import LOOP
|
||||||
from lancedb.db import LanceDBConnection
|
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)"
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_close(mem_db_async: lancedb.AsyncConnection):
|
async def test_close(mem_db_async: lancedb.AsyncConnection):
|
||||||
assert mem_db_async.is_open()
|
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)
|
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")
|
@pytest.mark.skipif(sys.platform == "win32", reason="Namespace client issues")
|
||||||
def test_namespace_client_native_storage(tmp_path):
|
def test_namespace_client_native_storage(tmp_path):
|
||||||
"""Test namespace_client() returns DirectoryNamespace for native storage."""
|
"""Test namespace_client() returns DirectoryNamespace for native storage."""
|
||||||
|
|||||||
@@ -60,6 +60,11 @@ class _NamespaceClient:
|
|||||||
return _ipc_file()
|
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:
|
def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
|
||||||
table = LanceTable.__new__(LanceTable)
|
table = LanceTable.__new__(LanceTable)
|
||||||
table._table = _FailingSyncInner()
|
table._table = _FailingSyncInner()
|
||||||
@@ -138,6 +143,24 @@ class TestNamespaceConnection:
|
|||||||
db.drop_namespace(["test_ns"])
|
db.drop_namespace(["test_ns"])
|
||||||
assert "test_ns" not in db.list_namespaces().namespaces
|
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):
|
def test_create_table_through_namespace(self):
|
||||||
"""Test creating a table through namespace."""
|
"""Test creating a table through namespace."""
|
||||||
db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
|
db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
|
||||||
@@ -639,6 +662,24 @@ class TestAsyncNamespaceConnection:
|
|||||||
await db.drop_namespace(["test_ns"])
|
await db.drop_namespace(["test_ns"])
|
||||||
assert "test_ns" not in (await db.list_namespaces()).namespaces
|
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):
|
async def test_async_namespace_client_is_lazy(self):
|
||||||
"""namespace_client() should still return the backing client on demand."""
|
"""namespace_client() should still return the backing client on demand."""
|
||||||
pytest.importorskip("lance")
|
pytest.importorskip("lance")
|
||||||
|
|||||||
@@ -8,14 +8,23 @@ use std::fs::create_dir_all;
|
|||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::{collections::HashMap, sync::Arc};
|
use std::{collections::HashMap, sync::Arc};
|
||||||
|
|
||||||
use lance::dataset::refs::Ref;
|
use lance::dataset::transaction::{Operation, Transaction};
|
||||||
use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
|
use lance::dataset::{
|
||||||
|
CommitBuilder, ReadParams, WriteDestination, WriteMode, builder::DatasetBuilder,
|
||||||
|
};
|
||||||
use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
|
use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
|
||||||
use lance_datafusion::utils::StreamingWriteSource;
|
use lance_datafusion::utils::StreamingWriteSource;
|
||||||
use lance_file::version::LanceFileVersion;
|
use lance_file::version::LanceFileVersion;
|
||||||
use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
|
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::local::LocalFileSystem;
|
||||||
|
use object_store::path::Path as ObjectPath;
|
||||||
use snafu::ResultExt;
|
use snafu::ResultExt;
|
||||||
|
|
||||||
use crate::blob::{ensure_blob_storage_version, has_blob_columns};
|
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::ReadConsistency;
|
||||||
use crate::database::namespace::LanceNamespaceDatabase;
|
use crate::database::namespace::LanceNamespaceDatabase;
|
||||||
use crate::error::{CreateDirSnafu, Error, Result};
|
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::table::NativeTable;
|
||||||
use crate::utils::validate_table_name;
|
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_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";
|
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
|
/// Controls how new tables should be created
|
||||||
#[derive(Clone, Debug, Default)]
|
#[derive(Clone, Debug, Default)]
|
||||||
pub struct NewTableConfig {
|
pub struct NewTableConfig {
|
||||||
@@ -355,20 +461,45 @@ impl ListingDatabase {
|
|||||||
url.to_string()
|
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(
|
async fn prepare_namespace_root(
|
||||||
uri: &str,
|
uri: &str,
|
||||||
storage_options: &HashMap<String, String>,
|
storage_options: &HashMap<String, String>,
|
||||||
session: Arc<lance::session::Session>,
|
session: Arc<lance::session::Session>,
|
||||||
|
prepare_native_directory: bool,
|
||||||
) -> Result<String> {
|
) -> Result<String> {
|
||||||
match url::Url::parse(uri) {
|
match url::Url::parse(uri) {
|
||||||
Ok(url) if url.scheme().len() == 1 && cfg!(windows) => {
|
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(
|
let (object_store, _) = ObjectStore::from_uri_and_params(
|
||||||
session.store_registry(),
|
session.store_registry(),
|
||||||
uri,
|
uri,
|
||||||
&ObjectStoreParams::default(),
|
&ObjectStoreParams::default(),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
if object_store.is_local() {
|
if object_store.is_local() && prepare_native_directory {
|
||||||
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
|
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
|
||||||
}
|
}
|
||||||
Ok(uri.to_string())
|
Ok(uri.to_string())
|
||||||
@@ -399,6 +530,13 @@ impl ListingDatabase {
|
|||||||
url.set_query(None);
|
url.set_query(None);
|
||||||
let plain_uri = url.to_string();
|
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 {
|
let os_params = ObjectStoreParams {
|
||||||
storage_options_accessor: if storage_options.is_empty() {
|
storage_options_accessor: if storage_options.is_empty() {
|
||||||
None
|
None
|
||||||
@@ -415,7 +553,7 @@ impl ListingDatabase {
|
|||||||
&os_params,
|
&os_params,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
if object_store.is_local() {
|
if object_store.is_local() && prepare_native_directory {
|
||||||
Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
|
Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
|
||||||
path: plain_uri.clone(),
|
path: plain_uri.clone(),
|
||||||
})?;
|
})?;
|
||||||
@@ -424,13 +562,16 @@ impl ListingDatabase {
|
|||||||
Ok(plain_uri)
|
Ok(plain_uri)
|
||||||
}
|
}
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
|
if prepare_native_directory {
|
||||||
|
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
|
||||||
|
}
|
||||||
let (object_store, _) = ObjectStore::from_uri_and_params(
|
let (object_store, _) = ObjectStore::from_uri_and_params(
|
||||||
session.store_registry(),
|
session.store_registry(),
|
||||||
uri,
|
uri,
|
||||||
&ObjectStoreParams::default(),
|
&ObjectStoreParams::default(),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
if object_store.is_local() {
|
if object_store.is_local() && prepare_native_directory {
|
||||||
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
|
Self::try_create_dir(uri).context(CreateDirSnafu { path: uri })?;
|
||||||
}
|
}
|
||||||
Ok(uri.to_string())
|
Ok(uri.to_string())
|
||||||
@@ -442,13 +583,14 @@ impl ListingDatabase {
|
|||||||
request: &ConnectRequest,
|
request: &ConnectRequest,
|
||||||
) -> Result<LanceNamespaceDatabase> {
|
) -> Result<LanceNamespaceDatabase> {
|
||||||
let options = ListingDatabaseOptions::parse_from_map(&request.options)?;
|
let options = ListingDatabaseOptions::parse_from_map(&request.options)?;
|
||||||
let session = request
|
let (session, owns_session) = session_or_default(request.session.clone());
|
||||||
.session
|
let namespace_root = Self::prepare_namespace_root(
|
||||||
.clone()
|
&request.uri,
|
||||||
.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
|
&options.storage_options,
|
||||||
let namespace_root =
|
session.clone(),
|
||||||
Self::prepare_namespace_root(&request.uri, &options.storage_options, session.clone())
|
owns_session || !cfg!(windows),
|
||||||
.await?;
|
)
|
||||||
|
.await?;
|
||||||
let ns_properties = Self::build_manifest_enabled_namespace_client_properties(
|
let ns_properties = Self::build_manifest_enabled_namespace_client_properties(
|
||||||
&namespace_root,
|
&namespace_root,
|
||||||
&options.storage_options,
|
&options.storage_options,
|
||||||
@@ -547,10 +689,14 @@ impl ListingDatabase {
|
|||||||
url.to_string()
|
url.to_string()
|
||||||
};
|
};
|
||||||
|
|
||||||
let session = request
|
let (session, owns_session) = session_or_default(request.session.clone());
|
||||||
.session
|
let prepare_native_directory = owns_session || !cfg!(windows);
|
||||||
.clone()
|
#[cfg(windows)]
|
||||||
.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
|
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 {
|
let os_params = ObjectStoreParams {
|
||||||
storage_options_accessor: if options.storage_options.is_empty() {
|
storage_options_accessor: if options.storage_options.is_empty() {
|
||||||
None
|
None
|
||||||
@@ -567,7 +713,7 @@ impl ListingDatabase {
|
|||||||
&os_params,
|
&os_params,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
if object_store.is_local() {
|
if object_store.is_local() && prepare_native_directory {
|
||||||
Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
|
Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
|
||||||
path: storage_base_uri.clone(),
|
path: storage_base_uri.clone(),
|
||||||
})?;
|
})?;
|
||||||
@@ -625,14 +771,18 @@ impl ListingDatabase {
|
|||||||
namespace_client_properties: HashMap<String, String>,
|
namespace_client_properties: HashMap<String, String>,
|
||||||
session: Option<Arc<lance::session::Session>>,
|
session: Option<Arc<lance::session::Session>>,
|
||||||
) -> Result<Self> {
|
) -> 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(
|
let (object_store, base_path) = ObjectStore::from_uri_and_params(
|
||||||
session.store_registry(),
|
session.store_registry(),
|
||||||
path,
|
path,
|
||||||
&ObjectStoreParams::default(),
|
&ObjectStoreParams::default(),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
if object_store.is_local() {
|
if object_store.is_local() && prepare_native_directory {
|
||||||
Self::try_create_dir(path).context(CreateDirSnafu { path })?;
|
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
|
/// Try to create a local directory to store the lancedb dataset
|
||||||
fn try_create_dir(path: &str) -> core::result::Result<(), std::io::Error> {
|
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
|
let filesystem_path = match url::Url::parse(path) {
|
||||||
// Note: file:///path becomes file:/path after url.to_string(), so we need to handle both
|
Ok(url) if url.scheme() == "file" => url.to_file_path().map_err(|_| {
|
||||||
let fs_path = if let Some(stripped) = path.strip_prefix("file://") {
|
std::io::Error::new(
|
||||||
// file:///path or file://host/path format
|
std::io::ErrorKind::InvalidInput,
|
||||||
stripped
|
format!("Unable to convert URL '{url}' to a local path"),
|
||||||
} 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
|
_ => Path::new(path).to_path_buf(),
|
||||||
stripped
|
|
||||||
} else {
|
|
||||||
path
|
|
||||||
};
|
};
|
||||||
|
let path = filesystem_path.as_path();
|
||||||
let path = Path::new(fs_path);
|
|
||||||
if !path.try_exists()? {
|
if !path.try_exists()? {
|
||||||
create_dir_all(path)?;
|
create_dir_all(path)?;
|
||||||
}
|
}
|
||||||
@@ -733,7 +879,10 @@ impl ListingDatabase {
|
|||||||
if let Some(query_string) = &self.query_string {
|
if let Some(query_string) = &self.query_string {
|
||||||
uri.push_str(&format!("?{}", 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 {
|
for name in names {
|
||||||
let dir_name = format!("{}.{}", name, LANCE_EXTENSION);
|
let dir_name = format!("{}.{}", name, LANCE_EXTENSION);
|
||||||
let full_path = self.base_path.clone().join(dir_name.clone());
|
let full_path = self.base_path.clone().join(dir_name.clone());
|
||||||
@@ -866,6 +1015,9 @@ impl ListingDatabase {
|
|||||||
write_params.mode = WriteMode::Overwrite;
|
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.session = Some(self.session.clone());
|
||||||
|
|
||||||
write_params
|
write_params
|
||||||
@@ -1112,30 +1264,90 @@ impl Database for ListingDatabase {
|
|||||||
},
|
},
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
|
let source_commit_handler = Self::rooted_commit_handler_for_uri(&request.source_uri);
|
||||||
let read_params = ReadParams {
|
let read_params = ReadParams {
|
||||||
store_options: Some(storage_params.clone()),
|
store_options: Some(storage_params.clone()),
|
||||||
session: Some(self.session.clone()),
|
session: Some(self.session.clone()),
|
||||||
|
commit_handler: source_commit_handler.clone(),
|
||||||
..Default::default()
|
..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())
|
.with_read_params(read_params.clone())
|
||||||
.load()
|
.load()
|
||||||
.await
|
.await
|
||||||
.map_err(|e| -> Error { e.into() })?;
|
.map_err(|e| -> Error { e.into() })?;
|
||||||
|
|
||||||
let version_ref = match (request.source_version, request.source_tag) {
|
let (ref_name, version_number) = match (request.source_version, request.source_tag) {
|
||||||
(Some(v), None) => Ok(Ref::Version(None, Some(v))),
|
(Some(version), None) => Ok((None, version)),
|
||||||
(None, Some(tag)) => Ok(Ref::Tag(tag)),
|
(None, Some(tag)) => {
|
||||||
(None, None) => Ok(Ref::Version(None, Some(source_dataset.version().version))),
|
let tag = source_dataset.tags().get(&tag).await?;
|
||||||
|
Ok((tag.branch, tag.version))
|
||||||
|
}
|
||||||
|
(None, None) => Ok((None, source_dataset.version().version)),
|
||||||
_ => Err(Error::InvalidInput {
|
_ => Err(Error::InvalidInput {
|
||||||
message: "Cannot specify both source_version and source_tag".to_string(),
|
message: "Cannot specify both source_version and source_tag".to_string(),
|
||||||
}),
|
}),
|
||||||
}?;
|
}?;
|
||||||
|
|
||||||
let target_uri = self.table_uri(&request.target_table_name)?;
|
let target_uri = self.table_uri(&request.target_table_name)?;
|
||||||
source_dataset
|
let source_location = source_dataset
|
||||||
.shallow_clone(&target_uri, version_ref, Some(storage_params))
|
.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
|
.await
|
||||||
.map_err(|e| -> Error { e.into() })?;
|
.map_err(|e| -> Error { e.into() })?;
|
||||||
|
|
||||||
@@ -1144,7 +1356,7 @@ impl Database for ListingDatabase {
|
|||||||
&request.target_table_name,
|
&request.target_table_name,
|
||||||
request.target_namespace_path,
|
request.target_namespace_path,
|
||||||
self.store_wrapper.clone(),
|
self.store_wrapper.clone(),
|
||||||
None,
|
Some(target_read_params),
|
||||||
self.read_consistency_interval,
|
self.read_consistency_interval,
|
||||||
request.namespace_client,
|
request.namespace_client,
|
||||||
HashSet::new(), // listing database doesn't support server-side queries
|
HashSet::new(), // listing database doesn't support server-side queries
|
||||||
@@ -1210,6 +1422,9 @@ impl Database for ListingDatabase {
|
|||||||
}
|
}
|
||||||
default_params
|
default_params
|
||||||
});
|
});
|
||||||
|
if read_params.commit_handler.is_none() {
|
||||||
|
read_params.commit_handler = self.rooted_commit_handler();
|
||||||
|
}
|
||||||
read_params.session(self.session.clone());
|
read_params.session(self.session.clone());
|
||||||
|
|
||||||
let native_table = Arc::new(
|
let native_table = Arc::new(
|
||||||
@@ -1307,6 +1522,60 @@ mod tests {
|
|||||||
use tokio::sync::Barrier;
|
use tokio::sync::Barrier;
|
||||||
use tokio::time::timeout;
|
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) {
|
async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
|
||||||
let tempdir = tempdir().unwrap();
|
let tempdir = tempdir().unwrap();
|
||||||
let uri = tempdir.path().to_str().unwrap();
|
let uri = tempdir.path().to_str().unwrap();
|
||||||
@@ -1329,6 +1598,199 @@ mod tests {
|
|||||||
(tempdir, db)
|
(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(
|
||||||
|
®istry.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 {
|
struct BarrierScannable {
|
||||||
batch: RecordBatch,
|
batch: RecordBatch,
|
||||||
barrier: Arc<Barrier>,
|
barrier: Arc<Barrier>,
|
||||||
@@ -1749,6 +2211,173 @@ mod tests {
|
|||||||
assert_eq!(source_count, 3);
|
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]
|
#[tokio::test]
|
||||||
async fn test_clone_table_with_storage_options() {
|
async fn test_clone_table_with_storage_options() {
|
||||||
let tempdir = tempdir().unwrap();
|
let tempdir = tempdir().unwrap();
|
||||||
@@ -2653,18 +3282,14 @@ mod tests {
|
|||||||
/// as `path=<base table>` + `query=/_mem_wal/...`, causing
|
/// as `path=<base table>` + `query=/_mem_wal/...`, causing
|
||||||
/// `Dataset::write` to find the base table dataset and falsely
|
/// `Dataset::write` to find the base table dataset and falsely
|
||||||
/// report `Dataset already exists`.
|
/// 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]
|
#[tokio::test]
|
||||||
async fn test_table_uri_url_path_has_no_trailing_question_mark() {
|
async fn test_table_uri_url_path_has_no_trailing_question_mark() {
|
||||||
let tempdir = tempdir().unwrap();
|
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 {
|
let request = ConnectRequest {
|
||||||
uri: uri.clone(),
|
uri: uri.clone(),
|
||||||
|
|||||||
@@ -59,6 +59,46 @@ fn is_table_already_exists_namespace_error(err: &lance::Error) -> bool {
|
|||||||
/// via the `delimiter` property.
|
/// via the `delimiter` property.
|
||||||
const DEFAULT_NAMESPACE_DELIMITER: &str = "$";
|
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
|
/// A database implementation that uses lance-namespace for table management
|
||||||
pub struct LanceNamespaceDatabase {
|
pub struct LanceNamespaceDatabase {
|
||||||
namespace: Arc<dyn LanceNamespace>,
|
namespace: Arc<dyn LanceNamespace>,
|
||||||
@@ -92,6 +132,17 @@ fn resolve_delimiter(ns_properties: &HashMap<String, String>) -> String {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl LanceNamespaceDatabase {
|
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(
|
pub fn from_namespace_client(
|
||||||
namespace_client: Arc<dyn LanceNamespace>,
|
namespace_client: Arc<dyn LanceNamespace>,
|
||||||
namespace_client_impl: String,
|
namespace_client_impl: String,
|
||||||
@@ -153,6 +204,25 @@ impl LanceNamespaceDatabase {
|
|||||||
pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
|
pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
|
||||||
new_table_config: NewTableConfig,
|
new_table_config: NewTableConfig,
|
||||||
) -> Result<Self> {
|
) -> 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);
|
let mut builder = ConnectBuilder::new(ns_impl);
|
||||||
for (key, value) in ns_properties.clone() {
|
for (key, value) in ns_properties.clone() {
|
||||||
builder = builder.property(key, value);
|
builder = builder.property(key, value);
|
||||||
@@ -310,6 +380,7 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
&self,
|
&self,
|
||||||
request: ListNamespacesRequest,
|
request: ListNamespacesRequest,
|
||||||
) -> Result<ListNamespacesResponse> {
|
) -> Result<ListNamespacesResponse> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.list_namespaces(request).await?)
|
Ok(self.namespace.list_namespaces(request).await?)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -317,10 +388,12 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
&self,
|
&self,
|
||||||
request: CreateNamespaceRequest,
|
request: CreateNamespaceRequest,
|
||||||
) -> Result<CreateNamespaceResponse> {
|
) -> Result<CreateNamespaceResponse> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.create_namespace(request).await?)
|
Ok(self.namespace.create_namespace(request).await?)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
|
async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.drop_namespace(request).await?)
|
Ok(self.namespace.drop_namespace(request).await?)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -328,10 +401,12 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
&self,
|
&self,
|
||||||
request: DescribeNamespaceRequest,
|
request: DescribeNamespaceRequest,
|
||||||
) -> Result<DescribeNamespaceResponse> {
|
) -> Result<DescribeNamespaceResponse> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.describe_namespace(request).await?)
|
Ok(self.namespace.describe_namespace(request).await?)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
|
async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
let ns_request = ListTablesRequest {
|
let ns_request = ListTablesRequest {
|
||||||
id: Some(request.namespace_path),
|
id: Some(request.namespace_path),
|
||||||
page_token: request.start_after,
|
page_token: request.start_after,
|
||||||
@@ -345,10 +420,12 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.list_tables(request).await?)
|
Ok(self.namespace.list_tables(request).await?)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn create_table(&self, request: DbCreateTableRequest) -> Result<Arc<dyn BaseTable>> {
|
async fn create_table(&self, request: DbCreateTableRequest) -> Result<Arc<dyn BaseTable>> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
let mut table_id = request.namespace_path.clone();
|
let mut table_id = request.namespace_path.clone();
|
||||||
table_id.push(request.name.clone());
|
table_id.push(request.name.clone());
|
||||||
let mut existing_table = None;
|
let mut existing_table = None;
|
||||||
@@ -518,6 +595,7 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
|
async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
let native_table = NativeTable::open_from_namespace(
|
let native_table = NativeTable::open_from_namespace(
|
||||||
self.namespace.clone(),
|
self.namespace.clone(),
|
||||||
&request.name,
|
&request.name,
|
||||||
@@ -547,6 +625,7 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
cur_namespace_path: &[String],
|
cur_namespace_path: &[String],
|
||||||
new_namespace_path: &[String],
|
new_namespace_path: &[String],
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
let mut cur_table_id = cur_namespace_path.to_vec();
|
let mut cur_table_id = cur_namespace_path.to_vec();
|
||||||
cur_table_id.push(cur_name.to_string());
|
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<()> {
|
async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
let mut table_id = namespace_path.to_vec();
|
let mut table_id = namespace_path.to_vec();
|
||||||
table_id.push(name.to_string());
|
table_id.push(name.to_string());
|
||||||
|
|
||||||
@@ -612,10 +692,12 @@ impl Database for LanceNamespaceDatabase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn namespace_client(&self) -> Result<Arc<dyn LanceNamespace>> {
|
async fn namespace_client(&self) -> Result<Arc<dyn LanceNamespace>> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok(self.namespace.clone())
|
Ok(self.namespace.clone())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn namespace_client_config(&self) -> Result<(String, HashMap<String, String>)> {
|
async fn namespace_client_config(&self) -> Result<(String, HashMap<String, String>)> {
|
||||||
|
self.ensure_storage_supported()?;
|
||||||
Ok((self.ns_impl.clone(), self.ns_properties.clone()))
|
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()
|
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]
|
#[tokio::test]
|
||||||
async fn test_namespace_connection_simple() {
|
async fn test_namespace_connection_simple() {
|
||||||
// Test that namespace connections work with simple connect_namespace(impl_type, properties)
|
// Test that namespace connections work with simple connect_namespace(impl_type, properties)
|
||||||
|
|||||||
@@ -1,12 +1,22 @@
|
|||||||
// SPDX-License-Identifier: Apache-2.0
|
// SPDX-License-Identifier: Apache-2.0
|
||||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
// 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};
|
use std::{fmt::Formatter, sync::Arc};
|
||||||
|
|
||||||
|
#[cfg(any(windows, test))]
|
||||||
|
use futures::TryStreamExt;
|
||||||
use futures::{StreamExt, TryFutureExt, stream::BoxStream};
|
use futures::{StreamExt, TryFutureExt, stream::BoxStream};
|
||||||
use lance::io::WrappingObjectStore;
|
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::{
|
use object_store::{
|
||||||
CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
|
CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
|
||||||
ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
|
ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
|
||||||
@@ -15,9 +25,797 @@ use object_store::{
|
|||||||
|
|
||||||
use async_trait::async_trait;
|
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)]
|
#[cfg(test)]
|
||||||
pub mod io_tracking;
|
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(), ¶ms)
|
||||||
|
.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(), ¶ms)
|
||||||
|
.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)]
|
#[derive(Debug)]
|
||||||
struct MirroringObjectStore {
|
struct MirroringObjectStore {
|
||||||
primary: Arc<dyn ObjectStore>,
|
primary: Arc<dyn ObjectStore>,
|
||||||
|
|||||||
Reference in New Issue
Block a user