"Feeding the Models" — Data Engineering for ML and AI

Skim Card

One thing to remember: a model learns from what the data said at the time, so every vector and every feature must follow the source and carry when it became known.

• Join features as they stood when the prediction was made: today’s profile scored hit@1 0.715; the as-of join 0.318, below popularity’s 0.450.
• Give each feature an event time and an availability time; serving matched availability-time features 2,000 times out of 2,000, event-time ones missed 17.
• Let CDC drive the index and hash text before embedding: replaying 7,000 rows cost no calls; an appended message re-embedded one chunk of three.
• Make replays safe with source LSNs and tombstones: 157 older events were skipped or superseded; both deleted documents stayed deleted.
• pgvector filters after the HNSW scan: 48 of 60 one-country queries came back short until iterative index scans.
• Score retrieval on a tagged eval set: the query instruction lifted recall@5 from 0.644 to 0.831; hybrid’s 0.864 is not a significant gain.

Use …	When …
pgvector	Vectors sit next to rows you already filter, join and delete in PostgreSQL
A dedicated vector database (Qdrant, Milvus, Weaviate)	Billions of vectors, sharding, heavy filtered search
Iceberg + DuckDB	Offline evaluation, batch jobs, small corpora: exact search is fast enough
A feature store (Feast)	Many models share features and need an online store

Verified against: pgvector 0.8.7 on PostgreSQL 18.6, Debezium 3.7.0, Kafka 4.4.0, Qwen3-Embedding-0.6B (8-bit) on LMForge (pre-release 0.3.0 build; latest release 0.2.0), DuckDB 2.0.0, PyIceberg 0.12.0, Feast 0.67 docs · vendor docs · 31 October 2026

Jump to: Concepts · Hands-On · Landscape · Design Drill

Story Opening

Support wanted an assistant that answers agents from past tickets and reviews; data science wanted a training set for “you might also like”. Rohan had a plan for both by lunch: a nightly job that re-embeds every ticket, and a join from orders to the customer profile table.

“The nightly job pays to embed 7,000 documents that didn’t change,” Ingrid said, “and it keeps answering from a ticket a customer asked us to erase until tonight. And your profile table already knows what each customer bought next. Your model will look brilliant offline and lose to ‘show everyone pantry staples’ in production.”

The Quick Map

Concepts, with use cases

Of Part 2’s five promises, this part strengthens correctness: a model is only as right as the data it saw, at the time it would have seen it. It builds on Part 16’s change data capture and Part 23’s erasure process.

Point-in-time correctness

A feature is a model input computed from data, such as “the category this customer buys most”. A training example pairs features with a label at a prediction time. Point-in-time correctness means each example uses the feature values as they stood at that time, never later ones. Breaking it is leakage: the label, or the future, leaks into the inputs.

Two clocks matter, as in streaming (Part 26): the event time a value became true and the availability time it reached serving. With a nightly feature job, today’s orders reach the model tomorrow; train on event time and it learns from values it won’t have when serving: training/serving skew.

What breaks without it. Offline metrics look excellent; production does worse than the baseline it replaced. Larkspur: joining today’s profile scored 0.715 on next-category prediction; the honest score is 0.318, below recommending the best-selling category (0.450) (Step 6). Elsewhere: a lender’s default model that uses “number of collection calls”, a field filled in after the default happened.

Feature stores

A feature store keeps feature definitions in a registry, value history in an offline store for training, latest values in an online store for serving, and point-in-time retrieval. Feast 0.67 (Apache 2.0) is the best-known open-source one: get_historical_features scans back from each row’s timestamp up to a TTL and, with filter_by_created_timestamp=True, also drops values created after it, the availability rule above (not every offline store supports the flag yet).

What breaks without one. Teams re-derive “favourite category” differently; serving code drifts from the training SQL. Larkspur: for one model, an Iceberg table with both timestamps and one SQL lookup shared by training and serving is enough; a registry and online store pay off at many models. Elsewhere: a ride-hailing company serves “driver’s acceptance rate, last hour” to dispatch and to its fraud model from one definition.

Unstructured data pipelines

