Bound LanceDB ingestion memory with an Arrow-native transactional sink

[jioffe502]

Summary

Batch ingestion currently treats IngestVdbOperator as a global-batch operator. Ray repartitions the complete embedded dataset into one block, converts that block for VDB.run(), and only then begins the LanceDB table-write lifecycle.

That handoff is not bounded by rows or bytes, so peak host memory grows with the complete embedded result. This issue is about making the terminal storage path bounded and observable. It is not a claim that an unmodified full-corpus run has been observed OOMing on a 251 GiB host.

Current evidence

Control	Production code	Result
Full BrowseComp-Plus validation	Validation stack containing #2454 and #2474	100,195 files produced 781,927 embedded rows and one 33.2 GiB Ray object-store block. The store actor reached 173.34 GiB RSS; host-available memory fell to 24.16 GiB. The run was manually stopped after about 930 seconds with zero rows committed. No OOM exception or kernel kill occurred.
BrowseComp-Plus 10k sink replay	Detached upstream main at c80f4a51	Replaying 75,065 already-embedded rows through LanceDB.run() under an 8 GiB/no-swap cgroup reached the sink and was OOM-killed (OOMKilled=true, exit 137).
Same BrowseComp-Plus replay	Detached upstream main at c80f4a51	The identical 75,065 rows completed under a 24 GiB/no-swap cgroup with a 12.378 GiB observed peak.
BRIGHT Biology 10k replay	Detached upstream main at c80f4a51	10,014 already-embedded rows completed under 8 GiB at roughly 1.5 GiB process RSS.

The controlled replay disables vector-index construction and does not repeat extraction, embedding, or Ray repartition. It isolates table creation/write and demonstrates a real memory-capacity failure at that boundary. It does not prove that full BrowseComp-Plus naturally OOMs on a 251 GiB host, nor does it reproduce a separate failure during vector/FTS index construction.

Related history should remain separate:

• The original BRIGHT nightly failure was serialized CPU text extraction and six-hour harness timeouts, not OOM.
• PR fix: preserve heterogeneous Ray Data rows across batch ingestion #2474 fixes Arrow/pandas representation correctness; it does not implement storage batching or memory admission. It allowed the full embedded result to reach the existing sink, where this downstream behavior became measurable.

Desired outcome

Replace the global materialization boundary with a terminal LanceDB sink whose in-flight memory is bounded by configuration and whose lifecycle is explicit:

1. Prepare
   • Resolve the canonical Arrow schema and vector dimension before writing.
   • Open or initialize the target once.
   • Select overwrite or append semantics before consuming any batch.
2. Consume with backpressure
   • Project only canonical VDB columns plus compact accounting/result columns.
   • Consume bounded Arrow batches using a byte-based in-flight budget and small prefetch count.
   • Use one logical writer/coordinator; do not turn every Ray block into an independent LanceDB commit.
3. Commit and finalize
   • Validate the committed table version and row count.
   • Build or update vector/FTS indices once after the data write succeeds.
   • Apply compaction/optimization only through an explicit policy informed by table statistics.
4. Abort and retry
   • Do not report success or build indices after a failed data write.
   • Define overwrite and append retry semantics explicitly. Append must not silently duplicate rows.

Evaluate the public LanceDB paths before selecting a low-level implementation:

1. one table.add() supplied by a bounded Arrow RecordBatch iterator;
2. bounded Arrow/Parquet staging exposed as a pyarrow.dataset.Dataset, followed by one coordinated table.add();
3. Lance's two-phase distributed-write model only if the public iterator/dataset paths cannot satisfy the gates.

Prior art

PR #2041 showed that streaming storage can reduce memory and wall time, but its per-Ray-block table.add() design should remain an experimental comparison rather than the default:

• every block can become a table version/commit and create small fragments;
• correctness depends on mutable “first batch overwrites, later batches append” state;
• retries can duplicate or partially publish work;
• increasing writer concurrency creates competing committers rather than backpressure;
• result materialization and public result types became coupled to the storage change.

Reference: #2041

Required ablations

Use immutable inputs and the same hardware:

Control	Purpose
Current global-batch sink	Establish peak memory, versions/fragments, and end-to-end baseline
PR #2041-style per-block append, experiment only	Quantify commit/version/fragment amplification
Bounded RecordBatch iterator	Evaluate the preferred public-API path
Staged pyarrow.dataset.Dataset	Evaluate the bounded-memory throughput fallback

Tune by bytes, not only rows. Test at least 64 MiB, 256 MiB, and 512 MiB in-flight budgets with prefetch 1 and 2 on the fixed 10k BrowseComp-Plus control. Use one representative heterogeneous PDF corpus to validate the Arrow schema path. Do not repeat the full BrowseComp-Plus run until a bounded candidate wins these controls.

Acceptance criteria

Correctness

• Exact file and stored-row parity on the fixed BrowseComp-Plus 10k control: 10,000 files / 75,065 rows.
• Exact expected stored rows on a representative heterogeneous PDF control.
• Full BrowseComp-Plus eventually commits all 781,927 expected rows.
• Vector and FTS index coverage includes every committed row before the job reports success.
• Failure injection leaves no visible partial overwrite, builds no index, and retry behavior is explicit and tested.
• Schema validation and bad-vector handling match the existing production contract.
• No changes to the deprecated NRL benchmark harness.

Resource bounds

• No global repartition(num_blocks=1) for the VDB sink.
• Peak sink RSS is bounded by the configured in-flight byte budget plus measured Arrow/LanceDB overhead, rather than corpus size.
• Ray object-store spill remains zero on the bounded controls.
• Table version and fragment growth do not scale linearly with the number of Ray input blocks.
• Peak RSS, object-store bytes/spill, versions, fragments, committed rows, and index duration are recorded.

Performance

• The 10k BrowseComp-Plus control remains within 10% of the current 854.510-second end-to-end result while materially reducing sink memory.
• The winning design completes the full BrowseComp-Plus sink without approaching the host safety boundary.
• Record rows/s, bytes/s, write duration, and index duration are first-class telemetry.

Tests

• Arrow batch iteration preserves schema and enforces the configured byte/prefetch bound.
• A real local LanceDB multi-batch write produces one complete table and correct indices.
• Failure/retry coverage proves visible-version and duplicate-row behavior.
• One bounded Ray integration test verifies backpressure and avoids global repartition.
• Existing ingestion and retrieval suites remain green.

Non-goals

• Diagnosing unrelated reports that fail specifically during vector-index construction; first capture their exact phase and memory evidence.
• Redesigning extraction routing or actor admission.
• Changing service-mode model residency.
• Adding behavior to the deprecated NRL benchmark harness.
• Solving legacy full-result client materialization in the same production diff.
• Selecting the low-level distributed-fragment API before the public LanceDB paths are measured.

References

• LanceDB large-ingest and iterator guidance: https://docs.lancedb.com/tables/create
• LanceDB concurrent-write limits: https://docs.lancedb.com/faq/faq-oss
• LanceDB Table.add, compaction, and optimization API: https://lancedb.github.io/lancedb/python/python/
• Lance distributed writes: https://lancedb.github.io/lance/distributed_write.html
• Prior experiment: Streaming VDB writes #2041