perf: share one lance Session per process

[FANNG1]

Changes Made

Every lance.dataset() opened without a session builds its own Arc<Manifest> holding the table's whole fragment list, plus its own index and metadata cache budgets and its own object-store client. Daft opens one per scan task, one per writer flush and one per maintenance instance, so a many-fragment table pays that again for every concurrently open dataset.

This passes one process-level lance.Session to every open instead (daft_lance.namespace.get_shared_session). The manifest is a metadata-cache entry keyed by (uri, version, e_tag), so every open of the snapshot the driver pinned hands back a clone of the same Arc<Manifest>.

• utils.construct_lance_dataset_handle, DatasetOpenContext._open and LanceDataSink (including the write_fragments call, whose Rust writer loads the dataset on every flush) now open through that session.
• read_blobs additionally stops closing over the driver's LanceDataset. LanceDataset.__reduce__ ships the whole re-encoded manifest per task, which both inflates the payload and bypasses the session's manifest cache; it now carries the uri plus the pinned open kwargs and reopens once per worker process.
• DAFT_LANCE_SHARE_SESSION=0 restores lance's per-open sessions. It is an escape hatch, and the only way to get pre-sharing behaviour inside a Ray worker, which the A/B benchmark needs: a driver-side monkeypatch never reaches the worker.

What it saves, per path

The three paths differ and only one of them multiplies by concurrency, so all three were measured rather than assumed.

Path	Opens per worker process	What sharing changes
Scans, read_blobs, writes	scantask_max_parallel concurrent opens in one process	Collapses them to one manifest.
Indexed (nearest, point lookups)	One index-driven task, whatever the concurrency	The index is loaded once per process instead of once per query.
Maintenance (compact_files, update_columns, merge_columns)	One, always	Nothing to collapse — see below.

The maintenance path is the exception, and it is covered so the exception stays documented. Daft runs each daft.cls instance in a process of its own (measured on both runners; use_process=False does not change it under Ray), so a maintenance worker opens the table exactly once however high the concurrency. An indexed table does not change that either: a task's open brings in the index's metadata entry only, and the index data is read once on the driver, when the compaction commit remaps every entry onto the new fragment ids. If Daft ever fuses those instances into one process, test_maintenance_instances_each_get_their_own_process fails and says so.

Measurements

In a Ray worker on a KubeRay cluster, read_lance → write_parquet. A write rather than an aggregation on purpose: a task holds its parquet writer's buffers at the same time as its manifest, so the two compete for one memory limit, which is what a real backfill does. An aggregation keeps almost nothing resident and flatters the worker.

	per-open	stage time	worker ru_maxrss growth	Daft's sampled peak
4000 fragments × 10k rows, concurrency 24, whole table	1.22 MB	11.3s → 10.4s	206.5 → 157.4 MB (−49.1)	487.8 → 442.1 MB
100k fragments × 10k rows, concurrency 8, 1% in groups of 10	46.68 MB	609.6s → 61.9s	1052.2 → 573.3 MB (−479.0)	1352.6 → 997.3 MB

At 100k fragments the cost of not sharing is mostly time: each unshared task re-parsed the same 8 MB manifest, about 5.5s of the 6.1s it spent. Repeated nearest queries on a 200k-row IVF_PQ index run 1.7–2.0× faster for the same reason.

Two knobs keep the large table affordable without weakening the measurement, because a manifest is the whole fragment list and an open costs the same whatever its task then reads: reading a prefix of the table (--read-fraction) prunes whole fragments, and read_lance(fragment_group_size=N) puts several fragments in one task. The benchmark refuses to report a number when the resulting task count is too low for the opens to overlap.

Related Issues

None.

Checklist

• Documented in README.md (new ## Environment Variables section, plus the per-path table above).
• If adding a new feature, unit tests were added in tests/io/lance/test_shared_session.py and tests/io/lance/test_ray_session_memory.py.

Tests assert the mechanism rather than the megabytes: Session.size_bytes() is exact, needs no large table, and is zero for any open that did not go through the process session — which is exactly what the pre-sharing behaviour looks like. benchmarks/lance_session_memory.py (make bench) is the counterpart that puts numbers on it.

make test also now sets RAY_ENABLE_UV_RUN_RUNTIME_ENV=0. Ray walks the driver's process tree, sees uv run, and replicates that environment to every worker by packaging the working directory and rebuilding the venv per worker; under DAFT_RUNNER=ray that made worker startup fail once the suite had created enough actors (277 ActorUnschedulableError). Workers inherit the interpreter from the raylet regardless.

Verified on this branch rebased onto main: DAFT_RUNNER=native make test 533 passed, DAFT_RUNNER=ray make test 532 passed, make lint and make check-format clean.