[codex] Add retry-aware image Lance workflow tutorial

pyproject.toml

@@ -137,6 +137,12 @@ image_cpu = [
     "torchvision"
 ]
 
+image_lance = [
+    "nemo_curator[image_cpu]",
+    "nemo_curator[lance]",
+    "s3fs>=2024.12.0",
+]
+
 # NVIDIA DALI (simplified; update the package to match your CUDA version if needed)
 image_cuda12 = [
     "nemo_curator[image_cpu]",

tests/stages/image/test_lance_writer_tutorial.py

@@ -0,0 +1,129 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import json
+import tarfile
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+from PIL import Image
+
+from tutorials.image.lance_writer.manifest import PhysicalTar, build_manifest, load_manifest
+from tutorials.image.lance_writer.pipeline import build_pipeline
+from tutorials.image.lance_writer.stages import (
+    IMAGE_SCHEMA,
+    candidate_is_better,
+    iter_tar_candidates,
+)
+
+
+def _inventory(path: Path, *, etag: str = "etag-a") -> None:
+    rows = [
+        {
+            "source_shard": shard,
+            "tar_id": tar_id,
+            "attempt": attempt,
+            "tar_uri": f"s3://source/shard_{shard:05d}/attempt_{attempt}/{tar_id}.tar",
+            "tar_size": size,
+            "tar_etag": etag if tar_id == "00000" else f"etag-{tar_id}-{attempt}",
+            "last_modified": "2026-01-01T00:00:00+00:00",
+        }
+        for shard, tar_id, attempt, size in [
+            (1, "00000", 1, 700),
+            (1, "00000", 2, 700),
+            (1, "00001", 1, 700),
+            (1, "00002", 1, 700),
+            (2, "00000", 1, 400),
+        ]
+    ]
+    pq.write_table(pa.Table.from_pylist(rows), path)
+
+
+def test_manifest_is_stable_and_packs_never_cross_source_shards(tmp_path: Path) -> None:
+    inventory_path = tmp_path / "inventory.parquet"
+    manifest_dir = tmp_path / "manifest"
+    _inventory(inventory_path)
+
+    first = build_manifest(str(inventory_path), str(manifest_dir), target_pack_bytes=1_000)
+    second = build_manifest(str(inventory_path), str(manifest_dir), target_pack_bytes=1_000)
+    metadata, packs = load_manifest(str(manifest_dir))
+
+    assert first == second == metadata
+    assert metadata["physical_tar_count"] == 5
+    assert metadata["fpp_count"] == 4
+    assert all({fpp.source_shard for fpp in pack.fpps} == {pack.source_shard} for pack in packs)
+    retried = next(fpp for pack in packs for fpp in pack.fpps if fpp.fpp_id == "shard_00001/00000")
+    assert [item.attempt for item in retried.attempts] == [1, 2]
+
+
+def test_winner_policy_prefers_resolution_then_newest_attempt() -> None:
+    base = {"width": 100, "height": 100, "source_attempt": 1, "source_id": "b"}
+    assert candidate_is_better({**base, "width": 101}, base)
+    assert candidate_is_better({**base, "source_attempt": 2}, base)
+    assert candidate_is_better({**base, "source_id": "a"}, base)
+    assert not candidate_is_better({**base, "source_attempt": 0}, base)
+
+
+def test_materializer_and_writer_are_fusible_one_cpu_stages() -> None:
+    pipeline = build_pipeline(
+        manifest_dir="/manifest",
+        dataset_uri="s3://output/images",
+        lance_commit_path="/checkpoints/lance",
+        source_storage_options={},
+        lance_storage_options={},
+    )
+
+    materializer, writer = pipeline.stages[1:]
+    assert materializer.resources == writer.resources
+    assert materializer.resources.cpus == 1
+    assert pa.types.is_large_binary(IMAGE_SCHEMA.field("image").type)
+
+
+def test_tar_reader_pairs_json_with_encoded_image_and_preserves_format() -> None:
+    image_buffer = io.BytesIO()
+    Image.new("RGB", (12, 8), color="red").save(image_buffer, format="JPEG")
+    metadata = json.dumps(
+        {
+            "status": "success",
+            "url": "https://example.test/image",
+            "width": 12,
+            "height": 8,
+        }
+    ).encode()
+    tar_buffer = io.BytesIO()
+    with tarfile.open(fileobj=tar_buffer, mode="w") as archive:
+        for name, payload in (("sample.json", metadata), ("sample.jpg", image_buffer.getvalue())):
+            info = tarfile.TarInfo(name)
+            info.size = len(payload)
+            archive.addfile(info, io.BytesIO(payload))
+    tar_buffer.seek(0)
+    physical_tar = PhysicalTar(
+        source_shard=1,
+        tar_id="00000",
+        attempt=1,
+        tar_uri="s3://source/shard_00001/attempt_1/00000.tar",
+        tar_size=len(tar_buffer.getvalue()),
+        tar_etag="etag",
+        last_modified="2026-01-01T00:00:00+00:00",
+    )
+
+    rows = list(iter_tar_candidates(tar_buffer, physical_tar))
+
+    assert len(rows) == 1
+    assert rows[0]["url"] == "https://example.test/image"
+    assert rows[0]["image_format"] == "JPEG"
+    assert rows[0]["mime_type"] == "image/jpeg"
+    assert (rows[0]["width"], rows[0]["height"]) == (12, 8)

tutorials/image/README.md

@@ -11,6 +11,7 @@ Hands-on tutorials for curating image data with NeMo Curator. Complete working e
 | Tutorial | Description | Files |
 |----------|-------------|-------|
 | **[Getting Started](getting-started/)** | Image curation fundamentals | `image_curation_example.py`, `image_dedup_example.py`, `helper.py` |
+| **[Lance Writer](lance_writer/)** | Retry-aware tar ingestion into URL-addressable Lance tables | `pipeline.py`, `manifest.py`, `stages.py`, `submit_array.sh` |
 
 ## Documentation Links
 
@@ -23,4 +24,4 @@ Hands-on tutorials for curating image data with NeMo Curator. Complete working e
 
 ## Support
 
-**Documentation**: [Main Docs](https://docs.nvidia.com/nemo/curator/latest/) • [API Reference](https://docs.nvidia.com/nemo/curator/latest/apidocs/index.html) • [GitHub Discussions](https://github.com/NVIDIA-NeMo/Curator/discussions)
+**Documentation**: [Main Docs](https://docs.nvidia.com/nemo/curator/latest/) • [API Reference](https://docs.nvidia.com/nemo/curator/latest/apidocs/index.html) • [GitHub Discussions](https://github.com/NVIDIA-NeMo/Curator/discussions)

tutorials/image/lance_writer/README.md

@@ -0,0 +1,159 @@
+# Retry-attempt image tars to Lance
+
+This workflow builds a URL-addressable Lance table from image/JSON pairs stored
+in tar files. It is designed for sources that retry the same logical tar: every
+physical attempt is processed together, and exactly one winner is selected per
+exact URL before the image is written.
+
+The source contract must keep every retry copy of a URL in the same stable
+`(source_shard, tar_id)` FPP. Cross-tar or cross-shard URL deduplication is a
+separate dataset-wide operation and is not performed by this ingestion recipe.
+
+The example focuses on ingestion. Dataset-wide validation, compaction,
+secondary-index construction, and benchmarking are intentionally outside its
+scope.
+
+## Pipeline
+
+```text
+immutable tar inventory
+        │
+        ▼
+stable (source_shard, tar_id) FPPs
+        │  group all attempts; pack within one source shard toward 1 GiB
+        ▼
+FppPackPartitioningStage
+        │  deterministic task IDs + Slurm-array filtering
+        ▼
+FppPackMaterializationStage ──fused──> LanceWriter
+        │                                │
+        │ Arrow large_binary images      └─ uncommitted fragments + checkpoints
+        ▼
+commit_lance_checkpoint (once, after every logical shard is complete)
+```
+
+The materializer and writer both request one CPU, allowing Ray Data to fuse
+them. This avoids retaining an additional image-heavy intermediate block.
+
+## Input contracts
+
+The inventory is a Parquet table with one row per physical tar:
+
+| Column | Type | Meaning |
+|---|---|---|
+| `source_shard` | integer | Source shard containing the logical tar |
+| `tar_id` | string | Stable logical tar ID |
+| `attempt` | integer | Monotonically increasing retry attempt |
+| `tar_uri` | string | Full `s3://` URI |
+| `tar_size` | integer | Object size in bytes |
+| `tar_etag` | string | Object ETag |
+| `last_modified` | string | Stable timestamp from the inventory pass |
+
+Each tar contains image and JSON members with the same member stem. JSON rows
+must contain `url`; `width`, `height`, `sha256`, and `status` are consumed when
+present. Pillow determines the actual encoded format, so JPEG, MPO, PNG, and
+WebP payloads remain distinguishable even when the member suffix is `.jpg`.
+
+For equal URLs, the winner policy is:
+
+1. largest pixel area;
+2. newest attempt;
+3. lexicographically smallest source ID as a deterministic tie-breaker.
+
+The output includes the encoded image as ordinary Arrow `large_binary`, MD5
+and SHA-256 hashes, image format/MIME/dimensions, the original JSON, and only
+the winning tar/member provenance.
+
+## Environment
+
+Install the image and Lance dependencies directly:
+
+```bash
+pip install -e '.[image_lance]'
+```
+
+Use the standard AWS environment variables for credentials. S3-compatible
+endpoint settings are explicit JSON objects because fsspec and Lance use
+different option shapes:
+
+```bash
+export SOURCE_STORAGE_OPTIONS='{"client_kwargs":{"endpoint_url":"https://s3.example"}}'
+export LANCE_STORAGE_OPTIONS='{"endpoint":"https://s3.example","aws_region":"us-east-1","virtual_hosted_style_request":"false","client_max_retries":"20"}'
+```
+
+## Build the manifest once
+
+Run this on a CPU node after the source inventory is frozen:
+
+```bash
+python -m tutorials.image.lance_writer.pipeline build-manifest \
+  --inventory /shared/inventory/physical_tars.parquet \
+  --manifest-dir /shared/manifests/images-v1 \
+  --target-pack-bytes 1073741824
+```
+
+The snapshot ID hashes the inventory descriptors and pack target. The command
+reuses an identical manifest and refuses to overwrite a different one.
+
+## Run with Slurm arrays and resumability
+
+Choose enough logical shards to give each array element many packs. Limit
+concurrent nodes from measured object-store throughput rather than CPU count.
+
+```bash
+export MANIFEST_DIR=/shared/manifests/images-v1
+export DATASET_URI=s3://output-bucket/lance/images-v1
+export LANCE_COMMIT_PATH=/shared/checkpoints/images-v1/lance
+export CHECKPOINT_PATH=/shared/checkpoints/images-v1/curator
+
+sbatch --account=<account> --partition=<cpu-partition> \
+  --array=0-499%60 --export=ALL \
+  tutorials/image/lance_writer/submit_array.sh
+```
+
+Both checkpoint paths should be on shared storage. Dataset payload is written
+only to Lance; source tars are read through fsspec and are never copied to an
+intermediate object-store prefix.
+
+After the array finishes, use Curator's Slurm retry helper with the same
+`CHECKPOINT_PATH`. The `fields` format preserves the original logical shard
+count when the physical retry array is sparse:
+
+```bash
+python tutorials/slurm/retry_array.py \
+  --checkpoint-path "${CHECKPOINT_PATH}" \
+  --format fields \
+  --max-array-size 1001
+```
+
+Each output line contains `array_expression shard_index_offset
+minimum_shard_index original_total_shards`. Resubmit only those indices while
+preserving the logical values:
+
+```bash
+while read -r array offset minimum total; do
+  sbatch --account=<account> --partition=<cpu-partition> \
+    --array="${array}%60" \
+    --export="ALL,SHARD_INDEX_OFFSET=${offset},MINIMUM_SHARD_INDEX=${minimum},TOTAL_SHARDS=${total}" \
+    tutorials/image/lance_writer/submit_array.sh
+done < <(
+  python tutorials/slurm/retry_array.py \
+    --checkpoint-path "${CHECKPOINT_PATH}" \
+    --format fields \
+    --max-array-size 1001
+)
+```
+
+Repeat until the helper prints no lines, then atomically commit all
+checkpointed fragments:
+
+```bash
+python -m tutorials.image.lance_writer.pipeline commit \
+  --dataset-uri "${DATASET_URI}" \
+  --lance-commit-path "${LANCE_COMMIT_PATH}" \
+  --lance-storage-options "${LANCE_STORAGE_OPTIONS}"
+```
+
+Do not run the commit while array shards are still writing. `LanceWriter`
+checkpoint records make individual packs idempotent; Curator's completion
+manifests determine which logical array shards need another attempt.

tutorials/image/lance_writer/__init__.py

@@ -0,0 +1 @@
+"""Build a URL-addressable Lance image table from retry-attempt tar shards."""