add lance writer

docs/nav.md

@@ -24,6 +24,7 @@
 - [Files and videos](reading-data/files-and-videos.md)
 - [Hugging Face](reading-data/hugging-face.md)
 - [TensorFlow](reading-data/tensorflow.md)
+- [Lance](reading-data/lance.md)
 - [Custom readers](reading-data/custom-readers.md)
 
 ## Episode data
@@ -65,6 +66,7 @@
 - [LeRobot](writing-data/lerobot.md)
 - [Zarr](writing-data/zarr.md)
 - [Parquet and JSONL](writing-data/parquet-and-jsonl.md)
+- [Lance](writing-data/lance.md)
 - [Media assets and reducers](writing-data/media-assets-and-reducers.md)
 
 ## Examples

docs/reading-data/index.md

@@ -26,6 +26,7 @@ pipeline = mdr.read_lerobot("hf://datasets/lerobot/aloha_sim_transfer_cube_human
 | Raw files or media files | `read_files`, `read_videos` | [Files and Videos](files-and-videos.md) |
 | Hugging Face datasets table | `read_hf_dataset` | [Hugging Face](hugging-face.md) |
 | TFRecord files or TensorFlow Datasets | `read_tfrecords`, `read_tfds` | [TensorFlow](tensorflow.md) |
+| Versioned Lance dataset | `load_lance` | [Lance](lance.md) |
 | Your own source system | `from_source` | [Custom Readers](custom-readers.md) |
 
 ## Core ideas

docs/reading-data/lance.md

@@ -0,0 +1,45 @@
+---
+title: "Lance"
+description: "Read immutable Lance dataset versions as fragment-aligned Refiner shards"
+---
+
+# Lance
+
+Lance support is optional:
+
+```bash
+pip install macrodata-refiner[lance]
+```
+
+Use `load_lance(...)` to read a pinned Lance dataset version:
+
+```python
+import refiner as mdr
+
+pipeline = mdr.load_lance(
+    "s3://my-bucket/hands.lance",
+    version=42,
+    columns=["image", "frame_id"],
+    batch_size=128,
+)
+```
+
+When `version` is omitted, Refiner resolves the latest version once and pins it
+for the pipeline. Column projection is pushed into Lance, and `batch_size`
+controls the streamed Arrow batch size. Use `blob_handling` to select Lance's
+blob materialization behavior when reading blob columns.
+
+`load_lance` uses Lance's native storage layer. It rejects configured fsspec
+filesystem objects and `storage_options`; provide a URI whose endpoint and
+credentials are available to Lance instead.
+
+Each Lance fragment becomes one Refiner shard. A worker may claim and process
+multiple fragments over its lifetime.
+
+## Internal Notes
+
+The source keeps the dataset URI and resolved version on the pipeline. It uses
+ordinary row-range shards to assign fragment indices and attaches protected
+fragment-ID and fragment-local-row-position columns to the rows. The
+`add_columns` writer uses those columns to restore output order and validate
+one-to-one row alignment.

docs/reference/optional-dependencies.md

@@ -14,6 +14,7 @@ Install extras based on the data and operations you use.
 | `hf` | Hugging Face Hub APIs and HF filesystem helpers. |
 | `hand_tracking` | Hand tracking with ego-vision. |
 | `hdf5` | HDF5 reader support. |
+| `lance` | Lance readers, writers, and distributed schema evolution. |
 | `zarr` | Zarr reader and writer support. |
 | `mcap` | MCAP robotics log reader support, including ROS2, protobuf, and H.264 video decoding. |
 | `video` | Video decode/write support. |
@@ -30,5 +31,6 @@ pip install macrodata-refiner[hf,video]
 pip install macrodata-refiner[datasets]
 pip install macrodata-refiner[hdf5,zarr]
 pip install macrodata-refiner[mcap]
+pip install macrodata-refiner[lance]
 pip install macrodata-refiner[hand_tracking]
 ```

docs/writing-data/index.md

@@ -13,6 +13,7 @@ reader, transforms, and writer stages.
 | [LeRobot](lerobot.md) | Training-ready robotics datasets. |
 | [Zarr](zarr.md) | Array stores and replay buffers. |
 | [Parquet and JSONL](parquet-and-jsonl.md) | Tabular outputs and logs. |
+| [Lance](lance.md) | Lance files, datasets, and distributed column additions. |
 | [Media Assets and Reducers](media-assets-and-reducers.md) | Asset uploads, video handling, and reducer stages. |
 
 ## Writer pattern
@@ -26,4 +27,3 @@ pipeline = (
 ```
 
 The writer does work when the pipeline is launched.
-

docs/writing-data/lance.md

@@ -0,0 +1,96 @@
+---
+title: "Lance"
+description: "Write Lance files, datasets, and distributed schema evolution results"
+---
+
+# Lance
+
+Lance support is optional:
+
+```bash
+pip install macrodata-refiner[lance]
+```
+
+## Standalone files
+
+Use `write_lance(...)` to create one independent Lance file per finalized
+Refiner shard:
+
+```python
+import refiner as mdr
+
+pipeline = (
+    mdr.read_parquet("s3://my-bucket/raw/*.parquet")
+    .write_lance("s3://my-bucket/lance-files/")
+)
+```
+
+## Lance datasets
+
+Use `write_lance_dataset(...)` for committed Lance datasets:
+
+```python
+pipeline = (
+    mdr.read_parquet("s3://my-bucket/raw/*.parquet")
+    .write_lance_dataset("s3://my-bucket/clean.lance", mode="create")
+)
+```
+
+Supported modes are `create`, `overwrite`, `append`, and `add_columns`.
+Empty `create` and `overwrite` jobs commit an empty dataset when Refiner can
+determine the output Arrow schema statically; otherwise they fail explicitly.
+
+Lance opens dataset URIs through its native storage layer. Configured fsspec
+filesystem objects and `storage_options` are therefore rejected instead of
+being silently ignored. Put the endpoint and credentials in the URI or Lance's
+supported environment/configuration.
+
+## Adding columns
+
+Use `add_columns` for row-preserving enrichment such as model inference:
+
+```python
+pipeline = (
+    mdr.load_lance(
+        "s3://my-bucket/hands.lance",
+        version=42,
+        columns=["image"],
+    )
+    .map(
+        detect_hands,
+        dtypes={
+            "hand_boxes": mdr.datatype.list(mdr.datatype.float32()),
+            "detector_score": mdr.datatype.float32(),
+        },
+    )
+    .write_lance_dataset(
+        "s3://my-bucket/hands.lance",
+        mode="add_columns",
+        columns=["hand_boxes", "detector_score"],
+    )
+)
+```
+
+The `columns` argument is required and only those columns are written. Existing
+columns, including large blob columns, remain referenced by their original
+files. Results may arrive out of order; Refiner restores fragment-local source
+order before writing. Missing or duplicate results fail execution and do not
+create a new dataset version.
+
+## Internal Notes
+
+Workers buffer only the requested output columns plus internal ordering columns
+as Arrow tables. At fragment completion, they reorder those tables with Arrow,
+write uncommitted column files, and record replacement-fragment metadata. The
+reducer commits finalized attempts once against the pinned read version with a
+Lance merge operation. Before committing, it verifies that every non-empty
+fragment in the pinned source version has exactly one finalized result. Cleanup
+records only files created by each attempt, so rejected retries cannot delete
+base dataset files.
+
+This follows the worker-output/coordinator-commit pattern used by Spark,
+Beam/Dataflow, Daft, and Ray Data. Refiner keeps Lance fragments as its work
+unit because Lance schema evolution is fragment-based; repartitioning first
+would require a keyed join or staging dataset. Hugging Face Datasets generally
+materializes a new dataset revision instead of attaching column files to
+existing fragments.

pyproject.toml

@@ -67,6 +67,9 @@ mcap = [
 s3 = [
     "s3fs",
 ]
+lance = [
+    "pylance>=4.0.1",
+]
 gcs = [
     "gcsfs",
 ]
@@ -87,6 +90,7 @@ all = [
     "macrodata-refiner[hdf5]",
     "macrodata-refiner[hf]",
     "macrodata-refiner[mcap]",
+    "macrodata-refiner[lance]",
     "macrodata-refiner[video]",
     "macrodata-refiner[zarr]",
     "macrodata-refiner[text]",

src/refiner/__init__.py

@@ -22,6 +22,7 @@
     "SUPPORTED_GPU_TYPES": "refiner.pipeline",
     "from_items": "refiner.pipeline",
     "from_source": "refiner.pipeline",
+    "load_lance": "refiner.pipeline",
     "read_csv": "refiner.pipeline",
     "read_files": "refiner.pipeline",
     "read_hf_dataset": "refiner.pipeline",
@@ -64,6 +65,7 @@
     "read_json",
     "read_jsonl",
     "read_lerobot",
+    "load_lance",
     "read_mcap",
     "read_parquet",
     "read_tfds",
@@ -131,6 +133,7 @@ def __dir__() -> list[str]:
         SUPPORTED_GPU_TYPES,
         from_items,
         from_source,
+        load_lance,
         read_csv,
         read_files,
         read_hdf5,

src/refiner/io/datafolder.py

@@ -45,6 +45,7 @@ def __init__(
         """
         AsyncFileSystem.__init__(self)
         self._fs = fs
+        self._explicit_fs = fs is not None
         # Keep string paths unresolved so cloud submission can inspect manifests
         # and infer extras without requiring local remote-storage credentials.
         self._path = fs._strip_protocol(path) if fs is not None else path
@@ -144,6 +145,10 @@ def abs_path(self, path: str = "") -> str:
     def required_refiner_extras(self) -> tuple[str, ...]:
         return required_refiner_extras(self._path, self._fs)
 
+    @property
+    def has_explicit_filesystem_configuration(self) -> bool:
+        return self._explicit_fs or bool(self._storage_options)
+
     def abs_paths(self, paths: str | Iterable[str]) -> str | list[str]:
         """
         Transform  a list of relative paths into a list of complete paths (including fs protocol and base path)

src/refiner/pipeline/__init__.py

@@ -9,6 +9,7 @@
     "RefinerPipeline": "refiner.pipeline.pipeline",
     "from_items": "refiner.pipeline.pipeline",
     "from_source": "refiner.pipeline.pipeline",
+    "load_lance": "refiner.pipeline.pipeline",
     "read_csv": "refiner.pipeline.pipeline",
     "read_files": "refiner.pipeline.pipeline",
     "read_hf_dataset": "refiner.pipeline.pipeline",
@@ -46,6 +47,7 @@
     "read_json",
     "read_jsonl",
     "read_lerobot",
+    "load_lance",
     "read_mcap",
     "read_parquet",
     "read_tfds",
@@ -79,6 +81,7 @@ def __dir__() -> list[str]:
         RefinerPipeline,
         from_items,
         from_source,
+        load_lance,
         read_csv,
         read_files,
         read_hdf5,