Data Pipeline & Indexing

The policy that answered three ways

Monday night, a cron job loads the company handbook into Qdrant. On Tuesday, HR changes the remote-work policy from two days a week to three, and deletes an old page about a 2019 pilot. On Wednesday morning, someone asks the assistant: "How many days a week can I work from home?"

Here is the job, in the form most teams write first. It uses the small open model all-MiniLM-L6-v2 and the course's local Qdrant server:

import uuid
from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")   # 384-d, unit vectors
client = QdrantClient(url="http://localhost:6333")     # or QdrantClient(":memory:")
COLL = "handbook"

MONDAY = {   # what the wiki connector lists on Monday night: doc id -> document
    "handbook/remote-work.md": {"acl": ["all-staff"], "text":
        "Remote work: you may work from home up to two days a week. Agree the days with your manager.\n\n"
        "Equipment: the company lends you a laptop and one monitor for home use.\n\n"
        "Internet: you can claim up to 30 euros a month for your home connection."},
    "handbook/remote-work-2019.md": {"acl": ["all-staff"], "text":
        "Working from home (2019 pilot): staff may work from home one day a week with written approval."},
    "handbook/travel.md": {"acl": ["all-staff"], "text":
        "Travel booking: book flights through the travel portal at least 14 days ahead.\n\n"
        "Hotels: the nightly limit is 150 euros in most cities and 220 euros in London and Paris."},
    "handbook/expenses.md": {"acl": ["all-staff"], "text":
        "Expense claims: submit receipts within 30 days of the purchase through the finance portal."},
    "handbook/security.md": {"acl": ["all-staff"], "text":
        "Laptops: lock your screen whenever you leave your desk.\n\n"
        "Passwords: use the company password manager and never reuse a password."},
}
TUESDAY = dict(MONDAY)   # HR edits the policy and deletes the 2019 pilot page
TUESDAY["handbook/remote-work.md"] = {"acl": ["all-staff"], "text":
    "Remote work: you may work from home up to three days a week. Agree the days with your manager.\n\n"
    "Equipment: the company lends you a laptop and one monitor for home use."}
del TUESDAY["handbook/remote-work-2019.md"]

def naive_ingest(source):
    """The script everyone writes first: random ids, append only."""
    for doc_id, doc in source.items():
        chunks = doc["text"].split("\n\n")
        vectors = model.encode(chunks)
        client.upsert(COLL, points=[
            models.PointStruct(id=str(uuid.uuid4()), vector=v.tolist(),
                               payload={"doc_id": doc_id, "text": c})
            for c, v in zip(chunks, vectors)])

if client.collection_exists(COLL):
    client.delete_collection(COLL)
client.create_collection(COLL, vectors_config=models.VectorParams(
    size=384, distance=models.Distance.COSINE))

for night, source in [("Mon", MONDAY), ("Tue", TUESDAY), ("Wed", TUESDAY)]:
    naive_ingest(source)
    print(night, "points:", client.count(COLL, exact=True).count)

q = model.encode("How many days a week can I work from home?")
for hit in client.query_points(COLL, query=q.tolist(), limit=5).points:
    print(f"{hit.score:.3f}  {hit.payload['doc_id']:29} {hit.payload['text'][:52]}")

Predict first: on Wednesday the wiki holds seven paragraphs. How many points does the collection hold after Wednesday's run? And how many different answers to the question sit in the top four?

Mon points: 9
Tue points: 16
Wed points: 23
0.695  handbook/remote-work.md       Remote work: you may work from home up to three days
0.695  handbook/remote-work.md       Remote work: you may work from home up to three days
0.668  handbook/remote-work.md       Remote work: you may work from home up to two days a
0.663  handbook/remote-work-2019.md  Working from home (2019 pilot): staff may work from 
0.339  handbook/remote-work.md       Internet: you can claim up to 30 euros a month for y

Nothing here is wrong with the model, the embeddings or the search. Retrieval did its job on the wrong corpus. RAG Architecture & Implementation calls this failure bucket Not indexed: the index no longer matches the source. The nine lines of naive_ingest hold three bugs, and every nightly run "succeeded":

1. Random ids. Every run appends a fresh copy of every chunk, so the top-k fills with duplicates.
2. No replacement. An edit adds a new version next to the old one instead of replacing it.
3. No deletion path. A page deleted at the source lives on in the index, still retrievable.

A fourth bug shows in no query: the acl never reached the payload, so a permission filter has nothing to filter on.

This lesson is the map of the offline half of a RAG system: how documents become searchable chunks, and how to keep the index equal to the source, run after run. One property carries most of the weight:

Idempotent ingestion: running the job again on an unchanged source changes nothing, and running it after a change makes the index match the source: every edit replaced, every deletion removed.

Each move below starts from a symptom you can see in results, logs or bills:

This lesson assumes you know how a chunk becomes a vector (Vector Embeddings) and the basics of Qdrant.

The map: from source to searchable

Every ingestion pipeline, whatever its tools, does the same eight jobs. Stage 1 lists the source once per run. Every listed document becomes one work item, which fetches the document (the rest of stage 1) and runs stages 2 to 7. After the work items, two steps run once per run: deleting the documents the listing no longer has (the end of stage 7), and stage 8:

source ─► 1 connect: list every doc id, once per run
                                        │  one work item per listed document
          ┌─────────────────────────────▼─────────────────────────────┐
          │ 1 connect   fetch bytes, updated_at, ACL                  │
          │ 2 parse     bytes → text with headings and tables         │
          │ 3 clean     strip boilerplate, normalize, drop copies     │
          │ 4 chunk     text → pieces the embedding model can read    │
          │ 5 enrich    payload fields; optional context header       │
          │ 6 embed     chunk text → vector, one model per collection │
          │ 7 index     upsert by deterministic id, delete leftovers  │
          └─────────────────────────────┬─────────────────────────────┘
                                        ▼  once per run, after the work items
                     7 index    delete documents the listing no longer has
                     8 verify   counts, coverage, golden queries → serve or hold