Text needs its own pipeline: parse (HTML, PDF, e-mail threads into text), redact personal data, chunk into retrievable passages, embed each chunk into a vector, and store vectors with metadata (country, status, model, chunker version, source position). RAG (retrieval-augmented generation) gives a language model the chunks nearest a question as context.

What breaks without it. One vector per document blurs five topics; e-mail addresses reach prompts; nobody knows which model made which vector. Larkspur: tickets are threads, so one chunk per message, each prefixed with a header line naming the ticket, issue, country and subject. Elsewhere: a hospital indexes discharge letters by section, so “medications on discharge” is retrieved without the history.

Incremental re-embedding driven by CDC

Embedding is the expensive step, so the index follows the source: on each change event (Part 16’s Debezium) the document is re-chunked and only chunks with a new content hash are embedded; a delete removes the chunks.

What breaks without it. A nightly full rebuild spends the embedding budget on unchanged text, serves stale answers all day and keeps deleted documents until the next run. Larkspur: the larkspur-support connector feeds a host consumer that writes to pgvector (Step 3). Elsewhere: a SaaS help centre re-indexes the one article an author edited, not all 40,000.

Where the vectors live

• pgvector: a vector type and HNSW/IVFFlat indexes inside PostgreSQL; indexes take up to 2,000 dimensions (halfvec 4,000).
• Lance / LanceDB (Apache 2.0): a versioned columnar format with vector indexes, and the embedded database on it.
• Dedicated vector databases (Qdrant, Milvus, Weaviate): sharding, quantisation, filtered search at billions of vectors.
• The lakehouse: Iceberg vectors and DuckDB’s exact search, enough offline (evaluation, deduplication) or when small.

What breaks if you pick wrong. A dedicated cluster for 15,000 chunks is an on-call rota for nothing; a lake table behind a chat box scans everything per question. Larkspur: pgvector in its own rag database, and a tagged Iceberg copy for evaluation. Elsewhere: a media company serves a billion image embeddings from a dedicated store and runs duplicate detection on the lake.

Evaluation data and dataset versions

An evaluation set is a fixed list of questions with known answers (here: queries and the tickets that answer them), used to score retrieval with recall@k, the share of expected documents in the top k. A dataset version pins exactly which rows a model or score came from: an Iceberg tag on a snapshot (Part 7).

What breaks without it. The chunker changes, recall “improves”, and nobody knows whether the index, the questions or the model moved. Larkspur: the eval set tagged rag-eval-v1, the training set reco-train-v1, and each run logged with model, chunker and the offsets the index had applied. Elsewhere: a search team at a marketplace gates every ranking change on a frozen query set.

Under the Hood

What the indexer does with one change event

A Debezium event carries op (r snapshot, c, u, d), the row after (for a delete, before holds the key, with "" and 1970-01-01 placeholders in other NOT NULL columns under the default replica identity) and source.lsn, the change’s position in the write-ahead log. The indexer:

1. Keeps each document’s last event per batch. Earlier ones are superseded: the last carries the whole row or the delete.
2. Skips older versions. A lower LSN than the index holds is a replay; applying it would roll the document back.
3. Deletes into a tombstone. A delete removes the chunks and keeps a row with the document ID and LSN but no customer, text or hash; without it, a replay re-inserts the ticket from its snapshot event. (Not a Kafka tombstone: tombstones.on.delete=false, cleanup.policy=delete.)
4. Hashes before it embeds. The document hash covers model, chunker version and the redacted chunks; equal hash means metadata only. Otherwise each chunk’s own hash decides: matching chunks keep their vectors, new ones are embedded.
5. Commits offsets with the data. Chunks and the next Kafka offsets go into one PostgreSQL transaction, so a crash re-reads at most one batch, and re-applying it is idempotent.

HNSW, filters and the planner

HNSW (hierarchical navigable small world) builds a layered graph of vectors; a search walks it greedily from the top layer and keeps hnsw.ef_search candidates (40 by default) at the bottom. It needs no training; m (16) and ef_construction (64) trade build time and memory for recall. IVFFlat clusters vectors into lists and searches probes of them: faster and smaller to build, but create it after the data exists.

