Vector Database Incremental Updates: CDC, Reindexing, and Stale Embeddings

By Fadi Updated Aug 20, 2025 13 min read Vector Databases

TL;DR — Vector database incremental updates: CDC pipelines, embedding versioning, and reindexing strategies. DBIServices: "By the time you trace it back to stale embeddings, trust is gone. CDC/event-driven pipelines are a precondition to AI data workflows." DBIServices: "IVFFlat builds 5-6x faster than HNSW. If recall drops without query pattern changes, or 30%+ new rows inserted, REINDEX to refresh centroids." Milvus: "Incremental updates involve adding, modifying, or removing data without full index rebuild. HNSW graphs handle this efficiently." Medium: "As vectors grow to millions, reindexing needs arise. Use CREATE INDEX CONCURRENTLY to avoid locking. Batch re-embedding to avoid API rate limits." Learn more with pgvector tutorial, vector search API, HNSW vs IVFFlat, and generate embeddings.

DBIServices frames the problem: "By the time you trace it back to stale embeddings, the trust is already gone. CDC/event-driven pipelines bridge two worlds — PostgreSQL CDC to JDBC Sink and event-driven architecture as a precondition to AI data workflows."

Milvus explains the core challenge: "Handling incremental updates in a vector database involves efficiently adding, modifying, or removing data without requiring a full rebuild of the index. Vector databases rely on indexing structures like HNSW graphs, tree-based partitions, or product quantization."

Incremental Update Architecture