Two habits make the rest of the lesson easier.

Keep what each stage produced. Store the fetched bytes, the parsed text and the chunks (with any LLM-written headers), keyed by document id and a hash of their inputs (text and chunker version, not the ACL). A rebuild for a new embedding model then starts at stage 6 instead of re-running OCR and LLM calls (Move 6 prices the difference). These stores are rebuild sources, so deletes must reach them too, or a rebuild brings deleted documents back (Data Freshness & Lifecycle).

One work item = one document. A document is small enough to retry cheaply and big enough to update as a whole: all its chunks change together. Batching for the embedding API happens across work items, inside stage 6.

Before any code: what is a document?

Stage 1 decides the identity everything else depends on. Every chunk should carry these fields from ingest to citation:

The rule for doc_id is short: stable across edits, unique across sources. Prefer the source system's own immutable id (most wikis, drives and ticket systems have one; prefix it with the system name so two systems cannot collide). A path or URL works if you canonicalize it (one case, no tracking parameters, one trailing-slash rule), but a rename or move then looks like a delete plus a new document. Never put a version, timestamp or content hash into doc_id. Those go in the payload.

Predict: a teammate sets doc_id = sha256(path + modified_at)[:16] "so every edit gets a fresh id". A five-chunk page is edited down to three chunks. The change event runs the update for the new id: upsert the new chunks, then delete this doc_id's chunks from older versions. How many points does the page have afterwards?

Move 1: Give every chunk an address

Symptom: re-running the job doubles the results.

naive_ingest called upsert, and upsert does overwrite: a point with the same id is replaced. It never overwrote anything because uuid4() makes a new id on every call. The fix is an id computed from what the chunk is, so the same chunk gets the same id on every run and on every machine.

Qdrant accepts only unsigned integers or UUIDs as point ids. Send a readable id such as handbook/remote-work.md:0 to the local 1.19.1 server and it answers 400 Bad Request with:

Format error in JSON body: value handbook/remote-work.md:0 is not a valid point ID, valid
values are either an unsigned integer or a UUID

A name-based UUID (version 5) bridges the two: uuid.uuid5(namespace, name) hashes a readable name into a UUID, deterministically. Keep the readable parts (doc_id, chunk_index) in the payload for filters and citations.

import uuid
print(uuid.uuid5(uuid.NAMESPACE_URL, "handbook/remote-work.md:0"))   # db327a63-a96d-554b-96c6-d1de36901cf5
print(uuid.uuid5(uuid.NAMESPACE_URL, "handbook/remote-work.md:0"))   # db327a63-a96d-554b-96c6-d1de36901cf5
print(uuid.uuid4(), uuid.uuid4())                                    # two new random UUIDs every run

Random ids break idempotent re-runs. They do not make deletes impossible: if every point carries doc_id in its payload, a filtered delete removes a document whatever its point ids are. That is why the payload field matters more than the id scheme, and why naive_ingest storing doc_id would have let you clean up this mess with one delete per document.

Which address for a chunk?

All three common schemes are deterministic, so all three make re-runs idempotent. They differ in what an edit does:

None of them alone handles a document that shrinks. Edit a five-chunk page down to three, upsert chunks 0 to 2, and chunks 3 and 4 are never touched. They are leftovers: stale text from a version that no longer exists, still retrievable. That is Move 2's job.

Your turn. A contracts team edits long agreements mostly by inserting clauses near the top. Queries cite clauses by section anchor. Which chunk id scheme keeps the most points unchanged on an edit, and what must you still do on every edit?

Move 2: Replace, don't append

Symptom: an edited page leaves its old paragraphs behind, and a page deleted at the source is still cited.

Here is the whole sync, continuing from the first block. It keeps three promises: skip what has not changed, replace what has, and remove what is gone.

import hashlib, json
from collections import Counter

CHUNKER = "paragraphs-v1"     # bump it when splitting or cleaning changes
MAX_REMOVED_SHARE = 0.25      # demo value; see "The listing that lied" below
QUARANTINE: dict[str, list[str]] = {}

def point_id(doc_id: str, i: int) -> str:
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{doc_id}:{i}"))   # same chunk -> same id

def fingerprint(doc: dict) -> str:
    """Changes when anything that shapes the indexed points changes."""
    raw = json.dumps([CHUNKER, doc["text"], sorted(doc["acl"])])
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

def of_doc(doc_id: str) -> models.FieldCondition:
    return models.FieldCondition(key="doc_id", match=models.MatchValue(value=doc_id))

def ingest_document(doc_id: str, doc: dict) -> int:
    """Make the index hold exactly this version of doc_id. Safe to run twice."""
    fp = fingerprint(doc)
    chunks = doc["text"].split("\n\n")
    vectors = model.encode(chunks)
    client.upsert(COLL, wait=True, points=[
        models.PointStruct(id=point_id(doc_id, i), vector=v.tolist(), payload={
            "doc_id": doc_id, "chunk_index": i, "fingerprint": fp,
            "acl": doc["acl"], "text": c})
        for i, (c, v) in enumerate(zip(chunks, vectors))])
    client.delete(COLL, wait=True, points_selector=models.Filter(      # leftovers of
        must=[of_doc(doc_id)],                                          # older versions
        must_not=[models.FieldCondition(key="fingerprint", match=models.MatchValue(value=fp))]))
    return len(chunks)

