Data pipelines

A pipeline is a job that moves data from where it is produced to where it is used, on a schedule, without anyone watching. The hard part is not moving it once but surviving everything that goes wrong afterwards: the run that fails halfway, the run that fires twice, the upstream column that quietly changes type. This section is about building one that can be re-run safely, which is the property that separates a pipeline from a script.

What went wrong on Monday                    What the pipeline needed
the job died after writing half the rows     replace a partition, never append
a bug was found in last quarter's numbers    a date parameter, not "yesterday"
upstream added a column                      a schema check that tolerates additions
upstream turned an integer into a string     a schema check that fails on types
the export ran but produced 40 rows          a volume check against recent history
an event arrived three days late             a trailing reprocess window

Pipelines fail partially#

The defining problem is that a pipeline stops halfway. Some data was written, some was not, and nothing recorded where it got to. Nothing in the target says which rows are from the failed run, because rows do not carry that. Every design decision follows from asking what happens on the second run.

Make reruns safe#

An idempotent pipeline produces the same result whether it runs once or five times. That is a property of how you write, not a claim you make in a docstring, so here it is in code. The target is a list of rows tagged with the day they belong to, and the difference between the two loaders is one line.

def load_append(table, day, rows):
    return table + [(day, r) for r in rows]

def load_replace(table, day, rows):
    kept = [row for row in table if row[0] != day]    # drop the day first
    return kept + [(day, r) for r in rows]

rows = ["a", "b", "c"]

t = []
t = load_append(t, "2026-03-01", rows)
t = load_append(t, "2026-03-01", rows)               # the retry after a timeout
print(len(t))                                        # 6: the day is counted twice

t = []
t = load_replace(t, "2026-03-01", rows)
t = load_replace(t, "2026-03-01", rows)              # the same retry
print(len(t))                                        # 3: identical to one run

That is only half of it. Delete-then-insert as two visible steps leaves a window in which the day exists but is empty, and a crash inside that window is worse than the duplicate it was meant to prevent: nothing errors, and a reader sees a real zero. Build the new state first, then publish it with one operation that either happens or does not.

def build(table, day, rows, fail_after=None):
    """Compute the whole new state. Publishing is the caller's single assignment."""
    staged = [row for row in table if row[0] != day]
    for i, r in enumerate(rows):
        if fail_after is not None and i == fail_after:
            raise RuntimeError("source timed out")
        staged.append((day, r))
    return staged

table = build([], "2026-03-01", ["a", "b", "c"])
try:
    table = build(table, "2026-03-02", ["d", "e", "f"], fail_after=1)
except RuntimeError:
    pass                                   # the assignment never ran

print(len(table))                          # 3: day two published nothing at all
print(sorted({d for d, _ in table}))       # ['2026-03-01']

Backfills#

At some point you will reprocess two years of history, usually because a transform was wrong. The loop over dates is the easy part. What actually bites is everything around it.

• The job must take a date. A pipeline that reads the clock cannot be backfilled at all, and rewriting it to take one is the first day of the work.
• Direction matters. Run forwards if a day's output depends on the day before it; run backwards from today if the recent months are the ones people are looking at.
• Throughput. Seven hundred days launched at once will saturate the source database and page whoever owns it. Cap the concurrency and run it in chunks.
• Cost. Seven hundred times the daily cost, paid in an afternoon rather than over two years. Work it out before you start, not from the invoice.
• The table is live while you rewrite it. For hours, readers see some days on the old logic and some on the new. Either say so, or backfill into a copy and swap at the end.
• It will be interrupted. Keep a ledger of finished days so the resume costs minutes rather than starting again.

import datetime as dt

def daterange(start, end):
    day = start
    while day <= end:
        yield day
        day += dt.timedelta(days=1)

def backfill(run, start, end, ledger, limit=None):
    """Resumable. Takes the same run(day) the scheduler calls every morning."""
    done = set(ledger.read())
    processed = 0
    for day in daterange(start, end):
        if day.isoformat() in done:
            continue
        run(day)                              # publishes the day, atomically
        ledger.mark(day.isoformat())          # only now is it done
        processed += 1
        if limit is not None and processed >= limit:
            break                             # one chunk per invocation
    return processed

class MemoryLedger:
    def __init__(self): self.done = []
    def read(self): return list(self.done)
    def mark(self, key): self.done.append(key)

seen, ledger = [], MemoryLedger()
backfill(seen.append, dt.date(2026, 3, 1), dt.date(2026, 3, 10), ledger, limit=4)
backfill(seen.append, dt.date(2026, 3, 1), dt.date(2026, 3, 10), ledger)
print(len(seen), len(set(seen)))              # 10 10: nothing ran twice

Schemas change#

Three things can happen to an upstream schema and only one of them is safe. A column is added, which nothing downstream reads and nothing breaks. A column is removed or renamed, which breaks every consumer that reads it, loudly and immediately. Or a column keeps its name and changes type, which breaks nothing at load time and quietly corrupts everything computed from it. The third is the one worth writing code against.

EXPECTED = {"order_id": "int", "user_id": "int", "total": "float", "country": "str"}

def kind_of(value):
    return {bool: "bool", int: "int", float: "float", str: "str"}.get(type(value), "other")

def diff_schema(expected, row):
    observed = {k: kind_of(v) for k, v in row.items() if v is not None}
    added = [k for k in observed if k not in expected]
    missing = [k for k in expected if k not in row]
    changed = [
        (k, expected[k], observed[k])
        for k in observed
        if k in expected and observed[k] != expected[k]
        and not (expected[k] == "float" and observed[k] == "int")   # widening is safe
    ]
    return added, missing, changed

row = {"order_id": 7, "user_id": 31, "total": "19.99", "country": "DE", "channel": "app"}
added, missing, changed = diff_schema(EXPECTED, row)
print(added)      # ['channel']                   log it, carry on
print(missing)    # []                            a removal would be here
print(changed)    # [('total', 'float', 'str')]   stop the run

The policy that follows is dull and works. Additions are allowed without ceremony. Type changes stop the run. A rename is not one change, it is three separated in time, and doing all three in one release is an outage.

week 0   producer writes both country and country_code     nothing breaks
week 1   consumers move to country_code, one at a time      country still written
week 3   producer stops writing country                     nothing reads it
         schema version bumped, old version deleted

Quality checks in the pipeline#

Two kinds, and most pipelines only have the first. Absolute checks say what must always be true of a batch. They catch broken data.

def check(df, expected_rows):
    assert len(df) > 0, "empty batch"
    assert df.id.is_unique, "duplicate ids"
    assert df.total.notna().all(), "missing totals"
    assert (df.total >= 0).all(), "negative totals"
    # Volume check catches upstream failures that produce valid but tiny data
    assert len(df) > 0.5 * expected_rows, f"only {len(df)} rows, expected {expected_rows}"

Relative checks compare today against what the last few runs looked like. They catch data that is not broken, just wrong: a source that silently started filtering, a partner that changed a currency, a region that stopped reporting. Compare against the median of recent runs rather than the mean, or one bad day raises the bar for the whole week after it.