pgvector applies a WHERE filter after the index scan: with 40 candidates and a country holding 15% of the chunks, a “top 10 in SG” query finds fewer than 10 (Step 4). Since 0.8.0, iterative index scans (hnsw.iterative_scan = strict_order or relaxed_order) keep walking the graph until enough rows pass the filter or hnsw.max_scan_tuples (20,000) is reached; relaxed_order may return results slightly out of order by distance (use strict_order, or re-sort in a materialized CTE). And the planner may not use the index at all: for the one-country query it chose a sequential scan (Step 4); forced both ways on the unfiltered top 10, the full scan took 32.6 ms against 1.3 ms (Step 5).

Embeddings are a dependency with a version

The lab embeds with Qwen3-Embedding-0.6B (Apache 2.0, 1,024 dimensions, 32K context), 8-bit, served by LMForge, a local OpenAI-compatible server (/v1/embeddings). Disclosure: LMForge is the author’s own MIT-licensed project. Ollama or another OpenAI-compatible server should work (not tested): set EMBEDDINGS_URL and EMBEDDING_MODEL; another quantisation shifts the numbers. The model card asks for an instruction on queries (Instruct: …\nQuery:…) and none on documents. The same text embedded alone and in a batch differed by up to 0.000275 in one dimension (batching), so the index detects changes by hashing text, never by comparing vectors, and stores the model per chunk.

Hands-On

The code is in the every-record-counts folder of the companion-workspaces repository (tag every-record-counts/part-32). make up-stream starts core, Kafka and Kafka Connect with Debezium; embeddings run on the host, outside Docker. The lab reads raw.orders and writes ml.* as root, a shortcut outside Part 23’s governed path; production gives the pipeline its own principal.

make up-stream
uv run python -m scripts.part32_models setup      # source tables, rag database, eval set
uv run python -m scripts.part32_models connect    # Debezium connector larkspur-support
uv run python -m scripts.part32_models index --label initial   # the CDC consumer
uv run python -m scripts.part32_models index --from-beginning --label replay
uv run python -m scripts.part32_models change     # edits and one erasure, in the source
uv run python -m scripts.part32_models index --label changes
uv run python -m scripts.part32_models erased
uv run python -m scripts.part32_models index --from-beginning --label replay-after
uv run python -m scripts.part32_models filtered
uv run python -m scripts.part32_models eval
uv run python -m scripts.part32_models batching
uv run python -m scripts.part32_models export
uv run python -m scripts.part32_models features

Step 1: text in the source, an index beside it

datagen/support.py writes 4,000 tickets on real orders (seed 42) and 3,000 reviews, with the customer’s e-mail or phone in about 30% of first messages. setup loads them into support.tickets and support.reviews through Part 23’s suppression list, so an erased customer never lands: 0 rows dropped here; in Step 6 it drops CGB0000113’s one order, back in raw.orders through a gap on Part 23’s “not covered” list (Part 16’s replay). The index lives in its own database:

CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE chunks (
    doc_id      text        NOT NULL REFERENCES documents ON DELETE CASCADE,
    chunk_no    int         NOT NULL,
    chunk_hash  text        NOT NULL,           -- sha256 of model + chunk text: the unit of re-use
    country     char(2)     NOT NULL,           -- copied from the document for filtered search
    text        text        NOT NULL,           -- header line + one message, personal data redacted
    tsv         tsvector    GENERATED ALWAYS AS (to_tsvector('english', text)) STORED,
    embedding   vector(1024) NOT NULL,
    model       text        NOT NULL,
    embedded_at timestamptz NOT NULL,
    PRIMARY KEY (doc_id, chunk_no)
);
CREATE INDEX chunks_embedding ON chunks USING hnsw (embedding vector_cosine_ops);  -- m 16, ef_construction 64
CREATE INDEX chunks_tsv ON chunks USING gin (tsv);

documents holds per ticket or review the content hash, source_lsn, source_ts (event time), indexed_at (ingestion time) and deleted_at.

Step 2: the connector and the first index

kafka/connectors/larkspur-support.json is Part 16’s connector with four keys changed: table.include.list, its own slot.name and publication.name, and topic.prefix larkspur.cx, which Debezium’s docs say “should be unique across all other connectors” (Troubleshooting). The snapshot reached Kafka in 2 s. The first index run embeds everything:

initial: 307 s (from the beginning)   (memory line trimmed)
  events r                            7,000
  chunks embedded                    15,006
  documents (re)indexed               7,000
  embedding requests                    241
  embedding tokens                  844,371
  redactions                          1,593
  index now: 7,000 documents (0 tombstones), 15,006 chunks

15,006 chunks in 5 minutes on an M3 Pro: about 50 chunks a second through 241 requests of up to 64 inputs. The 1,593 redactions replace e-mail addresses and phone numbers with [email] and [phone] before hashing. They are two regular expressions for the patterns the generator plants: order numbers, cities and names stay, and an order number leads back to the customer. Production uses a PII detector and still treats chunks as personal data.

Step 3: replays, edits and an erasure

Replaying both topics from offset 0 (index --from-beginning): 7,000 events, 7,000 documents unchanged, 0 embedding requests in 4 s. Then change edits the source: it closes 153 open or pending tickets, appends an agent message to one, adds “(escalated)” to another’s subject, and erases a customer with one ticket and one review. The indexer, from its stored offsets:

changes: 1 s (from the stored offsets)   (token and memory lines trimmed)
  events d                                2
  events u                              155
  chunks deleted                          5
  chunks embedded                         5
  chunks re-used                          2
  documents (re)indexed                   2
    TGB000003: embedded 4, re-used 0
    TGB000034: embedded 1, re-used 2
  documents deleted                       2
  documents unchanged                   153
  embedding requests                      1
  redactions                              4
  index now: 6,998 documents (2 tombstones), 15,002 chunks

The subject edit re-embedded all four chunks of TGB000003, because the subject is in every chunk’s header: the header buys retrieval context and costs re-embedding on header edits.

erased checks where the customer still is:

CGB0006207: source rows 0; index: documents naming them 0, chunks of TGB003140, RGB002847 0, tombstones (no customer, no text) 2
Kafka still holds 2 events with their rows ({'r': 2}). Retention deletes whole segments: a segment rolls after segment.ms (7 days) and is deleted once its newest record is older than retention.ms (7 days); cleanup.policy=delete

The snapshot events still carry the full rows, unredacted, until their segment rolls (segment.ms, 7 days by default) and then ages past retention.ms: up to about two weeks here. Set both below the erasure deadline (Part 23). A last replay from offset 0 made no embedding calls: 132 events were older than the index and 25 were superseded by a later event in the same batch, the two deleted documents’ snapshot rows among them. Nothing came back.

Step 4: filtered search and the planner

filtered asks each of the 60 eval queries for the 10 nearest SG chunks:

SG share of chunks: 14.9%
planner's choice, top 10 in SG: Seq Scan; with enable_seqscan = off: Index Scan using chunks_embedding
planner's choice, top 10 unfiltered: Seq Scan
hnsw.iterative_scan = off            rows returned (asked for 10): min 0, mean 5.8, queries short of 10: 48
hnsw.iterative_scan = relaxed_order  rows returned (asked for 10): min 10, mean 10.0, queries short of 10: 0
hnsw.iterative_scan = strict_order   rows returned (asked for 10): min 10, mean 10.0, queries short of 10: 0

Either iterative mode fixed all 48 short queries. Left alone, the planner picked a sequential scan for both queries: exact, and slower than HNSW (Step 5), so the lab forces each path to compare them.

Step 5: evaluate, and version what you evaluated

The evaluation set comes from the generator, so read it as a lab benchmark: each of the 60 queries is written from one of 20 templates that name the product and paraphrase the issue, and its expected tickets are every ticket with the same issue and SKU (one to three, by construction). eval reads the set at its tag and scores vector search (with and without the model’s instruction), full-text (any of the query’s words) and hybrid, both lists merged by reciprocal rank fusion (a document scores the sum of 1/(60 + rank)), all in sql/part32/postgres/search.sql:

