Sync Vector Indexes Incrementally from Change Data Capture

Tier: Advanced · Workflow area: RAG data pipelines

Detect real content changes with hashes, re-chunk and re-embed only what changed, propagate deletes, and cut over versioned indexes for breaking changes. This playbook builds that sync loop and the reconciliation checks that prove the index tracks the source of truth.

When to reach for this

• •
  Users cite stale answers ("that policy changed last month") from an index rebuilt on a slow cadence.
• •
  Full re-embeds are burning meaningful money or exceeding the maintenance window as the corpus grows.
• •
  Deleted or superseded documents still surface in retrieval — a correctness and sometimes compliance failure.
• •
  You are wiring the vector store into an existing Debezium/warehouse-stream/orchestrator estate as a new sink.

The workflow

1. 1.

   Establish the source-of-truth change feed. Options in order of preference: an existing CDC stream (Debezium topics, warehouse streams/change tables), source-system webhooks, or — the universal fallback — a periodic crawl that diffs against stored state. Whatever the feed, normalize events to (doc_id, op ∈ {upsert, delete}, payload_ref, event_ts).

2. 2.

   Detect content change with hashes, not timestamps. Files get re-saved, re-exported, and touched without changing meaning; updated_at alone triggers useless re-embeds. Diff a normalized content hash (same normalization as pb-27 dedup) against the pipeline's chunk table (pb-30/pb-31 schema already stores content_hash):

   Copy

   import duckdb, hashlib
   
   def content_hash(text: str) -> str:
       return hashlib.sha256(" ".join(text.split()).lower().encode()).hexdigest()
   
   con = duckdb.connect("pipeline.db")
   
   def classify(doc_id: str, new_text: str) -> str:
       row = con.execute(
           "SELECT DISTINCT content_hash FROM chunks WHERE doc_id = ?", [doc_id]).fetchall()
       if not row:
           return "new"
       return "unchanged" if content_hash(new_text) in {r[0] for r in row} else "changed"
   

   Refine to chunk-level diffing for long documents: re-chunk the new version, hash each chunk, and re-embed only chunks whose hash is absent — on large documents with small edits this cuts re-embed volume by 90%+.

3. 3.

   Make upserts a transaction of three steps. For a changed document: (a) re-chunk and re-embed new/changed chunks via the pb-30 idempotent pipeline, (b) upsert vectors keyed by deterministic chunk_id, (c) delete vectors whose chunk_id belonged to the old version but not the new (WHERE doc_id = ? AND chunk_id NOT IN (...)). Skipping (c) is how indexes accumulate orphan chunks that retrieve stale text forever.

4. 4.

   Propagate deletes as first-class events. A source delete must remove all chunks for the doc_id in the vector store and mark the chunk table. Deletes are the events most often dropped by ad-hoc sync scripts — and the ones with compliance teeth (right-to-erasure). Dead-letter failed deletes loudly; a failed upsert is staleness, a failed delete is exposure.

5. 5.

   Reserve versioned cutover for breaking changes. Model swaps (pb-29), chunking changes (pb-28), enrichment rollout (pb-35): build a complete shadow collection (rag_chunks_v2), run the golden-set eval against it, then flip the retrieval function's target atomically and keep v1 for rollback. Never mix representations in one collection — incremental sync is for content changes, cutover is for representation changes.

6. 6.

   Reconcile continuously — this is the metric layer. The measures that prove the sync works:

   • •
     Freshness lag: now() - max(event_ts applied), per source; alert at your SLA (e.g. 15 min for streams, 24 h for crawls).
   • •
     Count reconciliation: live source doc count vs distinct doc_id in the index — drift beyond a small tolerance pages someone.
   • •
     Deletion probe: a scheduled test deletes a canary doc and asserts it stops retrieving within the SLA.

   Copy

   SELECT (SELECT count(DISTINCT doc_id) FROM source_docs WHERE NOT is_deleted)
        - (SELECT count(DISTINCT doc_id) FROM chunks) AS doc_drift;
   

   Expected output: doc_drift = 0 in steady state; any persistent nonzero value is a dropped-event bug, not noise.

Verify before you trust it

• End-to-end trace testedit one source document, watch it re-embed, and confirm the new text retrieves (and the old text does not) within the freshness SLA — measured, with timestamps.
• Delete trace testremove a document at the source; all its chunks are gone from the vector store and chunk table within SLA; zero results for its distinctive content.
• Chunk-level diffing verifieda one-paragraph edit to a 100-chunk document re-embeds only the affected chunks (check embedded_at timestamps), not all 100.
• Replay safetyre-deliver a day of CDC events; counts and vectors end identical (idempotency under at-least-once delivery, which every CDC system eventually exercises).

Pitfalls

• Timestamp-triggered re-embedding. Bulk migrations and re-exports touch every updated_at at once — without content hashing, one upstream housekeeping job triggers a full-corpus re-embed bill.
• Upsert-only sync. Handling creates and updates but not deletes yields an index that only ever grows and increasingly retrieves documents that no longer exist.
• Non-deterministic chunk IDs. If re-chunking the same content yields new IDs, every sync duplicates instead of upserting. IDs must derive from doc_id + position + content_hash (pb-31).
• In-place representation changes. Changing chunk size or embedding model through the incremental path creates a mixed index that no eval can interpret. Shadow build + cutover, always.
• Reconciliation as a launch-week activity. Drift accumulates from rare dropped events; without the standing count check and deletion probe, you find out from a user citing a document deleted eight months ago.

Tools

• Debezium / warehouse streamsthe change feed: Debezium for operational databases, native streams/change tables for Snowflake/BigQuery/Databricks sources.
• Airflow / Dagsterorchestrate the sync DAG, own the freshness SLAs and reconciliation checks as scheduled sensors/assets.
• Qdrant / pgvector upsertsdeterministic-ID upsert and filtered delete APIs on the sink side; both handle the transaction pattern in step 3 cleanly.

Client-safe variant

CDC-driven sync usually runs entirely inside the client's estate — the feed, the queue, and the store are theirs — so the sensitive deliverable is not data movement but guarantees: put the freshness SLA, deletion SLA, and reconciliation report into the handover document, because "how fast does an erasure request leave the AI index" is a question their DPO will ask. Build the canary deletion probe with synthetic documents so it can run in production without touching real content, and if enrichment or hosted embeddings are part of re-embed (pb-30/pb-35), note that every incremental sync inherits those providers' approval scope — a new document class flowing into the CDC stream silently expands what gets sent externally unless the pipeline filters by classification.