Data Indexing Pipelines for RAG: Common Challenges

Data indexing turns raw data into a form built for retrieval. An application creates new data, and that data is the source of truth. An indexing pipeline processes data that already exists, and every row it writes must stay traceable to its source. Because the index is derived from its source, it has requirements that other pipelines do not. How indexing pipelines differ from other data pipelines compares it with ETL and streaming; this post covers what a good indexing pipeline needs and where pipelines break.

Characteristics of a good indexing pipeline

1. Ease of building

You should be able to build a new indexing pipeline without first learning database access, stream processing, parallelization and fault recovery. Transformation steps should compose and be reusable across pipelines.

2. Maintainability

The pipeline should be easy to understand, change and debug, even when the transformation logic is complex.

An indexing pipeline is also a stateful system. Besides the logic, it should show its state: how many entries it holds, how fresh they are, and which source item produced a given row.

3. Cost-effectiveness

Transformations, and the tracking of which output came from which input, should not cost more compute or storage than they need. Reuse earlier computation where you can. If 1% of the documents change, or a new chunking strategy changes 1% of the chunks, the pipeline should not run the embedding model over the whole dataset again. CocoIndex reuses earlier results through memoization.

4. Indexing freshness

In many applications the source changes all the time, so the index must follow it promptly. CocoIndex does this with live mode.

Common challenges in indexing pipelines

Incremental processing is challenging

Processing only new or changed data, instead of everything, matters for both cost and freshness, and it matters more as the data grows.

To make incremental processing work, the pipeline must track its state, decide which part of the data to process again, and make sure that everything derived from an old version is deleted or replaced. Fan-in and fan-out between steps, out-of-order processing and recovery after a run stops early all make this harder to get right.

Upgradability is often overlooked

Many implementations focus on the first version of the pipeline and not on how it will change. When requirements change, or you add a step, the pipeline should adapt without a full rebuild.

In a traditional implementation, adding or changing a step usually means reprocessing all the data. That is expensive, and someone usually has to run it by hand.

The deterministic logic trap

Many pipelines find the rows to replace by computing their keys again: the same input gives the same keys, so a run overwrites the rows of the run before it. This works only if the processing is deterministic. Three common cases break it:

• A source item is deleted. There is no input left to compute keys from, so nothing finds the rows it produced.
• The logic changes. A new chunk size or a new prompt produces different chunks or facts, with different keys.
• The logic is non-deterministic. An LLM extractor run twice on the same text words some facts differently, or returns a different number of them.

In each case, the keys from the last run no longer match the keys from this run. A pipeline that upserts by key writes the new rows and leaves the old ones in the index. An agent that retrieves from that index then sees old and new facts side by side, and some of the old ones are no longer true.

Try it below. The document is a short set of meeting notes, and an LLM extracts facts from it as (subject, predicate, object) rows keyed by their text. Re-run the extractor, edit the document, then delete it, and compare the two indexes.

CocoIndex does not find old rows by computing their keys again. Each source item is a processing component with a stable component path, such as the file path. After the component runs, CocoIndex compares the target states it declared in this run with the ones it declared in the last run at the same path, then applies the inserts, updates and deletes as one unit. When the file is deleted, its component is no longer mounted, and CocoIndex deletes its target states. The row keys can change from run to run; the component path stays stable.

Here is the extractor from the demo as a CocoIndex app. extract_facts stands for your LLM call (a @coco.fn function), and a lifespan provides the Postgres pool for PG_DB:

import pathlib
from dataclasses import dataclass

import asyncpg
import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

PG_DB = coco.ContextKey[asyncpg.Pool]("facts_db")

@dataclass
class Fact:
    doc: str  # primary key, with id
    id: str  # the fact as the LLM worded it
    subject: str
    predicate: str
    object: str

@coco.fn(memo=True)
async def process_doc(file: FileLike, table: postgres.TableTarget[Fact]) -> None:
    for f in await extract_facts(await file.read_text()):
        table.declare_row(row=Fact(
            id=f"{f.subject}|{f.predicate}|{f.object}",
            doc=file.file_path.path.name,
            subject=f.subject, predicate=f.predicate, object=f.object,
        ))

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    table = await postgres.mount_table_target(
        PG_DB,
        table_name="facts",
        table_schema=await postgres.TableSchema.from_class(Fact, primary_key=["doc", "id"]),
    )
    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
    )
    await coco.mount_each(process_doc, files.items(), table)

app = coco.App(coco.AppConfig(name="Facts"), app_main, sourcedir=pathlib.Path("./notes"))

mount_each makes one component per file. With memo=True, an unchanged file is skipped and its rows carry over. The component runs again when the file changes, or when the code of process_doc or of a @coco.fn function it calls changes (such as a decorated extract_facts). That run is when CocoIndex compares and deletes. The re-run button in the demo stands for that case, or for cocoindex update --full-reprocess.

How does CocoIndex solve these challenges?

You write the transformation as ordinary Python and declare the target state you want. CocoIndex keeps the state and computes the changes. The core concepts explain how an app, its processing components and their target states fit together. For the three challenges above:

• Incremental processing. For memoized components, CocoIndex re-runs only the ones whose input or code changed. Memoized functions inside a component reuse their cached results, so an unchanged chunk keeps its embedding.
• Upgradability. CocoIndex fingerprints the code of each @coco.fn function. When you change one step, only the calls that depend on it run again, and results whose inputs and code did not change stay cached.
• Non-deterministic logic. Stale rows are deleted by comparing declared target states, not by computing keys again, as the demo shows. Each target state belongs to the component that declared it, so every row traces back to its source item; see lineage.

CocoIndex also handles recovery. It writes a component’s target states only after the component finishes, so a failed run leaves no partial output from processing. If a write fails partway, the next run computes the declared state again and brings the target to it.

The mental model shift

React changed how developers update a UI: you declare what the UI should look like for a given state, and React applies the changes. CocoIndex applies the same idea to indexing. You declare the transformation and the target state, and CocoIndex works out which inserts, updates and deletes the target needs.

About the author

CocoIndex TeamThe team behind CocoIndex · @cocoindex_io

Posts from the CocoIndex team — product launches, release notes, and announcements.