recall@5  vector, instruction          0.831
recall@5  keyword                      0.628  vs vector with instruction: 4 queries better, 20 worse
recall@5  hybrid                       0.864  vs vector with instruction: 4 queries better, 3 worse
recall@5  vector, no instruction       0.644  vs vector with instruction: 1 queries better, 14 worse
HNSW top-10 vs exact top-10: 0.978 of the exact neighbours found; 1.3 ms vs 32.6 ms per query over 15,002 chunks
logged to ml.rag_eval_runs: eval snapshot 6323619975482053401, offsets {'larkspur.cx.support.reviews/0': 1011, … (six partitions)}

The instruction’s +0.19 is robust: without it 14 queries got worse and one better. Hybrid’s +0.033 (four queries better, three worse) is not significant: a reason to test hybrid on your own data, not proof that it wins. HNSW found 98% of the exact top 10 at about 25 times the speed. The run log records the eval snapshot and the six partition offsets the index had applied, so each score names the corpus it measured.

export then writes the 15,002 chunks with their vectors to an Iceberg table, ml.support_chunks, tagged rag-corpus-v1. DuckDB’s exact search over that snapshot (array_cosine_distance on FLOAT[1024], no index) took 16.8 ms per query and returned the same 10 chunks as pgvector’s exact scan for all 60 queries: for offline evaluation, the lakehouse is enough.

Step 6: features as of the moment of prediction

The task: at each repeat online order, predict a category the customer will buy from their favourite so far, else the best-seller. features reads Part 16’s CDC replica of orders (raw.orders), builds customer_features (one row per customer per order) and attaches it three ways:

SELECT e.customer_id, e.order_id, e.event_ts, e.bought,
       l.favourite_category AS fav_naive,          -- today's profile
       fe.favourite_category AS fav_event_time,    -- as of event time
       fa.favourite_category AS fav_available, fa.available_at,
       p.top_category AS popular
FROM examples e
JOIN latest l USING (customer_id)
ASOF LEFT JOIN customer_features fe ON e.customer_id = fe.customer_id AND e.event_ts > fe.feature_ts
ASOF LEFT JOIN published fa ON e.customer_id = fa.customer_id AND e.event_ts >= fa.available_at
ASOF LEFT JOIN popularity p ON e.event_ts >= p.available_at;

ASOF JOIN takes the latest feature row before each order. available_at is the 02:00 UTC run after the order’s day; published keeps the last value per run, as the online store would.

suppression list: 1 rows dropped, 0 identifiers tombstoned
customer_features: 175,667 rows for 154,041 customers (one per order, as it stood after it)
training set                                 examples with a feature   hit@1
naive: today's profile                         21,626         21,626   0.715
as of event time (feature_ts < order)          21,626         21,626   0.317
as of availability (available_at <= order)     21,626         21,493   0.318
popularity only                                21,626         21,626   0.450
served value vs training value, 2,000 examples: event-time features differ in 17, availability-time features differ in 0
ml.customer_features 175,667 rows; ml.reco_training 21,626 rows, tag reco-train-v1 -> snapshot 8768366725202028920

The naive join scores 0.715 because today’s profile includes the order being predicted. The honest 0.318 shows what it hid: the generator draws each order’s products independently of the customer, so history carries no signal and loses to popularity. Leakage didn’t inflate a good model; it invented one. The skew check calls the serving lookup for 2,000 examples. Availability-time features match by construction (same published table, same <= rule): a parity check against drift between two code paths. Event-time features differ in 17 (0.85%: repeat orders placed before the previous order’s features were published); fast-moving features would skew far more. The tag reco-train-v1 lets the next run compare on the same rows. tests/test_part32.py asserts the behaviours above (11 passed in 6 min 40 s, most of it embedding).

The Landscape

Current product names as of 31 October 2026. A cell that couldn’t be verified says so.

Tips & Gotchas

Gotcha — deletes don’t reach the copies you forgot. The index dropped the erased customer at once, but the topic keeps their snapshot rows until the segment rolls and ages out (Step 3), and the tags rag-corpus-v1 and reco-train-v1 pin snapshots that hold their text or orders. Put topics, eval sets and tagged training sets in Part 23’s erasure inventory, and re-point the tags (Part 23’s repoint) when erasing.

Gotcha — a model change is a full re-embed. Vectors from two models don’t share a space; mixed in one column, they quietly ruin search. The model is in the document hash here, so a change re-embeds everything.