flowchart TD subgraph Source["Source Data"] Docs["Source Documents
CMS, files, database"] Trigger["Postgres Trigger
or Logical Replication"] CDC["CDC Stream
detect content changes"] end subgraph Pipeline["Update Pipeline"] Detect["Detect Changes
content_updated_at > embedding_at"] Queue["Job Queue
pgmq / Redis / RQ"] Embed["Re-embed Changed
batch 100-1000"] Version["Track Version
embedding_version column"] end subgraph DB["PostgreSQL + pgvector"] Upsert["Upsert
ON CONFLICT DO UPDATE"] HNSW["HNSW Index
absorbs inserts, no rebuild"] IVFFlat["IVFFlat Index
may need REINDEX after 30% new"] Stale["Stale Detection
content_updated > embedding_generated"] end subgraph Monitor["Monitoring"] Recall["Recall Monitor
drop without query change = stale"] Count["Stale Count
SELECT count(*) WHERE stale"] Alert["Alert
stale count > threshold"] Reindex["Reindex Trigger
pg_repack off-peak"] end Docs --> Trigger --> CDC CDC --> Detect --> Queue --> Embed --> Version Version --> Upsert --> HNSW Upsert --> IVFFlat Stale --> Count --> Alert Recall --> Alert Alert --> Reindex Reindex --> IVFFlat

Update Strategy Comparison

Strategy	HNSW	IVFFlat	When to Use
Insert new	Absorbs without quality loss	May drift, track new row %	Always for HNSW, monitor for IVFFlat
Upsert (update existing)	ON CONFLICT DO UPDATE	ON CONFLICT DO UPDATE	Both — idempotent re-embedding
Delete	Removes from graph	Removes from cluster	Both — cleanup removed documents
Reindex	Rarely needed	After 30% new rows or recall drop	IVFFlat primarily
Full rebuild	Switch index type	Switch index type	Algorithm change, model change
CDC pipeline	Event-driven re-embed	Event-driven re-embed	Production with frequent updates

Implementation

import os
import time
from dataclasses import dataclass
from typing import Optional
from enum import Enum

class UpdateStrategy(Enum):
    HNSW_ABSORB = "hnsw_absorb"
    IVFFLAT_REINDEX = "ivfflat_reindex"
    CDC_PIPELINE = "cdc_pipeline"
    FULL_REBUILD = "full_rebuild"

@dataclass
class IncrementalUpdateManager:
    """Manage incremental updates to pgvector vector database."""

    def get_schema_sql(self) -> str:
        """Schema with embedding versioning."""
        return (
            "CREATE EXTENSION IF NOT EXISTS vector;\n"
            "\n"
            "CREATE TABLE IF NOT EXISTS documents (\n"
            "    id BIGSERIAL PRIMARY KEY,\n"
            "    source TEXT NOT NULL,\n"
            "    title TEXT,\n"
            "    content TEXT NOT NULL,\n"
            "    content_hash TEXT,\n"
            "    embedding vector(1536),\n"
            "    embedding_version INT DEFAULT 0,\n"
            "    embedding_model TEXT,\n"
            "    content_updated_at TIMESTAMPTZ\n"
            "        DEFAULT now(),\n"
            "    embedding_generated_at TIMESTAMPTZ,\n"
            "    created_at TIMESTAMPTZ DEFAULT now()\n"
            ");\n"
            "\n"
            "-- Trigger to track content changes\n"
            "CREATE OR REPLACE FUNCTION\n"
            "    update_content_timestamp()\n"
            "RETURNS TRIGGER AS $$\n"
            "BEGIN\n"
            "    NEW.content_updated_at = now();\n"
            "    RETURN NEW;\n"
            "END;\n"
            "$$ LANGUAGE plpgsql;\n"
            "\n"
            "CREATE TRIGGER track_content_change\n"
            "    BEFORE UPDATE OF content ON documents\n"
            "    FOR EACH ROW\n"
            "    EXECUTE FUNCTION update_content_timestamp();\n"
            "\n"
            "-- HNSW index (absorbs inserts)\n"
            "CREATE INDEX IF NOT EXISTS docs_embedding_hnsw\n"
            "ON documents USING hnsw (embedding vector_cosine_ops)\n"
            "WITH (m = 16, ef_construction = 200);"
        )

    def get_upsert_sql(self) -> str:
        """Upsert with embedding versioning."""
        return (
            "-- Upsert with version tracking\n"
            "INSERT INTO documents\n"
            "    (source, title, content, content_hash,\n"
            "     embedding, embedding_version,\n"
            "     embedding_model, embedding_generated_at)\n"
            "VALUES ($1, $2, $3, $4, $5::vector,\n"
            "        $6, $7, now())\n"
            "ON CONFLICT (id) DO UPDATE SET\n"
            "    content = EXCLUDED.content,\n"
            "    content_hash = EXCLUDED.content_hash,\n"
            "    embedding = EXCLUDED.embedding,\n"
            "    embedding_version =\n"
            "        documents.embedding_version + 1,\n"
            "    embedding_model = EXCLUDED.embedding_model,\n"
            "    embedding_generated_at = now()\n"
            "WHERE documents.content_hash\n"
            "      IS DISTINCT FROM EXCLUDED.content_hash"
        )

    def get_stale_detection_sql(self) -> str:
        """Detect stale embeddings."""
        return (
            "-- Find documents with stale embeddings\n"
            "-- (content changed after embedding was generated)\n"
            "SELECT id, source, title,\n"
            "       content_updated_at,\n"
            "       embedding_generated_at,\n"
            "       embedding_version\n"
            "FROM documents\n"
            "WHERE content_updated_at\n"
            "      > embedding_generated_at\n"
            "   OR embedding_generated_at IS NULL\n"
            "ORDER BY content_updated_at DESC;\n"
            "\n"
            "-- Count stale embeddings\n"
            "SELECT count(*) AS stale_count\n"
            "FROM documents\n"
            "WHERE content_updated_at\n"
            "      > embedding_generated_at\n"
            "   OR embedding_generated_at IS NULL;\n"
            "\n"
            "-- Find documents with old embedding model\n"
            "SELECT count(*) AS old_model_count\n"
            "FROM documents\n"
            "WHERE embedding_model != 'text-embedding-3-small'\n"
            "  AND embedding_model IS NOT NULL;"
        )

    def get_reindex_sql(self) -> str:
        """Reindex strategies for HNSW and IVFFlat."""
        return (
            "-- Reindex IVFFlat (after 30% new rows or recall drop)\n"
            "-- Use CONCURRENTLY to avoid locking\n"
            "CREATE INDEX CONCURRENTLY\n"
            "    docs_embedding_ivf_new\n"
            "ON documents USING ivfflat\n"
            "    (embedding vector_cosine_ops)\n"
            "WITH (lists = 1000);\n"
            "\n"
            "-- Drop old index after new one is built\n"
            "DROP INDEX CONCURRENTLY\n"
            "    docs_embedding_ivf_old;\n"
            "\n"
            "-- Rename new index\n"
            "ALTER INDEX docs_embedding_ivf_new\n"
            "    RENAME TO docs_embedding_ivf;\n"
            "\n"
            "-- For HNSW: rarely needs reindex\n"
            "-- Only when switching index type or parameters\n"
            "-- Use pg_repack for bloat cleanup\n"
            "-- pg_repack --no-order -t documents"
        )

    def get_cdc_pipeline(self) -> str:
        """CDC pipeline for event-driven updates."""
        return (
            "# CDC pipeline using Postgres logical replication\n"
            "# or triggers + job queue\n"
            "\n"
            "import asyncpg\n"
            "import json\n"
            "from embedding_generator import EmbeddingGenerator\n"
            "\n"
            "async def process_cdc_event(event):\n"
            "    '''Process a CDC event for a document.'''\n"
            "    doc_id = event['id']\n"
            "    content = event['content']\n"
            "    \n"
            "    # Re-embed the changed document\n"
            "    gen = EmbeddingGenerator()\n"
            "    embedding = gen.embed_single(content)\n"
            "    \n"
            "    # Upsert with version increment\n"
            "    async with pool.acquire() as conn:\n"
            "        await conn.execute('''\n"
            "            UPDATE documents\n"
            "            SET embedding = $1::vector,\n"
            "                embedding_version =\n"
            "                    embedding_version + 1,\n"
            "                embedding_generated_at = now()\n"
            "            WHERE id = $2\n"
            "        ''', str(embedding), doc_id)\n"
            "\n"
            "# Batch re-embed stale documents\n"
            "async def reembed_stale(batch_size=100):\n"
            "    async with pool.acquire() as conn:\n"
            "        stale = await conn.fetch('''\n"
            "            SELECT id, content FROM documents\n"
            "            WHERE content_updated_at\n"
            "                  > embedding_generated_at\n"
            "            LIMIT $1\n"
            "        ''', batch_size)\n"
            "        \n"
            "        texts = [r['content'] for r in stale]\n"
            "        embeddings = gen.embed_batch(texts)\n"
            "        \n"
            "        for row, emb in zip(stale, embeddings):\n"
            "            await conn.execute('''\n"
            "                UPDATE documents\n"
            "                SET embedding = $1::vector,\n"
            "                    embedding_version =\n"
            "                        embedding_version + 1,\n"
            "                    embedding_generated_at = now()\n"
            "                WHERE id = $2\n"
            "            ''', str(emb), row['id'])"
        )

    def recommend_strategy(self, index_type: str,
                           vector_count: int,
                           new_rows_pct: float,
                           recall_drop: bool,
                           has_cdc: bool) -> dict:
        """Recommend update strategy based on conditions."""
        if recall_drop and index_type == "ivfflat":
            return {
                "strategy": UpdateStrategy.IVFFLAT_REINDEX.value,
                "reason": "Recall dropped without query change. "
                          "IVFFlat centroids are stale. "
                          "REINDEX with CONCURRENTLY.",
                "sql": self.get_reindex_sql(),
            }

        if new_rows_pct > 30 and index_type == "ivfflat":
            return {
                "strategy": UpdateStrategy.IVFFLAT_REINDEX.value,
                "reason": f"{new_rows_pct}% new rows inserted. "
                          "IVFFlat needs REINDEX after 30% growth. "
                          "Centroids have drifted.",
                "sql": self.get_reindex_sql(),
            }

        if has_cdc:
            return {
                "strategy": UpdateStrategy.CDC_PIPELINE.value,
                "reason": "CDC pipeline detects source changes "
                          "and triggers re-embedding. "
                          "Event-driven updates.",
                "sql": self.get_cdc_pipeline(),
            }

        if index_type == "hnsw":
            return {
                "strategy": UpdateStrategy.HNSW_ABSORB.value,
                "reason": "HNSW absorbs inserts without "
                          "quality loss. No reindex needed. "
                          "Use ON CONFLICT DO UPDATE for upserts.",
                "sql": self.get_upsert_sql(),
            }

        return {
            "strategy": UpdateStrategy.FULL_REBUILD.value,
            "reason": "Full rebuild recommended. "
                      "Switch index type or embedding model. "
                      "Use CREATE INDEX CONCURRENTLY.",
            "sql": self.get_reindex_sql(),
        }

    def get_monitoring_queries(self) -> dict:
        """Monitoring queries for incremental updates."""
        return {
            "stale_count": (
                "SELECT count(*) FROM documents "
                "WHERE content_updated_at > "
                "embedding_generated_at"
            ),
            "index_size": (
                "SELECT pg_size_pretty("
                "pg_relation_size("
                "'docs_embedding_hnsw'))"
            ),
            "dead_tuples": (
                "SELECT n_dead_tup FROM pg_stat_user_tables "
                "WHERE relname = 'documents'"
            ),
            "new_rows_pct": (
                "SELECT round(100.0 * count(*) / "
                "(SELECT count(*) FROM documents), 1) "
                "FROM documents "
                "WHERE created_at > now() - interval '7 days'"
            ),
            "embedding_versions": (
                "SELECT embedding_model, "
                "embedding_version, count(*) "
                "FROM documents "
                "GROUP BY embedding_model, "
                "embedding_version "
                "ORDER BY count(*) DESC"
            ),
        }

Incremental Update Checklist

[ ] HNSW absorbs inserts without quality loss — no reindex needed for new data
[ ] IVFFlat needs REINDEX after 30% new rows or when recall drops without query pattern changes
[ ] IVFFlat builds 5-6x faster than HNSW but centroids drift with new data
[ ] Use INSERT ... ON CONFLICT DO UPDATE for idempotent upserts
[ ] Track embedding_version column for each document
[ ] Track embedding_model to detect old model embeddings
[ ] Track content_updated_at and embedding_generated_at timestamps
[ ] Use Postgres triggers to auto-update content_updated_at on content changes
[ ] Use content_hash to detect actual content changes (avoid unnecessary re-embedding)
[ ] WHERE documents.content_hash IS DISTINCT FROM EXCLUDED.content_hash — skip if unchanged
[ ] CDC pipeline: use logical replication or triggers to detect source changes
[ ] Event-driven architecture is a precondition to AI data workflows
[ ] By the time you trace it back to stale embeddings, trust is already gone
[ ] CDC/event-driven pipelines bridge source data and vector database
[ ] Use job queue (pgmq, Redis, Python RQ) for async re-embedding
[ ] Batch re-embedding: 100-1000 per batch to avoid API rate limits
[ ] Detect stale: SELECT count(*) WHERE content_updated_at > embedding_generated_at
[ ] Monitor recall over time — sudden drops indicate stale embeddings or index degradation
[ ] Alert when stale count exceeds threshold
[ ] Reindex IVFFlat: CREATE INDEX CONCURRENTLY, drop old, rename new
[ ] Use CREATE INDEX CONCURRENTLY to avoid locking production table
[ ] Use pg_repack for bloat cleanup without locks
[ ] pg_repack --no-order -t documents for table bloat
[ ] Schedule reindex off-peak (3 AM cron)
[ ] As vector database grows from thousands to millions, reindexing needs arise
[ ] Algorithm tuning: switch from HNSW to IVF-PQ to save RAM at scale
[ ] Track new row percentage: compare recent inserts to total count
[ ] Monitor dead tuples after frequent updates: pg_stat_user_tables
[ ] Set statement_timeout for long-running upserts
[ ] Embedding versioning enables rollback to previous model
[ ] Keep old embeddings temporarily when switching models
[ ] Gradual model migration: re-embed in batches, verify quality, switch
[ ] Monitor index size: pg_size_pretty(pg_relation_size(...))
[ ] Monitor embedding model distribution: GROUP BY embedding_model
[ ] pgvector adds vectors to PostgreSQL — revolutionary for keeping relational and vector data together
[ ] Query vectors and relational data in one transaction
[ ] Multiple index types: HNSW, IVFFlat, pgvectorscale DiskANN
[ ] Handling incremental updates involves adding, modifying, or removing without full rebuild
[ ] Vector databases rely on HNSW graphs, tree-based partitions, or product quantization
[ ] Read pgvector tutorial for setup
[ ] Read vector search API for API
[ ] Read HNSW vs IVFFlat for index selection
[ ] Read generate embeddings for embedding
[ ] Read semantic search for search
[ ] Read pgvector vs Pinecone for comparison
[ ] Test: HNSW absorbs 1000 inserts without recall degradation
[ ] Test: IVFFlat recall drops after 30% new rows without reindex
[ ] Test: ON CONFLICT DO UPDATE increments embedding_version
[ ] Test: stale detection query finds changed-but-not-re-embedded documents
[ ] Test: CDC pipeline triggers re-embedding on content change
[ ] Test: REINDEX CONCURRENTLY does not lock table
[ ] Document update strategy, reindex schedule, CDC pipeline, and monitoring alerts

FAQ

How do you handle incremental updates in a vector database?

Handle incremental updates by using HNSW indexes that absorb inserts without rebuild, upserting embeddings with ON CONFLICT, and using CDC pipelines for event-driven updates. Milvus: "Handling incremental updates involves efficiently adding, modifying, or removing data without requiring a full rebuild of the index. Vector databases rely on indexing structures like HNSW graphs, tree-based partitions, or product quantization." DBIServices: "By the time you trace it back to stale embeddings, the trust is already gone. CDC/event-driven pipelines bridge two worlds — PostgreSQL CDC to JDBC Sink and event-driven architecture as a precondition to AI data workflows." Strategies: (1) HNSW absorbs inserts without quality loss. (2) Use ON CONFLICT DO UPDATE for upserts. (3) CDC pipeline detects source changes. (4) Re-embed only changed documents. (5) Track embedding versions for rollback.

When do you need to reindex a pgvector HNSW or IVFFlat index?

HNSW rarely needs reindexing — it absorbs inserts without quality loss. IVFFlat needs REINDEX when data distribution shifts or after inserting 30%+ new rows. DBIServices: "IVFFlat builds 5-6x faster than HNSW. If your data distribution shifts materially, plan a REINDEX to refresh clustering quality (centroid drift). A practical signal: if recall drops without any change in query patterns, or if you have inserted more than ~30% new rows since the last build, the centroids are stale and a REINDEX is warranted." Medium: "As your vector database grows from thousands to millions of production vectors, you will hit reindexing needs. Algorithm tuning: switch from HNSW to IVF-PQ to save RAM. When reindexing, use CREATE INDEX CONCURRENTLY to avoid locking." HNSW: no reindex needed unless switching index type. IVFFlat: reindex after 30% new rows or when recall drops.

How do you implement embedding versioning with CDC in pgvector?

Use PostgreSQL CDC (logical replication or triggers) to detect source document changes, then re-embed only changed documents with version tracking. DBIServices: "CDC/event-driven pipelines bridge two worlds. PostgreSQL CDC to JDBC Sink and event-driven architecture as a precondition to AI data workflows. By the time you trace it back to stale embeddings, the trust is already gone." Implementation: (1) Add embedding_version column to documents table. (2) Use Postgres triggers or logical replication to detect source changes. (3) On change event, re-embed the document and update embedding_version. (4) Keep old embeddings for rollback. (5) Track which embedding model version was used. (6) Use a job queue (pgmq, Redis, or Python RQ) for async re-embedding. (7) Monitor for stale embeddings: documents where content_updated_at > embedding_updated_at.

How do you detect and fix stale embeddings in a vector database?

Detect stale embeddings by comparing document content update timestamps with embedding generation timestamps, and by monitoring recall drops. DBIServices: "By the time you trace it back to stale embeddings, the trust is already gone. Event-driven architecture is a precondition to AI data workflows." DBIServices: "A practical signal: if recall drops without any change in query patterns, or if you have inserted more than ~30% new rows since the last build, the centroids are stale." Detection: (1) Compare content_updated_at with embedding_generated_at. (2) Monitor recall over time — sudden drops indicate stale embeddings. (3) Track embedding model version — old model embeddings may be stale. (4) Query for stale: SELECT count(*) FROM documents WHERE content_updated_at > embedding_generated_at. Fix: (1) Re-embed stale documents. (2) Update embedding_version. (3) Use CDC pipeline for automatic detection. (4) Set up alerts for stale count > threshold.

What is the best upsert strategy for pgvector in production?

Use INSERT ... ON CONFLICT DO UPDATE for idempotent upserts, batch embeddings for efficiency, and track embedding versions for rollback. DBIServices: "Event-driven architecture is a precondition to AI data workflows. CDC pipelines detect source changes and trigger re-embedding." Medium: "When reindexing, use CREATE INDEX CONCURRENTLY to avoid locking. Batch re-embedding to avoid API rate limits." Strategy: (1) Use ON CONFLICT (id) DO UPDATE SET embedding = EXCLUDED.embedding, embedding_version = EXCLUDED.embedding_version. (2) Batch upserts: 100-1000 per transaction. (3) Track embedding_version for each document. (4) Use a job queue for async re-embedding of changed documents. (5) Set statement_timeout for long-running upserts. (6) Monitor for dead tuples after frequent updates. (7) Use pg_repack for bloat cleanup. (8) Use CREATE INDEX CONCURRENTLY when rebuilding indexes.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →