From 9a61cb3c1d93cd83795c5afa3ee6cfa2d9652d0c Mon Sep 17 00:00:00 2001 From: Martin Prammer Date: Sat, 10 Oct 2026 16:57:22 -0400 Subject: [PATCH 1/2] Add single-file Lance exports and Python loading Signed-off-by: Martin Prammer --- .github/workflows/ci.yml | 2 +- AGENTS.md | 10 +- CHANGELOG.md | 8 ++ README.md | 27 +++- pyproject.toml | 4 +- raincloud/_readers.py | 26 +++- raincloud/_registry.py | 3 + raincloud/pipeline/export/__init__.py | 3 +- raincloud/pipeline/export/base.py | 4 +- raincloud/pipeline/export/bounded.py | 12 +- raincloud/pipeline/export/exporters.py | 58 ++++++++ raincloud/pipeline/export/readers.py | 17 +++ raincloud/pipeline/spec.py | 8 ++ sidecars/README.md | 18 +++ sources.schema.json | 2 + sources.schema.md | 7 + tests/test_lance.py | 176 +++++++++++++++++++++++++ tests/test_manifest_policy.py | 12 +- 18 files changed, 379 insertions(+), 18 deletions(-) create mode 100644 tests/test_lance.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 10c0222..52d25fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,7 +38,7 @@ jobs: # kaggle/huggingface -- those deps are only # needed at fetch time and neither test path imports them. - name: Install dependencies - run: uv sync --extra dev --extra tui --extra build --extra pandas --extra osm --extra sas --extra excel --extra archives --inexact + run: uv sync --extra dev --extra tui --extra lance --extra build --extra pandas --extra osm --extra sas --extra excel --extra archives --inexact - name: Lint (ruff) run: uv run ruff check diff --git a/AGENTS.md b/AGENTS.md index 3f83bbb..70c66d8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -157,7 +157,7 @@ becomes part of its toolchain, and a writer whose library cannot do what it asks that export as a measurement rather than writing something else; `sidecars/README.md` tabulates which writer honours what, and the `/raincloud-write-settings` skill is the procedure (a setting changes only files this install writes, so an existing file must be -re-exported or rebuilt to gain it). ORC, Avro and Vortex have the same kind of settings +re-exported or rebuilt to gain it). ORC, Avro, Lance and Vortex have the same kind of settings (`spec.FORMAT_SETTINGS`, `RAINCLOUD_ORC_*`, `RAINCLOUD_AVRO_*`, `RAINCLOUD_VORTEX_*`), read and refused the same way; their codec, unset, is the zstd raincloud has always written. @@ -180,6 +180,11 @@ Run an unattended build as a systemd unit with `OOMPolicy=continue`: the default with a bare `--format vortex` it records the failure and exits 1, and a named cell's failure (`--format vortex@rs`) exits 1 and records nothing. +Lance (`lance@py`, optional `[lance]`) writes one `.lance` data file, not a Lance +dataset directory. Its exporter uses a spawned process, including on Linux, because +Lance is not fork-safe. The loader supports projected batches and `to_arrow()`; +`dataset()` has no lazy single-file adapter and raises `UnsupportedType`. + Every writer reads back what it writes before its file is promoted. An in-process writer reads its file with the same format's in-process reader and compares it to the canonical (`exporters.read_back`), streamed window by window (`compare.stream_equal`: @@ -294,6 +299,9 @@ its own `outputs/` rather than a machine's shared store. | `RAINCLOUD_AVRO_COMPRESSION` | Avro codec in every Avro writer: `zstd`, `deflate`, `snappy`, `bzip2`, `xz`, `none` | unset (zstd, as always) | | `RAINCLOUD_AVRO_COMPRESSION_LEVEL` | level for zstd (1-22), deflate or xz (0-9); Avro Java only, arrow-avro refuses | unset (each writer's own) | | `RAINCLOUD_AVRO_BLOCK_BYTES` | Avro block (sync interval) size; Avro Java only, arrow-avro refuses | unset (each writer's own) | +| `RAINCLOUD_LANCE_FILE_VERSION` | single-file Lance format version (`2.1` or `2.2`) | unset (2.2) | +| `RAINCLOUD_LANCE_PAGE_BYTES` | Lance page byte limit; `lance@py` refuses it because the structural encoder ignores this option | unset (no page limit requested) | +| `RAINCLOUD_LANCE_DATA_CACHE_BYTES` | Lance total column buffer budget (`0`: no configured limit) | unset (library default: 8 MiB per top-level column) | | `RAINCLOUD_VORTEX_COMPACT` | `1`: BtrBlocks' compact encodings (vortex@py and vortex@rs; vortex@jni refuses) | unset (each writer's own: default) | | `RAINCLOUD_VORTEX_ROW_BLOCK_ROWS` | Vortex row block size (vortex@rs only) | unset (each writer's own) | | `RAINCLOUD_VORTEX_DATA_BLOCK_BYTES` | Vortex data block size target (vortex@rs only) | unset (each writer's own) | diff --git a/CHANGELOG.md b/CHANGELOG.md index f1f9005..0c1584d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +### Added + +- `lance@py`: single-file Lance 2.2 artifacts using optional `raincloud[lance]` + (`pylance` 13.0.0), with streamed canonical input, bounded read-back verification, + projected loader batches, and file-version and buffer-budget settings. Explicit + page-size limits are refused because the structural encoder ignores that option. + Exports spawn a fresh process because Lance is not fork-safe. + ### Changed - **`parquet@py` writes a page index and page checksums by default.** With diff --git a/README.md b/README.md index 7c4fff5..f0451ad 100644 --- a/README.md +++ b/README.md @@ -304,7 +304,7 @@ served as it is, and `build=True` builds only what is missing, so rebuild (or re from a kept canonical, `python -m raincloud.pipeline.export uci-iris --format parquet`) to rewrite one. A writer that cannot do what a setting asks refuses it, and the format is recorded unavailable with the reason. Every setting, its default, and which writer honours -it: the `RAINCLOUD_PARQUET_*`, `_ORC_*`, `_AVRO_*` and `_VORTEX_*` rows of +it: the `RAINCLOUD_PARQUET_*`, `_ORC_*`, `_AVRO_*`, `_LANCE_*` and `_VORTEX_*` rows of [AGENTS.md](AGENTS.md#data-locations) and [`sidecars/README.md`](sidecars/README.md). A build does not fail because one format's writer cannot handle a dataset. Every @@ -323,6 +323,31 @@ its own. To try again with nothing changed, pass `--retry-errors` (`raincloud build --retry-errors`, `raincloud load --retry-errors`, or `load(..., retry_errors=True)`). +### Single-file Lance + +Install `raincloud[lance]` (`uv sync --extra lance --extra build --inexact` in a +checkout), then export an existing canonical with: + +```bash +python -m raincloud.pipeline.export --format lance +``` + +This writes `lance/.lance`, a single Lance data file in format version 2.2, +using `pylance` 13.0.0. It is read back before publication and available through +`raincloud.load(slug, format="lance")`, `batches()`, `to_arrow()` and `to_pandas()`. +Projected batches read only requested columns. The single-file API has no lazy +PyArrow Dataset adapter, so `dataset()` raises `UnsupportedType`; this is not a +Lance dataset directory for `lance.dataset()`. + +Lance is opt-in (included by `formats=all` when installed) and does not change +`auto` selection. `RAINCLOUD_LANCE_FILE_VERSION` chooses `2.1` or `2.2`, +`RAINCLOUD_LANCE_DATA_CACHE_BYTES` sets the total column buffer budget (unset: +8 MiB per top-level column). `RAINCLOUD_LANCE_PAGE_BYTES` is explicitly refused: +the pinned structural encoder ignores the library's `max_page_bytes` option. +Parquet settings do not apply to Lance. +Unsupported data types become recorded export failures, as for every format. + + ## Licensing and redistribution Each entry carries an SPDX identifier, a source URL, and an explicit diff --git a/pyproject.toml b/pyproject.toml index 0aee654..fd05bdd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,8 @@ raincloud = "raincloud.cli:main" # Exact: the Vortex file format is still versioned by its writer, and the # snapshot's checksums are recorded against this one. vortex = ["vortex-data==0.86.1"] +# Single-file Lance artifacts, with reproducible writer provenance. +lance = ["pylance==13.0.0"] s3 = ["s3fs>=2024.0"] http = ["aiohttp>=3.9"] pandas = ["pandas>=2.0"] @@ -61,7 +63,7 @@ huggingface = ["huggingface-hub>=0.25"] tui = ["textual>=0.80"] dev = ["pytest>=8.0", "ruff>=0.13"] all = [ - "raincloud[s3,http,pandas,build,osm,sas,excel,archives,generated,kaggle,huggingface,tui]", + "raincloud[s3,http,pandas,build,lance,osm,sas,excel,archives,generated,kaggle,huggingface,tui]", ] [tool.ruff] diff --git a/raincloud/_readers.py b/raincloud/_readers.py index 4784b47..1013fe8 100644 --- a/raincloud/_readers.py +++ b/raincloud/_readers.py @@ -63,6 +63,9 @@ def open_dataset(fmt: str, path): """A lazy pyarrow Dataset over the open file at `path`.""" import pyarrow as pa import pyarrow.dataset as pads + if fmt == "lance": + from .exceptions import UnsupportedType + raise UnsupportedType("Single-file Lance has no PyArrow Dataset adapter; use batches() or to_arrow()") if fmt == "vortex": import vortex return vortex.open(str(path)).to_dataset() @@ -126,5 +129,26 @@ def _orc_batches(path, stack, *, columns, batch_size, check_columns): return schema, (file.read_stripe(i, columns=columns) for i in range(file.nstripes)) +def _lance_batches(path, stack, *, columns, batch_size, check_columns): + from lance.file import LanceFileReader + + try: + file = LanceFileReader(str(path), columns=columns or None) + except OSError: + if columns is not None: + # The constructor validates projection. Recover the full schema + # only on this error path to report UnknownColumn consistently. + check_columns(LanceFileReader(str(path)).metadata().schema) + raise + if columns is not None: + check_columns(file.metadata().schema) + reader = stack.enter_context(file.read_all(batch_size=batch_size, batch_readahead=1).to_batches()) + if columns == []: + # Lance treats no columns as all columns; preserve the loader's empty projection. + import pyarrow as pa + return pa.schema([], metadata=reader.schema.metadata), (batch.select([]) for batch in reader) + return reader.schema, reader + + _BATCHES = {"parquet": _parquet_batches, "arrow": _arrow_batches, "vortex": _vortex_batches, - "orc": _orc_batches} + "orc": _orc_batches, "lance": _lance_batches} diff --git a/raincloud/_registry.py b/raincloud/_registry.py index 00e2843..a18b882 100644 --- a/raincloud/_registry.py +++ b/raincloud/_registry.py @@ -75,6 +75,8 @@ "orc": {"ext": "orc", "auto": False, "reader": "pyarrow._orc", "implementation": "pyarrow"}, # pyarrow reads no Avro: the loader serves the file's path. "avro": {"ext": "avro", "auto": False, "reader": None, "implementation": None}, + "lance": {"ext": "lance", "auto": False, "reader": "lance", + "implementation": "lance-python", "extra": "lance"}, # Nimble is C++ only: served by path. "nimble": {"ext": "nimble", "auto": False, "reader": None, "implementation": None}, } @@ -86,6 +88,7 @@ "parquet@py": "exporters:ParquetExporter", "vortex@py": "exporters:VortexExporter", "orc@py": "exporters:OrcExporter", + "lance@py": "exporters:LanceExporter", } # Opt-in exporter cells that shell out to a PATH-discovered reference writer, diff --git a/raincloud/pipeline/export/__init__.py b/raincloud/pipeline/export/__init__.py index e78e2f9..732ec69 100644 --- a/raincloud/pipeline/export/__init__.py +++ b/raincloud/pipeline/export/__init__.py @@ -194,7 +194,8 @@ def _error_text(text: str) -> str: # The Python distributions an in-process writer runs. -_DISTRIBUTIONS = {"parquet@py": ("pyarrow",), "vortex@py": ("vortex-data", "pyarrow"), "orc@py": ("pyarrow",)} +_DISTRIBUTIONS = {"parquet@py": ("pyarrow",), "vortex@py": ("vortex-data", "pyarrow"), "orc@py": ("pyarrow",), + "lance@py": ("pylance", "pyarrow")} def writer_toolchain(exporter: Exporter) -> dict[str, str]: diff --git a/raincloud/pipeline/export/base.py b/raincloud/pipeline/export/base.py index af7b8b9..4ceb06a 100644 --- a/raincloud/pipeline/export/base.py +++ b/raincloud/pipeline/export/base.py @@ -93,7 +93,9 @@ class Exporter(Protocol): that is a subprocess already sets `bounds_itself = True` and applies `bounded.export_timeout()` itself, and may define `toolchain()` -> the versions that identify it, recorded when it cannot produce a format - (`export.writer_toolchain` names an in-process writer's libraries). + (`export.writer_toolchain` names an in-process writer's libraries). A native + library that is not fork-safe sets `process_start_method = "spawn"`; that + choice also applies to read-back of files written earlier. """ format_id: str diff --git a/raincloud/pipeline/export/bounded.py b/raincloud/pipeline/export/bounded.py index 0ba2d8c..dce3686 100644 --- a/raincloud/pipeline/export/bounded.py +++ b/raincloud/pipeline/export/bounded.py @@ -18,7 +18,8 @@ runs native code that a Python thread cannot interrupt, so it runs in a child process instead: forked on Linux, so the child shares the parent's imports, configuration and catalog selection and costs no interpreter start; spawned -elsewhere (macOS cannot fork a threaded process safely). The child writes into +elsewhere (macOS cannot fork a threaded process safely). A writer may require +spawn on every platform (`lance@py`, whose native runtime is not fork-safe). The child writes into a private directory beside the destination, and the parent moves the file into place only after the child reports success: a child that is killed, crashes or raises leaves nothing at the destination, and the caller's `Publication` @@ -81,7 +82,7 @@ def run_bounded(exporter, spec: dict, canonical: Path, dest: Path) -> ExportResu written = work / dest.name try: result = _bounded_call(f"export {exporter.cell_id}", exporter.cell_id, exporter.export, - (spec, canonical, written)) + (spec, canonical, written), start_method=getattr(exporter, "process_start_method", None)) if result.out_path != written or not written.is_file(): raise ExportFailed(f"{exporter.cell_id}: the writer reported {result.out_path}, " f"not the file it was given") @@ -109,15 +110,16 @@ def read_back_bounded(exporter, artifact: Path, canonical: Path) -> tuple[bool, if missing is not None: raise BuildToolingMissing(f"{exporter.cell_id} {missing}") return _bounded_call(f"read back {exporter.cell_id}", exporter.cell_id, read_back, - (exporter.cell_id, artifact, canonical)) + (exporter.cell_id, artifact, canonical), + start_method=getattr(exporter, "process_start_method", None)) -def _bounded_call(name: str, cell: str, fn, args: tuple): +def _bounded_call(name: str, cell: str, fn, args: tuple, *, start_method: str | None = None): """`fn(*args)` in a child process under the time and memory ceilings: its return value, or ExportFailed naming what ended it.""" limit = export_timeout() memory = export_memory() - context = multiprocessing.get_context(_START_METHOD) + context = multiprocessing.get_context(start_method or _START_METHOD) receiver, sender = context.Pipe(duplex=False) child = context.Process(target=_call_in_child, name=name, args=(sender, fn, args, os.getpid())) started = reported = False diff --git a/raincloud/pipeline/export/exporters.py b/raincloud/pipeline/export/exporters.py index b1d0865..e85b86c 100644 --- a/raincloud/pipeline/export/exporters.py +++ b/raincloud/pipeline/export/exporters.py @@ -554,6 +554,64 @@ def export(self, spec: dict, canonical: Path, dest: Path | None = None) -> Expor ) +class LanceExporter: + """A single Lance data file, streamed from canonical IPC (not a dataset directory).""" + + format_id = "lance" + cell_id = "lance@py" + # Lance owns runtime threads and explicitly warns against fork. + process_start_method = "spawn" + + def unavailable(self) -> str | None: + if importlib.util.find_spec("lance") is not None: + return None + return "needs pylance; install `raincloud[lance]`" + + def out_path(self, slug: str) -> Path: + return prepared_artifact(slug, self.format_id) + + def export(self, spec: dict, canonical: Path, dest: Path | None = None) -> ExportResult: + missing = self.unavailable() + if missing: + from raincloud.exceptions import BuildToolingMissing + raise BuildToolingMissing(f"{self.cell_id} {missing}") + from lance.file import LanceFileWriter + + settings = write_settings("lance") + if settings["page_bytes"] is not None: + raise UnsupportedOption( + f"lance@py cannot honour RAINCLOUD_LANCE_PAGE_BYTES={settings['page_bytes']}: " + "pylance's structural encoder does not enforce max_page_bytes in formats 2.1/2.2; " + "RAINCLOUD_LANCE_DATA_CACHE_BYTES controls column buffering") + options = {"version": settings["file_version"] or "2.2"} + if settings["data_cache_bytes"] is not None: + options["data_cache_bytes"] = settings["data_cache_bytes"] + dest = dest or self.out_path(slug_from_canonical(canonical)) + dest.parent.mkdir(parents=True, exist_ok=True) + tmp = tmp_path(dest) + print(f"[export:{self.cell_id}] {display_path(dest)}") + try: + with pa.ipc.open_file(str(canonical)) as reader: + variant = has_variant(reader.schema) + with LanceFileWriter(str(tmp), schema=reader.schema, **options) as writer: + for i in range(reader.num_record_batches): + writer.write_batch(reader.get_batch(i)) + roundtrip, why = read_back(self.cell_id, tmp, canonical) + nbytes, sha = tmp.stat().st_size, sha256_file(tmp) + if not roundtrip: + raise RuntimeError(why) + tmp.replace(dest) + finally: + tmp.unlink(missing_ok=True) + note = "VARIANT column stored as its shredded struct — Lance has no Parquet VARIANT logical type" if variant else "" + return ExportResult( + format_id=self.cell_id, out_path=dest, nbytes=nbytes, sha256=sha, + compliance=Compliance(roundtrip=roundtrip, variant_faithful=not variant, + note="; ".join(n for n in (why, note) if n)), + ) + + register(ParquetExporter()) register(VortexExporter()) register(OrcExporter()) +register(LanceExporter()) diff --git a/raincloud/pipeline/export/readers.py b/raincloud/pipeline/export/readers.py index b137c61..6420ebd 100644 --- a/raincloud/pipeline/export/readers.py +++ b/raincloud/pipeline/export/readers.py @@ -297,6 +297,22 @@ def _run() -> Verdict: return _guarded(self.reader_id, _run) +class LancePyReader: + """Single-file Lance reader; one batch in flight during verification.""" + + reader_id = "lance@py" + formats = {"lance"} + + def read_conformance(self, artifact: Path, canonical: Path) -> Verdict: + def run() -> Verdict: + from lance.file import LanceFileReader + + file = LanceFileReader(str(artifact)) + with file.read_all(batch_size=1024, batch_readahead=1).to_batches() as batches: + return stream_verdict(self.reader_id, (batches.schema, file.num_rows(), batches), canonical) + return _guarded(self.reader_id, run) + + class PyarrowOrcReader: """In-process `orc@py` reader — pyarrow's ORC stripes vs the canonical's batches, streamed.""" @@ -472,6 +488,7 @@ def run_reader( register_reader(PyarrowParquetReader()) register_reader(VortexPyReader()) register_reader(PyarrowOrcReader()) +register_reader(LancePyReader()) register_reader(SidecarReader("parquet@rs", {"parquet"}, "raincloud-read-parquet-rs")) register_reader(SidecarReader("vortex@rs", {"vortex"}, "raincloud-read-vortex-rs")) register_reader(SidecarReader("parquet@java", {"parquet"}, "raincloud-read-parquet-java")) diff --git a/raincloud/pipeline/spec.py b/raincloud/pipeline/spec.py index b2c707a..e4c3922 100644 --- a/raincloud/pipeline/spec.py +++ b/raincloud/pipeline/spec.py @@ -411,6 +411,9 @@ def parquet_options(spec: dict) -> ParquetOptions: AVRO_COMPRESSION = "RAINCLOUD_AVRO_COMPRESSION" AVRO_COMPRESSION_LEVEL = "RAINCLOUD_AVRO_COMPRESSION_LEVEL" AVRO_BLOCK_BYTES = "RAINCLOUD_AVRO_BLOCK_BYTES" +LANCE_FILE_VERSION = "RAINCLOUD_LANCE_FILE_VERSION" +LANCE_PAGE_BYTES = "RAINCLOUD_LANCE_PAGE_BYTES" +LANCE_DATA_CACHE_BYTES = "RAINCLOUD_LANCE_DATA_CACHE_BYTES" VORTEX_COMPACT = "RAINCLOUD_VORTEX_COMPACT" VORTEX_ROW_BLOCK_ROWS = "RAINCLOUD_VORTEX_ROW_BLOCK_ROWS" VORTEX_DATA_BLOCK_BYTES = "RAINCLOUD_VORTEX_DATA_BLOCK_BYTES" @@ -434,6 +437,11 @@ def read(var: str) -> str | None: # (field, variable, reader) per format, as `_PARQUET_SETTINGS`. FORMAT_SETTINGS = { + "lance": ( + ("file_version", LANCE_FILE_VERSION, _env_choice("2.1", "2.2")), + ("page_bytes", LANCE_PAGE_BYTES, _env_limit), + ("data_cache_bytes", LANCE_DATA_CACHE_BYTES, _env_limit), + ), "orc": ( ("compression", ORC_COMPRESSION, _env_choice(*ORC_CODECS)), ("compression_strategy", ORC_COMPRESSION_STRATEGY, _env_choice("speed", "compression")), diff --git a/sidecars/README.md b/sidecars/README.md index 65931fd..ad810e9 100644 --- a/sidecars/README.md +++ b/sidecars/README.md @@ -407,3 +407,21 @@ Upgrade their pins together and re-measure supported cells; a package version change alone is not evidence of conformance. Preserve skipped and unsupported lanes in the results. Software updates do not move stored data or refresh an active catalog revision. + +## Single-file Lance + +`lance@py` is in-process and installed by `raincloud[lance]`; it does not need a +sidecar binary. It uses `lance.file.LanceFileWriter` / `LanceFileReader` from +pylance 13.0.0, stores a single `.lance` file, and supports file formats 2.1 and +2.2 (default 2.2). Every export and read-back runs in a spawned, bounded process. + +| setting | `lance@py` | +|---|---| +| `RAINCLOUD_LANCE_FILE_VERSION` | `2.1` or `2.2` | +| `RAINCLOUD_LANCE_DATA_CACHE_BYTES` | total buffer budget, split across columns; unset is 8 MiB per column | +| `RAINCLOUD_LANCE_PAGE_BYTES` | refused: the structural encoder ignores `max_page_bytes` | + +The buffer budget affects when pages are flushed. It is not a hard bound on the +size of one value. VARIANT storage structs can survive, but the writer does not +claim a Parquet VARIANT logical annotation. Unsupported types are measurements, +not permanent format exclusions in the manifest. diff --git a/sources.schema.json b/sources.schema.json index 8f96921..58c86f8 100644 --- a/sources.schema.json +++ b/sources.schema.json @@ -712,6 +712,7 @@ "parquet", "vortex", "orc", + "lance", "avro", "nimble" ] @@ -791,6 +792,7 @@ "parquet", "vortex", "orc", + "lance", "avro", "nimble" ] diff --git a/sources.schema.md b/sources.schema.md index 92148a4..19157ac 100644 --- a/sources.schema.md +++ b/sources.schema.md @@ -276,3 +276,10 @@ A hydrated dataset derives from another entry instead of an upstream, so it carr Its recipe folds in the parent's, so rebuilding the parent makes it stale. See [`HYDRATING.md`](HYDRATING.md). + +`export.formats` may include `lance`: one single-file Lance artifact at +`lance/.lance`, written by `lance@py` with the optional `[lance]` extra. +As with other formats, `export.priority` may name its writer (`py`) and the +install chooses whether to export it. This is a data file, not a Lance dataset +directory. The writer's file-version and page settings are install settings; +see the [settings table](AGENTS.md#data-locations). diff --git a/tests/test_lance.py b/tests/test_lance.py new file mode 100644 index 0000000..edd9e3c --- /dev/null +++ b/tests/test_lance.py @@ -0,0 +1,176 @@ +# SPDX-FileCopyrightText: 2026 Raincloud Maintainers +# SPDX-License-Identifier: Apache-2.0 +"""Single-file Lance artifacts: streaming, settings, loader, and measured failures.""" +from __future__ import annotations + +import pyarrow as pa +import pytest + +import raincloud +from raincloud import _builds +from raincloud._resolve import artifact_key +from raincloud.catalogs import operation +from raincloud.pipeline import build +from raincloud.pipeline.export import get_exporter, get_reader, run_bounded, writer_toolchain +from raincloud.pipeline.export.bounded import ExportFailed, read_back_bounded +from raincloud.pipeline.export.compare import stream_equal +from raincloud.pipeline.spec import FORMAT_SETTINGS, prepared_artifact, write_settings +from tests._helpers import write_ipc +from tests.test_pipeline_contracts import TABLE, _catalog +from tests.test_pipeline_contracts import stages as stages # noqa: F401 + +pytest.importorskip("lance") +from lance.file import LanceFileReader, LanceFileWriter # noqa: E402 + + +@pytest.fixture(autouse=True) +def settings(monkeypatch): + for _, var, _ in FORMAT_SETTINGS["lance"]: + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("RAINCLOUD_EXPORT_TIMEOUT", "15") + + +@pytest.mark.parametrize("empty", [False, True]) +def test_lance_streams_nested_data_and_empty_schema(tmp_path, empty): + table = pa.table({"x": pa.array([0., -0., float("nan"), None]), + "s": ["a", None, "", "é"], + "nested": pa.array([[1, None], None, [], [3]], pa.list_(pa.int32()))}) + if empty: + table = table.slice(0, 0) + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", table, max_chunksize=1) + result = get_exporter("lance@py").export({"slug": "tiny"}, canonical, tmp_path / "tiny.lance") + assert result.compliance.roundtrip is True, result.compliance + assert result.out_path.is_file() + reader = LanceFileReader(str(result.out_path)) + meta = reader.metadata() + assert (meta.major_version, meta.minor_version) == (2, 2) + batches = reader.read_all(batch_size=2, batch_readahead=1).to_batches() + assert stream_equal(batches.schema, batches, table.schema, table.to_batches())[0] + assert get_reader("lance@py").read_conformance(result.out_path, canonical).status == "pass" + + +def test_lance_export_runs_after_parent_has_used_lance(tmp_path): + # Native runtime threads initialized above must not deadlock a later bounded export. + table = pa.table({"x": [1, 2]}) + with LanceFileWriter(str(tmp_path / "parent.lance")) as writer: + writer.write_batch(table) + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", table) + result = run_bounded(get_exporter("lance@py"), {"slug": "tiny"}, canonical, tmp_path / "bounded.lance") + assert result.compliance.roundtrip is True + assert read_back_bounded(get_exporter("lance@py"), result.out_path, canonical)[0] + + +def test_lance_build_load_and_record(tmp_path, stages): + spec = {"slug": "tiny"} + cfg = _catalog(tmp_path, "lance", [spec]) + with operation(cfg): + assert build.run_one(spec, strict=False, formats=["lance"]) + assert prepared_artifact("tiny", "lance").is_file() + ds = raincloud.load("tiny", format="lance", config=cfg, offline=True) + assert ds.path().name == "tiny.lance" + assert ds.to_arrow().equals(TABLE) + assert ds.schema.equals(TABLE.schema) + with ds.batches(batch_size=1, columns=["url", "x"]) as batches: + assert pa.Table.from_batches(list(batches)).equals(TABLE.select(["url", "x"])) + with ds.batches(batch_size=1) as batches: + assert next(batches).num_rows == 1 # closes an early-terminated reader + with pytest.raises(raincloud.UnknownColumn): + with ds.batches(columns=["absent"]): + pass + with pytest.raises(raincloud.UnsupportedType, match="batches"): + ds.dataset() + entry = _builds.read(cfg.data_dir)[artifact_key("tiny", "lance", 2)] + assert entry["verified"] is True and entry["writer"] == "py" + assert "pylance" in writer_toolchain(get_exporter("lance@py")) + + +def test_lance_reader_keeps_open_generation(tmp_path, stages): + cfg = _catalog(tmp_path, "generation", [{"slug": "tiny"}]) + with operation(cfg): + assert build.run_one({"slug": "tiny"}, strict=False, formats=["lance"]) + ds = raincloud.load("tiny", format="lance", config=cfg, offline=True) + other = tmp_path / "replacement.lance" + with LanceFileWriter(str(other)) as writer: + writer.write_batch(pa.table({"x": [9, 10], "url": ["changed", "changed"]})) + with ds.batches(batch_size=1) as batches: + first = next(batches) + other.replace(ds.path()) + assert pa.Table.from_batches([first, *batches]).equals(TABLE) + + +@pytest.mark.parametrize("var,value", [("RAINCLOUD_LANCE_PAGE_BYTES", "bad"), + ("RAINCLOUD_LANCE_DATA_CACHE_BYTES", "-1"), + ("RAINCLOUD_LANCE_FILE_VERSION", "9.0")]) +def test_lance_invalid_setting_is_named(monkeypatch, var, value): + monkeypatch.setenv(var, value) + with pytest.raises(ValueError, match=var): + write_settings("lance") + + +@pytest.mark.parametrize("setting,value", [("DATA_CACHE_BYTES", "8192")]) +def test_lance_page_settings_change_real_file_layout(tmp_path, monkeypatch, setting, value): + # Fixed width, incompressible values, so reducing the page budget produces more pages. + import numpy as np + table = pa.table({"x": np.random.default_rng(0).random(30_000)}) + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", table, max_chunksize=1000) + exporter = get_exporter("lance@py") + default = exporter.export({"slug": "tiny"}, canonical, tmp_path / "default.lance") + monkeypatch.setenv(f"RAINCLOUD_LANCE_{setting}", value) + small = exporter.export({"slug": "tiny"}, canonical, tmp_path / "small.lance") + assert small.compliance.roundtrip is True + before = LanceFileReader(str(default.out_path)).metadata() + after = LanceFileReader(str(small.out_path)).metadata() + assert (after.major_version, after.minor_version) == (2, 2) + assert len(after.columns[0].pages) > len(before.columns[0].pages) + toolchain = writer_toolchain(exporter) + assert toolchain[f"lance_{setting.lower()}"] == value + + +def test_lance_reader_rejects_wrong_values(tmp_path): + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", pa.table({"x": [1]})) + artifact = tmp_path / "wrong.lance" + with LanceFileWriter(str(artifact)) as writer: + writer.write_batch(pa.table({"x": [2]})) + assert get_reader("lance@py").read_conformance(artifact, canonical).status == "fail" + + +def test_lance_explicit_file_version(tmp_path, monkeypatch): + monkeypatch.setenv("RAINCLOUD_LANCE_FILE_VERSION", "2.1") + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", pa.table({"x": [1]})) + result = get_exporter("lance@py").export({"slug": "tiny"}, canonical, tmp_path / "v21.lance") + metadata = LanceFileReader(str(result.out_path)).metadata() + assert (metadata.major_version, metadata.minor_version) == (2, 1) + assert result.compliance.roundtrip is True + + +def test_lance_unsupported_type_preserves_previous_file(tmp_path): + arr = pa.UnionArray.from_dense(pa.array([0, 0], pa.int8()), pa.array([0, 1], pa.int32()), + [pa.array([1, 2])], field_names=["i"]) + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", pa.table({"u": arr})) + previous = tmp_path / "tiny.lance" + with LanceFileWriter(str(previous)) as writer: + writer.write_batch(pa.table({"x": [1, 2]})) + saved = previous.read_bytes() + with pytest.raises(ExportFailed): + run_bounded(get_exporter("lance@py"), {"slug": "tiny"}, canonical, previous) + assert previous.read_bytes() == saved + assert not list(tmp_path.glob(".*.export")) + + +def test_lance_refuses_ineffective_page_limit(tmp_path, monkeypatch): + from raincloud.pipeline.export.exporters import UnsupportedOption + monkeypatch.setenv("RAINCLOUD_LANCE_PAGE_BYTES", "4096") + canonical = write_ipc(tmp_path / "tiny.arrow.zstd", pa.table({"x": [1]})) + with pytest.raises(UnsupportedOption, match="cannot honour RAINCLOUD_LANCE_PAGE_BYTES=4096"): + get_exporter("lance@py").export({"slug": "tiny"}, canonical, tmp_path / "out.lance") + assert not (tmp_path / "out.lance").exists() + + +def test_lance_preserves_variant_storage_without_claiming_parquet_annotation(tmp_path): + from tests.test_parquet_variant_sidecars import _variant_table + table = _variant_table(null_row=True) + canonical = write_ipc(tmp_path / "variant.arrow.zstd", table, max_chunksize=1) + result = get_exporter("lance@py").export({"slug": "variant"}, canonical, tmp_path / "variant.lance") + assert result.compliance.roundtrip is True, result.compliance + assert result.compliance.variant_faithful is False + assert "shredded struct" in result.compliance.note diff --git a/tests/test_manifest_policy.py b/tests/test_manifest_policy.py index 7b20eb0..f0f8c9e 100644 --- a/tests/test_manifest_policy.py +++ b/tests/test_manifest_policy.py @@ -87,10 +87,10 @@ def test_priority_rejects_a_bad_shape(): def test_export_cells_per_format(): s = {"export": {"priority": {"parquet": ["rs", "py"]}}} - assert export_cells(s) == ["parquet@rs", "vortex@py", "orc@py", "avro@rs", "nimble@cpp"] - assert export_cells({"export": {"priority": ["java", "py"]}}) == ["parquet@java", "vortex@py", "orc@py", "avro@java", "nimble@cpp"] + assert export_cells(s) == ["parquet@rs", "vortex@py", "orc@py", "lance@py", "avro@rs", "nimble@cpp"] + assert export_cells({"export": {"priority": ["java", "py"]}}) == ["parquet@java", "vortex@py", "orc@py", "lance@py", "avro@java", "nimble@cpp"] assert export_cells({}, {"schema_version": 2, "export_priority": {"vortex": ["rs"]}}) == \ - ["parquet@py", "vortex@rs", "orc@py", "avro@rs", "nimble@cpp"] + ["parquet@py", "vortex@rs", "orc@py", "lance@py", "avro@rs", "nimble@cpp"] def test_resolve_export_cell_skips_uninstalled_and_unknown(): @@ -112,7 +112,7 @@ def test_sf100_specs_prefer_rs_for_parquet_only(): assert len(sf100) == 32 for d in sf100: assert d["export"]["priority"] == {"parquet": ["rs", "py"]}, d["slug"] - assert export_cells(d, m) == ["parquet@rs", "vortex@py", "orc@py", "avro@rs", "nimble@cpp"] + assert export_cells(d, m) == ["parquet@rs", "vortex@py", "orc@py", "lance@py", "avro@rs", "nimble@cpp"] # ---------- export.formats is the one v2 declaration ---------- @@ -412,12 +412,12 @@ class Unset: def test_a_priority_naming_no_writer_for_a_format_falls_back_to_the_default(): - assert export_cells({"export": {"priority": ["java"]}}) == ["parquet@java", "vortex@py", "orc@py", "avro@java", "nimble@cpp"] + assert export_cells({"export": {"priority": ["java"]}}) == ["parquet@java", "vortex@py", "orc@py", "lance@py", "avro@java", "nimble@cpp"] def test_format_sets_derive_from_the_writers(): from raincloud._formats import ALL_FORMATS, EXPORTED_FORMATS - assert set(EXPORTED_FORMATS) == {base for base in WRITERS if base != "arrow"} == {"parquet", "vortex", "orc", "avro", "nimble"} + assert set(EXPORTED_FORMATS) == {base for base in WRITERS if base != "arrow"} == {"parquet", "vortex", "orc", "lance", "avro", "nimble"} assert set(ALL_FORMATS) == set(WRITERS) From 336adb650d13101e382cdaad43b7ac461a6fc902 Mon Sep 17 00:00:00 2001 From: Martin Prammer Date: Sat, 10 Oct 2026 17:38:24 -0400 Subject: [PATCH 2/2] Preserve Lance filesystem access errors Signed-off-by: Martin Prammer --- AGENTS.md | 5 +++-- raincloud/__init__.py | 7 +++++++ tests/test_lance.py | 28 ++++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 70c66d8..688d017 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -158,8 +158,9 @@ that export as a measurement rather than writing something else; `sidecars/READM tabulates which writer honours what, and the `/raincloud-write-settings` skill is the procedure (a setting changes only files this install writes, so an existing file must be re-exported or rebuilt to gain it). ORC, Avro, Lance and Vortex have the same kind of settings -(`spec.FORMAT_SETTINGS`, `RAINCLOUD_ORC_*`, `RAINCLOUD_AVRO_*`, `RAINCLOUD_VORTEX_*`), read -and refused the same way; their codec, unset, is the zstd raincloud has always written. +(`spec.FORMAT_SETTINGS`, `RAINCLOUD_ORC_*`, `RAINCLOUD_AVRO_*`, `RAINCLOUD_LANCE_*`, +`RAINCLOUD_VORTEX_*`), read and refused the same way. ORC, Avro and Vortex default to +zstd; Lance leaves encoding to its structural encoder. `export.formats` lists the formats a dataset wants. When the planned writer cannot produce one for the dataset -- it raises, dies, reports a failed round-trip, or exceeds diff --git a/raincloud/__init__.py b/raincloud/__init__.py index fd32bfc..fc70bc9 100644 --- a/raincloud/__init__.py +++ b/raincloud/__init__.py @@ -57,6 +57,7 @@ def _settings(config: Config | str | Path | None) -> Config: # How vortex-data reports an operating-system failure: its Rust io::Error, # surfaced as a RuntimeError ("Io: Permission denied (os error 13)"). _VORTEX_OS_ERROR = re.compile(r"Io: (.*) \(os error (\d+)\)") +_LANCE_OS_ERROR = re.compile(r"\(os error (\d+)\)") @contextmanager @@ -90,9 +91,15 @@ def corrupt(exc): except OSError as exc: if exc.errno is not None: raise + if fmt == "lance" and (system := _LANCE_OS_ERROR.search(str(exc))): + raise OSError(int(system.group(1)), str(exc), str(path)) from exc raise corrupt(exc) from exc except pa.ArrowInvalid as exc: raise corrupt(exc) from exc + except ValueError as exc: + if fmt == "lance" and str(exc).startswith("Not found:"): + raise FileNotFoundError(2, str(exc), str(path)) from exc + raise except RuntimeError as exc: if fmt != "vortex": raise diff --git a/tests/test_lance.py b/tests/test_lance.py index edd9e3c..820332d 100644 --- a/tests/test_lance.py +++ b/tests/test_lance.py @@ -3,11 +3,15 @@ """Single-file Lance artifacts: streaming, settings, loader, and measured failures.""" from __future__ import annotations +import os +from contextlib import ExitStack + import pyarrow as pa import pytest import raincloud from raincloud import _builds +from raincloud._readers import _lance_batches from raincloud._resolve import artifact_key from raincloud.catalogs import operation from raincloud.pipeline import build @@ -23,6 +27,30 @@ from lance.file import LanceFileReader, LanceFileWriter # noqa: E402 +@pytest.mark.parametrize("missing", [False, True]) +@pytest.mark.parametrize("columns", [None, ["x"]]) +def test_lance_access_errors_are_not_corruption(tmp_path, missing, columns): + if not missing and (os.name != "posix" or os.geteuid() == 0): + pytest.skip("needs file permissions that bind") + path = tmp_path / "access.lance" + with LanceFileWriter(str(path)) as writer: + writer.write_batch(pa.table({"x": [1]})) + if missing: + path.unlink() + else: + path.chmod(0) + try: + with pytest.raises(FileNotFoundError if missing else PermissionError) as error: + with raincloud._decoding(path, "lance"), ExitStack() as stack: + _, batches = _lance_batches(path, stack, columns=columns, batch_size=1, + check_columns=lambda schema: None) + list(batches) + assert error.value.filename == str(path) + finally: + if not missing: + path.chmod(0o644) + + @pytest.fixture(autouse=True) def settings(monkeypatch): for _, var, _ in FORMAT_SETTINGS["lance"]: