Keep a vector index in sync with a Delta table: MERGE and change data feed

Part 1 wrote the help articles as a table of chunks, 1 per section. Help articles change: a price goes up, a section is added, an article is retired. The index the assistant searches has to follow, and embedding every chunk again on every change costs money and time for nothing. This part loads only what changed and lets the index pick up exactly that. It uses part 1's help folder and chunk_help.py. To rebuild from nothing, run python make_help.py and python chunk_help.py.

Record the changes

Delta tables can record every row a write inserts, updates or deletes, in the change data feed. It is off by default; turn it on for the table once:

spark.sql("ALTER TABLE help_chunks SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")

From the next write on, each commit's changed rows can be read back. That only helps if the writes change the rows that changed, and nothing else. Part 1's load overwrites the table, which deletes every chunk and writes every chunk again, so the feed would say all of them changed.

Load only what changed

Replace the last line of chunk_help.py, the overwrite, with a MERGE on the chunk id:

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "help_chunks")
(
    target.alias("t")
    .merge(chunks.drop("words").alias("s"), "t.chunk_id = s.chunk_id")
    .whenMatchedUpdateAll(condition="t.text <> s.text")  # only a chunk whose text changed
    .whenNotMatchedInsertAll()                           # a new section
    .whenNotMatchedBySourceDelete()                      # a section or article that is gone
    .execute()
)

The 3 clauses map to the 3 ways a help centre changes:

• A chunk whose text changed is updated. The condition t.text <> s.text matters: without it, every matched row would be rewritten, changed or not.
• A new section has an id the table does not have, and is inserted.
• A removed section is in the table but not in the files, and is deleted. Without this clause, an answer about a retired policy would stay searchable for ever.

This is why part 1 built the chunk id from the article and the section's place. With a random id, every run would look like all new chunks, and every old one would be deleted.

See what a load changed

Run the new chunk_help.py once, with nothing changed, and then change the help centre in 2 ways. In make_help.py, raise the standard delivery cost from 3.95 to 4.50, and delete the gift cards article's "How long a gift card lasts" section, then run both scripts again:

python make_help.py
python chunk_help.py

Read the commits the change feed recorded since it was turned on:

from pyspark.sql import functions as F

# The feed starts at the commit that turned it on.
enabled = (
    spark.sql("DESCRIBE HISTORY help_chunks")
    .where("operation = 'SET TBLPROPERTIES'")
    .agg(F.max("version")).first()[0]
)
changes = (
    spark.read.option("readChangeFeed", "true").option("startingVersion", enabled)
    .table("help_chunks")
)
changes.select("_commit_version", "_change_type", "chunk_id").orderBy("_commit_version", "chunk_id").show()

+---------------+----------------+------------+
|_commit_version|    _change_type|    chunk_id|
+---------------+----------------+------------+
|              3| update_preimage|  delivery-2|
|              3|update_postimage|  delivery-2|
|              3|          delete|gift-cards-2|
+---------------+----------------+------------+

_change_type says what happened to each row: insert, delete, or a pair of update_preimage and update_postimage, the row before and after an update. The first MERGE, with nothing changed, added no rows to the feed; its commit is there, empty. The second added 1 update pair for delivery-2, the delivery costs section, and 1 delete for gift-cards-2. The other 8 chunks are not in it. That short list is all the index has to process.

It is short only because part 1 removes each file's last newline. Without that, gift-cards-1, the section before the removed one, would be in the feed too: it became the article's last section, its text gained a newline, and t.text <> s.text was true. A change the reader cannot see is still a change to the text, and costs an embedding.

The index on Databricks

On Databricks, AI Search, formerly called Vector Search, builds a search index from a Delta table. A Delta Sync index reads the table's change data feed and applies it: it embeds new and changed rows with the model you name, and removes deleted ones. On Databricks the chunks are a Unity Catalog table, such as workspace.shop.help_chunks, written by the same script with the change data feed on.

An index lives on an endpoint, which serves it. Create both from a notebook, once:

%pip install databricks-ai-search
dbutils.library.restartPython()

from databricks.ai_search.client import AISearchClient

client = AISearchClient()
client.create_endpoint(name="shop", endpoint_type="STANDARD")

index = client.create_delta_sync_index(
    endpoint_name="shop",
    source_table_name="workspace.shop.help_chunks",
    index_name="workspace.shop.help_chunks_index",
    pipeline_type="TRIGGERED",
    primary_key="chunk_id",
    embedding_source_column="text",
    embedding_model_endpoint_name="databricks-gte-large-en",
)

• primary_key is the chunk id, how the index matches a change to its entry.
• embedding_source_column is the text with its title and heading from part 1. Databricks computes the embeddings with embedding_model_endpoint_name, a model it hosts.
• pipeline_type is the sync mode. CONTINUOUS follows the table within seconds, on compute that runs all the time, at a higher cost. TRIGGERED catches up when you ask, which fits a table loaded by a scheduled job.

The first sync embeds every chunk. After that, a triggered index syncs with 1 call, as the step after the load in the job:

AISearchClient().get_index(index_name="workspace.shop.help_chunks_index").sync()

On a standard endpoint each sync is incremental: only the rows changed since the last sync are processed, and the embeddings of unchanged rows are kept. With part 2's MERGE, that is the 2 changed chunks, not all 10. Free Edition has 1 AI Search endpoint, enough for this module.

The same idea without Databricks

The pattern is not tied to 1 product. With your own vector database, such as pgvectorpgvectorA PostgreSQL extension that adds vector similarity search., a job reads the change feed since the last version it processed, embeds the inserts and the post-images of updates, upserts them by chunk id and deletes the ids the feed deleted. The Embedding PipelineEmbedding PipelineThe pipeline that turns source documents into searchable vectors: it reads new and changed content, chunks it, embeds each chunk and keeps the vector index in step with the source. concept walks through it, and the embedding cost calculator prices the embeddings, first for the whole collection and then for the share that changes each day. Part 3 calls a model from SQL.