Add single-file Lance exports and loading

.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

AGENTS.md

@@ -157,9 +157,10 @@ 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
-(`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.
+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_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
@@ -180,6 +181,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 +300,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) |

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### 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.
+
 - `parquet@go`: Apache Arrow Go 18.8.0's `pqarrow` writer and reader sidecars,
   with streaming IPC input, shared Parquet settings, page indexes on with
   statistics, and read-back verification before publication. Explicit page

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
@@ -330,6 +330,31 @@ The optional `parquet@go` sidecar writes and verifies Parquet with Apache Arrow
 statistics, and refuses requests to write page checksums or enforce an exact page-row limit. See
 [the build instructions and option matrix](sidecars/README.md#apache-arrow-go).
 
+### 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 <slug> --format lance
+```
+
+This writes `lance/<slug>.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

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]

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

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}

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,

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]:

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

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

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())

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@go", {"parquet"}, "raincloud-read-parquet-go"))
 register_reader(SidecarReader("parquet@rs", {"parquet"}, "raincloud-read-parquet-rs"))
 register_reader(SidecarReader("vortex@rs", {"vortex"}, "raincloud-read-vortex-rs"))

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")),