def remove_document(doc_id: str) -> None:
    client.delete(COLL, wait=True, points_selector=models.Filter(must=[of_doc(doc_id)]))

def index_state() -> dict[str, set[str]]:
    """doc_id -> the fingerprints its points carry right now."""
    state, offset = {}, None
    while True:
        points, offset = client.scroll(COLL, limit=1000, offset=offset,
                                       with_payload=["doc_id", "fingerprint"])
        for p in points:
            state.setdefault(p.payload["doc_id"], set()).add(p.payload["fingerprint"])
        if offset is None:
            return state

def sync(source: dict[str, dict], check=lambda doc: []) -> Counter:
    state, stats = index_state(), Counter(listed=len(source))
    for doc_id, doc in source.items():
        if problems := check(doc):              # Move 4 plugs quality checks in here
            QUARANTINE[doc_id] = problems       # keep the last good version indexed
            stats["quarantined"] += 1
            continue
        if state.get(doc_id) == {fingerprint(doc)}:
            stats["unchanged"] += 1
            continue
        stats["written"] += 1
        stats["chunks_embedded"] += ingest_document(doc_id, doc)
    gone = state.keys() - source.keys()         # in the index, no longer at the source
    if len(gone) > MAX_REMOVED_SHARE * len(state):
        raise RuntimeError(f"listing lacks {len(gone)} of {len(state)} indexed documents; "
                           "refusing to delete them")
    for doc_id in gone:
        remove_document(doc_id)
        stats["removed"] += 1
    stats["points"] = client.count(COLL, exact=True).count
    return stats

client.delete_collection(COLL)                  # start again from an empty collection
client.create_collection(COLL, vectors_config=models.VectorParams(
    size=384, distance=models.Distance.COSINE))
client.create_payload_index(COLL, "doc_id", models.PayloadSchemaType.KEYWORD)   # before loading

for night, source in [("Mon", MONDAY), ("Mon again", MONDAY), ("Tue", TUESDAY)]:
    print(f"{night:9}", dict(sync(source)))
for hit in client.query_points(COLL, query=q.tolist(), limit=3).points:
    print(f"{hit.score:.3f}  {hit.payload['doc_id']:24} {hit.payload['text'][:52]}")

It prints:

Mon       {'listed': 5, 'written': 5, 'chunks_embedded': 9, 'points': 9}
Mon again {'listed': 5, 'unchanged': 5, 'points': 9}
Tue       {'listed': 4, 'written': 1, 'chunks_embedded': 2, 'unchanged': 3, 'removed': 1, 'points': 7}
0.695  handbook/remote-work.md  Remote work: you may work from home up to three days
0.262  handbook/remote-work.md  Equipment: the company lends you a laptop and one mo
0.232  handbook/travel.md       Hotels: the nightly limit is 150 euros in most citie

That is the idempotency test, passed three ways:

The question now has one answer, three days, at 0.695. The next hit scores 0.262.

Why it works

ingest_document ends in one state whatever state it started from: every point with this doc_id carries the current fingerprint, one point per chunk. The upsert overwrites chunks 0 to n−1 in place, and the filtered delete removes anything of this document with another fingerprint, which is exactly the leftovers. sync then makes the set of documents match: whatever is in the index but no longer listed is removed. Run either step twice and nothing more happens.

Order matters. Writing the new version first means there is never a moment when the page has no chunks. Deleting by doc_id first and then upserting is also correct and slightly simpler, as the Qdrant lesson shows, but the document is missing from search between the two calls, and if embedding fails in between, it stays missing until the retry succeeds.

Why the fingerprint includes the ACL and the chunker version. It must change whenever the indexed points would change. Hash only the text, and a permission change (a page restricted from all-staff to hr) is "unchanged" forever: the page stays visible to everyone. Leave out CHUNKER, and a new splitting rule never reaches old documents.

Detecting a permission change is not the same as re-embedding for it. This demo has one write path, so an ACL change runs ingest_document and re-embeds the whole document: the index ends up right, but the embedding calls are wasted. A production pipeline also stores a hash of the text part (chunker version and text) on each point. When the fingerprint changes and that hash does not, only the permissions changed: one filtered set_payload writes the new acl and fingerprint to all of the document's points, and nothing is embedded (Metadata & Filtering, Data Freshness & Lifecycle). An embedding cache keyed by model and chunk text also makes such a re-embed free.

What it costs. index_state scrolls the whole collection: fine for thousands of points, slow for tens of millions. At that scale keep a manifest table (doc_id, fingerprint, chunk count, status, last error), update it after each confirmed write, and reconcile it against the index on a schedule. Each changed document costs one filtered delete (cheap with the payload index on doc_id) and a moment in which a leftover sits next to the new chunks.

Your turn. The worker is killed between the upsert and the delete while applying Tuesday's edit to remote-work.md (starting from Monday's index). What does the index hold for that page, and what does the next sync(TUESDAY) do?

sync(MONDAY)                                     # back to Monday's index
real_delete, client.delete = client.delete, None # the "worker" dies before its delete
try:
    ingest_document("handbook/remote-work.md", TUESDAY["handbook/remote-work.md"])
except TypeError:
    print("killed after the upsert")
client.delete = real_delete
print(len(index_state()["handbook/remote-work.md"]), "fingerprints")   # 2 fingerprints
print(dict(sync(TUESDAY)))
# {'listed': 4, 'written': 1, 'chunks_embedded': 2, 'unchanged': 3, 'removed': 1, 'points': 7}

The listing that lied

The deletion step trusts the connector's listing. If a token expires and the listing comes back empty, "everything in the index is gone at the source" is exactly what a naive sync concludes, and it deletes the whole index at 3 a.m. That is why sync has a delete brake: it refuses to remove more than a share of the indexed documents in one run:

try:
    sync({})                     # the connector's token expired: it listed nothing
except RuntimeError as e:
    print(e)
# listing lacks 4 of 4 indexed documents; refusing to delete them

A quarter is a demo value for five documents. In production use a few percent, alert when the guard trips, and require an explicit override for a planned mass delete (a retired wiki space, a migration). Also make the connector raise when listing fails rather than return an empty list: "I could not list" and "the source is empty" must never look the same.

⚠️ An empty filter deletes everything. In Qdrant, a delete whose filter has no conditions matches every point. On the 1.19.1 server, client.delete(COLL, points_selector=models.Filter(must=[])) and models.Filter(should=[]) both emptied a test collection. A helper that builds its conditions from a list of ids that happens to be empty is one line away from wiping the index. Guard against empty lists, or use MatchAny(any=ids), which matched nothing when ids was empty. The same rule protects access control: an ACL filter builder that can produce an empty filter fails open (Metadata & Filtering).

The index stage, in one screen

The rest of the index stage belongs to the database lessons. Five facts matter to a pipeline, and they agree with those lessons:

• Create payload indexes before the first load for every field you filter on (doc_id, acl, tenant, dates). Qdrant does not post-filter, but it adds filter-aware graph links only for fields indexed when the graph is built. Metadata & Filtering measured filtered ANN recall (overlap with an exact filtered search) of 1.00 with the index created first and 0.06 to 0.16 with it missing or created later, for a filter matching 10% of points.
• Wait for the writes the next step reads. client.upsert waits by default; upload_points and the REST API do not. The verify stage must count what is really there.
• Bulk loads leave work behind. After a large load the optimizer is still building graphs. Before sending traffic, check that GET /collections/{name}/optimizations shows nothing queued and that indexed_vectors_count is close to points_count. Close, not equal: both counts are approximate, and segments smaller than the indexing threshold stay unindexed on purpose and are searched exactly. Advanced Qdrant has the details.
• A delete is a mark first. Deleted points leave the results at once, but their bytes stay in the segment until it is rewritten, and every snapshot taken until then copies them. On Qdrant 1.19.1 the original write can outlive even the rewrite in the write-ahead log, which snapshots include; a rebuild into a fresh collection removes both (Vector Database Architecture, Data Freshness & Lifecycle).
• Low ANN recall is a query-time knob before it is a rebuild. Raise hnsw_ef first; a small m makes a recall target costlier to reach, not unreachable (ANN Algorithms).

Move 3: Make failure boring

Symptom: a 3 a.m. rate limit silently drops chunks, or one corrupt PDF crashes the job at 60% and the next night starts from zero.

Scripts fail as a whole. Pipelines fail per document, and they sort failures into two kinds before doing anything else:

• Transient: HTTP 429, 503, a timeout, a database that is restarting. The same call may succeed in a minute. Retry, with exponential backoff and random jitter so that a hundred workers do not retry in lockstep (Marc Brooker's Exponential Backoff And Jitter, AWS, 2015, compares the variants; the code below uses its "full jitter"). If the server sends Retry-After, wait at least that long.
• Permanent: a corrupt file, input over the model's limit (OpenAI's embeddings API rejects over-length input rather than truncating it), a 400. Retrying cannot help. Set it aside with its reason, and let the rest of the run continue.

Anything that still fails after its last retry is set aside too, with its last error, so an outage cannot stall the run. The place where set-aside items wait is a dead-letter queue (DLQ): a list of work items the pipeline gave up on, each with its error. A bad input waits there for a human or a fix; an item that only ran out of retries needs nothing but the dependency, so re-drive it automatically once the dependency recovers (Embedding Pipeline shows the same rule for embedding batches).

 change events / nightly listing
            │  one message per document: {op, doc_id}
            ▼
      ┌───────────┐     ┌──────────── workers (N) ────────────┐
      │   queue   │ ──► │ fetch current doc → ingest_document │ ──► index
      └───────────┘     │ transient error: back off, retry    │
            ▲           │ permanent, or too many attempts     │ ──► dead-letter queue
            │           └─────────────────────────────────────┘             │
            └─────────────────── re-drive after the fix ────────────────────┘

Here is that loop with one worker, wrapped around ingest_document and with failures injected on purpose:

import random, time

class Transient(Exception):
    """429, timeout, 503: the same call may succeed later."""

class Permanent(Exception):
    """Corrupt file, input over the model's limit: retrying cannot help."""

def with_retries(task, item, max_attempts=5, base=0.05, cap=2.0):
    for attempt in range(1, max_attempts + 1):
        try:
            return task(item)
        except Transient as e:
            if attempt == max_attempts:
                raise Permanent(f"gave up after {attempt} attempts ({e})")
            time.sleep(random.uniform(0, min(cap, base * 2 ** attempt)))   # full jitter

def worker(queue, task):
    done, dead_letters = [], []
    for item in queue:
        try:
            with_retries(task, item)
            done.append(item["doc_id"])
        except Permanent as e:
            dead_letters.append((item["doc_id"], str(e)))   # keep the item and the reason
    return done, dead_letters

FAULTS = {"handbook/travel.md":   [Transient("429 Too Many Requests")] * 2,
          "handbook/security.md": [Transient("503 Service Unavailable")] * 9,
          "handbook/expenses.md": [Permanent("400 input rejected")]}

