Add Lance reader and writer stages

[VibhuJawa]

Description

Adds a first-pass Lance integration for Curator text IO:

• LanceReader partitions Lance datasets by fragment and returns Arrow-backed DocumentBatches.
• LanceWriter writes Curator batches to Lance and records checkpointed commit payloads.
• LanceAnnotationWriter appends/validates annotation columns with Lance add_columns(...), writes fragment-local row updates, and records checkpointed commit payloads.
• Writer-side driver commit helpers live in writer/lance.py and publish checkpointed writes with commit_lance_checkpoint(...) and annotation-column updates with commit_lance_annotation_checkpoint(...).
• Shared Lance metadata/checkpoint utilities live under nemo_curator/utils/lance.py and reuse Curator filesystem/hash helpers.
• Includes focused Lance functionality coverage for read filters/projection, blob v2 round trips, write checkpoint commits, annotation prepare/update, duplicate row-address rejection, and split-fragment update rejection.

Usage

from nemo_curator.stages.text.io.reader import LanceReader
from nemo_curator.stages.text.io.writer import LanceWriter, commit_lance_checkpoint

reader = LanceReader(
    "s3://bucket/dataset.lance",
    fields=["url", "text", "content_zlib"],
    read_kwargs={"filter": "snapshot_id = 'CC-MAIN-2025-26'"},
)

writer = LanceWriter(
    path="s3://bucket/output.lance",
    commit_path="s3://bucket/checkpoints/output",
    mode="overwrite",
)

# Run the Curator pipeline with the writer, then commit on the driver.
version = commit_lance_checkpoint(
    "s3://bucket/output.lance",
    "s3://bucket/checkpoints/output",
)

Append annotation columns back to an existing Lance dataset:

import pyarrow as pa

from nemo_curator.stages.text.io.writer import LanceAnnotationWriter, commit_lance_annotation_checkpoint

annotation_writer = LanceAnnotationWriter(
    path="s3://bucket/dataset.lance",
    commit_path="s3://bucket/checkpoints/annotations",
    schema=pa.schema([pa.field("word_count", pa.int64())]),
    create_columns=True,
)
annotation_writer.prepare()

# Run the Curator pipeline with `annotation_writer`, then commit on the driver.
version = commit_lance_annotation_checkpoint(
    "s3://bucket/dataset.lance",
    "s3://bucket/checkpoints/annotations",
)

Testing

• uv run ruff check nemo_curator/utils/lance.py nemo_curator/stages/text/io/reader/lance.py nemo_curator/stages/text/io/writer/lance.py nemo_curator/stages/text/io/writer/__init__.py tests/stages/text/io/reader/test_lance.py tests/stages/text/io/writer/test_lance.py pyproject.toml
• bash -n tests/L0_Unit_Test_CPU.sh
• uv run pytest tests/stages/text/io/reader/test_lance.py tests/stages/text/io/writer/test_lance.py -q

Checklist

• I am familiar with the Contributing Guide.
• New or Existing tests cover these changes.
• The documentation is up to date with these changes.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[VibhuJawa]

@claude review

[Copilot]

Pull request overview

Adds first-pass Lance integration for Curator text IO, enabling fragment-partitioned reading into Arrow-backed DocumentBatch tasks, checkpointed writes/annotation updates, and driver-side commit helpers to publish staged writes.

Changes:

• Introduces LanceReader (partition + read stages) with support for filter/projection, blob v2 restoration, and optional Lance metadata columns.
• Adds LanceWriter and LanceAnnotationWriter plus checkpoint/commit helpers (commit_lance_checkpoint, commit_lance_annotation_checkpoint) to publish staged writes.
• Wires in a new optional lance extra (lance-ray>=0.4), updates CI/unit test env sync, and adds targeted pytest coverage.

Reviewed changes

Copilot reviewed 10 out of 11 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
uv.lock	Adds locked dependencies for lance-ray and its transitive requirements; registers lance extra and includes it in all.
pyproject.toml	Defines the new optional dependency extra lance and includes it under all.
tests/L0_Unit_Test_CPU.sh	Ensures CPU unit test env installs the lance extra.
nemo_curator/stages/text/io/reader/__init__.py	Exposes LanceReader from the text IO reader package.
nemo_curator/stages/text/io/writer/__init__.py	Exposes LanceWriter / LanceAnnotationWriter from the text IO writer package.
nemo_curator/stages/text/io/reader/lance.py	Implements Lance dataset partitioning by fragment and fragment-scoped scanning into DocumentBatch (with optional metadata + blob v2 restore).
nemo_curator/stages/text/io/writer/lance.py	Implements staged fragment writes and fragment-local annotation updates with checkpoint record emission.
nemo_curator/stages/text/io/lance_utils.py	Adds checkpoint record/marker read-write helpers and reserved Lance metadata column constants.
nemo_curator/stages/text/io/lance_commit.py	Adds driver-side commit helpers to publish checkpointed writes/updates to a Lance dataset.
tests/stages/text/io/reader/test_lance.py	Adds coverage for partitioning/version pinning, filter/projection behavior, blob v2, and metadata columns.
tests/stages/text/io/writer/test_lance.py	Adds coverage for checkpoint commits, blob v2 round trips, annotation prepare/update flows, and rejection of invalid update sets.

[VibhuJawa]

Initial review

[VibhuJawa]

Some more cleanup

[VibhuJawa]

Its fine, lets just assume we will get version from task._metadata.get("lance") !

[VibhuJawa]

Why dont we just get schema from dataset.schema ?

[VibhuJawa]

Closing in favor of 3 PRs , i have started #2111 , #2112, #2113