Pipeline

• 1: Pipeline Internals

1 - Pipeline Internals

What a Pipeline Is

A Pipeline is an ordered sequence of Stages. When a Document enters the Pipeline, it flows through each Stage in order. Each Stage modifies the Document in place and may optionally generate child Documents. The Pipeline returns an Iterator over all Documents that emerged from the processing — the original (possibly modified) Document plus any children generated along the way.

The key insight is that the Pipeline does not eagerly process everything and return a collection. It returns a lazy Iterator that processes Documents on demand as next() is called. This has profound implications for memory usage, especially when stages generate children.

How Pipeline.processDocument() Works

The implementation is deceptively simple:

public Iterator<Document> processDocument(Document document) throws StageException {
    Iterator<Document> result = document.iterator();

    for (Stage stage : stages) {
        result = stage.apply(result);
    }

    return result;
}

This builds a chain of lazy iterators — one per stage — and returns the outermost one. No processing has happened yet. The stages are not called until next() is called on the returned Iterator.

What happens when next() is called

Consider a pipeline with stages S1, S2, S3. When the Worker calls next() on the returned Iterator:

1. The outermost iterator (S3’s wrapper) calls next() on its input iterator (S2’s wrapper).
2. S2’s wrapper calls next() on its input iterator (S1’s wrapper).
3. S1’s wrapper calls next() on its input iterator (the original document’s singleton iterator).
4. The original document is returned.
5. S1’s wrapper calls apply(document) — which calls processConditional(document) — which calls S1’s processDocument(document). S1 modifies the document in place and returns an iterator of children (or null).
6. S1’s apply(document) returns an IteratorChain(children, parent) — children first, then the parent.
7. S2’s wrapper receives the first element from S1’s output (a child, if any, or the parent) and calls apply() on it.
8. This continues up the chain until S3 produces its first output document.

The critical property: each document is processed through the full pipeline one at a time. The pipeline doesn’t process all documents through S1, then all through S2, then all through S3. It processes one document through all stages before starting the next.

Children flow through downstream stages only

If S2 generates a child document, that child is passed through S3 but NOT back through S1. This is because the iterator chain is built left-to-right: S1’s output feeds S2, S2’s output feeds S3. A child generated by S2 enters the chain at S2’s output position and only flows forward.

This is the correct semantic for search ingestion: if S1 extracts text from a PDF and S2 chunks that text into pieces, the chunks should flow through S3 (which might generate embeddings) but should NOT flow back through S1 (which would try to extract text from them again).

Why Iterators Instead of Lists

The memory problem with lists

Consider what would happen if processDocument() returned a List<Document> instead of an Iterator<Document>:

// HYPOTHETICAL: if stages returned lists
List<Document> processDocument(Document doc) {
    List<Document> results = List.of(doc);
    for (Stage stage : stages) {
        List<Document> nextResults = new ArrayList<>();
        for (Document d : results) {
            nextResults.addAll(stage.process(d));  // returns list of doc + children
        }
        results = nextResults;
    }
    return results;
}

This approach has a critical flaw: all documents must be held in memory simultaneously.

Consider a pipeline that chunks a large document into 1,000 pieces (S1), then generates an embedding for each piece (S2), then formats each for indexing (S3):

• After S1: 1,000 documents in memory
• After S2: still 1,000 documents in memory (each now with an embedding vector)
• After S3: still 1,000 documents in memory

With the list approach, all 1,000 chunks exist in memory at once. With the iterator approach, only one chunk at a time flows through S2 and S3. The previous chunk has already been handed off to the Worker (which sends it to the indexing queue) before the next chunk is generated.

Compounding with multiple child-generating stages

The problem compounds when multiple stages generate children. Consider:

• S1: Extracts 10 files from a zip archive (10 children)
• S2: Extracts text from each file, producing 5 pages each (50 children)
• S3: Chunks each page into 20 pieces (1,000 children)

With lists: after S3, you’d have 1,000 documents in memory simultaneously.

With iterators: at any given moment, you have at most one document at each level of the chain being processed. The zip is opened lazily, files are extracted one at a time, pages are produced one at a time, chunks are produced one at a time. Memory usage is proportional to the depth of the pipeline, not the breadth of the output.

The lazy evaluation model

The iterator chain implements pull-based lazy evaluation. Nothing happens until the consumer (the Worker) calls next(). Each next() call pulls exactly one document through the full pipeline. This means:

• Documents are produced one at a time
• Each document is fully processed (through all stages) before the next begins
• Memory usage is bounded by the pipeline depth, not the number of children
• The Worker can send each document to the indexing queue immediately after receiving it, freeing memory

In-Place Modification vs. Copying

Lucille stages modify Documents in place rather than creating new copies. This is a deliberate design choice with significant implications.

How it works

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
    // modifies doc directly — no copy is made
    String value = doc.getString("input");
    doc.setField("output", value.toUpperCase());
    return null;
}

The same Document object flows through all stages. S1 modifies it, then S2 sees the modifications S1 made, then S3 sees the modifications both S1 and S2 made.

Why in-place modification is the right choice for search ETL

Memory efficiency. A search ingestion document can be large — it might contain extracted text from a PDF (megabytes), binary content, or dozens of fields. Copying the entire document at each stage would multiply memory usage by the number of stages. With 20 stages and a 5MB document, that’s 100MB per document in flight vs. 5MB.

Simplicity for stage authors. The mental model is straightforward: “I receive a document, I modify it, I’m done.” There’s no need to construct a new Document, copy all existing fields, add new fields, and return it. The boilerplate savings are significant across dozens of stages.

