Incremental Vector Data Pipeline

The problem

Vector indexes drift out of sync with source data. The typical response is a nightly full re-embed of the entire corpus, which is both wasteful (60-80% of documents didn't change meaningfully) and stale (queries between runs are served outdated context). The harder problem is incremental, consistent, versioned embedding at scale: re-embed only what changed, prove no vectors are orphaned or missing, and support instant model rollback without re-embedding the corpus again.

How this project solves it

A Change Data Capture pipeline decides, for each source write, whether it requires new embeddings and at what granularity. Two layers of delta filtering eliminate unnecessary embedding API calls:

• Document-level filter (cdc/change_filter.py): a decision ladder tests hash equality, then metadata-only changes, then text similarity. Only documents with genuinely significant content changes reach the embedding stage.
• Chunk-level diff (chunk/chunker.py): chunk identities are content-addressed (doc_id:sha256(text)[:16]), so inserting a paragraph at the top of a document re-embeds one chunk, not the whole document. diff_chunks returns exactly the chunks to embed and the chunk IDs to tombstone.
• Versioned store (store/vector_store.py): superseding a vector sets valid_to and preserves the old row. Old vectors stay available for rollback and audit. Three invariants are enforced structurally: unique-current per (chunk_id, model_version), dimension lock per model version, and monotonically increasing version numbers per chunk.
• Blue-green migration (version/version_manager.py): register a new model alongside the active one, backfill incrementally, then promote. can_promote refuses until every current chunk is covered under the new model. Rollback is an O(1) pointer switch back to the previous model because its vectors are still in the store.
• Consistency checker (consistency/checker.py): six properties verified independently of the write path after every batch: no duplicate-current vectors, no orphaned vectors, no missing vectors, no source drift, dimension uniformity, and version monotonicity.
• Oracle persistence (orchestration/pipeline.py): checkpoint() serializes the expected-chunk oracle and the LSN watermark to a JSON-serializable dict. restore_checkpoint() reconstructs the pipeline on restart, preventing all stored vectors from appearing orphaned after a crash.
• Carry-forward on rollback (orchestration/pipeline.py): rollback_and_carry_forward() detects chunks that exist only under the now-former model and re-embeds them under the rollback target, closing the data-loss gap for documents added after a promote.
• Version-aware expiration (version/version_manager.py): expire_model() refuses to delete vectors for the active model or the immediate rollback target, preserving the O(1) recovery guarantee.
• Cost accounting (cost/cost_report.py): tracks chunks actually embedded vs. total corpus size across batches and reports realized savings percentage.
• Freshness SLA (sla/freshness_monitor.py): two-tier alerting (warn/critical) on index age and CDC lag.

How it scales

Data volume. The in-process store is a plain list, suitable for development and CI. Production replaces it with LanceDB or an Iceberg table; the store interface is the same. Both support columnar scans and predicate pushdown, so current_rows(model) scales independently of historical row count.

Embedding throughput. The DeterministicEmbedder used in tests is single-call. APIEmbedder wraps an external embedding API and can be replaced with any implementation of the Embedder protocol. For high throughput, replace the serial embed([text, ...]) loop in _reembed_doc with a batched call and a rate-limited async queue; the orchestrator's interface does not change.

Migration cost. A same-dimension (non-breaking) model upgrade is a background incremental backfill: only new and changed documents require dual-write during the migration window. A dimension-changing (breaking) upgrade requires a full corpus re-embed, but rollback to the previous model remains O(1) until the old vectors are explicitly expired.

Consistency at scale. The six-property checker runs in a single pass over current rows and the expected-chunk map. For very large corpora, this check can be partitioned by doc_id range and run in parallel workers, each checking its shard's subset of expected_chunks.

Oracle durability. The checkpoint dict is JSON and fits in a single object store PUT. For continuous pipelines, checkpoint after every batch and use the indexed_lsn watermark as the Kafka consumer-group offset. On restart, restore_checkpoint() reloads the oracle and incremental diffs resume correctly from the next event.

How to demo it

From a clean checkout:

pip install -r requirements.txt

# Run the full test suite (no cluster or API keys needed)
make test
# Expected: 68 passed

# Run the end-to-end demo
make demo

Expected demo output:

[1] Initial ingest of 2 documents
   embedded 2 chunks across 2 docs
   OK [after ingest] CONSISTENT (2 current vectors == 2 expected chunks)

[2] Cosmetic edit to docA (whitespace only)
   chunks embedded: 0 (skipped_cosmetic=0, skipped_unchanged=1)

[3] Significant edit to docA (rewrite one paragraph)
   chunks embedded: 1, chunks tombstoned: 1  (only the changed chunk)
   OK [after significant edit] CONSISTENT (2 current vectors == 2 expected chunks)

[4] Delete docB
   tombstoned 1 chunks; current vectors now: 1
   OK [after delete (no orphans)] CONSISTENT (1 current vectors == 1 expected chunks)

[5] Blue-green migration v1.0.0 -> v1.1.0
   migration: non-breaking -> incremental backfill ok
   can promote v2? True (fully covered: 1/1)
   active model now: [email protected]

[6] Rollback to v1.0.0 (simulated quality regression on v2)
   rolled back to: [email protected]

[7] Cost: incremental vs nightly full re-embed
   incremental embedded 3 chunks ($0.0001) vs full re-embed 7 chunks ($0.0002) = 57.1% savings

No environment variables or external services are required to run the tests or demo. The Flink CDC job (cdc/flink_cdc_job.py) and Airflow DAG (airflow/dags/index_health_dag.py) are wired for the docker-compose stack; start it with make up if you have Docker.

Project structure

src/vectorpipe/
  cdc/           change-significance filter + Flink CDC job stub
  chunk/         content-addressed chunker and chunk diff
  embed/         model registry, DeterministicEmbedder (CI), APIEmbedder (prod)
  store/         versioned multimodal store with expiration
  version/       blue-green migration, O(1) rollback, version-aware expiration
  consistency/   six-property correctness checker
  sla/           freshness alerting (index age + CDC lag)
  cost/          incremental vs. full re-embed cost accounting
  orchestration/ closed incremental loop with checkpoint and carry-forward