def flaky_ingest(item):          # ingest_document with injected failures
    if FAULTS.get(item["doc_id"]):
        raise FAULTS[item["doc_id"]].pop(0)
    return ingest_document(item["doc_id"], item["doc"])

# The demo puts each document in its work item; a real worker fetches it (see below).
queue = [{"op": "upsert", "doc_id": d, "doc": doc} for d, doc in TUESDAY.items()]
done, dead = worker(queue, flaky_ingest)
print("done:", done)
for doc_id, reason in dead:
    print("dead letter:", doc_id, "->", reason)
print("points:", client.count(COLL, exact=True).count)

done: ['handbook/remote-work.md', 'handbook/travel.md']
dead letter: handbook/expenses.md -> 400 input rejected
dead letter: handbook/security.md -> gave up after 5 attempts (503 Service Unavailable)
points: 7

Three paths, all visible: the travel page survived two 429s on retry; the security page exhausted its attempts; the expenses page failed permanently on the first try and was not retried. Nothing was dropped silently, and the run finished. The demo's waits add up to at most 1.8 s: 0.1 + 0.2 on the travel page, 0.1 + 0.2 + 0.4 + 0.8 on the security page. Real values are larger (a base of a second, a cap of a minute), tuned to how long your provider's rate-limit windows and outages last.

And notice the last line: 7 points. Every document in this queue was already indexed, so running them again changed nothing. That is not luck, and it is what makes the next section safe.

At least once, effectively once

Most queues deliver at least once. A worker that writes to the index and then crashes before acknowledging the message gets the same message again after a timeout. Amazon SQS standard queues, for example, document at-least-once delivery, the occasional duplicate and the occasional out-of-order message (docs). So every work item will sometimes run twice, and the only defence is a write that does no harm the second time: at-least-once delivery plus an idempotent write gives you effectively once. Move 2 is what makes Move 3 safe.

Two more rules for workers:

• Events say what changed; fetch the current state. Two edits to one page can arrive out of order. A worker that indexes the text carried in the event can write the older version last. A worker that fetches the current document when it runs cannot. For the same reason, route all work for one doc_id to one worker at a time (a message group or partition key per document).
• Checkpoint after the write, never before. Mark a document done in the manifest only after the index acknowledged the write. The other order turns a crash into a document that is recorded as indexed and missing from the index.

Dead letters are a coverage gap, not a trash can

Every item in the DLQ is a document users cannot find. Alert when the DLQ is not empty, show the reasons grouped by error, and re-drive the items into the main queue once their cause is fixed or their dependency is back (safe, because the write is idempotent). Queues have this built in: SQS moves a message to its DLQ after maxReceiveCount receives (docs), and RabbitMQ dead-letters rejected or expired messages to a dead-letter exchange (docs). A parser that can crash the process (a malformed PDF can exhaust memory) belongs in a subprocess with a time and memory limit, so one poison document costs one work item, not the night.

When a cron job and a loop are no longer enough, orchestrators such as Apache Airflow, Dagster, Prefect or Temporal add schedules, retries, run history and backfills. They run your tasks; they do not make them idempotent. Concurrency per stage and batching under rate limits are covered in Embedding Pipeline.

When to run: schedule, events or stream

Pick the slowest trigger that meets a freshness requirement someone can actually state. Whatever the trigger, keep a periodic full listing and sync: webhooks are dropped when your endpoint is down, and only a full comparison of source and index finds what the events missed. Its interval, not the event path, bounds how stale a page can get when its event is lost, so a guarantee like "within one hour, even when an event is lost" needs a full listing whose interval plus run time stays under the hour, not a nightly one. Change detection (CDC, webhooks, polling) and freshness targets are the subject of Data Freshness & Lifecycle.

Move 4: Count everything, then decide

Symptom: the nightly job "succeeded" in 4 minutes instead of 40, or search quietly lost a folder, and nobody knows since when.

A pipeline that finishes without an exception has proved one thing: it did not crash. Whether it produced a good index is a separate question, and it needs numbers. Three kinds:

1. Per-stage counts, like with like. Documents listed → parsed → passed checks → written; chunks produced → embedded → indexed. Compare documents with documents and chunks with chunks. "10,000 documents in, 9,847 chunks out" says nothing: a document yields several chunks.
2. Quality checks per document, before anything is embedded.
3. A verdict per run: coverage against the source, and a few golden queries.

Here are the checks, plugged into sync through its check parameter. Thursday's listing adds three documents with typical problems:

LIMIT = model.max_seq_length       # 256 word pieces, [CLS] and [SEP] included

def check(doc: dict) -> list[str]:
    problems = []
    if not doc.get("acl"):
        problems.append("no ACL, and a missing ACL never means public")
    if not doc["text"].strip():
        problems.append("empty text: a scan without a text layer?")
    for i, chunk in enumerate(doc["text"].split("\n\n")):
        n = len(model.tokenizer(chunk)["input_ids"])
        if n > LIMIT:
            problems.append(f"chunk {i} has {n} word pieces; the model reads {LIMIT}")
    return problems

ROTA = "On-call rota 2026: " + "; ".join(
    f"week {w}: team {'ABCD'[w % 4]}, escalate to the duty manager" for w in range(1, 41))
THURSDAY = {**TUESDAY,
    "handbook/benefits-2024.pdf": {"acl": ["all-staff"], "text": ""},
    "handbook/payroll.md": {"text": "Payroll: salaries are paid on the 25th of each month."},
    "handbook/on-call.md": {"acl": ["engineering"], "text": ROTA}}

print(dict(sync(THURSDAY, check=check)))
for doc_id, problems in QUARANTINE.items():
    print("quarantined:", doc_id, problems)
searchable = set(index_state())
print(f"coverage: {len(searchable)} of {len(THURSDAY)} listed documents are searchable")

