feat(vdb): bound batch LanceDB ingestion with Arrow streaming

[jioffe502]

Why

Batch VDB ingestion currently repartitions the complete embedded Ray dataset into one block before LanceDB starts writing. This creates a corpus-sized handoff and leaves the VDB write and index work until the end of the pipeline.

What this changes

This PR replaces that handoff with one coordinated LanceDB stream:

• reads native Ray batches without repartition(num_blocks=1);
• projects and packs canonical Arrow batches under a byte budget;
• sends one lazy RecordBatchReader to one LanceDB data mutation;
• validates the committed rows, schema, and index coverage before reporting success;
• preserves the existing pandas ingest result without running upstream work twice;
• records incomplete operations so retries fail safely instead of silently duplicating append data.

This is not a per-Ray-block append design. Ray workers do not independently commit, overwrite, or build indices.

The implementation is LanceDB-specific, but the Ray executor is not: it asks the ingest operator whether bounded terminal streaming is supported. Custom VDB implementations and PutVdbOperator keep their existing behavior.

Measured result

We ran the existing implementation and this PR through the real RayDataExecutor on the same source-frozen, CPU-only set of 12 nightly datasets: 197,796 stored rows per arm and 24 completed/validated runs.

Measured sink-slice result	Existing global handoff	This PR	Change
Aggregate graph wall time	130.10 s	99.13 s	-23.8%
Maximum sampled process-tree RSS	17.92 GiB	7.74 GiB	-56.8%
Maximum sampled Ray object-store use	1.72 GiB	0.93 GiB	-45.5%

Correctness checks showed:

• exact stored row-count, multiset-digest, schema-fingerprint, and dense-index-coverage parity;
• one table data mutation, one fragment, and one data file per run;
• no candidate-side repartition or materialization.

Validation

Local validation after merging current main:

• 53 real-Lance sink, recovery, concurrent-write, and index tests passed;
• 164 operator and Ray executor tests passed;
• the real Ray 2.56.1 multi-block integration passed on CPU;
• Black and Flake8 passed.

Scope of the measurements

These are single-repetition sink-slice measurements, not end-to-end extraction/embedding results or statistically qualified performance claims. The retained inputs contain canonical VDB rows rather than the original wide graph rows.

The public pandas result remains corpus-sized. The configured byte limit bounds canonical Arrow batches owned by the sink; it does not claim a corpus-independent total process RSS bound. The performance matrix covers dense overwrite and indexing.

Detailed design and evidence: bounded Ray Data LanceDB sink research.

Relates to #2476.

[jioffe502]

Refreshed this branch onto current upstream/main (bc9f2ce5, including #2566) and resolved the LanceDB index-wait overlap without dropping either side's behavior.

Post-merge validation:

• 53 real-Lance sink, recovery, concurrent-write, and IVF tests passed;
• 163 operator and Ray executor tests passed;
• the CPU-only real Ray 2.56.1 multi-block integration passed;
• merge commit hooks passed (Black and Flake8).

The 12-dataset measurements in the PR body remain the source-frozen 41ab2753 control/candidate run; I did not relabel or silently rerun that evidence after the merge.

Architectural scope is now explicit: this PR is LanceDB-first. The Arrow schema, RecordBatchReader write, retry markers, version/index validation, and optimization lifecycle remain Lance-owned. Decoupling terminal streaming capability detection from concrete LanceDB knowledge is tracked separately in #2570 and is not being folded into this PR.

[greptile-apps]

Greptile Summary

The PR replaces the global LanceDB handoff with coordinated Arrow batch streaming and adds operation tracking, commit validation, and index finalization.

• Adds bounded canonical Arrow batch projection and a single LanceDB mutation.
• Integrates terminal streaming into the Ray executor while preserving the pandas result.
• Adds retry markers, readiness checks, validation, and focused sink and Ray tests.

Confidence Score: 5/5

The PR appears safe to merge because no blocking failure remains.

No blocking failure remains.

Important Files Changed

Filename	Overview
nemo_retriever/src/nemo_retriever/common/vdb/sink.py	Implements canonical Arrow batching, coordinated LanceDB writes, validation, and finalization.
nemo_retriever/src/nemo_retriever/common/vdb/sink_operation.py	Adds durable operation identities and lifecycle markers for safe retry handling.
nemo_retriever/src/nemo_retriever/graph/executor.py	Adds capability-driven terminal batch streaming to Ray execution.
nemo_retriever/src/nemo_retriever/operators/vdb.py	Exposes bounded-batch consumption for supported VDB ingestion operators.
nemo_retriever/src/nemo_retriever/common/vdb/lancedb.py	Adds sink readiness enforcement and phase-level index timing support.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
    Ray[Ray Dataset blocks] --> Iter[iter_batches]
    Iter --> Project[Canonical row projection]
    Project --> Pack[Byte-budgeted Arrow batches]
    Pack --> Reader[RecordBatchReader]
    Reader --> Lance[Single LanceDB mutation]
    Lance --> Validate[Validate rows and schema]
    Validate --> Index[Build and validate indexes]
    Index --> Success[Record operation success]
    Iter --> Result[Retained pandas result]

Loading

_{Reviews (2): Last reviewed commit: "Merge remote-tracking branch 'upstream/m..." | Re-trigger Greptile}