From 663fa1538792094e4257a7f3c79071061f5e0db8 Mon Sep 17 00:00:00 2001 From: fanng <“fanng@apache.org”> Date: Fri, 19 Jun 2026 18:01:54 +0800 Subject: [PATCH 1/2] feat: add lance namespace read write support --- README.md | 23 +++ daft_lance/__init__.py | 52 +++++ daft_lance/_lance.py | 15 +- daft_lance/lance_data_sink.py | 72 +++++-- daft_lance/lance_scan.py | 21 +- daft_lance/namespace.py | 217 ++++++++++++++++++++ daft_lance/utils.py | 40 +++- tests/io/lancedb/test_namespace.py | 84 ++++++++ tests/io/lancedb/test_namespace_rest_e2e.py | 94 +++++++++ 9 files changed, 597 insertions(+), 21 deletions(-) create mode 100644 daft_lance/namespace.py create mode 100644 tests/io/lancedb/test_namespace.py create mode 100644 tests/io/lancedb/test_namespace_rest_e2e.py diff --git a/README.md b/README.md index 11e7c3f..a7133b6 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,29 @@ from daft_lance import merge_columns_df merge_columns_df(df, "s3://bucket/my_dataset") ``` +### Namespace Tables + +```python +import daft +import daft_lance # installs daft.read_lance / DataFrame.write_lance namespace support + +table_id = ["my_table"] +namespace_properties = {"root": "/tmp/lance_tables"} + +df.write_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + mode="create", +).collect() + +df = daft.read_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, +) +``` + ## Migration The migration only requires replacing `daft.io.lance` with `daft_lance`. diff --git a/daft_lance/__init__.py b/daft_lance/__init__.py index cb07862..8a1ecba 100644 --- a/daft_lance/__init__.py +++ b/daft_lance/__init__.py @@ -1,3 +1,7 @@ +from __future__ import annotations + +from typing import Any, Literal + try: import daft except ImportError: @@ -11,6 +15,54 @@ merge_columns_df, read_lance, ) +from .lance_data_sink import LanceDataSink + + +def _patch_daft_lance_api() -> None: + """Expose the daft-lance implementation through Daft's convenience APIs.""" + daft.read_lance = read_lance # type: ignore[assignment] + + from daft.dataframe import DataFrame + + original_write_lance = getattr(DataFrame, "write_lance") + if getattr(original_write_lance, "_daft_lance_namespace_patch", False): + return + + def write_lance( + self: DataFrame, + uri: Any = None, + mode: Literal["create", "append", "overwrite", "merge"] = "create", + io_config: Any | None = None, + schema: Any | None = None, + left_on: str | None = None, + right_on: str | None = None, + **kwargs: Any, + ) -> DataFrame: + if mode == "merge": + if any(k in kwargs for k in ("namespace_impl", "namespace_properties", "table_id")): + raise ValueError("write_lance(mode='merge') does not support namespace parameters yet.") + return original_write_lance( + self, + uri, + mode=mode, + io_config=io_config, + schema=schema, + left_on=left_on, + right_on=right_on, + **kwargs, + ) + + if schema is None: + schema = self.schema() + sanitized_kwargs = {k: v for k, v in kwargs.items() if k not in ("left_on", "right_on")} + sink = LanceDataSink(uri, schema, mode, io_config, **sanitized_kwargs) + return self.write_sink(sink) + + write_lance._daft_lance_namespace_patch = True # type: ignore[attr-defined] + DataFrame.write_lance = write_lance # type: ignore[assignment] + + +_patch_daft_lance_api() __all__ = [ "compact_files", diff --git a/daft_lance/_lance.py b/daft_lance/_lance.py index 21b7c67..7b5f55c 100644 --- a/daft_lance/_lance.py +++ b/daft_lance/_lance.py @@ -30,7 +30,7 @@ @PublicAPI def read_lance( - uri: str | pathlib.Path, + uri: str | pathlib.Path | None = None, io_config: IOConfig | None = None, version: str | int | None = None, asof: str | None = None, @@ -42,6 +42,10 @@ def read_lance( fragment_group_size: int | None = None, include_fragment_id: bool | None = None, checkpoint: CheckpointConfig | None = None, + *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, ) -> DataFrame: """Create a DataFrame from a LanceDB table. @@ -123,8 +127,8 @@ def read_lance( >>> df = daft.read_lance("s3://daft-oss-public-data/lance/words-test-dataset", io_config=io_config) >>> df.show() """ - uri_str = str(uri) - if uri_str.startswith("rest://"): + uri_str = str(uri) if uri is not None else None + if uri_str is not None and uri_str.startswith("rest://"): raise ValueError( "rest:// Lance URIs are no longer supported by daft.read_lance. " "The previous REST-namespace integration did not match the real " @@ -133,11 +137,14 @@ def read_lance( ) io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config - storage_options = io_config_to_storage_options(io_config, uri_str) + storage_options = io_config_to_storage_options(io_config, uri_str) if uri_str is not None else None ds = construct_lance_dataset( uri_str, storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, version=version, asof=asof, block_size=block_size, diff --git a/daft_lance/lance_data_sink.py b/daft_lance/lance_data_sink.py index 62c36bb..2ec69a7 100644 --- a/daft_lance/lance_data_sink.py +++ b/daft_lance/lance_data_sink.py @@ -26,6 +26,13 @@ detect_blob_v2_columns, resolve_storage_version, ) +from daft_lance.namespace import ( + get_namespace_kwargs, + get_write_fragments_kwargs, + merge_storage_options, + resolve_namespace_table, + validate_uri_or_namespace, +) if TYPE_CHECKING: from collections.abc import Iterator @@ -40,11 +47,14 @@ class LanceDataSink(DataSink[list[FragmentMetadata]]): def __init__( self, - uri: str | pathlib.Path, + uri: str | pathlib.Path | None, schema: Schema | pa.Schema, mode: Literal["create", "append", "overwrite"] = "create", io_config: IOConfig | None = None, *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, blob_columns: list[str] | None = None, max_rows_per_file: int = 1024 * 1024, max_rows_per_group: int = 1024, @@ -57,17 +67,26 @@ def __init__( compact_after_write: bool = True, ) -> None: self._reject_unsupported_modes(mode, use_legacy_format) - if not isinstance(uri, (str, pathlib.Path)): + validate_uri_or_namespace(uri, namespace_impl, table_id) + if uri is not None and not isinstance(uri, (str, pathlib.Path)): raise TypeError(f"Expected URI to be str or pathlib.Path, got {type(uri)}") - self._table_uri = str(uri) self._mode = mode + self._namespace_impl = namespace_impl + self._namespace_properties = namespace_properties + self._table_id = table_id self._io_config = get_context().daft_planning_config.default_io_config if io_config is None else io_config - self._storage_options = ( - storage_options - if storage_options is not None - else io_config_to_storage_options(self._io_config, self._table_uri) + base_storage_options = ( + ( + storage_options + if storage_options is not None + else io_config_to_storage_options(self._io_config, str(uri)) + ) + if uri is not None + else storage_options ) + self._table_uri, namespace_storage_options = self._resolve_table_uri(uri) + self._storage_options = merge_storage_options(base_storage_options, namespace_storage_options) self._init_lance_knobs( max_rows_per_file=max_rows_per_file, max_rows_per_group=max_rows_per_group, @@ -109,6 +128,28 @@ def __init__( ] ) + @property + def _namespace_kwargs(self) -> dict[str, object]: + return get_namespace_kwargs(self._namespace_impl, self._namespace_properties, self._table_id) + + @property + def _dataset_uri_arg(self) -> str | None: + return None if self._namespace_impl is not None and self._table_id is not None else self._table_uri + + def _resolve_table_uri(self, uri: str | pathlib.Path | None) -> tuple[str, dict[str, str] | None]: + if uri is not None: + return str(uri), None + mode = "create" if self._mode == "create" else "overwrite" if self._mode == "overwrite" else "read" + resolved_uri, namespace_storage_options = resolve_namespace_table( + namespace_impl=self._namespace_impl, + namespace_properties=self._namespace_properties, + table_id=self._table_id, + mode=mode, + ) + if resolved_uri is None: + raise ValueError("Unable to resolve Lance dataset URI from namespace.") + return resolved_uri, namespace_storage_options + @staticmethod def _reject_unsupported_modes( mode: Literal["create", "append", "overwrite"], use_legacy_format: bool | None @@ -167,7 +208,9 @@ def _absorb_existing_dataset(self) -> lance.LanceDataset | None: """ dataset: lance.LanceDataset | None try: - dataset = lance.dataset(self._table_uri, storage_options=self._storage_options) + dataset = lance.dataset( + self._dataset_uri_arg, storage_options=self._storage_options, **self._namespace_kwargs + ) except (ValueError, FileNotFoundError, OSError) as e: # Pinned to the Rust message format; lance has no typed exception. See test_lance_message_format_unchanged. if "was not found" in str(e): @@ -230,6 +273,7 @@ def _write_arrow_table(self, table: pa.Table) -> WriteResult[list[FragmentMetada data_storage_version=self._data_storage_version, use_legacy_format=self._use_legacy_format, enable_stable_row_ids=self._enable_stable_row_ids, + **get_write_fragments_kwargs(self._namespace_impl, self._namespace_properties, self._table_id), ) # Sum on-disk sizes from fragment metadata. Lance Blob V2 sidecar .blob # files are not tracked in FragmentMetadata.files (out of scope here). @@ -240,7 +284,7 @@ def _write_arrow_table(self, table: pa.Table) -> WriteResult[list[FragmentMetada def _ensure_mem_wal_dataset(self) -> lance.LanceDataset: try: - ds = lance.dataset(self._table_uri, storage_options=self._storage_options) + ds = lance.dataset(self._dataset_uri_arg, storage_options=self._storage_options, **self._namespace_kwargs) except (ValueError, FileNotFoundError, OSError): ds = None @@ -250,11 +294,12 @@ def _ensure_mem_wal_dataset(self) -> lance.LanceDataset: {f.name: pa.array([], type=f.type) for f in self._effective_pyarrow_schema}, schema=self._effective_pyarrow_schema, ), - self._table_uri, + self._dataset_uri_arg, mode="create", storage_options=self._storage_options, data_storage_version=self._data_storage_version, use_legacy_format=self._use_legacy_format, + **self._namespace_kwargs, ) details = ds.mem_wal_index_details() @@ -335,6 +380,7 @@ def _finalize_cow(self, write_results: list[WriteResult[list[FragmentMetadata]]] operation, read_version=self._version, storage_options=self._storage_options, + **self._namespace_kwargs, ) stats = dataset.stats.dataset_stats() stats_dict = MicroPartition.from_pydict( @@ -348,7 +394,7 @@ def _finalize_cow(self, write_results: list[WriteResult[list[FragmentMetadata]]] return stats_dict def _finalize_mem_wal(self, write_results: list[WriteResult[list[FragmentMetadata]]]) -> MicroPartition: - dataset = lance.dataset(self._table_uri, storage_options=self._storage_options) + dataset = lance.dataset(self._dataset_uri_arg, storage_options=self._storage_options, **self._namespace_kwargs) if self._compact_after_write: logger.info( @@ -359,7 +405,9 @@ def _finalize_mem_wal(self, write_results: list[WriteResult[list[FragmentMetadat from daft_lance.lance_compaction import compact_files_internal compact_files_internal(dataset) - dataset = lance.dataset(self._table_uri, storage_options=self._storage_options) + dataset = lance.dataset( + self._dataset_uri_arg, storage_options=self._storage_options, **self._namespace_kwargs + ) stats = dataset.stats.dataset_stats() return MicroPartition.from_pydict( diff --git a/daft_lance/lance_scan.py b/daft_lance/lance_scan.py index de3349f..f52de22 100644 --- a/daft_lance/lance_scan.py +++ b/daft_lance/lance_scan.py @@ -16,6 +16,7 @@ from daft.recordbatch import RecordBatch from ._metadata import convert_lance_schema +from .namespace import get_namespace_kwargs from .point_lookup import detect_point_lookup_columns from .utils import combine_filters_to_arrow @@ -40,7 +41,15 @@ def _lancedb_table_factory_function( "Use nearest with fragment_ids=None for index-driven global vector search." ) - ds = lance.dataset(ds_uri, **(open_kwargs or {})) + open_kwargs = dict(open_kwargs or {}) + namespace_impl = open_kwargs.pop("namespace_impl", None) + namespace_properties = open_kwargs.pop("namespace_properties", None) + table_id = open_kwargs.pop("table_id", None) + ds = lance.dataset( + None if namespace_impl is not None and table_id is not None else ds_uri, + **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), + **open_kwargs, + ) def _iter_batches() -> Iterator[PyRecordBatch]: # Iterate fragments individually; append a fragment_id column only when requested @@ -117,7 +126,15 @@ def _lancedb_count_result_function( filter: pa.compute.Expression | None = None, ) -> Iterator[PyRecordBatch]: """Use LanceDB's API to count rows and return a record batch with the count result.""" - ds = lance.dataset(ds_uri, **(open_kwargs or {})) + open_kwargs = dict(open_kwargs or {}) + namespace_impl = open_kwargs.pop("namespace_impl", None) + namespace_properties = open_kwargs.pop("namespace_properties", None) + table_id = open_kwargs.pop("table_id", None) + ds = lance.dataset( + None if namespace_impl is not None and table_id is not None else ds_uri, + **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), + **open_kwargs, + ) logger.debug("Using metadata for counting all rows") count = ds.count_rows(filter=filter) diff --git a/daft_lance/namespace.py b/daft_lance/namespace.py new file mode 100644 index 0000000..2abe99c --- /dev/null +++ b/daft_lance/namespace.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import os +from functools import lru_cache +from typing import Any +from urllib.parse import urlparse + +import lance + +_NAMESPACE_CACHE_SIZE = int(os.environ.get("DAFT_LANCE_NAMESPACE_CACHE_SIZE", "16")) +_PYLANCE_5 = (5, 0, 0) + + +@lru_cache(maxsize=1) +def _pylance_version() -> tuple[int, ...]: + version = getattr(lance, "__version__", "0.0.0") + parts = [] + for part in version.split(".")[:3]: + digits = "".join(ch for ch in part if ch.isdigit()) + parts.append(int(digits or "0")) + while len(parts) < 3: + parts.append(0) + return tuple(parts) + + +def has_namespace_params(namespace_impl: str | None, table_id: list[str] | None) -> bool: + return namespace_impl is not None and table_id is not None + + +def validate_uri_or_namespace( + uri: str | os.PathLike[str] | None, + namespace_impl: str | None, + table_id: list[str] | None, +) -> None: + has_uri = uri is not None + has_ns = has_namespace_params(namespace_impl, table_id) + + if namespace_impl is not None and table_id is None: + raise ValueError("'table_id' must be provided when 'namespace_impl' is provided.") + if table_id is not None and namespace_impl is None: + raise ValueError("'namespace_impl' must be provided when 'table_id' is provided.") + if has_uri and has_ns: + raise ValueError( + "Cannot provide both 'uri' and namespace parameters. Use either 'uri' OR ('namespace_impl' + 'table_id')." + ) + if not has_uri and not has_ns: + raise ValueError("Must provide either 'uri' OR ('namespace_impl' + 'table_id').") + + +def _normalize_file_uri(location: str) -> str: + parsed = urlparse(location) + if parsed.scheme == "file": + return parsed.path + return location + + +@lru_cache(maxsize=_NAMESPACE_CACHE_SIZE) +def _get_cached_namespace(namespace_impl: str, namespace_properties_tuple: tuple[tuple[str, str], ...] | None) -> Any: + import lance_namespace as ln + + namespace_properties = dict(namespace_properties_tuple) if namespace_properties_tuple else {} + return ln.connect(namespace_impl, namespace_properties) + + +def get_or_create_namespace(namespace_impl: str | None, namespace_properties: dict[str, str] | None) -> Any | None: + if namespace_impl is None: + return None + namespace_properties_tuple = tuple(sorted(namespace_properties.items())) if namespace_properties else None + return _get_cached_namespace(namespace_impl, namespace_properties_tuple) + + +def _create_storage_options_provider( + namespace_impl: str | None, + namespace_properties: dict[str, str] | None, + table_id: list[str] | None, +) -> Any | None: + if not has_namespace_params(namespace_impl, table_id): + return None + namespace = get_or_create_namespace(namespace_impl, namespace_properties) + if namespace is None or not hasattr(lance, "LanceNamespaceStorageOptionsProvider"): + return None + return lance.LanceNamespaceStorageOptionsProvider(namespace=namespace, table_id=table_id) + + +def get_namespace_kwargs( + namespace_impl: str | None, + namespace_properties: dict[str, str] | None, + table_id: list[str] | None, +) -> dict[str, Any]: + if not has_namespace_params(namespace_impl, table_id): + return {} + + namespace = get_or_create_namespace(namespace_impl, namespace_properties) + if namespace is None: + return {} + + kwargs: dict[str, Any] = {"table_id": table_id} + if _pylance_version() >= _PYLANCE_5: + kwargs["namespace_client"] = namespace + else: + kwargs["namespace"] = namespace + provider = _create_storage_options_provider(namespace_impl, namespace_properties, table_id) + if provider is not None: + kwargs["storage_options_provider"] = provider + return kwargs + + +def get_write_fragments_kwargs( + namespace_impl: str | None, + namespace_properties: dict[str, str] | None, + table_id: list[str] | None, +) -> dict[str, Any]: + if not has_namespace_params(namespace_impl, table_id): + return {} + if _pylance_version() >= _PYLANCE_5: + namespace = get_or_create_namespace(namespace_impl, namespace_properties) + if namespace is None: + return {} + return {"namespace_client": namespace, "table_id": table_id} + provider = _create_storage_options_provider(namespace_impl, namespace_properties, table_id) + return {"storage_options_provider": provider} if provider is not None else {} + + +def _storage_options(response: Any) -> dict[str, str] | None: + storage_options = getattr(response, "storage_options", None) + if storage_options is None: + return None + return dict(storage_options) + + +def _response_location(response: Any) -> str: + location = getattr(response, "location", None) or getattr(response, "table_uri", None) + if not location: + raise ValueError("Namespace response did not include a table location.") + return _normalize_file_uri(str(location)) + + +def _declare_table(namespace: Any, table_id: list[str]) -> tuple[str, dict[str, str] | None]: + try: + from lance_namespace import DeclareTableRequest + + response = namespace.declare_table(DeclareTableRequest(id=table_id, location=None)) + return _response_location(response), _storage_options(response) + except (AttributeError, NotImplementedError): + from lance_namespace import CreateEmptyTableRequest + + response = namespace.create_empty_table(CreateEmptyTableRequest(id=table_id)) + return _response_location(response), _storage_options(response) + + +def resolve_namespace_table( + *, + namespace_impl: str | None, + namespace_properties: dict[str, str] | None, + table_id: list[str] | None, + mode: str = "read", +) -> tuple[str | None, dict[str, str] | None]: + namespace = get_or_create_namespace(namespace_impl, namespace_properties) + if namespace is None or table_id is None: + return None, None + + if mode == "create": + return _declare_table(namespace, table_id) + + from lance_namespace import DescribeTableRequest + + try: + response = namespace.describe_table(DescribeTableRequest(id=table_id)) + return _response_location(response), _storage_options(response) + except Exception: + if mode == "overwrite": + return _declare_table(namespace, table_id) + raise + + +def merge_storage_options( + storage_options: dict[str, Any] | None, + namespace_storage_options: dict[str, str] | None, +) -> dict[str, Any] | None: + merged: dict[str, Any] = {} + if storage_options: + merged.update(storage_options) + if namespace_storage_options: + merged.update(namespace_storage_options) + return merged or None + + +def open_lance_dataset( + uri: str | os.PathLike[str] | None, + *, + storage_options: dict[str, Any] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, + table_id: list[str] | None = None, + mode: str = "read", + **kwargs: Any, +) -> lance.LanceDataset: + validate_uri_or_namespace(uri, namespace_impl, table_id) + resolved_uri = str(uri) if uri is not None else None + namespace_storage_options = None + if resolved_uri is None: + resolved_uri, namespace_storage_options = resolve_namespace_table( + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, + mode=mode, + ) + if resolved_uri is None: + raise ValueError("Unable to resolve Lance dataset URI.") + + merged_storage_options = merge_storage_options(storage_options, namespace_storage_options) + return lance.dataset( + None if has_namespace_params(namespace_impl, table_id) else resolved_uri, + storage_options=merged_storage_options, + **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), + **kwargs, + ) diff --git a/daft_lance/utils.py b/daft_lance/utils.py index 0488262..799fe4c 100644 --- a/daft_lance/utils.py +++ b/daft_lance/utils.py @@ -7,6 +7,13 @@ from daft.dependencies import pa from daft.logical.schema import Schema as DaftSchema +from daft_lance.namespace import ( + get_namespace_kwargs, + has_namespace_params, + merge_storage_options, + resolve_namespace_table, + validate_uri_or_namespace, +) if TYPE_CHECKING: import pathlib @@ -82,12 +89,29 @@ def distribute_fragments_balanced(fragments: list[Any], fragment_group_size: int def construct_lance_dataset( - uri: str | pathlib.Path, + uri: str | pathlib.Path | None, version: int | str | None = None, storage_options: dict[str, Any] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, + table_id: list[str] | None = None, **kwargs: Any, ) -> lance.LanceDataset: """Construct a Lance dataset with common options.""" + validate_uri_or_namespace(uri, namespace_impl, table_id) + resolved_uri = str(uri) if uri is not None else None + namespace_storage_options = None + if resolved_uri is None: + resolved_uri, namespace_storage_options = resolve_namespace_table( + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, + mode="read", + ) + if resolved_uri is None: + raise ValueError("Unable to resolve Lance dataset URI.") + merged_storage_options = merge_storage_options(storage_options, namespace_storage_options) + original_default_scan_options = kwargs.pop("default_scan_options", None) safe_default_scan_options = None if isinstance(original_default_scan_options, dict): @@ -98,11 +122,21 @@ def construct_lance_dataset( # Non-dict defaults are forwarded as-is. kwargs["default_scan_options"] = original_default_scan_options - ds = lance.dataset(uri, storage_options=storage_options, version=version, **kwargs) + dataset_uri = None if has_namespace_params(namespace_impl, table_id) else resolved_uri + ds = lance.dataset( + dataset_uri, + storage_options=merged_storage_options, + version=version, + **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), + **kwargs, + ) effective_kwargs = { - "storage_options": storage_options, + "storage_options": merged_storage_options, "version": version, + "namespace_impl": namespace_impl, + "namespace_properties": namespace_properties, + "table_id": table_id, } effective_kwargs.update(kwargs or {}) try: diff --git a/tests/io/lancedb/test_namespace.py b/tests/io/lancedb/test_namespace.py new file mode 100644 index 0000000..c573a56 --- /dev/null +++ b/tests/io/lancedb/test_namespace.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import pytest + +import daft +import daft_lance + + +def test_namespace_write_read_append_roundtrip(tmp_path): + namespace_properties = {"root": str(tmp_path)} + table_id = ["roundtrip"] + + df1 = daft.from_pydict({"id": [1, 2], "label": ["a", "b"]}) + df2 = daft.from_pydict({"id": [3], "label": ["c"]}) + + df1.write_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + mode="create", + ).collect() + df2.write_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + mode="append", + ).collect() + + result = daft.read_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + ).to_pydict() + + assert result == {"id": [1, 2, 3], "label": ["a", "b", "c"]} + + +def test_namespace_read_supports_pushdowns(tmp_path): + namespace_properties = {"root": str(tmp_path)} + table_id = ["pushdowns"] + + daft.from_pydict( + { + "id": [1, 2, 3], + "label": ["a", "b", "c"], + "score": [10, 20, 30], + } + ).write_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + mode="create", + ).collect() + + result = ( + daft.read_lance( + namespace_impl="dir", + namespace_properties=namespace_properties, + table_id=table_id, + ) + .where(daft.col("score") > 10) + .select("label") + .to_pydict() + ) + + assert result == {"label": ["b", "c"]} + + +def test_namespace_rejects_uri_and_namespace(tmp_path): + with pytest.raises(ValueError, match="Cannot provide both 'uri' and namespace parameters"): + daft_lance.read_lance( + str(tmp_path / "dataset"), + namespace_impl="dir", + namespace_properties={"root": str(tmp_path)}, + table_id=["tbl"], + ) + + +def test_namespace_requires_table_id(tmp_path): + with pytest.raises(ValueError, match="'table_id' must be provided"): + daft_lance.read_lance( + namespace_impl="dir", + namespace_properties={"root": str(tmp_path)}, + ) diff --git a/tests/io/lancedb/test_namespace_rest_e2e.py b/tests/io/lancedb/test_namespace_rest_e2e.py new file mode 100644 index 0000000..d9f27e9 --- /dev/null +++ b/tests/io/lancedb/test_namespace_rest_e2e.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import os +import uuid + +import pytest + +import daft +import daft_lance # noqa: F401 - patches daft.read_lance and DataFrame.write_lance. + +pytestmark = pytest.mark.skipif( + os.environ.get("DAFT_LANCE_REST_URI") is None, + reason="Set DAFT_LANCE_REST_URI to run the Lance REST namespace integration test.", +) + + +def test_rest_namespace_write_read_append_roundtrip() -> None: + import lance + import lance_namespace as ln + from lance_namespace import CreateNamespaceRequest, DescribeTableRequest, NamespaceExistsRequest + + namespace_properties = {"uri": os.environ["DAFT_LANCE_REST_URI"]} + catalog = os.environ.get("DAFT_LANCE_REST_CATALOG", "lance_catalog") + schema = os.environ.get("DAFT_LANCE_REST_SCHEMA", "daft_ns_e2e") + table_id = [catalog, schema, f"orders_{uuid.uuid4().hex[:8]}"] + + namespace = ln.connect("rest", namespace_properties) + try: + namespace.namespace_exists(NamespaceExistsRequest(id=[catalog, schema])) + except Exception: + namespace.create_namespace(CreateNamespaceRequest(id=[catalog, schema], mode="CREATE")) + + daft.from_pydict( + { + "id": [1, 2, 3], + "label": ["a", "b", "c"], + "score": [10, 20, 30], + } + ).write_lance( + namespace_impl="rest", + namespace_properties=namespace_properties, + table_id=table_id, + mode="create", + ).collect() + + daft.from_pydict( + { + "id": [4, 5], + "label": ["d", "e"], + "score": [40, 50], + } + ).write_lance( + namespace_impl="rest", + namespace_properties=namespace_properties, + table_id=table_id, + mode="append", + ).collect() + + describe = namespace.describe_table(DescribeTableRequest(id=table_id)) + location = getattr(describe, "location", None) or getattr(describe, "table_uri", None) + assert location + + result = daft.read_lance( + namespace_impl="rest", + namespace_properties=namespace_properties, + table_id=table_id, + ).to_pydict() + assert result == { + "id": [1, 2, 3, 4, 5], + "label": ["a", "b", "c", "d", "e"], + "score": [10, 20, 30, 40, 50], + } + + filtered = ( + daft.read_lance( + namespace_impl="rest", + namespace_properties=namespace_properties, + table_id=table_id, + ) + .where(daft.col("score") >= 30) + .select("id", "label") + .to_pydict() + ) + assert filtered == {"id": [3, 4, 5], "label": ["c", "d", "e"]} + + assert ( + daft.read_lance( + namespace_impl="rest", + namespace_properties=namespace_properties, + table_id=table_id, + ).count_rows() + == 5 + ) + assert lance.dataset(None, namespace_client=namespace, table_id=table_id).count_rows() == 5 From 383592a980341b726ea6273cae5dfe2497148dad Mon Sep 17 00:00:00 2001 From: fanng <“fanng@apache.org”> Date: Fri, 3 Jul 2026 18:24:14 +0900 Subject: [PATCH 2/2] feat(namespace): extend Lance Namespace support and harden implementation Build on the initial namespace read/write support: - Extend namespace params (namespace_impl/namespace_properties/table_id) to merge_columns, merge_columns_df, create_scalar_index, compact_files, threading namespace kwargs into their commit sites. - Add an explicit daft_lance.write_lance() and make the daft monkeypatch opt-in via patch_daft(); wire write_lance(mode="merge") through namespace. - Fix _declare_table fallback that always raised ImportError on the CreateEmptyTableRequest import; keep only declare_table. - Narrow overwrite resolution to only declare on TableNotFound instead of swallowing every describe_table exception. - Drop dead code (open_lance_dataset, pylance<5 compat) and dedupe the worker dataset-reopen logic into open_dataset_from_open_kwargs. - Cover all entry points with dir-namespace tests plus a Gravitino REST namespace e2e roundtrip. - Work around an upstream daft+lance native teardown SIGSEGV (unrelated to test outcomes) via a pytest_unconfigure hard-exit after reporting. --- README.md | 55 ++++- daft_lance/__init__.py | 52 ++--- daft_lance/_lance.py | 237 ++++++++++++++++++-- daft_lance/lance_merge_column.py | 5 + daft_lance/lance_scalar_index.py | 12 +- daft_lance/lance_scan.py | 22 +- daft_lance/namespace.py | 152 ++++++------- tests/io/lancedb/conftest.py | 27 +++ tests/io/lancedb/test_namespace.py | 193 +++++++++++++--- tests/io/lancedb/test_namespace_rest_e2e.py | 65 +++--- 10 files changed, 580 insertions(+), 240 deletions(-) diff --git a/README.md b/README.md index a7133b6..e88e5f1 100644 --- a/README.md +++ b/README.md @@ -40,25 +40,60 @@ merge_columns_df(df, "s3://bucket/my_dataset") ### Namespace Tables +Address Lance tables through a [Lance Namespace](https://lancedb.github.io/lance-namespace/) +(catalog) instead of a raw URI. Pass `namespace_impl` + `namespace_properties` + `table_id` +in place of `uri` — the namespace resolves the table's storage location and vends any storage +credentials. This works across `read_lance`, `write_lance`, `merge_columns_df`, +`create_scalar_index`, and `compact_files`. + ```python import daft -import daft_lance # installs daft.read_lance / DataFrame.write_lance namespace support +import daft_lance table_id = ["my_table"] -namespace_properties = {"root": "/tmp/lance_tables"} +namespace = {"namespace_impl": "dir", "namespace_properties": {"root": "/tmp/lance_tables"}} -df.write_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, +daft_lance.write_lance( + daft.from_pydict({"id": [1, 2, 3]}), table_id=table_id, mode="create", + **namespace, ).collect() -df = daft.read_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, - table_id=table_id, -) +df = daft_lance.read_lance(table_id=table_id, **namespace) +``` + +`uri` and the namespace parameters are mutually exclusive: provide exactly one of `uri` or +(`namespace_impl` + `table_id`). + +#### Using a REST namespace (e.g. Gravitino Lance REST server) + +```python +namespace = { + "namespace_impl": "rest", + "namespace_properties": {"uri": "http://127.0.0.1:9101/lance"}, +} +table_id = ["lance_catalog", "sales", "orders"] + +daft_lance.write_lance(df, table_id=table_id, mode="create", **namespace).collect() +daft_lance.read_lance(table_id=table_id, **namespace).show() +``` + +When the catalog holds the storage configuration (bucket, endpoint, credentials), the +`describe_table` response vends `storage_options` to the client, so you do not need to pass +object-store credentials yourself. + +#### Drop-in Daft APIs + +If you prefer Daft's own entry points, call `daft_lance.patch_daft()` once to route +`daft.read_lance` and `DataFrame.write_lance` through daft-lance (namespace parameters included): + +```python +import daft, daft_lance + +daft_lance.patch_daft() +df.write_lance(table_id=["my_table"], mode="create", **namespace).collect() +daft.read_lance(table_id=["my_table"], **namespace).show() ``` ## Migration diff --git a/daft_lance/__init__.py b/daft_lance/__init__.py index 8a1ecba..5af90b5 100644 --- a/daft_lance/__init__.py +++ b/daft_lance/__init__.py @@ -14,21 +14,25 @@ merge_columns, merge_columns_df, read_lance, + write_lance, ) from .lance_data_sink import LanceDataSink -def _patch_daft_lance_api() -> None: - """Expose the daft-lance implementation through Daft's convenience APIs.""" +def patch_daft() -> None: + """Route ``daft.read_lance`` and ``DataFrame.write_lance`` through daft-lance. + + Optional convenience for drop-in use of namespace parameters with Daft's own + APIs, e.g. ``df.write_lance(namespace_impl="dir", ...)``. Idempotent. + """ daft.read_lance = read_lance # type: ignore[assignment] from daft.dataframe import DataFrame - original_write_lance = getattr(DataFrame, "write_lance") - if getattr(original_write_lance, "_daft_lance_namespace_patch", False): + if getattr(DataFrame.write_lance, "_daft_lance_patch", False): return - def write_lance( + def _write_lance( self: DataFrame, uri: Any = None, mode: Literal["create", "append", "overwrite", "merge"] = "create", @@ -38,37 +42,29 @@ def write_lance( right_on: str | None = None, **kwargs: Any, ) -> DataFrame: - if mode == "merge": - if any(k in kwargs for k in ("namespace_impl", "namespace_properties", "table_id")): - raise ValueError("write_lance(mode='merge') does not support namespace parameters yet.") - return original_write_lance( - self, - uri, - mode=mode, - io_config=io_config, - schema=schema, - left_on=left_on, - right_on=right_on, - **kwargs, - ) - - if schema is None: - schema = self.schema() - sanitized_kwargs = {k: v for k, v in kwargs.items() if k not in ("left_on", "right_on")} - sink = LanceDataSink(uri, schema, mode, io_config, **sanitized_kwargs) - return self.write_sink(sink) - - write_lance._daft_lance_namespace_patch = True # type: ignore[attr-defined] - DataFrame.write_lance = write_lance # type: ignore[assignment] + return write_lance( + self, + uri, + mode=mode, + io_config=io_config, + schema=schema, + left_on=left_on, + right_on=right_on, + **kwargs, + ) + _write_lance._daft_lance_patch = True # type: ignore[attr-defined] + DataFrame.write_lance = _write_lance # type: ignore[method-assign] -_patch_daft_lance_api() __all__ = [ + "LanceDataSink", "compact_files", "create_scalar_index", "merge_columns", "merge_columns_df", + "patch_daft", "read_lance", "take_blobs", + "write_lance", ] diff --git a/daft_lance/_lance.py b/daft_lance/_lance.py index 7b5f55c..12bdf90 100644 --- a/daft_lance/_lance.py +++ b/daft_lance/_lance.py @@ -2,9 +2,7 @@ import pathlib from collections.abc import Callable -from typing import TYPE_CHECKING, Any - -import lance +from typing import TYPE_CHECKING, Any, Literal from daft import context from daft.api_annotations import PublicAPI @@ -15,9 +13,11 @@ from daft.logical.builder import LogicalPlanBuilder from .lance_compaction import compact_files_internal +from .lance_data_sink import LanceDataSink from .lance_merge_column import merge_columns_from_df, merge_columns_internal from .lance_scalar_index import create_scalar_index_internal from .lance_scan import LanceDBScanOperator +from .namespace import is_table_not_found, validate_uri_or_namespace from .utils import construct_lance_dataset if TYPE_CHECKING: @@ -28,6 +28,19 @@ from daft.dependencies import pa +def _effective_uri(lance_ds: LanceDataset, uri: str | pathlib.Path | None) -> str: + """The dataset location: the user-supplied URI, or the namespace-resolved one.""" + return str(uri) if uri is not None else str(lance_ds.uri) + + +def _effective_storage_options( + lance_ds: LanceDataset, storage_options: dict[str, Any] | None +) -> dict[str, Any] | None: + """Storage options actually used to open the dataset (includes namespace-vended ones).""" + open_kwargs = getattr(lance_ds, "_lance_open_kwargs", None) or {} + return open_kwargs.get("storage_options") or storage_options + + @PublicAPI def read_lance( uri: str | pathlib.Path | None = None, @@ -168,9 +181,12 @@ def read_lance( @PublicAPI def merge_columns( - uri: str | pathlib.Path, + uri: str | pathlib.Path | None = None, io_config: IOConfig | None = None, *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, transform: dict[str, str] | BatchUDF | Callable[[pa.lib.RecordBatch], pa.lib.RecordBatch] | None = None, read_columns: list[str] | None = None, reader_schema: pa.Schema | None = None, @@ -229,12 +245,16 @@ def merge_columns( ) io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config - storage_options = storage_options or io_config_to_storage_options(io_config, uri) + if uri is not None: + storage_options = storage_options or io_config_to_storage_options(io_config, uri) # Build Lance dataset handle for committing lance_ds = construct_lance_dataset( uri, storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, version=version, asof=asof, block_size=block_size, @@ -246,11 +266,11 @@ def merge_columns( return merge_columns_internal( lance_ds, - uri, + _effective_uri(lance_ds, uri), transform=transform, read_columns=read_columns, reader_schema=reader_schema, - storage_options=storage_options, + storage_options=_effective_storage_options(lance_ds, storage_options), daft_remote_args=daft_remote_args, concurrency=concurrency, ) @@ -259,9 +279,12 @@ def merge_columns( @PublicAPI def merge_columns_df( df: DataFrame, - uri: str | pathlib.Path, + uri: str | pathlib.Path | None = None, io_config: IOConfig | None = None, *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, read_columns: list[str] | None = None, reader_schema: pa.Schema | None = None, storage_options: dict[str, Any] | None = None, @@ -325,12 +348,16 @@ def merge_columns_df( >>> daft_lance.merge_columns_df(df, "s3://my-lancedb-bucket/data/") """ io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config - storage_options = storage_options or io_config_to_storage_options(io_config, uri) + if uri is not None: + storage_options = storage_options or io_config_to_storage_options(io_config, uri) # Build Lance dataset handle for committing lance_ds = construct_lance_dataset( uri, storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, version=version, asof=asof, block_size=block_size, @@ -349,10 +376,10 @@ def merge_columns_df( return merge_columns_from_df( df, lance_ds=lance_ds, - uri=uri, + uri=_effective_uri(lance_ds, uri), read_columns=read_columns, reader_schema=reader_schema, - storage_options=storage_options, + storage_options=_effective_storage_options(lance_ds, storage_options), daft_remote_args=daft_remote_args, concurrency=concurrency, left_on=left_on, @@ -363,9 +390,12 @@ def merge_columns_df( @PublicAPI def create_scalar_index( - uri: str | pathlib.Path, + uri: str | pathlib.Path | None = None, io_config: IOConfig | None = None, *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, column: str, index_type: str = "INVERTED", name: str | None = None, @@ -463,11 +493,15 @@ def create_scalar_index( >>> daft_lance.create_scalar_index("s3://my-bucket/dataset/", column="title", replace=False) """ io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config - storage_options = storage_options or io_config_to_storage_options(io_config, str(uri)) + if uri is not None: + storage_options = storage_options or io_config_to_storage_options(io_config, str(uri)) lance_ds = construct_lance_dataset( uri, storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, version=version, asof=asof, block_size=block_size, @@ -479,12 +513,12 @@ def create_scalar_index( create_scalar_index_internal( lance_ds=lance_ds, - uri=uri, + uri=_effective_uri(lance_ds, uri), column=column, index_type=index_type, name=name, replace=replace, - storage_options=storage_options, + storage_options=_effective_storage_options(lance_ds, storage_options), fragment_group_size=fragment_group_size, num_partitions=num_partitions, max_concurrency=max_concurrency, @@ -495,9 +529,12 @@ def create_scalar_index( @PublicAPI def compact_files( - uri: str | pathlib.Path, + uri: str | pathlib.Path | None = None, io_config: IOConfig | None = None, *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, storage_options: dict[str, Any] | None = None, version: int | str | None = None, asof: str | None = None, @@ -548,13 +585,17 @@ def compact_files( RuntimeError: When compaction fails or no successful results """ io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config - storage_options = storage_options or io_config_to_storage_options( - io_config, str(uri) if isinstance(uri, pathlib.Path) else uri - ) + if uri is not None: + storage_options = storage_options or io_config_to_storage_options( + io_config, str(uri) if isinstance(uri, pathlib.Path) else uri + ) - lance_ds = lance.dataset( + lance_ds = construct_lance_dataset( uri, storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, version=version, asof=asof, block_size=block_size, @@ -570,3 +611,159 @@ def compact_files( partition_num=partition_num, concurrency=concurrency, ) + + +def _is_missing_dataset_error(error: Exception) -> bool: + """Whether opening a dataset failed because it does not exist yet.""" + if isinstance(error, (FileNotFoundError,)): + return True + if isinstance(error, (ValueError, OSError)) and "was not found" in str(error): + return True + return is_table_not_found(error) + + +def _stats_dataframe(lance_ds: LanceDataset) -> DataFrame: + """Build the same stats DataFrame that LanceDataSink.finalize returns.""" + import pyarrow as _pa + + from daft.recordbatch import MicroPartition + + stats = lance_ds.stats.dataset_stats() + return DataFrame._from_micropartitions( + MicroPartition.from_pydict( + { + "num_fragments": _pa.array([stats["num_fragments"]], type=_pa.int64()), + "num_deleted_rows": _pa.array([stats["num_deleted_rows"]], type=_pa.int64()), + "num_small_files": _pa.array([stats["num_small_files"]], type=_pa.int64()), + "version": _pa.array([lance_ds.version], type=_pa.int64()), + } + ) + ) + + +@PublicAPI +def write_lance( + df: DataFrame, + uri: str | pathlib.Path | None = None, + mode: Literal["create", "append", "overwrite", "merge"] = "create", + io_config: IOConfig | None = None, + schema: Any | None = None, + left_on: str | None = None, + right_on: str | None = None, + *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, + **kwargs: Any, +) -> DataFrame: + """Write a DataFrame to a Lance table, addressed by URI or by Lance Namespace. + + Args: + df: The DataFrame to write. + uri: The URI of the Lance table. Mutually exclusive with the namespace parameters. + mode: One of "create", "append", "overwrite", or "merge" (add new columns to + existing rows; requires ``fragment_id`` and a join key column in ``df``). + io_config: A custom IOConfig to use when accessing Lance data. + schema: Desired schema to enforce during write; defaults to the DataFrame schema. + left_on/right_on: Join keys for ``mode="merge"``; default "_rowaddr". + table_id: Table identifier within the namespace, e.g. ["catalog", "schema", "table"]. + namespace_impl: Lance Namespace implementation, e.g. "dir" or "rest". + namespace_properties: Properties for connecting to the namespace, e.g. + {"root": "/data"} for "dir" or {"uri": "http://host:port"} for "rest". + **kwargs: Additional arguments forwarded to the Lance writer. + + Returns: + DataFrame: write statistics (num_fragments, num_deleted_rows, num_small_files, version). + + Examples: + >>> import daft, daft_lance + >>> df = daft.from_pydict({"id": [1, 2]}) + >>> daft_lance.write_lance( + ... df, namespace_impl="dir", namespace_properties={"root": "/tmp/tables"}, table_id=["t"] + ... ).collect() # doctest: +SKIP + """ + validate_uri_or_namespace(uri, namespace_impl, table_id) + + if schema is None: + schema = df.schema() + + if mode != "merge": + sanitized_kwargs = {k: v for k, v in kwargs.items() if k not in ("left_on", "right_on")} + sink = LanceDataSink( + uri, + schema, + mode, + io_config, + table_id=table_id, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + **sanitized_kwargs, + ) + return df.write_sink(sink) + + io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config + storage_options = kwargs.pop("storage_options", None) + if uri is not None: + storage_options = storage_options or io_config_to_storage_options(io_config, str(uri)) + + lance_ds: LanceDataset | None + try: + lance_ds = construct_lance_dataset( + uri, + storage_options=storage_options, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, + ) + except Exception as error: + if not _is_missing_dataset_error(error): + raise + lance_ds = None + + sanitized_kwargs = {k: v for k, v in kwargs.items() if k not in ("left_on", "right_on")} + + if lance_ds is None: + # Merge against a missing table degenerates to create. + sink = LanceDataSink( + uri, + schema, + "create", + io_config, + table_id=table_id, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + storage_options=storage_options, + **sanitized_kwargs, + ) + return df.write_sink(sink) + + existing_fields = {getattr(f, "name", str(f)) for f in lance_ds.schema} + meta_exclusions = {"fragment_id", "_rowaddr", "_rowid"} + new_cols = [c for c in df.column_names if c not in existing_fields and c not in meta_exclusions] + + if len(new_cols) == 0: + # No schema evolution: merge degenerates to append. + sink = LanceDataSink( + uri, + schema, + "append", + io_config, + table_id=table_id, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + storage_options=storage_options, + **sanitized_kwargs, + ) + return df.write_sink(sink) + + join_left = left_on or "_rowaddr" + join_right = right_on or join_left + merged = merge_columns_from_df( + df, + lance_ds=lance_ds, + uri=_effective_uri(lance_ds, uri), + left_on=join_left, + right_on=join_right, + storage_options=_effective_storage_options(lance_ds, storage_options), + ) + return _stats_dataframe(merged) diff --git a/daft_lance/lance_merge_column.py b/daft_lance/lance_merge_column.py index c0e0cb9..ea98553 100644 --- a/daft_lance/lance_merge_column.py +++ b/daft_lance/lance_merge_column.py @@ -14,6 +14,8 @@ from daft.udf import cls as daft_cls from daft.udf import method +from .namespace import namespace_kwargs_for_dataset + if TYPE_CHECKING: import pathlib from collections.abc import Callable @@ -93,6 +95,7 @@ def merge_columns_internal( op, read_version=lance_ds.version, storage_options=storage_options, + **namespace_kwargs_for_dataset(lance_ds), ) @@ -459,6 +462,7 @@ def _merge_fast_path( op, read_version=lance_ds.version, storage_options=storage_options, + **namespace_kwargs_for_dataset(lance_ds), ) @@ -507,4 +511,5 @@ def _merge_slow_path( op, read_version=lance_ds.version, storage_options=storage_options, + **namespace_kwargs_for_dataset(lance_ds), ) diff --git a/daft_lance/lance_scalar_index.py b/daft_lance/lance_scalar_index.py index 7f63115..9fe8468 100644 --- a/daft_lance/lance_scalar_index.py +++ b/daft_lance/lance_scalar_index.py @@ -14,6 +14,7 @@ import lance from daft.dependencies import pa +from daft_lance.namespace import namespace_kwargs_for_dataset from daft_lance.utils import distribute_fragments_balanced logger = logging.getLogger(__name__) @@ -330,7 +331,10 @@ def _create_segmented_index( # Reload dataset to pick up the latest version (segment files were written # by workers against the version that was current at their invocation time). - lance_ds = lance.LanceDataset(uri, storage_options=storage_options) + namespace_kwargs = namespace_kwargs_for_dataset(lance_ds) + lance_ds = lance.dataset( + None if namespace_kwargs else uri, storage_options=storage_options, **namespace_kwargs + ) lance_ds.commit_existing_index_segments(name, column, index_metas) logger.info("Segmented index %s committed successfully", name) @@ -384,7 +388,10 @@ def _create_partitioned_index( df.collect() logger.info("Starting index metadata merging by reloading dataset to get latest state") - lance_ds = lance.LanceDataset(uri, storage_options=storage_options) + namespace_kwargs = namespace_kwargs_for_dataset(lance_ds) + lance_ds = lance.dataset( + None if namespace_kwargs else uri, storage_options=storage_options, **namespace_kwargs + ) lance_ds.merge_index_metadata(index_id, index_type) logger.info("Starting atomic index creation and commit") @@ -430,6 +437,7 @@ def _create_partitioned_index( create_index_op, read_version=lance_ds.version, storage_options=storage_options, + **namespace_kwargs, ) logger.info("Index %s created successfully with ID %s", name, index_id) diff --git a/daft_lance/lance_scan.py b/daft_lance/lance_scan.py index f52de22..959dacf 100644 --- a/daft_lance/lance_scan.py +++ b/daft_lance/lance_scan.py @@ -16,7 +16,7 @@ from daft.recordbatch import RecordBatch from ._metadata import convert_lance_schema -from .namespace import get_namespace_kwargs +from .namespace import open_dataset_from_open_kwargs from .point_lookup import detect_point_lookup_columns from .utils import combine_filters_to_arrow @@ -41,15 +41,7 @@ def _lancedb_table_factory_function( "Use nearest with fragment_ids=None for index-driven global vector search." ) - open_kwargs = dict(open_kwargs or {}) - namespace_impl = open_kwargs.pop("namespace_impl", None) - namespace_properties = open_kwargs.pop("namespace_properties", None) - table_id = open_kwargs.pop("table_id", None) - ds = lance.dataset( - None if namespace_impl is not None and table_id is not None else ds_uri, - **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), - **open_kwargs, - ) + ds = open_dataset_from_open_kwargs(ds_uri, open_kwargs) def _iter_batches() -> Iterator[PyRecordBatch]: # Iterate fragments individually; append a fragment_id column only when requested @@ -126,15 +118,7 @@ def _lancedb_count_result_function( filter: pa.compute.Expression | None = None, ) -> Iterator[PyRecordBatch]: """Use LanceDB's API to count rows and return a record batch with the count result.""" - open_kwargs = dict(open_kwargs or {}) - namespace_impl = open_kwargs.pop("namespace_impl", None) - namespace_properties = open_kwargs.pop("namespace_properties", None) - table_id = open_kwargs.pop("table_id", None) - ds = lance.dataset( - None if namespace_impl is not None and table_id is not None else ds_uri, - **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), - **open_kwargs, - ) + ds = open_dataset_from_open_kwargs(ds_uri, open_kwargs) logger.debug("Using metadata for counting all rows") count = ds.count_rows(filter=filter) diff --git a/daft_lance/namespace.py b/daft_lance/namespace.py index 2abe99c..3fd19c7 100644 --- a/daft_lance/namespace.py +++ b/daft_lance/namespace.py @@ -8,19 +8,6 @@ import lance _NAMESPACE_CACHE_SIZE = int(os.environ.get("DAFT_LANCE_NAMESPACE_CACHE_SIZE", "16")) -_PYLANCE_5 = (5, 0, 0) - - -@lru_cache(maxsize=1) -def _pylance_version() -> tuple[int, ...]: - version = getattr(lance, "__version__", "0.0.0") - parts = [] - for part in version.split(".")[:3]: - digits = "".join(ch for ch in part if ch.isdigit()) - parts.append(int(digits or "0")) - while len(parts) < 3: - parts.append(0) - return tuple(parts) def has_namespace_params(namespace_impl: str | None, table_id: list[str] | None) -> bool: @@ -69,56 +56,26 @@ def get_or_create_namespace(namespace_impl: str | None, namespace_properties: di return _get_cached_namespace(namespace_impl, namespace_properties_tuple) -def _create_storage_options_provider( - namespace_impl: str | None, - namespace_properties: dict[str, str] | None, - table_id: list[str] | None, -) -> Any | None: - if not has_namespace_params(namespace_impl, table_id): - return None - namespace = get_or_create_namespace(namespace_impl, namespace_properties) - if namespace is None or not hasattr(lance, "LanceNamespaceStorageOptionsProvider"): - return None - return lance.LanceNamespaceStorageOptionsProvider(namespace=namespace, table_id=table_id) - - def get_namespace_kwargs( namespace_impl: str | None, namespace_properties: dict[str, str] | None, table_id: list[str] | None, ) -> dict[str, Any]: + """Kwargs wiring a namespace client into pylance APIs (``lance.dataset``, ``commit``, ...). + + Requires pylance >= 7 which accepts ``namespace_client`` + ``table_id`` natively. + """ if not has_namespace_params(namespace_impl, table_id): return {} namespace = get_or_create_namespace(namespace_impl, namespace_properties) if namespace is None: return {} - - kwargs: dict[str, Any] = {"table_id": table_id} - if _pylance_version() >= _PYLANCE_5: - kwargs["namespace_client"] = namespace - else: - kwargs["namespace"] = namespace - provider = _create_storage_options_provider(namespace_impl, namespace_properties, table_id) - if provider is not None: - kwargs["storage_options_provider"] = provider - return kwargs + return {"namespace_client": namespace, "table_id": table_id} -def get_write_fragments_kwargs( - namespace_impl: str | None, - namespace_properties: dict[str, str] | None, - table_id: list[str] | None, -) -> dict[str, Any]: - if not has_namespace_params(namespace_impl, table_id): - return {} - if _pylance_version() >= _PYLANCE_5: - namespace = get_or_create_namespace(namespace_impl, namespace_properties) - if namespace is None: - return {} - return {"namespace_client": namespace, "table_id": table_id} - provider = _create_storage_options_provider(namespace_impl, namespace_properties, table_id) - return {"storage_options_provider": provider} if provider is not None else {} +# `lance.fragment.write_fragments` accepts the same namespace kwargs as `lance.dataset`. +get_write_fragments_kwargs = get_namespace_kwargs def _storage_options(response: Any) -> dict[str, str] | None: @@ -136,16 +93,27 @@ def _response_location(response: Any) -> str: def _declare_table(namespace: Any, table_id: list[str]) -> tuple[str, dict[str, str] | None]: - try: - from lance_namespace import DeclareTableRequest + from lance_namespace import DeclareTableRequest - response = namespace.declare_table(DeclareTableRequest(id=table_id, location=None)) - return _response_location(response), _storage_options(response) - except (AttributeError, NotImplementedError): - from lance_namespace import CreateEmptyTableRequest + response = namespace.declare_table(DeclareTableRequest(id=table_id, location=None)) + return _response_location(response), _storage_options(response) - response = namespace.create_empty_table(CreateEmptyTableRequest(id=table_id)) - return _response_location(response), _storage_options(response) + +def is_table_not_found(error: Exception) -> bool: + """Whether an exception from ``describe_table`` means the table does not exist. + + Native implementations raise the typed ``TableNotFoundError``; the Rust-backed + REST client surfaces untyped errors, so fall back to message matching. + """ + try: + from lance_namespace.errors import TableNotFoundError + + if isinstance(error, TableNotFoundError): + return True + except ImportError: + pass + message = str(error).lower() + return "not found" in message or "does not exist" in message or "no such table" in message def resolve_namespace_table( @@ -155,6 +123,11 @@ def resolve_namespace_table( table_id: list[str] | None, mode: str = "read", ) -> tuple[str | None, dict[str, str] | None]: + """Resolve a namespace table to ``(location, storage_options)``. + + ``mode="create"`` declares the table; ``mode="overwrite"`` declares it only when + it does not exist yet; other modes require the table to exist. + """ namespace = get_or_create_namespace(namespace_impl, namespace_properties) if namespace is None or table_id is None: return None, None @@ -167,8 +140,8 @@ def resolve_namespace_table( try: response = namespace.describe_table(DescribeTableRequest(id=table_id)) return _response_location(response), _storage_options(response) - except Exception: - if mode == "overwrite": + except Exception as error: + if mode == "overwrite" and is_table_not_found(error): return _declare_table(namespace, table_id) raise @@ -185,33 +158,40 @@ def merge_storage_options( return merged or None -def open_lance_dataset( - uri: str | os.PathLike[str] | None, - *, - storage_options: dict[str, Any] | None = None, - namespace_impl: str | None = None, - namespace_properties: dict[str, str] | None = None, - table_id: list[str] | None = None, - mode: str = "read", - **kwargs: Any, -) -> lance.LanceDataset: - validate_uri_or_namespace(uri, namespace_impl, table_id) - resolved_uri = str(uri) if uri is not None else None - namespace_storage_options = None - if resolved_uri is None: - resolved_uri, namespace_storage_options = resolve_namespace_table( - namespace_impl=namespace_impl, - namespace_properties=namespace_properties, - table_id=table_id, - mode=mode, - ) - if resolved_uri is None: - raise ValueError("Unable to resolve Lance dataset URI.") +def pop_namespace_params(open_kwargs: dict[str, Any]) -> tuple[str | None, dict[str, str] | None, list[str] | None]: + """Remove and return the namespace triple from a ``_lance_open_kwargs`` dict.""" + return ( + open_kwargs.pop("namespace_impl", None), + open_kwargs.pop("namespace_properties", None), + open_kwargs.pop("table_id", None), + ) + + +def namespace_kwargs_for_dataset(ds: Any) -> dict[str, Any]: + """Namespace kwargs for commit-style calls, recovered from ``ds._lance_open_kwargs``. + + ``lance.LanceDataset.commit`` is a static method, so a dataset opened through a + namespace does not carry its client into commits; re-derive it from the open + kwargs stashed by ``construct_lance_dataset``. + """ + open_kwargs = getattr(ds, "_lance_open_kwargs", None) or {} + return get_namespace_kwargs( + open_kwargs.get("namespace_impl"), + open_kwargs.get("namespace_properties"), + open_kwargs.get("table_id"), + ) + + +def open_dataset_from_open_kwargs(ds_uri: str | None, open_kwargs: dict[str, Any] | None) -> lance.LanceDataset: + """Re-open a dataset on a worker from serialized ``_lance_open_kwargs``. - merged_storage_options = merge_storage_options(storage_options, namespace_storage_options) + The namespace client is not picklable, so only the (impl, properties, table_id) + triple travels with the task; the client is re-created here via the lru cache. + """ + open_kwargs = dict(open_kwargs or {}) + namespace_impl, namespace_properties, table_id = pop_namespace_params(open_kwargs) return lance.dataset( - None if has_namespace_params(namespace_impl, table_id) else resolved_uri, - storage_options=merged_storage_options, + None if has_namespace_params(namespace_impl, table_id) else ds_uri, **get_namespace_kwargs(namespace_impl, namespace_properties, table_id), - **kwargs, + **open_kwargs, ) diff --git a/tests/io/lancedb/conftest.py b/tests/io/lancedb/conftest.py index dda9665..464714d 100644 --- a/tests/io/lancedb/conftest.py +++ b/tests/io/lancedb/conftest.py @@ -1,6 +1,33 @@ from __future__ import annotations +import os +import sys + import pytest # Try to import lance; if it fails, all tests in this directory will be skipped. lance = pytest.importorskip("lance") + + +_exitstatus = 0 + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + global _exitstatus + _exitstatus = exitstatus + + +@pytest.hookimpl(trylast=True) +def pytest_unconfigure(config: pytest.Config) -> None: + """Hard-exit after reporting to dodge an upstream teardown crash. + + Running enough ``daft`` + ``lance`` write/read cycles in one process makes the + interpreter SIGSEGV during native finalization (in lance's runtime, inside + OpenSSL cert teardown) *after* every test has already run and been reported. + This reproduces with plain ``daft.DataFrame.write_lance`` and is unrelated to + test outcomes. Runs last, once tracebacks and the summary are already flushed, + so it never hides a failure — it only skips the crashing native finalization. + """ + sys.stdout.flush() + sys.stderr.flush() + os._exit(_exitstatus) diff --git a/tests/io/lancedb/test_namespace.py b/tests/io/lancedb/test_namespace.py index c573a56..46ad10c 100644 --- a/tests/io/lancedb/test_namespace.py +++ b/tests/io/lancedb/test_namespace.py @@ -6,64 +6,177 @@ import daft_lance +def _dir_ns(tmp_path): + return {"namespace_impl": "dir", "namespace_properties": {"root": str(tmp_path)}} + + def test_namespace_write_read_append_roundtrip(tmp_path): - namespace_properties = {"root": str(tmp_path)} + ns = _dir_ns(tmp_path) table_id = ["roundtrip"] df1 = daft.from_pydict({"id": [1, 2], "label": ["a", "b"]}) df2 = daft.from_pydict({"id": [3], "label": ["c"]}) - df1.write_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, + daft_lance.write_lance(df1, table_id=table_id, mode="create", **ns).collect() + daft_lance.write_lance(df2, table_id=table_id, mode="append", **ns).collect() + + result = daft_lance.read_lance(table_id=table_id, **ns).to_pydict() + + assert result == {"id": [1, 2, 3], "label": ["a", "b", "c"]} + + +def test_namespace_overwrite(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["overwrite_tbl"] + + daft_lance.write_lance(daft.from_pydict({"id": [1, 2]}), table_id=table_id, mode="create", **ns).collect() + daft_lance.write_lance(daft.from_pydict({"id": [7, 8, 9]}), table_id=table_id, mode="overwrite", **ns).collect() + + result = daft_lance.read_lance(table_id=table_id, **ns).to_pydict() + assert result == {"id": [7, 8, 9]} + + +def test_namespace_overwrite_missing_table_declares(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["overwrite_fresh"] + + daft_lance.write_lance(daft.from_pydict({"id": [1]}), table_id=table_id, mode="overwrite", **ns).collect() + + result = daft_lance.read_lance(table_id=table_id, **ns).to_pydict() + assert result == {"id": [1]} + + +def test_namespace_read_supports_pushdowns(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["pushdowns"] + + daft_lance.write_lance( + daft.from_pydict( + { + "id": [1, 2, 3], + "label": ["a", "b", "c"], + "score": [10, 20, 30], + } + ), table_id=table_id, mode="create", + **ns, ).collect() - df2.write_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, + + result = daft_lance.read_lance(table_id=table_id, **ns).where(daft.col("score") > 10).select("label").to_pydict() + + assert result == {"label": ["b", "c"]} + + +def test_namespace_count_pushdown(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["count_tbl"] + + daft_lance.write_lance(daft.from_pydict({"id": list(range(10))}), table_id=table_id, mode="create", **ns).collect() + + assert daft_lance.read_lance(table_id=table_id, **ns).count_rows() == 10 + + +def test_namespace_patch_daft(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["patched"] + + daft_lance.patch_daft() + daft_lance.patch_daft() # idempotent + + daft.from_pydict({"id": [1, 2]}).write_lance(table_id=table_id, mode="create", **ns).collect() + result = daft.read_lance(table_id=table_id, **ns).to_pydict() + assert result == {"id": [1, 2]} + + # Plain-URI writes keep working through the patched method. + uri = str(tmp_path / "plain.lance") + daft.from_pydict({"id": [5]}).write_lance(uri).collect() + assert daft.read_lance(uri).to_pydict() == {"id": [5]} + + +def test_namespace_write_lance_merge_new_columns(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["merge_tbl"] + + daft_lance.write_lance( + daft.from_pydict({"id": [1, 2, 3], "score": [10, 20, 30]}), table_id=table_id, - mode="append", + mode="create", + **ns, ).collect() - result = daft.read_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, + df = daft_lance.read_lance( table_id=table_id, - ).to_pydict() + default_scan_options={"with_row_address": True}, + include_fragment_id=True, + **ns, + ) + df = df.with_column("doubled", df["score"] * 2) + daft_lance.write_lance(df, table_id=table_id, mode="merge", **ns).collect() - assert result == {"id": [1, 2, 3], "label": ["a", "b", "c"]} + result = daft_lance.read_lance(table_id=table_id, **ns).sort("id").to_pydict() + assert result["doubled"] == [20, 40, 60] -def test_namespace_read_supports_pushdowns(tmp_path): - namespace_properties = {"root": str(tmp_path)} - table_id = ["pushdowns"] +def test_namespace_merge_columns_df(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["merge_cols_df"] - daft.from_pydict( - { - "id": [1, 2, 3], - "label": ["a", "b", "c"], - "score": [10, 20, 30], - } - ).write_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, + daft_lance.write_lance( + daft.from_pydict({"id": [1, 2, 3], "score": [1, 2, 3]}), table_id=table_id, mode="create", + **ns, ).collect() - result = ( - daft.read_lance( - namespace_impl="dir", - namespace_properties=namespace_properties, - table_id=table_id, - ) - .where(daft.col("score") > 10) - .select("label") - .to_pydict() + df = daft_lance.read_lance( + table_id=table_id, + default_scan_options={"with_row_address": True}, + include_fragment_id=True, + **ns, ) + df = df.with_column("tripled", df["score"] * 3) + daft_lance.merge_columns_df(df.select("fragment_id", "_rowaddr", "tripled"), table_id=table_id, **ns) - assert result == {"label": ["b", "c"]} + result = daft_lance.read_lance(table_id=table_id, **ns).sort("id").to_pydict() + assert result["tripled"] == [3, 6, 9] + + +def test_namespace_create_scalar_index(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["indexed"] + + daft_lance.write_lance( + daft.from_pydict({"id": list(range(100)), "price": [i * 2 for i in range(100)]}), + table_id=table_id, + mode="create", + **ns, + ).collect() + + daft_lance.create_scalar_index(table_id=table_id, column="price", index_type="BTREE", **ns) + + import lance + import lance_namespace as ln + from lance_namespace import DescribeTableRequest + + namespace = ln.connect("dir", {"root": str(tmp_path)}) + location = namespace.describe_table(DescribeTableRequest(id=table_id)).location + indices = lance.dataset(location).list_indices() + assert any(idx["fields"] == ["price"] for idx in indices) + + +def test_namespace_compact_files(tmp_path): + ns = _dir_ns(tmp_path) + table_id = ["compacted"] + + daft_lance.write_lance(daft.from_pydict({"id": [1]}), table_id=table_id, mode="create", **ns).collect() + for i in range(3): + daft_lance.write_lance(daft.from_pydict({"id": [i + 2]}), table_id=table_id, mode="append", **ns).collect() + + daft_lance.compact_files(table_id=table_id, compaction_options={"target_rows_per_fragment": 1024}, **ns) + + result = daft_lance.read_lance(table_id=table_id, **ns).sort("id").to_pydict() + assert result == {"id": [1, 2, 3, 4]} def test_namespace_rejects_uri_and_namespace(tmp_path): @@ -82,3 +195,13 @@ def test_namespace_requires_table_id(tmp_path): namespace_impl="dir", namespace_properties={"root": str(tmp_path)}, ) + + +def test_namespace_requires_impl(): + with pytest.raises(ValueError, match="'namespace_impl' must be provided"): + daft_lance.read_lance(table_id=["tbl"]) + + +def test_namespace_requires_uri_or_namespace(): + with pytest.raises(ValueError, match="Must provide either 'uri' OR"): + daft_lance.read_lance() diff --git a/tests/io/lancedb/test_namespace_rest_e2e.py b/tests/io/lancedb/test_namespace_rest_e2e.py index d9f27e9..a193e2e 100644 --- a/tests/io/lancedb/test_namespace_rest_e2e.py +++ b/tests/io/lancedb/test_namespace_rest_e2e.py @@ -6,7 +6,7 @@ import pytest import daft -import daft_lance # noqa: F401 - patches daft.read_lance and DataFrame.write_lance. +import daft_lance pytestmark = pytest.mark.skipif( os.environ.get("DAFT_LANCE_REST_URI") is None, @@ -23,6 +23,7 @@ def test_rest_namespace_write_read_append_roundtrip() -> None: catalog = os.environ.get("DAFT_LANCE_REST_CATALOG", "lance_catalog") schema = os.environ.get("DAFT_LANCE_REST_SCHEMA", "daft_ns_e2e") table_id = [catalog, schema, f"orders_{uuid.uuid4().hex[:8]}"] + ns = {"namespace_impl": "rest", "namespace_properties": namespace_properties} namespace = ln.connect("rest", namespace_properties) try: @@ -30,41 +31,25 @@ def test_rest_namespace_write_read_append_roundtrip() -> None: except Exception: namespace.create_namespace(CreateNamespaceRequest(id=[catalog, schema], mode="CREATE")) - daft.from_pydict( - { - "id": [1, 2, 3], - "label": ["a", "b", "c"], - "score": [10, 20, 30], - } - ).write_lance( - namespace_impl="rest", - namespace_properties=namespace_properties, + daft_lance.write_lance( + daft.from_pydict({"id": [1, 2, 3], "label": ["a", "b", "c"], "score": [10, 20, 30]}), table_id=table_id, mode="create", + **ns, ).collect() - daft.from_pydict( - { - "id": [4, 5], - "label": ["d", "e"], - "score": [40, 50], - } - ).write_lance( - namespace_impl="rest", - namespace_properties=namespace_properties, + daft_lance.write_lance( + daft.from_pydict({"id": [4, 5], "label": ["d", "e"], "score": [40, 50]}), table_id=table_id, mode="append", + **ns, ).collect() describe = namespace.describe_table(DescribeTableRequest(id=table_id)) location = getattr(describe, "location", None) or getattr(describe, "table_uri", None) assert location - result = daft.read_lance( - namespace_impl="rest", - namespace_properties=namespace_properties, - table_id=table_id, - ).to_pydict() + result = daft_lance.read_lance(table_id=table_id, **ns).to_pydict() assert result == { "id": [1, 2, 3, 4, 5], "label": ["a", "b", "c", "d", "e"], @@ -72,23 +57,23 @@ def test_rest_namespace_write_read_append_roundtrip() -> None: } filtered = ( - daft.read_lance( - namespace_impl="rest", - namespace_properties=namespace_properties, - table_id=table_id, - ) - .where(daft.col("score") >= 30) - .select("id", "label") - .to_pydict() + daft_lance.read_lance(table_id=table_id, **ns).where(daft.col("score") >= 30).select("id", "label").to_pydict() ) assert filtered == {"id": [3, 4, 5], "label": ["c", "d", "e"]} - assert ( - daft.read_lance( - namespace_impl="rest", - namespace_properties=namespace_properties, - table_id=table_id, - ).count_rows() - == 5 - ) + assert daft_lance.read_lance(table_id=table_id, **ns).count_rows() == 5 assert lance.dataset(None, namespace_client=namespace, table_id=table_id).count_rows() == 5 + + +def test_rest_namespace_patch_daft_roundtrip() -> None: + """The opt-in ``patch_daft()`` path: ``daft.read_lance`` / ``df.write_lance``.""" + namespace_properties = {"uri": os.environ["DAFT_LANCE_REST_URI"]} + catalog = os.environ.get("DAFT_LANCE_REST_CATALOG", "lance_catalog") + schema = os.environ.get("DAFT_LANCE_REST_SCHEMA", "daft_ns_e2e") + table_id = [catalog, schema, f"patched_{uuid.uuid4().hex[:8]}"] + ns = {"namespace_impl": "rest", "namespace_properties": namespace_properties} + + daft_lance.patch_daft() + + daft.from_pydict({"id": [1, 2]}).write_lance(table_id=table_id, mode="create", **ns).collect() + assert daft.read_lance(table_id=table_id, **ns).to_pydict() == {"id": [1, 2]}