Natural accumulation. In search ingestion, stages typically add information to a document: S1 extracts text, S2 detects language, S3 extracts entities, S4 generates embeddings. Each stage enriches the document with new fields. The in-place model makes this accumulation natural — each stage sees everything previous stages added.

Performance. No allocation, no copying, no garbage collection pressure from intermediate Document objects. For pipelines processing millions of documents, this matters.

The tradeoff: no rollback on failure

If S3 throws an exception after S1 and S2 have already modified the document, the document is in a partially-enriched state. There’s no way to “undo” the modifications from S1 and S2. In Lucille, this is acceptable because:

1. A failed document is routed to a failure state — it’s not indexed in its partial form.
2. Search ingestion is generally idempotent — if the document is retried, it will be processed from scratch (a fresh copy from the source).
3. The alternative (copying at each stage for rollback capability) would impose a performance cost on every document for a scenario that affects a tiny minority.

The tradeoff: stage ordering matters

Because stages see each other’s modifications, the order of stages in the pipeline is significant. S2 can depend on fields that S1 created. This is a feature, not a bug — it’s how enrichment pipelines naturally work (extract text before you can detect language, detect language before you can apply language-specific NER). But it means reordering stages can change behavior, and a stage that expects a field to exist will fail if the stage that creates it is moved downstream.

When copies are necessary

The one case where copies are needed is child document creation. When a stage generates a child, it creates a new Document object (via Document.create(childId)). The child is a separate object from the parent — modifications to the child don’t affect the parent and vice versa. This is correct because children are independent documents that will be indexed separately.

The Stage.apply() Contract

Each Stage has two apply methods that implement the iterator chaining:

apply(Document doc) — process one document

public Iterator<Document> apply(Document doc) throws StageException {
    Iterator<Document> children = processConditional(doc);
    Iterator<Document> parent = doc.iterator();

    if (children == null) {
        return parent;  // no children — just the (modified) parent
    }

    // wrap children to copy run ID and count metrics
    Iterator<Document> wrappedChildren = new Iterator<>() { ... };

    return new IteratorChain(wrappedChildren, parent);  // children first, then parent
}

Key details:

• Children come before the parent in the returned iterator. This ensures the Worker sends CREATE events for children before the parent completes, so the Publisher knows about children before it might declare the parent done.
• The run ID is copied from parent to child automatically.
• Child count metrics are incremented as children are produced (lazily, as next() is called).

apply(Iterator<Document> docs) — wrap an iterator

public Iterator<Document> apply(Iterator<Document> docs) throws StageException {
    return new Iterator<>() {
        Iterator<Document> current = null;

        public boolean hasNext() {
            return (current != null && current.hasNext()) || docs.hasNext();
        }

        public Document next() {
            if (current != null && current.hasNext()) {
                return current.next();
            }
            Document d = docs.next();
            current = apply(d);  // process this document, get iterator of children + parent
            return current.next();  // return first element
        }
    };
}

This wraps an input iterator so that each document pulled from it is processed through the stage. If a document produces children, they are returned before the next document from the input iterator is pulled. This is how the “children flow through downstream stages” behavior works — a child produced by S2 enters S3’s input iterator and is processed by S3 before the next document from S2’s output.

What the Pipeline Framework Gives You

As a stage author, you don’t think about:

• Iterator mechanics. You implement processDocument(Document doc) and return null or an iterator of children. The framework handles the chaining.
• Conditional execution. The framework evaluates conditions before calling your method. If conditions don’t match, your code is never called.
• Dropped/skipped documents. The framework checks these flags before calling your method.
• Metrics. Processing time, error count, and child count are tracked automatically.
• Logging. Stage entry/exit is logged per document automatically.
• Child document lifecycle. Run ID copying, CREATE event sending, and downstream routing are handled by the framework.
• Thread safety. Each Worker thread has its own Pipeline instance with its own Stage instances.
• Memory management. The lazy iterator model ensures bounded memory usage regardless of how many children are generated.

As a pipeline designer, you get:

• Composability. Stages are independent units that can be reordered, added, or removed without modifying other stages.
• Conditional execution via config. Skip stages for certain documents without writing code.
• Heterogeneous document handling. A single pipeline can process different document types differently using conditions.
• Predictable ordering. Stages execute in config order. Children flow through downstream stages only.
• Bounded memory. Even pipelines that generate millions of children from a single input document operate in bounded memory.

Compared to writing your own processing loop:

A naive processing loop (for each stage: stage.process(doc)) would need to handle:

• What if a stage produces children? Do you process them through remaining stages?
• What if children produce grandchildren?
• How do you avoid holding all children in memory?
• How do you ensure children are emitted before parents for accounting?
• How do you handle conditional execution?
• How do you handle dropped/skipped documents?
• How do you track metrics per stage?
• How do you handle errors in one stage without affecting others?

Lucille’s Pipeline handles all of this in ~50 lines of iterator-chaining code that stage authors never see. The stage author’s world is simple: receive a document, modify it, optionally return children. Everything else is the framework’s problem.

Summary

The Pipeline’s design rests on three key decisions:

1. Lazy iterators instead of eager lists. Documents are produced one at a time, bounding memory usage regardless of how many children are generated. Processing is pull-based — nothing happens until the consumer asks for the next document.

2. In-place modification instead of copying. Stages enrich a document by adding fields to it directly. No allocation overhead, no copy overhead, natural accumulation of enrichment across stages. The tradeoff (no rollback on failure) is acceptable because failed documents are discarded, not indexed in a partial state.

3. Children before parents in the output order. This ensures the accounting system learns about children before it might consider the parent complete, preventing premature run-completion detection.

Together, these decisions produce a pipeline framework that is memory-efficient, simple for stage authors, and correct for the accounting system — without requiring stage authors to understand any of the underlying mechanics.