def check_relative(today, history, tolerance=0.5):
    """history: row counts from the last successful runs, most recent last."""
    if len(history) < 7:
        return                                   # not enough to judge yet
    typical = sorted(history)[len(history) // 2]
    if today < typical * tolerance or today > typical / tolerance:
        raise ValueError(f"{today} rows against a typical {typical}")

recent = [4000, 4200, 3900, 4100, 4050, 4300, 3950]
check_relative(4100, recent)                     # fine
try:
    check_relative(40, recent)                   # the export that finished early
except ValueError as e:
    print(e)                                     # 40 rows against a typical 4050

Batch or streaming#

Batch accumulates data and processes a period of it. Streaming processes each event as it arrives. The difference people quote is latency, but the difference that decides the design is what you can redo. A batch job's input for a period is a fixed set of files that still exist, so a rerun is exact. A streaming job's input is a position in a log, redoing means replaying from an offset, and everything the job already emitted has already been acted on downstream.

                     Batch                        Streaming
unit of work         a period: an hour, a day     one event
rerun                reprocess the period         rewind to an offset, replay
correct on retry     free, if you replace         only if every write is idempotent
late data            lands in a later run         needs a watermark and a policy
cost when idle       nothing                      the cluster runs all night
debugging            rerun it locally on files    reproduce a moment in a stream
people to operate    one                          a rota

So the question is whether freshness changes a decision. Fraud, alerting and live personalisation, yes. Reports, training data and nearly all analytics, no. Ask whether anyone would act differently if this data were an hour old, and if the answer is no, run it hourly. A five-minute batch gets most of the benefit of streaming while keeping batch's recovery story, which is why micro-batch is where a lot of supposedly streaming systems end up.

A pipeline that runs once is a script. Making it run every day, in the right order, and tell you exactly which step failed is a different job. This section covers choosing between batch and streaming honestly, since streaming sounds better and is considerably harder to operate; what an orchestrator gives you beyond a cron entry; what to do with events that arrive late or out of order; and lineage, which is being able to answer where a number came from when someone disputes it.

Batch or streaming#

Batch processes accumulated data on a schedule. Streaming processes each event as it arrives. Streaming sounds better and is much harder to operate. Choose it only when freshness genuinely matters to the decision the data feeds.

Batch, hourly or daily       reports, training data, most analytics
Streaming, seconds           fraud, alerting, live personalisation

Ask: would anyone act differently if this data were an hour old?
If not, batch.

The part that is easy to underestimate is the operating cost. A batch job that fails is a red square and a rerun in the morning. A streaming job that fails is consumer lag climbing while someone works out whether it is safe to restart from the last committed offset, and lag that grows faster than the job can catch up is an outage with no natural end. Exactly-once, where a framework offers it, holds inside that framework's own state and its own sinks. The moment the job calls an external API or writes to a database it does not manage, you are back to at-least-once and the write has to be idempotent yourself.

Orchestration#

Pipelines have dependencies: this table needs that one to finish first. An orchestrator runs steps in dependency order, retries failures, and records what ran. The value is less in running things and more in knowing, after a failure, exactly which steps completed. The core of one is smaller than the products suggest, and writing it once makes clear what you are actually buying.

def run_dag(tasks, deps, attempts=2):
    """tasks: name -> callable. deps: name -> names that must finish first."""
    state = {name: "pending" for name in tasks}

    def ready(name):
        return all(state[d] == "ok" for d in deps.get(name, ()))

    progressed = True
    while progressed:
        progressed = False
        for name, fn in tasks.items():
            if state[name] != "pending" or not ready(name):
                continue
            for attempt in range(attempts):
                try:
                    fn()
                    state[name] = "ok"
                    break
                except Exception as e:
                    state[name] = f"failed: {e}"
            progressed = True

    # Anything still pending was blocked by an upstream failure, not broken itself.
    for name in state:
        if state[name] == "pending":
            state[name] = "skipped: upstream failed"
    return state

calls, tried = [], {"clean": 0}

def extract():
    calls.append("extract")

def clean():
    tried["clean"] += 1
    if tried["clean"] == 1:
        raise RuntimeError("bad row")      # fails the first time, then works
    calls.append("clean")

def report():
    calls.append("report")

tasks = {"extract": extract, "clean": clean, "report": report}
deps = {"clean": ["extract"], "report": ["clean"]}
print(run_dag(tasks, deps))
# {'extract': 'ok', 'clean': 'ok', 'report': 'ok'}  -- clean failed once and retried
print(calls)                               # ['extract', 'clean', 'report']

What a real orchestrator adds on top is a scheduler that knows about dates, parallel execution with a concurrency cap, a record of every past run, and a page that shows which day of which task is red. Cron gives you none of that and is fine for one independent job. The moment a second job reads the first one's output, you need the dependency edge, because otherwise the second job runs on yesterday's data whenever the first one is late and nothing anywhere says so.

Late and out-of-order data#

Events arrive late, out of order, or twice. A phone was in a tunnel; a partner batches their exports; a retry duplicated a send. A pipeline that assumes ordered arrival produces aggregates that were right when computed and are wrong an hour later, which is the worst combination because nothing errors and the number moves after someone quoted it.

There are two clocks and keeping them apart is most of the battle. Event time is when the thing happened. Processing time is when your job saw it. Aggregate by event time or the Monday figure includes whatever straggled in on Tuesday. A watermark is your declared position in event time: the point past which you promise not to accept anything more.

from collections import defaultdict

def aggregate(events, allowed_lateness=2):
    """events: (event_time, processing_order) pairs, in arrival order.
    Counts per event-time bucket, with a watermark that closes old buckets."""
    counts, watermark, dropped = defaultdict(int), None, []
    for event_time, _ in events:
        watermark = event_time if watermark is None else max(watermark, event_time)
        if event_time < watermark - allowed_lateness:
            dropped.append(event_time)          # too late: count it separately
            continue
        counts[event_time] += 1
    return dict(counts), dropped

arrivals = [(10, 0), (11, 1), (9, 2), (12, 3), (8, 4), (12, 5)]
print(aggregate(arrivals))
# ({10: 1, 11: 1, 9: 1, 12: 2}, [8])  -- the event from time 8 arrived after the
# watermark reached 12, so it is dropped rather than silently changing a closed bucket

Dropping is one policy of three and you must choose explicitly. Drop, and the bucket is stable but slightly under-counts. Accept and restate, and the bucket is accurate but a number in a dashboard changes after someone screenshotted it. Send late events to a side output and correct in a nightly batch, which is what most systems that care about both end up doing. Say which you chose in the documentation the analysts read, because they will find the discrepancy before you do.

Lineage#

Lineage records which inputs produced which outputs. When a number in a dashboard is wrong, lineage lets you trace it to the source rather than guessing. For ML it answers the question every audit asks: exactly which data was this model trained on. It is cheap to record at write time and impossible to reconstruct afterwards, which is why it is always missing when it is first needed.

import hashlib, json, datetime as dt

def manifest(output_path, inputs, code_version, row_count):
    """Write this next to every output. It is the whole of lineage at one step."""
    return {
        "output": output_path,
        "written_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "code_version": code_version,          # git sha of the transform
        "inputs": [
            {"path": p, "rows": n, "content_hash": h} for p, n, h in inputs
        ],
        "rows": row_count,
    }

def digest(rows):
    return hashlib.sha256(json.dumps(rows, sort_keys=True).encode()).hexdigest()[:12]

raw = [{"id": 1, "total": 10}, {"id": 2, "total": 5}]
print(manifest(
    "warehouse/daily/date=2026-03-01",
    [("raw/orders/date=2026-03-01", len(raw), digest(raw))],
    code_version="a91f0c3",
    row_count=len(raw),
))

A scheduled job that loads yesterday's data will fail one morning, and someone will run it again. This section is about writing it so the rerun is harmless: partition the output by date, compute into a temporary location, check the result, and replace a whole day at once instead of appending. Get that property and every failure has the same fix, which is to run the day again, and backfilling a bug fix across three months is the same code in a loop rather than a special path.

The shape#

extract day D from source  ->  transform  ->  check  ->  replace partition D in the target
                                                     |
                                                     fail: publish nothing, alert

In code#

import duckdb, datetime as dt, pathlib, shutil

def run(day: dt.date):
    src = f"raw/calls/date={day}/*.json"
    out = pathlib.Path(f"warehouse/daily_usage/date={day}")
    tmp = out.with_name(out.name + ".tmp")

    con = duckdb.connect()
    df = con.sql(f"""
        SELECT user_id, model, sum(input_tokens) AS input_tokens,
               sum(output_tokens) AS output_tokens, count(*) AS calls
        FROM read_json_auto('{src}')
        WHERE status = 'ok'
        GROUP BY ALL
    """).df()

    check(df, day)                                  # raises on bad data
    tmp.mkdir(parents=True, exist_ok=True)
    df.to_parquet(tmp / "part.parquet")
    if out.exists():
        shutil.rmtree(out)
    tmp.rename(out)                                 # publish: replace the whole day

def check(df, day):
    assert len(df) > 0, f"{day}: no rows"
    assert (df["input_tokens"] >= 0).all(), f"{day}: negative tokens"
    assert df["user_id"].notna().all(), f"{day}: missing user ids"

Because each run replaces exactly one day's partition, running it twice gives the same result as running it once. That property, idempotency, is what makes failures boring: rerun the day. Three details in that code are doing the work and are worth naming, because leaving any one of them out gives you a pipeline that looks idempotent and is not.

• The check runs on the staged result, before anything is published. Check after publishing and you have already corrupted the day you are about to complain about.
• The write goes to a temporary directory with a different name, so a half-written file is never at the path readers use.
• The last step is a rename, which the filesystem performs as one operation. Writing directly into the output directory and deleting the extra files afterwards is two operations, and a crash between them leaves a day that is part old and part new.

That third point has a limit worth knowing. The rmtree then rename above is itself two operations, so a crash in the gap leaves the day missing entirely rather than mixed. Missing is recoverable and mixed is not, which is why it is written in that order, but if you need the swap to be genuinely atomic then write to a new versioned directory and move a pointer, which is what table formats such as Iceberg and Delta do for you.

Proving it, rather than believing it#

Idempotency is a claim about behaviour under failure, so the test has to inject a failure. This runs without a warehouse: a dict stands in for the target, and the load is killed at a chosen row.

def load(target, day, rows, die_at=None):
    """Stage the whole day, then publish with one assignment."""
    staged = []
    for i, row in enumerate(rows):
        if die_at is not None and i == die_at:
            raise RuntimeError("killed mid-run")
        staged.append(row)
    target[day] = staged                       # the publish
    return target

target, rows = {}, [1, 2, 3, 4]

for die_at in (None, 0, 2, 3):                 # clean run, then three crash points
    try:
        load(target, "2026-03-01", rows, die_at=die_at)
    except RuntimeError:
        pass

print(target)                                  # {'2026-03-01': [1, 2, 3, 4]}
print(sum(target["2026-03-01"]))               # 10, after four runs and three crashes

When the target is a table, not files#

Same property, different mechanism. Delete the day's rows and insert the new ones inside one transaction, so no reader ever sees the gap and a failure rolls the whole thing back.

BEGIN;
DELETE FROM daily_usage WHERE usage_date = :day;
INSERT INTO daily_usage
SELECT :day, user_id, model, sum(input_tokens), sum(output_tokens), count(*)
FROM raw_calls
WHERE call_date = :day AND status = 'ok'
GROUP BY user_id, model;
COMMIT;

The alternative is a merge on a natural key, which upserts row by row rather than replacing a period. It is the right shape when rows update individually, for instance a customer record that changes. It is the wrong shape for a daily aggregate, because a row that disappears from the source is never deleted by a merge: it simply keeps its old value forever, and nobody notices until someone reconciles the totals.

Data that keeps arriving#

Most sources are not finished with a day when the day ends. Payments settle, a partner exports late, a mobile client syncs when it next has signal. If the job only ever runs yesterday, those records land in a partition that was already published and are never counted.

def daily(today: dt.date, lookback: int = 7):
    """Reprocess a trailing window, not one day. Safe only because run() replaces."""
    for offset in range(1, lookback + 1):
        run(today - dt.timedelta(days=offset))

This is the concrete payoff of idempotency. Without it, reprocessing a week every night would multiply a week of data by seven; with it, the trailing window is free of consequences and the only question is compute. It is also why the cost of the property is worth paying up front, before you know you need the window.

Backfills#

def backfill(start: dt.date, end: dt.date):
    day = start
    while day <= end:
        run(day)
        day += dt.timedelta(days=1)

That loop is correct and will not survive contact with two years of history. Add three things before you start one: a ledger of days already finished so an interruption resumes, a concurrency cap so the source database is not saturated, and a number for what it will cost, obtained by timing one day and multiplying. The last is the one people skip, and the invoice arrives a month later attached to nobody's name.

Scheduling#

An orchestrator such as Airflow, Dagster or Prefect runs run(yesterday) every morning, retries on failure, runs backfills in parallel, and shows which days succeeded. Cron works for one job; an orchestrator earns its keep when jobs depend on each other.

Two settings decide whether it wakes you at three in the morning. Retries should be bounded and spaced, because the usual cause of a failure is an upstream system that is down and will not be up in ten seconds. And a job needs an alert on not having run at all, not only on having failed: a scheduler that stopped firing produces no failures, no red squares, and a table that quietly stops growing.

A vector index is a copy of your documents, and copies go stale. This section is the ingestion job that keeps one honest as the source changes: working out which documents actually changed, replacing their chunks, removing the ones that have been deleted, and keeping permissions attached to every chunk. Deletion is the half that is usually missing, and it is the half that has an assistant confidently citing a policy withdrawn last month to someone who was never allowed to read it.

Only process what changed#

import hashlib

def sync(source_docs, index):
    seen = set()
    for doc in source_docs:                       # id, text, updated_at, acl
        seen.add(doc.id)
        digest = hashlib.sha256(doc.text.encode()).hexdigest()
        if index.stored_hash(doc.id) == digest:
            continue                              # unchanged: skip parsing and embedding
        chunks = chunk(parse(doc))
        index.replace_document(doc.id, embed(chunks), digest=digest, acl=doc.acl)

    for stale_id in index.document_ids() - seen:  # gone from the source
        index.delete_document(stale_id)

The hash is what makes this affordable. Embedding is the expensive step, and on a corpus of any size almost nothing changes between runs, so skipping unchanged documents takes a nightly sync from hours to minutes. Hash the content you actually embed, after parsing and normalising, not the raw bytes: a PDF re-exported with a new timestamp has different bytes and identical text, and hashing the bytes re-embeds the whole corpus every time someone touches the export job.

import hashlib

def fingerprint(text, acl, chunk_size, model):
    """Everything that, if changed, means the stored chunks are wrong."""
    parts = [text, ",".join(sorted(acl)), str(chunk_size), model]
    return hashlib.sha256("|".join(parts).encode()).hexdigest()

doc = dict(text="Refunds are issued within 14 days.", acl=["support", "billing"])
a = fingerprint(doc["text"], doc["acl"], 512, "embed-v3")
b = fingerprint(doc["text"], ["billing", "support"], 512, "embed-v3")
c = fingerprint(doc["text"], ["billing"], 512, "embed-v3")
d = fingerprint(doc["text"], doc["acl"], 256, "embed-v3")
print(a == b, a == c, a == d)      # True False False

Replacing a document means deleting its chunks first#

A document that shrinks from twelve chunks to eight leaves four orphans behind if you only overwrite by chunk id. They are still in the index, still retrievable, and they contain text that was removed from the source, which is the same incident as failing to delete a document, arrived at more subtly. Delete every chunk belonging to the document, then insert the new set.

def replace_document(index, doc_id, chunks, digest, acl):
    index.delete_where(doc_id=doc_id)              # all of them, by document
    for i, chunk in enumerate(chunks):
        index.insert(
            chunk_id=f"{doc_id}:{i}",
            vector=embed(chunk),
            text=chunk,
            doc_id=doc_id,                          # the key deletion needs
            digest=digest,
            acl=acl,
        )

Deletions#

If a document is deleted or unpublished at the source but its chunks stay in the index, the assistant keeps quoting it. For a withdrawn policy or a document a user lost access to, that is a real incident. Every sync must compare the full set of source ids with the index and remove the difference.

The trap in that set difference is a partial source listing. If the crawl fails halfway and returns two hundred of two thousand documents, the difference is eighteen hundred documents and the sync cheerfully empties the index. Guard it: require the listing to have completed, and refuse to delete more than a small share of the index in one run without a human saying so.

def safe_delete(index_ids, source_ids, listing_complete, max_share=0.05):
    if not listing_complete:
        raise RuntimeError("source listing incomplete, refusing to delete")
    missing = index_ids - source_ids
    if len(missing) > max_share * len(index_ids):
        raise RuntimeError(
            f"{len(missing)} of {len(index_ids)} documents missing, needs review"
        )
    return missing

print(safe_delete({"a", "b", "c", "d"}, {"a", "b", "c"}, True, max_share=0.5))  # {'d'}
try:
    safe_delete({"a", "b", "c", "d"}, {"a"}, True, max_share=0.5)
except RuntimeError as e:
    print(e)                          # 3 of 4 documents missing, needs review

When parsing fails#

Some documents will not parse cleanly: scanned pages, broken encodings, huge tables. Do not index garbage. Send them to a quarantine list with the error, alert when it grows, and review it regularly. The reason to alert on the size rather than on each failure is that the interesting signal is a change: three files that never parse are a known annoyance, and three hundred means the source system changed its export format last night.

def sync_one(doc, index, quarantine):
    try:
        chunks = chunk(parse(doc))
    except Exception as e:
        quarantine[doc.id] = f"{type(e).__name__}: {e}"
        return "quarantined"
    if not chunks or sum(len(c) for c in chunks) < 40:
        quarantine[doc.id] = "parsed to nothing, probably a scan"
        return "quarantined"
    replace_document(index, doc.id, chunks, doc.digest, doc.acl)
    return "indexed"

Permissions travel with the chunk#

Store who may see each document on every chunk, and filter on it at query time. When permissions change at the source, the sync must update them even if the text did not change, which is why the permission list is inside the fingerprint above rather than beside it.

Filter before the search, not after. Retrieving the top twenty and then discarding the ones this user may not see returns four results for one person and twenty for another, and on a document set where most of it is restricted it returns nothing at all while the index plainly contains the answer. Every serious vector store takes a metadata filter that is applied during the search; use it, and test it with an account that should see almost nothing.

New embedding model#

Changing the embedding model means re-embedding everything into a new index, comparing retrieval quality, and switching over, as with any migration. Store the model name with each vector so a mixed index can never happen silently. A mixed index does not error; it returns nearest neighbours in a space where half the points were placed by a different function, and the results are plausible and wrong.

1  build a second index with the new model, from the same source
2  replay a fixed set of real queries against both
3  compare recall at 10 on the same judgements, and eyeball the disagreements
4  switch reads to the new index, keep the old one for a week
5  delete the old index, once nobody has asked for a rollback

Cost it before you agree to it. A million chunks at roughly five hundred tokens each is five hundred million tokens through an embedding model, plus the storage for a second full index while both exist, plus whatever the re-crawl costs at the source. None of that is prohibitive and all of it is a surprise if the first estimate appears after the migration has started.