Tip — store the model name per chunk (chunks.model), so a migration can run two indexes side by side and switch on measured recall.

Gotcha — ASOF joins on a timestamp shared by several rows pick one arbitrarily. Two orders on one day share an available_at; deduplicate to the last value per key and timestamp (published) before joining.

Troubleshooting

Design Drill

The question: “We are a legal-tech company. Build the data pipeline for a RAG assistant over 20 million constantly changing documents: contracts, court filings and e-mail from 3,000 law firms. Documents are edited, re-versioned and deleted every day; each answer must cite the version that is current, and a lawyer may only retrieve documents from matters they are staffed on.”

Clarify first

• How fast must edits and deletions reach answers? Do deletions have legal deadlines?
• How much changes per day, and in which formats (scans, DOCX, e-mail with attachments)?
• How is access defined: per firm, per matter, per document? How often do staffing lists change?
• Must answers be reproducible later (corpus version, model)? Peak query volume and the embedding budget?

The shape of a strong answer

• Change capture: CDC or events per document version, keyed by ID with a monotonic version; no nightly crawl. Deletes and legal-hold releases are first-class events; a hold preserves the source bytes, not their index entries.
• A parsing tier with retries: OCR, structure-aware chunks (clauses, e-mail messages), redaction, a parser version per chunk.
• Hash before you embed: document and chunk hashes; unchanged chunks keep vectors; ordering by version so replays can’t roll back; tombstones so deletes survive replays.
• Sizing out loud: 20 M documents × ~15 chunks = 300 M vectors; 1,024 float32 dimensions = 4 KB each, about 1.2 TB raw (half with halfvec, less with quantisation). If 2% of documents change daily and half their chunks are new, that is 3 M embeddings a day, about 35 per second around the clock (assumptions).
• Partition the index by tenant: a filter after an approximate scan returns too little (Step 4), so shard or partition by firm; a matter filter inside a firm still under-returns, so add iterative scans or use an engine that pre-filters. Staffing changes update metadata, never vectors.
• Two stores: a dedicated vector database or sharded pgvector for serving; a versioned corpus in the lakehouse for evaluation, re-indexing and audits.
• Evaluation as a gate: a golden question set per practice area, recall@k per release, and a blue/green index for model migrations, switched only on measured recall.

Trade-offs to say out loud

• Structure-aware chunks retrieve better but make the parser a product; fixed-size chunks are cheap and dumb.
• Per-tenant indexes isolate data and deletion; one big index with filters is cheaper and harder to keep correct.

Common weak answers

• “Re-embed everything nightly”, with no sizing and no deletion story.
• Access control in the prompt (“only answer from documents the user may see”).
• No evaluation set, so the first model change is judged by anecdotes.

Further Reading

1. pgvector README (0.8.7): HNSW, IVFFlat, filtering and iterative index scans. https://github.com/pgvector/pgvector/blob/v0.8.7/README.md
2. Feast documentation, “Point-in-time joins”. https://docs.feast.dev/getting-started/concepts/point-in-time-joins
3. Debezium documentation, “Debezium connector for PostgreSQL”. https://debezium.io/documentation/reference/stable/connectors/postgresql.html
4. Qwen3-Embedding-0.6B model card. https://huggingface.co/Qwen/Qwen3-Embedding-0.6B
5. Malkov and Yashunin, “Efficient and robust approximate nearest neighbor search using Hierarchical Navigable Small World graphs” (2016). https://arxiv.org/abs/1603.09320
6. Lewis et al., “Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks” (2020). https://arxiv.org/abs/2005.11401

Story Closing

Data science deleted the 0.715 from their slides and kept the popularity baseline: in this data, customer history predicts nothing. Support got an index as fresh as the consumer runs instead of a day behind, and an erasure that reached it in the same run.

“So the assistant can find the right tickets,” Farah said. “Can it answer ‘what was UK net revenue in Q3’ without inventing a number?”

“Not by retrieving tickets,” Ingrid said. “It needs the metric definitions, the catalog and the contracts, and permission to say no.” That is Part 33: metadata as the context an agent needs, served over MCP.

This is Part 32 of “Every Record Counts: Data Engineering for Backend Engineers” — Season 3: Real Time, Serving and AI.