Problem
searchIndexWriter (from #534/#565) buffers a whole dataset's extracted quads – and the groupByRoot subject index it feeds – before projecting, so peak memory is O(dataset). Fine for the current scale (millions of objects on Typesense), but it caps large-object indexing and pairs with the disk-backed engine work in #572: a 500M-object index wants streamed ingestion, not a dataset buffered in the indexer.
Root cause: Stage.runWithSelector merges every selector batch into a single per-dataset write(), which erases the self-contained framing unit the selector already produced. Each selector batch (per-root CONSTRUCT with its one-hop closure) is independently framable; the merge is what forces the whole-dataset buffer.
Decision (settled in design discussion)
Keep @lde/pipeline quad-typed – no Stage<Out> / documents-as-pipeline-payload (that trades away the RDF-pipeline integrity for genericity we do not want). Instead:
@lde/pipeline: add an opt-in per-batch (unmerged) write mode on Stage – each self-contained selector batch is delivered as its own write(dataset, …) call, writes serialized (readers still concurrent).@lde/search-pipeline: searchIndexWriter projects each write() chunk and streams the documents on, dropping the per-dataset buffer → peak memory O(batch). No @lde/search change (projectGraph on a bounded chunk suffices).
No Writer-interface change: the chunk is the existing write() call; the engine Writer<SearchDocument> already streams (BatchImporter).
Why opt-in, not the default
- Concurrency:
runWithSelector reads batches concurrently through one single-consumer queue, which serializes the write. Per-batch-by-default would force write serialization on the shared path, and firing concurrent write()s into the existing writers is unsafe (SparqlUpdateWriter's CLEAR-once check races; FileWriter's buffer interleaves).
- Efficiency:
SparqlUpdateWriter coalesces INSERT DATA by its own batchSize; per-batch writes mean more, smaller INSERTs. The production quad pipelines (NDE DKG, Linked Open Limburg) would pay that for no benefit.
Opt-in confines both to the search pipeline that benefits, at zero blast radius on the production pipelines.
Correctness contract
Per-batch projection is correct only if each root's CONSTRUCT inlines its one-hop closure, so a batch is self-contained to frame – which frameByType already assumes. State it explicitly for extraction authors.
Scope
@lde/pipeline: opt-in unmerged write mode on Stage (serialize writes, keep concurrent reads); tests; confirm SparqlUpdateWriter / FileWriter / @lde/pipeline-void unchanged in default mode.@lde/search-pipeline: rework searchIndexWriter to project + stream per write(); enable the mode; drop the passes buffer.- ADR for the per-batch mode + one-hop contract (supersedes the
searchIndexWriter-buffers-per-dataset note).
Relationships
Non-goal
- Genericizing the pipeline over payload type (
Stage<Out>, documents in the pipeline). Payload stays Quad.
Problem
searchIndexWriter(from #534/#565) buffers a whole dataset's extracted quads – and thegroupByRootsubject index it feeds – before projecting, so peak memory is O(dataset). Fine for the current scale (millions of objects on Typesense), but it caps large-object indexing and pairs with the disk-backed engine work in #572: a 500M-object index wants streamed ingestion, not a dataset buffered in the indexer.Root cause:
Stage.runWithSelectormerges every selector batch into a single per-datasetwrite(), which erases the self-contained framing unit the selector already produced. Each selector batch (per-rootCONSTRUCTwith its one-hop closure) is independently framable; the merge is what forces the whole-dataset buffer.Decision (settled in design discussion)
Keep
@lde/pipelinequad-typed – noStage<Out>/ documents-as-pipeline-payload (that trades away the RDF-pipeline integrity for genericity we do not want). Instead:@lde/pipeline: add an opt-in per-batch (unmerged) write mode onStage– each self-contained selector batch is delivered as its ownwrite(dataset, …)call, writes serialized (readers still concurrent).@lde/search-pipeline:searchIndexWriterprojects eachwrite()chunk and streams the documents on, dropping the per-dataset buffer → peak memory O(batch). No@lde/searchchange (projectGraphon a bounded chunk suffices).No Writer-interface change: the chunk is the existing
write()call; the engineWriter<SearchDocument>already streams (BatchImporter).Why opt-in, not the default
runWithSelectorreads batches concurrently through one single-consumer queue, which serializes the write. Per-batch-by-default would force write serialization on the shared path, and firing concurrentwrite()s into the existing writers is unsafe (SparqlUpdateWriter's CLEAR-once check races;FileWriter's buffer interleaves).SparqlUpdateWritercoalescesINSERT DATAby its ownbatchSize; per-batch writes mean more, smaller INSERTs. The production quad pipelines (NDE DKG, Linked Open Limburg) would pay that for no benefit.Opt-in confines both to the search pipeline that benefits, at zero blast radius on the production pipelines.
Correctness contract
Per-batch projection is correct only if each root's
CONSTRUCTinlines its one-hop closure, so a batch is self-contained to frame – whichframeByTypealready assumes. State it explicitly for extraction authors.Scope
@lde/pipeline: opt-in unmerged write mode onStage(serialize writes, keep concurrent reads); tests; confirmSparqlUpdateWriter/FileWriter/@lde/pipeline-voidunchanged in default mode.@lde/search-pipeline: reworksearchIndexWriterto project + stream perwrite(); enable the mode; drop thepassesbuffer.searchIndexWriter-buffers-per-dataset note).Relationships
searchIndexWriter).Non-goal
Stage<Out>, documents in the pipeline). Payload staysQuad.