{'listed': 7, 'unchanged': 4, 'quarantined': 3, 'points': 7}
quarantined: handbook/benefits-2024.pdf ['empty text: a scan without a text layer?']
quarantined: handbook/payroll.md ['no ACL, and a missing ACL never means public']
quarantined: handbook/on-call.md ['chunk 0 has 569 word pieces; the model reads 256']
coverage: 4 of 7 listed documents are searchable

(The tokenizer also logs a warning that 569 is longer than 256: the same fact, in a log line nobody reads at 3 a.m.)

Each check catches a failure that no error message would:

• Empty text. A scanned PDF without a text layer parses to an empty string, embeds happily, and matches nothing. Its fix is OCR (Document Processing). Scans often carry a stamped header and footer, which defeats an emptiness check; Document Processing counts only the text that is not repeated on most pages.
• No ACL. Fail closed: a document whose permissions the connector could not read is not public. Keep it out and alert. (One exception to "keep the last good version": if a document that is already indexed loses its readable ACL, hide it with remove_document until the ACL is readable again, because the old permissions may be exactly what someone just revoked.)
• Over the model's limit. all-MiniLM-L6-v2 reads 256 word pieces, including the two special tokens, and silently ignores the rest. 569 word pieces means the last 313 of them would never reach the vector. The fix is a better split (Smart Chunking).

Other cheap checks: chunks that duplicate another document's chunks (one page crawled under two paths), text per page far below the source's average, and payload fields whose type changed (a date that was a string yesterday and an integer today breaks range filters silently, as Metadata & Filtering shows).

Coverage: the report the Not-indexed bucket asks for

The last line above is a coverage report: documents listed at the source against documents searchable in the index. Every gap must be explained by a quarantine entry or a dead letter. A gap nobody can explain is a bug: a connector that skips a folder, ids that do not match between the listing and the index (a trailing slash, a changed case), writes that were never acknowledged. Print the unexplained ids, not only their count, so someone can trace five of them.

Compare the other way too: indexed documents the source no longer lists should be 0 after every run. If not, the deletion step did not run or its guard tripped.

Golden queries: does it still answer?

Counts prove the documents arrived. They do not prove retrieval still works after a parser upgrade or a chunker change. For that, keep a handful of questions whose right document you know, and run them after every sync:

GOLDEN = {"How many days a week can I work from home?": "handbook/remote-work.md",
          "What is the hotel limit in Paris?": "handbook/travel.md",
          "How long do I have to hand in receipts?": "handbook/expenses.md"}

def golden_hit_rate(k: int = 3) -> float:
    hits = 0
    for question, want in GOLDEN.items():
        top = client.query_points(COLL, query=model.encode(question).tolist(), limit=k,
                                  with_payload=["doc_id"]).points
        hits += want in {p.payload["doc_id"] for p in top}
    return hits / len(GOLDEN)

print("golden hit rate@3:", golden_hit_rate())
# golden hit rate@3: 1.0

Hit rate@k is the share of questions with at least one relevant result in the top k. Three questions are a smoke test. A real gate uses 50 to 200 labeled questions from your logs, sliced by source, and compares with the previous run; Evaluation & Quality Metrics and Retrieval Metrics cover building that set and reading the numbers.

The run report, and what should wake someone up

A run that changes nothing is also a signal. The "4 minutes instead of 40" night often has perfectly normal listing counts and zero written documents, because the connector read a stale cache.

Your turn. Tonight's report: 48,210 documents listed, 47,913 searchable. 12 documents are quarantined and 9 are in the dead-letter queue; none of those 21 was indexed before. What do you do next?

Move 5: Rebuild or patch

Symptom: something changed that touches every chunk (a new embedding model, a new chunker, a new filter field), and two million old chunks have to follow.

The question is always the same: can old and new points share one vector field while the change rolls out?

The dangerous row is the model change with the same dimension. Nothing fails: every query returns results, some still look right and others quietly do not, which is exactly why nobody notices. Vector Embeddings measures that damage and gives the migration recipe, and Qdrant shows the named-vector variant inside one collection; at scale it is a pipeline job (Embedding Pipeline). The shape of a full rebuild:

1. Create handbook_v2 with the new model, payload indexes first.
2. Fill it from the stored chunks (stage 6 onwards; no re-parsing, no OCR).
3. Replay what changed during the build. A rebuild takes hours. Edits and deletes that arrive meanwhile must reach the new collection too, so run sync against it right before the switch; skip this and the switch resurrects every document deleted during the build. Dual-writing live changes to both collections while it builds also works, but only if the rebuild and the live writes for one doc_id go through the same per-document queue; otherwise the rebuild can overwrite a fresh edit with an older copy or bring back a deleted page (Embedding Pipeline).
4. Verify it: counts, coverage and golden queries against the old collection's numbers.
5. Switch the alias the application queries, in one call, and the query encoder with it: queries must be embedded with the new model from the moment the alias points at handbook_v2, and with the old one again after a rollback. A config that maps each alias to its model makes that one change; with a second named vector, using and the query encoder switch together in the same way. The ingestion writer moves at the same moment (new collection, new model). Keep the old collection for a rollback window and keep syncing it with the old model, or a rollback serves a stale index. (This only works if the application has queried an alias, not a collection name, from day one.)

client.update_collection_aliases(change_aliases_operations=[      # one atomic switch
    models.DeleteAliasOperation(delete_alias=models.DeleteAlias(alias_name="handbook_live")),
    models.CreateAliasOperation(create_alias=models.CreateAlias(
        collection_name="handbook_v2", alias_name="handbook_live"))])

The cost is a second full copy of the index while both exist, and the re-embedding bill (Move 6).

Backfills

A backfill runs existing data through a new or fixed pipeline step: a new source's five-year archive, a new payload field for old chunks, re-parsing the documents a parser bug damaged. Three rules:

• Same code path, lower priority. Push backfill work items through the same queue and workers as daily updates, rate-limited or on a separate low-priority queue, so a two-million-document backfill does not delay this morning's policy change by a day.
• Resumable and idempotent. A backfill will be interrupted. With idempotent writes you can simply run it again; a checkpoint (the last doc_id range done) saves the re-reading.
• Selectable. Store the parser and pipeline version in the payload, and "re-process everything the March parser touched" becomes a filter, not an archaeology project.

A payload backfill looks like one call. Here is the trap in it:

client.set_payload(COLL, payload={"department": "hr"}, wait=True,     # backfill: no re-embedding
                   points=models.Filter(must=[of_doc("handbook/remote-work.md")]))
print(sorted({p.payload.get("department") for p in client.scroll(COLL, limit=100)[0]
              if p.payload["doc_id"] == "handbook/remote-work.md"}))
ingest_document("handbook/remote-work.md", TUESDAY["handbook/remote-work.md"])   # next edit
print(sorted({p.payload.get("department") for p in client.scroll(COLL, limit=100)[0]
              if p.payload["doc_id"] == "handbook/remote-work.md"}, key=str))
# ['hr']
# [None]

The field arrived without touching a vector, and then the next edit erased it, because an upsert replaces the whole point, payload included. A backfill is two changes, in this order:

1. Teach the pipeline to produce the field on every write. If its value can change at the source (a page moves to another department), put it in the fingerprint too, or the change never reaches the index: the ACL story from Move 2 again.
2. Then fill it in for the points that already exist, together with the new fingerprint. A new input to the fingerprint changes every document's fingerprint. Backfill only the field, and the next sync finds every stored fingerprint stale and re-embeds the whole corpus, which is the bill the backfill was meant to avoid.

# Sketch, run after the new pipeline is deployed: fingerprint() now also hashes "department"
# and ingest_document() writes it; fingerprint_v1 is the old function; source is the listing.
for doc_id, doc in source.items():
    client.set_payload(COLL, wait=True,
        payload={"department": doc["department"], "fingerprint": fingerprint(doc)},
        points=models.Filter(must=[of_doc(doc_id), models.FieldCondition(
            key="fingerprint", match=models.MatchValue(value=fingerprint_v1(doc)))]))

On this lesson's four-document index, the next sync printed 'unchanged': 4 and embedded nothing. When the backfill wrote only department, it printed 'written': 4, 'chunks_embedded': 7: every chunk again. The match on the old fingerprint matters too. A document edited at the source since its last sync carries an older fingerprint, so the backfill leaves it alone and the next sync re-indexes it the normal way. Without that condition, the backfill stamps the stale version as current: in the test run, the hotel limit stayed at 150 euros after the source said 160.

Only a field that goes into the embedded text needs re-embedding.

Your turn. For each change, in place or a new collection? (a) Switch from all-MiniLM-L6-v2 to another 384-dimensional model. (b) Add a region filter field taken from the HR system. (c) Prefix every chunk with its page title before embedding. (d) Keep only the first 256 of a model's 1,024 dimensions to save memory.

Move 6: Price the corpus before you run it

Symptom: the embedding line on the bill is small, and the total is not.

Teams budget for embeddings because that is the step everyone talks about. It is rarely the expensive one. Do the arithmetic for a whole corpus before the first full run, with every assumption written down:

# Assumptions: round numbers for the exercise, not quotes. Replace every one with yours.
DOCS, PAGES_PER_DOC, TOKENS_PER_PAGE = 120_000, 14, 520   # measure a sample with your tokenizer
SCANNED_SHARE = 0.10                  # pages that need OCR
OCR_USD_PER_1K_PAGES = 1.50
CHUNK_TOKENS = 400                    # average chunk, fits an 8,192-token embedding model
HEADER_TOKENS = 60                    # LLM-written context header, embedded with its chunk
EMBED_USD_PER_M = 0.02                # text-embedding-3-small list price, September 2026
DIM, BYTES = 1536, 4                  # float32 vectors
LLM_IN_USD_PER_M, LLM_OUT_USD_PER_M = 1.00, 5.00   # a small LLM for context headers
CACHE_WRITE, CACHE_READ = 1.25, 0.10  # prompt-caching multipliers on the input price

pages = DOCS * PAGES_PER_DOC
tokens = pages * TOKENS_PER_PAGE
chunks = tokens // CHUNK_TOKENS
chunks_per_doc = chunks / DOCS
doc_tokens = PAGES_PER_DOC * TOKENS_PER_PAGE

ocr = pages * SCANNED_SHARE / 1000 * OCR_USD_PER_1K_PAGES
embedded = tokens + chunks * HEADER_TOKENS     # every chunk is embedded with its header
embed = embedded / 1e6 * EMBED_USD_PER_M
vectors_gb = chunks * DIM * BYTES / 1e9
graph_gb = chunks * 2 * 16 * 4 / 1e9            # HNSW base layer, m = 16
text_gb = chunks * (CHUNK_TOKENS + HEADER_TOKENS) * 4 / 1e9   # about 4 bytes per token

# One LLM call per chunk: the whole document + the chunk in, the header out.
per_call_in = doc_tokens + CHUNK_TOKENS + 100
out_tok = chunks * HEADER_TOKENS
plain_in = chunks * per_call_in
enrich_plain = plain_in / 1e6 * LLM_IN_USD_PER_M + out_tok / 1e6 * LLM_OUT_USD_PER_M
# With caching: each document is written to the cache once and read by its other chunks.
cached_in = (DOCS * doc_tokens * CACHE_WRITE + (chunks - DOCS) * doc_tokens * CACHE_READ
             + chunks * (CHUNK_TOKENS + 100))
enrich_cached = cached_in / 1e6 * LLM_IN_USD_PER_M + out_tok / 1e6 * LLM_OUT_USD_PER_M

print(f"pages {pages:,}  tokens {tokens/1e6:,.1f}M  chunks {chunks:,} ({chunks_per_doc:.1f} per doc)")
print(f"OCR                  USD {ocr:>9,.2f}")
print(f"embeddings           USD {embed:>9,.2f}  ({embedded/1e6:,.1f}M tokens, headers included)")
print(f"context headers      USD {enrich_plain:>9,.2f}  without caching")
print(f"context headers      USD {enrich_cached:>9,.2f}  with the document prefix cached")
print(f"storage: vectors {vectors_gb:.2f} GB, graph {graph_gb:.2f} GB, text + headers {text_gb:.2f} GB")

pages 1,680,000  tokens 873.6M  chunks 2,184,000 (18.2 per doc)
OCR                  USD    252.00
embeddings           USD     20.09  (1,004.6M tokens, headers included)
context headers      USD 17,646.72  without caching
context headers      USD  4,341.79  with the document prefix cached
storage: vectors 13.42 GB, graph 0.28 GB, text + headers 4.02 GB

What the numbers say:

• Embeddings are the cheap line: 1,004.6 million tokens (873.6 million of chunk text plus a 60-token header on each of 2,184,000 chunks, since the header is embedded with its chunk) at USD 0.02 per million (OpenAI's list price for text-embedding-3-small, September 2026) is USD 20.09. A self-hosted model trades that for GPU time: at an assumed 500 chunks per second, about 73 minutes.
• Context headers dominate. Writing a short context for each chunk (Context Augmentation covers what it buys) sends the whole 7,280-token document with every one of its 18.2 chunks. That is 17 billion input tokens: USD 17,646.72, nearly 900 times the embedding bill. Caching the document prefix brings it to USD 4,341.79, still over 200 times. Both OpenAI's Batch API and Anthropic's Message Batches API advertise 50% off asynchronous jobs (September 2026), which suits an offline pipeline that can wait up to a day.
• OCR is per page, so a mostly scanned archive changes the picture: at 100% scanned, the OCR line is USD 2,520.
• Storage: 2,184,000 × 1,536 × 4 bytes = 13.42 GB of float32 vectors, plus about 0.28 GB of HNSW links and 4.02 GB of chunk text with headers. Quantization shrinks the vector line; Vector Database Architecture does that arithmetic.

Time is a budget too. Suppose your account's embedding limit is 1,000,000 tokens per minute: 1,004.6 minutes, about 16.7 hours, before retries. A full rebuild that must finish over a weekend fits; a nightly one does not.

The fingerprint is a cost control. A job without the "unchanged" check re-sends everything every night: USD 4,341.79 in headers per night, about USD 130,000 per 30-day month. With it, if 0.5% of documents change each night, headers cost about USD 21.71 a night, USD 651 a month. Store the headers keyed by hashes of the document's text and the chunk's text (not by the fingerprint, or a permission change regenerates them), and a model migration re-pays only the USD 20.09 of embeddings. Track cost per 1,000 documents processed per run; a jump is the earliest sign of an unplanned re-embed.

Your turn. Same corpus and prices, with the cached headers. On average 2% of documents change every night. What do embeddings and headers cost per 30-day month?

Final round: no label on the problem

Real incidents do not say which move they need. Read the symptom, then pick.

Challenge 1: the retry that doubled

An ingestion Job crashes at 80% and the scheduler retries it automatically. The next morning, every FAQ answer from the first 80% of the corpus appears twice in search, with identical scores.

Challenge 2: the wiki that moved

The wiki moves from /wiki/<page> to /kb/<page>, and doc_id is the URL. The nightly sync stops with "listing lacks 38,112 of 38,112 indexed documents; refusing to delete them", after writing every page under its new URL. Search shows every page twice.

Challenge 3: forty minutes became four

The nightly run finishes in 4 minutes instead of the usual 40. Listing count normal, 0 quarantined, 0 dead letters, 0 written. The wiki's audit log shows 600 edits yesterday.

Challenge 4: the deleted document that came back

In June, a customer's documents are deleted under a data-deletion request. In September, a full rebuild for a model upgrade restores them to search.

Cheat sheet: symptom → move → cost

Before moving on, take a pipeline you own (or plan) and write down: where doc_id comes from and what happens to it when a document is renamed; what one re-run, one edit, one delete and one crash between two writes each do to the index; which failures are retried and where the others go; the counts your run report prints; and the cost of one full run and of one typical night.

How would you measure it on your own data? Build a fixture corpus of 20 to 50 real documents, including a scan, a huge table, a document without an ACL and one that shrinks when edited. In CI, run the pipeline twice and assert that the second run writes nothing; edit one fixture and assert that only its points changed and its point count equals its chunk count; delete one and assert that it returns nothing; empty the listing and assert that the run refuses to delete. In production, gate every run on coverage (unexplained gap of 0), dead letters and golden hit rate@k against the previous run, and chart cost per 1,000 documents.

Next: Document Processing for parsing and cleaning, Smart Chunking for where to cut, Embedding Pipeline for throughput, rate limits and re-embedding at scale, and Data Freshness & Lifecycle for change detection, deletes and keeping the index current over months.