Chapter 13: Batch Ingestion

"The extract that runs every night for two years and then loses four hundred rows is worse than the one that never worked, because you trusted the first one."

Overview

Batch ingestion is the least glamorous chapter in this book and the one whose lessons you will use most often.

The code is simple. SELECT some rows, write them somewhere. A working version takes twenty minutes and it will run correctly for months. The difficulty is entirely in the cases that arrive later: the row that was updated during the extract, the row that was deleted with no trace, the day the source was three minutes behind, the backfill that ran while people were querying, the schema that gained a column, and the night the job died at 40% and someone restarted it.

Every one of those has a known answer, and this chapter is the answers.

Two threads run through it. The first is the watermark — the record of where you got to — which Chapter 2 §2.3 introduced as a failure and Chapter 4's Case Study 2 turned into a six-week data loss. §13.4 finally pays that debt: the four ways updated_at lies, how to detect each, and the three properties that make a watermark safe.

The second is deletes, which is the case that breaks naive ingestion completely and which Chapter 2 §2.2 flagged and deferred. A deleted row leaves no trace; absence is indistinguishable from "not in this batch." §13.5 covers what you can do about it in batch, and is honest that the general answer is Chapter 14.

In this chapter, you will learn to:

• Choose full or incremental using the three conditions that decide it.
• Name the five change-detection strategies and the source property each requires.
• Explain the four ways updated_at lies, and detect each.
• Design a watermark that cannot lose rows.
• Handle hard deletes in a source that gives you nothing.
• Design a backfill safe to run against a table people are querying.
• Make an extract restartable, and distinguish that from idempotent.

Who needs this chapter: everyone; it is on the Quick Start path. If you build one pipeline in your career it will be this one.

13.1 The Shape of a Batch Extract

Every batch extract, regardless of source, has the same seven steps. Naming them is useful because each one has a characteristic failure.

  1. DECIDE the range      what am I extracting?          → watermark bugs (§13.4)
  2. ACQUIRE a lease       am I the only one running?     → duplicate runs
  3. READ from the source  in chunks, with a timeout      → source outages (Ch. 7 §7.3)
  4. LAND raw              exactly as received            → parse-at-landing (Ch. 9 §9.4)
  5. VERIFY                did I get what I expected?     → silent partial extracts
  6. COMMIT the range      atomically, with the data      → watermark ahead of data
  7. RECORD what happened  a manifest                     → unanswerable questions later

Steps 2, 5, and 7 are the ones people omit, and each omission is a specific class of incident.

Step 2, the lease. Two copies of an extract running simultaneously — a manual run during a scheduled one, a retry that overlapped its predecessor, two schedulers after a migration. Without a lease they interleave, and with a non-idempotent write they duplicate. A row in a metadata table with UPDATE ... WHERE state = 'idle' is sufficient.

Step 5, verification. Did I get what I expected? A row count against a source count, a non-empty check, a maximum-timestamp check. Without it, an extract that returned zero rows because of a connection reset looks identical to one that returned zero rows because nothing changed.

Step 6, atomicity. The watermark and the data must move together. If the watermark commits and the write fails, you have permanently skipped a range — and nothing errors, because the next run starts from the advanced watermark. This is the single most damaging ordering mistake in ingestion.

Step 7, the manifest. A row per run: range extracted, rows read, rows written, duration, source version, code version. It costs nothing and it is the difference between answering "what happened on the night of the 14th" in thirty seconds and not being able to answer it at all.

13.2 Full or Incremental

Full load

Read everything, replace the target.

What it gets right, and it is more than people credit: it is idempotent by construction. Run it twice, get the same result. It handles deletes for free — a row absent from the source is absent from the target. It self-heals: any past bug is corrected on the next run. And it needs no watermark, so none of §13.4 applies.

What it costs: time and load, both proportional to table size rather than to change volume.

Full loads are underrated. A 2 GB dimension table takes ninety seconds to reload and is guaranteed correct. The engineering effort to make it incremental — watermarks, deletes, reconciliation, backfill logic — is days, and the maintenance is forever.

Incremental load

Read only what changed.

What it costs, and this is the honest list, because it is longer than it looks:

• A watermark, with all of §13.4's failure modes.
• A delete strategy, because absence is not detectable (§13.5).
• A backfill mechanism, for when you find a bug (§13.10).
• Reconciliation, because the two can drift and nothing tells you.
• Idempotency, because the same range will be read twice.

The three conditions that decide it

Not "is the table big." Three specific questions:

1. Does a full load fit in the window? If reloading takes eleven minutes and you have an hour, it fits. This is the only question most tables need.

2. Does a full load put unacceptable load on the source? A full scan of a 340 GB production table is Chapter 7 §7.3's problem. This is usually what forces incremental before size does.

3. Do you need history the source does not keep? If the source overwrites and you need the prior value, you need incremental capture — or CDC.

Kestrel's answers, and note that most tables stay full:

📐 Design Decision — Default to full, graduate to incremental

The instinct on seeing a large table is to write an incremental extract. This book's position is the opposite: start full, and move to incremental only when one of the three conditions forces it.

The case against, stated fairly: full loads scale badly, and a table that is fine today will not be in two years, so you will do the work eventually and doing it twice is waste.

The case for, which wins: a full load has no watermark bugs, no delete problem, no backfill mechanism, and no reconciliation drift — four entire categories of incident that simply do not exist. Every one of the failures in this chapter is an incremental-load failure. You are not avoiding work; you are avoiding a class of bug for as long as the arithmetic permits.

What deferring costs: you will convert some tables later, under time pressure, when the window is already tight. Mitigate by measuring full-load duration on every run and alerting when it exceeds half the window — which converts a future emergency into a scheduled piece of work.

Kestrel keeps four of six tables on full loads at 2.4 million orders a year.

13.3 The Five Change-Detection Strategies

If you must go incremental, you need to know what changed. Five ways, each requiring something from the source.

Strategy 2 is the one people get wrong. Ranging over an auto-increment primary key catches inserts and silently misses every update. It is correct only for genuinely append-only tables, and "append-only" is a claim about the application that should be verified rather than assumed — Chapter 2 §2.2's question 2.

Strategy 4, full compare, deserves more credit than it gets. Extract the key and a hash of the row from source and target, compare, and act on the difference:

-- source side: cheap if the hash can be computed in the source
SELECT customer_id, md5(customers::text) AS row_hash FROM customers;

It catches everything including deletes, requires nothing from the source but a stable key, and costs a full scan of two key-and-hash columns rather than of the whole table. For a mid-size table with no reliable updated_at and no CDC access, this is frequently the right answer and it is rarely considered.

13.4 The Four Ways updated_at Lies

The debt owed since Chapter 2 §2.3.

The pattern looks unimpeachable:

SELECT * FROM orders WHERE updated_at > :watermark;

It fails in four distinct ways. Each is silent, each loses rows, and each has a detection method.

Lie 1: it is set at transaction start, not at commit

Chapter 2 §2.3's failure, and the most common.

 t=100.0  txn A begins; the app sets updated_at = 100.0
 t=100.5  txn B begins AND commits, updated_at = 100.5
 t=100.6  extract reads B, records watermark = 100.5
 t=101.0  txn A COMMITS — with updated_at = 100.0, which is < 100.5
          A is now invisible to every future extract.

Why it happens: updated_at is a value the application computes and writes; visibility is governed by commit order. Those are two different clocks (Chapter 4 §4.6).

Detection. Look for rows whose updated_at is older than a watermark you have already passed:

-- Run this daily against the source. Any result is a lost row.
SELECT COUNT(*) FROM orders
 WHERE updated_at < (SELECT watermark FROM extract_state WHERE tbl = 'orders')
   AND order_id NOT IN (SELECT order_id FROM bronze.orders);

Fix: overlap the window, or use a commit-ordered column, or CDC. §13.4's summary below.

Lie 2: it is not maintained on every write path

updated_at is set by an ORM's onUpdate hook, or by an application trigger. Anything that bypasses that path leaves it unchanged: a direct UPDATE in a migration, a bulk fix run by an engineer, a stored procedure, a second application written by a different team.

Detection. Compare a hash of the row against what you last extracted, for a sample:

-- 1,000 random rows; any mismatch means a write path is not maintaining updated_at
SELECT COUNT(*) FROM (
  SELECT s.order_id FROM orders s
    JOIN bronze.orders b USING (order_id)
   WHERE md5(s::text) <> b.row_hash
     AND s.updated_at <= b.extracted_updated_at
   ORDER BY random() LIMIT 1000
) x;

Fix: a database trigger rather than an application hook. It is a source-system change and it is worth asking for, because it is the only fix that cannot be bypassed.

Lie 3: its granularity is too coarse

A TIMESTAMP with one-second precision on a table taking hundreds of writes per second means many rows share a value. If your watermark is > last_max, you skip every other row with that same second; if it is >= last_max, you re-read them, which is fine only if your write is idempotent.

Detection:

SELECT updated_at, COUNT(*) FROM orders
 GROUP BY 1 HAVING COUNT(*) > 1 ORDER BY 2 DESC LIMIT 5;

Fix: use >= with an idempotent write, or add a tiebreaker to the ordering — (updated_at, order_id) as a composite watermark.

Lie 4: it moves backwards

Clock skew across application servers, an NTP correction, or a manual data fix with an explicit timestamp. A row written now can carry a timestamp from an hour ago.

Detection: track the maximum updated_at you have seen and alert when a row arrives below it by more than a threshold.

Fix: the overlap window absorbs small excursions. Large ones need CDC.

⚠️ Failure Mode — All four at once, and none of them alerting

The four lies are not alternatives; a real source frequently exhibits several. Kestrel's orders table had lie 1 (application-set timestamps) and lie 3 (second granularity at 6,575 orders/day bursting to 41,300 on Black Friday).

The combined loss rate was about 0.005% of rows — a few hundred a night. Too small for a row-count check to notice, and large enough that reconciliation to the cent (Chapter 1 §1.7) can never pass.

The general shape: a watermark bug produces a loss rate that is invisible to volume monitoring and fatal to reconciliation. That is why the acceptance criterion for the whole platform is reconciliation rather than "the pipeline runs" — a volume check cannot see this and a reconciliation cannot miss it.

The three properties of a safe watermark

Everything above reduces to three requirements. A watermark is safe if:

1. Its upper bound comes from an authority, not from the data read. Read now() - 30 seconds from the primary (Chapter 4's Case Study 2), and extract the half-open range (lower, upper]. Never max(updated_at) of what you happened to read.

2. Its lower bound overlaps. watermark - N minutes, where N exceeds your worst observed lag and transaction duration. This converts loss into duplicates, and duplicates are handled by an idempotent write — Chapter 4 §4.5's central trade.

3. It advances whether or not rows were found. An empty range still moves the watermark to the range's upper bound. The else now() branch that caused Chapter 4's six-week loss exists precisely because someone tried to solve this without a bounded range.

# All three properties. Compare with Chapter 7 §7.7.
upper = primary.execute("SELECT now() - interval '30 seconds'").fetchone()[0]
lower = read_watermark(table) - OVERLAP          # property 2
rows  = replica.execute(
    f"SELECT * FROM {table} WHERE updated_at > %s AND updated_at <= %s",
    [lower, upper])                              # property 1: bounded both sides
land_idempotently(rows)
write_watermark(table, upper)                    # property 3: not max(rows)

13.5 Handling Deletes

The case that breaks naive ingestion completely.

A hard-deleted row leaves no trace. It is simply absent on the next read, and "absent" is indistinguishable from "not in this batch." No timestamp strategy can detect it, because there is no row to carry a timestamp.

Four approaches, in ascending order of correctness.

1. Soft deletes. The source sets deleted_at instead of deleting. Now a delete is an update and every strategy in §13.3 catches it.

This is the right answer and it is a source-system change. Ask for it. Product teams frequently agree, because soft deletes are useful to them too — undo, audit, and accidental-deletion recovery. Chapter 17's contract is where this conversation becomes structured rather than a favour.

2. Periodic full reconciliation. Extract the full key set on a schedule, compare, and tombstone what has vanished:

-- Nightly. Cheap: one column from the source, one from the target.
INSERT INTO bronze.orders_deleted (order_id, detected_at)
SELECT b.order_id, now()
  FROM bronze.orders b
 WHERE NOT EXISTS (SELECT 1 FROM staging.order_keys s
                    WHERE s.order_id = b.order_id)
   AND b.order_id NOT IN (SELECT order_id FROM bronze.orders_deleted);

This is what most batch pipelines actually do, and it works. The costs: a full key scan of the source (usually acceptable — one indexed column), and a detection delay equal to the reconciliation interval.

3. Full loads. Deletes are handled for free. §13.2.

4. CDC. The delete is an event in the log. Chapter 14, and it is the only approach with no detection delay.

🔁 Idempotency Check — Deletes and the re-read window

A subtlety that catches people combining §13.4's overlap window with §13.5's reconciliation.

Your extract re-reads the last five minutes on every run, deliberately. Your reconciliation tombstones keys absent from the source. If the reconciliation runs between two overlapping extracts, a row can be tombstoned and then re-inserted by the overlap, resurrecting a deleted record.

Three fixes:

1. Tombstones win. The upsert checks the tombstone table and refuses to re-insert a tombstoned key. Simple, and it means an un-delete in the source is never picked up — usually acceptable and worth knowing.
2. Order the operations. Reconcile only after the extract completes, in the same orchestrated unit. Chapter 24's dependency model.
3. Version the rows. The upsert takes the row with the highest source version, and a tombstone is a version. Correct in every case and it requires a version column.

Kestrel uses 2 and 3. The general lesson: two independently correct mechanisms can be jointly wrong, and the interaction is invisible in each one's own tests.

13.6 Extracting From Files

A large share of real ingestion is files: a partner's nightly drop, an SFTP directory, an export from a system with no API.

Five problems, each with a standard answer:

1. Knowing when a file is complete. A file appearing in a directory may still be being written. Never process a file you saw appear — wait for a marker file, a rename from .tmp, or a stability check (size unchanged for N seconds). This is Chapter 9 §9.3's _SUCCESS problem in a different costume.

2. Knowing what you have already processed. A manifest table keyed by filename plus a content hash. Filename alone is insufficient — partners overwrite products.csv daily with different content, and some re-send the same file after a failure.

3. Files arriving late, out of order, or not at all. Expect 2025-11-28 and get nothing, or get it on the 30th, or get 2025-11-26 again. Alert on absence, not only on failure — a file that never arrives produces no error anywhere.

4. The format changing without notice. A column added, a header removed, an encoding changed. Chapter 11's second case study, from the receiving end. Validate the header against an expected schema and fail loudly.

5. Partial or corrupt files. Truncated at the source, or a transfer that failed. A checksum in a sidecar file is the standard answer and partners frequently provide one; if they do not, ask.

13.7 Schema Drift

The source changes shape, nobody tells you, and your extract keeps running.

This is the most common recurring annoyance in batch ingestion, and it has a clean taxonomy. Five changes, and what each does to a SELECT * extract:

Why SELECT * is both right and wrong here

SELECT * in an extract is conventionally bad practice, and in bronze it is correct — Chapter 9 §9.4's argument. You want whatever arrives; a column you did not know about is data you would otherwise have lost permanently.

The cost is that you find out about changes at the wrong time. A renamed column produces a bronze table with both cust_id (now all null) and customer_id (populated from the rename date onward), and nothing complains until a silver model notices — which may be a while, because a null-heavy column looks like a null-heavy column.

The resolution is to keep SELECT * and add a schema check beside it, rather than narrowing the projection:

def check_schema(conn, table: str) -> list[str]:
    """Compare the source's current columns against what we last saw.

    This does not FAIL the extract -- bronze must accept whatever arrives. It
    records the change and alerts, so that the drift is discovered by a person
    on the day it happens rather than by a silver model six weeks later.
    """
    current = {(r[0], r[1]) for r in conn.execute("""
        SELECT column_name, data_type FROM information_schema.columns
         WHERE table_name = %s ORDER BY ordinal_position""", [table])}
    known = read_known_schema(table)
    added, removed = current - known, known - current
    changed = {(c, t) for c, t in current
               if any(c == kc and t != kt for kc, kt in known)}
    if added or removed or changed:
        record_schema_change(table, added, removed, changed)
        alert(f"{table}: +{len(added)} -{len(removed)} ~{len(changed)} columns")
    write_known_schema(table, current)
    return sorted(c for c, _ in added)

Ten minutes to write, and it converts an entire class of six-week mystery into a same-day notification. The change is not blocked — bronze still accepts everything — it is merely announced. Chapter 17 turns the announcement into a negotiated contract; this is the version you can have this afternoon.

⚠️ Failure Mode — The type that widened, then narrowed

A source changed postal_code from INTEGER to VARCHAR(10), correctly, because Canadian and UK postcodes contain letters.

The extract kept working. Bronze accepted both. The silver model cast to INTEGER, which had been correct for three years, and the cast now failed on any non-numeric postcode — which in PostgreSQL raises, in some engines returns null, and in a permissive pipeline was configured to null on error.

Every Canadian and UK customer's postcode became null. 12% of customers. No error, because nulling on cast failure was a deliberate configuration choice made years earlier for a different reason.

Two lessons:

A type change is the dangerous one because it is the only drift that can produce wrong values rather than missing or extra columns. Additions and removals are structural and visible; a type change is semantic.

A permissive cast setting is a decision to fail silently, made once, applying forever, to situations nobody imagined. Kestrel now casts strictly in silver and routes failures to a quarantine table with the raw value and the reason — Chapter 12 §12.3's pattern, applied to types.

13.8 Parallelism and Throughput

An extract that takes four hours has a different risk profile from one that takes twenty minutes (Chapter 7 §7.3), so throughput is a correctness concern rather than only a performance one.

Four levers, in the order to reach for them.

1. Chunk by a range and run chunks concurrently. The obvious one, and it has a constraint people miss: concurrency multiplies your load on the source. Eight parallel chunks is eight connections running eight scans, and Chapter 7 §7.1's guest relationship applies. Kestrel caps extraction concurrency at 4 against the replica.

2. Choose the chunk key for even distribution. Chunking orders by order_id range gives even chunks; chunking by placed_at gives wildly uneven ones, because Black Friday's chunk is 6.28× the median. This is Chapter 4 §4.2's skew, appearing in the extract rather than in the processing — and the symptom is identical: seven chunks finish in a minute and one takes twenty.

3. Push the projection down. Even in bronze, there is a difference between SELECT * and SELECT * on a table with a 40 KB JSONB blob you do not need yet. Land what you need to be able to reprocess; you do not have to land what the source happens to store. This is a genuine qualification to §13.7's rule and it should be a deliberate, documented exception.

4. Read from a snapshot rather than a live table. Chapter 7 §7.1's third option. Complete isolation, at the cost of freshness and a restore process.

📏 Scale Note — When the extract stops fitting

The progression, and the thresholds are approximate but the order is reliable:

Symptom	Response
Full load exceeds half the window	Move that table to incremental
Incremental exceeds half the window	Parallelize chunks (cap concurrency)
Parallel extract loads the source unacceptably	Move to a dedicated replica
A dedicated replica still cannot keep up	CDC — Chapter 14
CDC's initial snapshot is the problem	Snapshot from a restored backup, then stream

Notice that the last two are not performance fixes; they are architecture changes. The first three buy you time and the fourth is where most growing platforms end up. Knowing the sequence lets you predict when you will need CDC rather than discovering it during a peak.

13.9 Extracting From Legacy Systems

The category that consumes disproportionate time and appears in no architecture diagram.

A mainframe or an ERP with no API. The answer is usually a scheduled export to a file, and then §13.6. Expect fixed-width formats, EBCDIC encoding, packed decimal fields, and a copybook that describes the layout and which someone has to find.

A vendor system whose database you can read but not touch. Read-only, no index requests, no schema visibility, and the vendor's support contract likely forbids direct access. Ask anyway, in writing — the answer is sometimes yes, and having the refusal in writing is useful when the alternative is a screen-scraper.

A system whose only interface is a report. More common than you would like. A scheduled report delivered as PDF or Excel. Parsing it is fragile and it is sometimes the only option; the mitigation is a strict schema check on every parse and a loud failure, because a layout change produces plausible garbage.

The universal advice for all three: get the data out once, land it raw, and never parse it in the extraction step. Chapter 9 §9.4's rule matters most here, because legacy formats are exactly where your parser will be wrong and you will need the original.

🏭 From the Pipeline — The Excel report that was load-bearing

A finance reconciliation depended on a weekly Excel file produced by a person, from a report in a vendor system, and emailed to a shared inbox. A pipeline parsed the attachment.

It worked for two years. Then the person went on leave, their replacement produced the file with the columns in a different order, and the parser — which read by position — silently mapped revenue into the tax column for six weeks.

Three things made this survivable in the end and none of them were in the original design:

• The file was landed raw before parsing, so all six weeks could be reprocessed. This was the only thing that saved it.
• A reconciliation eventually caught the discrepancy.
• Someone had written down that the process existed.

The fixes: parse by header name, not by position; assert the expected header set and fail loudly on a mismatch; and — the one that actually mattered — record in the catalogue that a human is a dependency of this pipeline, with a named owner and a backup.

A data pipeline's dependencies include human processes, and they are the ones missing from every diagram. Chapter 2's Case Study 1 found the same thing with a quarterly upload.

13.10 Backfills

Reprocessing historical data, and the operation with the worst incident record in this book — Chapter 1's duplicate-rows incident was a backfill.

Four rules.

1. A backfill is a different program from an incremental load, even when it shares code. It processes a bounded historical range rather than a moving window, it may run at much higher concurrency, and it must be idempotent because it will be run more than once. Chapter 1's incident was a one-off backfill promoted to a schedule without being re-reviewed as one.

2. It must be idempotent, and the mechanism must be explicit. Delete-insert the target range, merge on a key, or replace partitions. Never append. Chapter 4 §4.5.

3. It must be chunked and resumable. A backfill of two years of fct_order_item is not one transaction. Process a day at a time, record each completed chunk, and make a restart skip what is done.

4. It must be rate-limited against the source and the target. A backfill running at full speed is a self-inflicted denial of service on both. Kestrel's runs at a configurable chunk-per-minute rate, defaulting low.

💸 Cost Check — What an unthrottled backfill costs, on two meters

A backfill of two years of Kestrel's clickstream — 10.2 billion events, 682 GB of Parquet.

Unthrottled, on 40 nodes running flat out for 6 hours:

$$40 \times 6 \times \$2.400 = \$576$$

Throttled to fit alongside production, 8 nodes over 30 hours:

$$8 \times 30 \times \$2.400 = \$576$$

Identical. Node-hours are node-hours, and the throttled version costs nothing extra.

What throttling actually buys is the other meter: the unthrottled version competes with the nightly production load and the 6am SLA, and the cost of missing that is not on any invoice. It also competes for the source database's capacity (Chapter 7 §7.3).

The general point: throttling a backfill is usually free in money and valuable in risk, which makes it one of the easier decisions in this book. Run it slow unless there is a reason not to.

13.11 Restartable, and Why That Is Not Idempotent

Two properties that are frequently conflated and are genuinely different.

Idempotent: running it twice produces the same result as running it once. Restartable: if it dies partway, it can resume without redoing completed work.

You want both, and they are achieved differently:

An extract can be idempotent and not restartable — a restart redoes everything, correctly, slowly. It can be restartable and not idempotent, which is worse: the resume duplicates the chunk that was in flight when it died.

The chunk manifest is what gives you both:

CREATE TABLE extract_chunks (
    run_id      UUID        NOT NULL,
    table_name  TEXT        NOT NULL,
    chunk_lo    TIMESTAMPTZ NOT NULL,
    chunk_hi    TIMESTAMPTZ NOT NULL,
    state       TEXT        NOT NULL,   -- pending | running | done | failed
    rows_read   BIGINT,
    rows_written BIGINT,
    started_at  TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    error       TEXT,
    PRIMARY KEY (run_id, table_name, chunk_lo)
);

A restart reads this table, skips done, and re-runs running and failed — which is safe because the write is idempotent. The two properties compose: restartability decides what to redo, and idempotency makes redoing harmless.

Negotiating with the source team

Three of §13.4's four lies have a fix that costs the source team something, which makes them negotiations rather than engineering problems. This section is about the negotiation, because it is the part that determines whether you spend a year working around a problem or a week having it fixed.

What you are actually asking for, ranked by what it costs them:

ask                                  their cost         your benefit
────────────────────────────────────────────────────────────────────────
tell us before you change the shape  a habit            enormous (ch 17)
a read-only replica we can use       capacity they      removes §7.1's
                                       may already have   biggest risk
a soft-delete column instead of
  a hard delete                      one migration,     removes §13.5's
                                       one code path      hardest problem
an updated_at set at COMMIT, not
  at statement time                  a trigger, and a   removes §13.4's
                                       write per row      lie 1 entirely
a monotonic version or sequence
  column                             a column and an    removes the
                                       index               watermark problem
logical replication enabled          a config change    CDC becomes
                                       and a RESTART      possible (ch 14)

Two observations about that table change how the conversation goes.

The cheapest ask is the most valuable one, and it is not technical. "Tell us before you change the shape" costs a habit and removes the class of problem that Chapter 17 spends a whole chapter on.

And the expensive asks are expensive in their currency, not yours. A trigger on updated_at is one migration to you and a write-amplification decision to them — Chapter 12 §12.5's trigger-based CDC cost, arriving as somebody else's throughput. Asking without knowing that is how a reasonable request becomes an adversarial one.

How to ask

Lead with what breaks, not with what you want. "We need a soft-delete column" is a demand about their schema. "When an order is cancelled and the row is deleted, our warehouse keeps showing it as revenue — here is the number it was wrong by last month" is a problem they can evaluate.

Bring the consumer list (Chapter 17 §17.4). The producer cannot know who reads them; this is the one piece of information you have and they do not, and handing it over changes the relationship more than any amount of process.

Offer the cheapest version first, and say it is the cheapest. "The ideal fix is a commit-time timestamp. The version that costs you nothing is telling us your longest transaction duration, and we will size our overlap from it." A request with a fallback gets a yes far more often than a request without one, and the fallback is usually good enough.

And write down what you agreed, where they can see it. §17.8's observed contract is exactly this artifact — a record of what you believe about their system, visible to them, correctable by them, and not claiming to be a promise.

When the answer is no

It frequently is, and it is frequently correct. A team under delivery pressure declining a schema change to help an analytics pipeline has made a defensible prioritisation.

What you do then is the whole of Part III: overlap the window (§13.4), full-compare for deletes (§13.5), reconcile against a total (Exercise 13.19), and write down that the mitigation is a mitigation — so that when the residual failure happens, it is a known accepted risk rather than a surprise.

The one thing not to do is to work around it silently. A workaround nobody has recorded becomes, within a year, a piece of unexplained complexity that the next engineer will remove.

🎓 Interview Angle — "how would you extract from a production database?"

A near-universal question, and the answer people give is a tool name. Airbyte, Fivetran, a Python script. That answers "with what" and the question is "how."

The strong answer starts with the relationship and ends with the check:

"First I'd want to know whether I'm reading a primary or a replica, and what the replica's lag looks like — because a watermark against a lagging replica loses rows permanently. Then I'd ask whether updated_at is set at commit or at statement time, and whether deletes are hard or soft. Those two answers decide whether a watermark works at all or whether I need CDC. I'd chunk the read so I'm not holding a snapshot for an hour, land it raw and idempotently, and — the part that matters — reconcile row counts and one measure against the source daily, with a check for a persistent same-signed variance rather than just a tolerance."

Four things that answer does that a tool name does not. It names the failure it is avoiding (permanent row loss) rather than the mechanism. It asks two questions of the source instead of assuming. It mentions the snapshot, which tells the interviewer you have been shouted at by a DBA. And it ends with the reconciliation, which is the thing most candidates never mention and every interviewer is waiting for.

The follow-ups, and what they are testing:

"What if there's no updated_at?" — full compare, CDC, or a version column. Naming the trade (full compare costs a full read every run) is the mark.

"How do you handle deletes?" — this is the question. Soft deletes if you can negotiate them, full compare if you cannot, CDC if you can have it. A candidate who says "we soft-delete" without noting that this requires the source team's agreement has not done it.

"How would you know if it silently stopped working?" — the reconciliation, and specifically the sign test. "We alert if the row count is zero" is the answer that reveals the gap, because the failures in this chapter all produce a non-zero, plausible count.

One thing worth volunteering if it does not come up: the extract needs to be able to say what it did — rows read, chunks, duration, connections — because when the database is slow, the unusual workload gets blamed (§7.1), and a pipeline with no record cannot defend itself.

🧪 Try It — break your own watermark, four ways

bash cd part-03-ingestion/chapter-13-batch-ingestion/code python watermark.py --self-check

Then reproduce each of §13.4's four lies against the seeded database, and record what the extract reports in each case. The point is that it reports success every time.

```sql -- LIE 1: a long transaction. In one session: BEGIN; UPDATE orders SET status = 'paid' WHERE order_id = 88214; -- ... leave it open, run the extract in another session, THEN commit

-- LIE 2: an update that bypasses the trigger UPDATE orders SET channel = 'ios' WHERE order_id = 88215; -- with updated_at deliberately not touched

-- LIE 3: a timestamp without an offset, interpreted twice SET timezone = 'America/New_York'; INSERT INTO orders (..., updated_at) VALUES (..., '2026-11-02 01:30:00'); -- -- a time that occurs TWICE on a DST fall-back day

-- LIE 4: the future UPDATE orders SET updated_at = now() + interval '3 days' WHERE order_id = 88216; ```

Then run the extract and record, for each:

text lie rows the extract read rows it SHOULD have read did it fail? 1 ____ ____ no 2 ____ ____ no 3 ____ ____ no 4 ____ ____ no

Lie 4 is the one to sit with. One row, three days in the future, and the watermark is now three days ahead of reality — so the extract reads nothing, every night, until the world catches up. Zero rows is not an error and the job exits successfully.

Then add the two-line guard — refuse to store a watermark greater than now() — and re-run. That guard is worth more than everything else in this callout, and it is two lines.

🧭 Version Note — the managed connector changed what "build an extractor" means

A decade ago every team wrote this chapter's code. Now a managed connector — Fivetran, Airbyte, a cloud provider's ingestion service — will do a watermark extract from most common sources in an afternoon, and it is frequently the right choice.

What has genuinely changed:

text the connector handles you still own ────────────────────────────────────────────────────────────────────── pagination, retries, auth whether updated_at lies (§13.4) schema drift detection what to do when it drifts incremental state the RECONCILIATION (Exercise 13.19) -- always parallelism and chunking the source team relationship (§7.1) a UI and a status page the fact that green is not correct

The right-hand column is this chapter. A managed connector removes the code and none of the reasoning, and its status page is exactly the kind of green that Chapter 1's incident was green in.

Three specific things a managed connector does not do, and each has produced an incident somewhere:

It does not know your source's transaction semantics. It reads updated_at because you told it to, and §13.4's first lie applies unchanged.

It does not reconcile. Row counts against the source, at the day grain, with a sign test — nobody ships that, and it is the acceptance criterion (§1.7).

And its per-row pricing changes the design. A connector billed per row makes a full compare (§13.3) or a wide re-fetch window (§16.8) a cost decision rather than a correctness one, which is the wrong basis and is the one you will be given.

The honest recommendation: buy the connector, and write the reconciliation yourself. That is the split that survives — the mechanical work is genuinely commodity, and the part that decides whether your numbers are right was never in the connector.

13.12 The Kestrel Extractor

Everything assembled.

"""platform/ingest/batch/extract_postgres.py

Watermarked, chunked, restartable, idempotent extraction from kestrel_app.
Chapter 7 §7.7 gave the shape; this is the version that goes in the platform.
Chapter 24 schedules it.
"""
from __future__ import annotations

import os
import uuid
from datetime import datetime, timedelta, timezone

import psycopg

CHUNK = timedelta(hours=1)
OVERLAP = timedelta(minutes=5)          # property 2 of a safe watermark
STATEMENT_TIMEOUT_MS = 30 * 60 * 1000
MAX_REPLICA_LAG_S = 60


def safe_upper_bound(primary_dsn: str) -> datetime:
    """Property 1: the upper bound comes from an AUTHORITY, not from the data.

    The replica's clock and the replica's applied position are different
    things. Recording a position from a lagging replica silently skips whatever
    arrives in the lag window -- six weeks of orders, Chapter 4 Case Study 2.
    """
    with psycopg.connect(primary_dsn, connect_timeout=5) as c:
        return c.execute("SELECT now() - interval '30 seconds'").fetchone()[0]


def check_replica_lag(replica_dsn: str) -> float:
    """Refuse to run when the replica is behind. Failing loudly beats
    succeeding incorrectly -- Chapter 2 Case Study 1."""
    with psycopg.connect(replica_dsn, connect_timeout=5) as c:
        lag = c.execute(
            "SELECT COALESCE(EXTRACT(EPOCH FROM "
            "  (now() - pg_last_xact_replay_timestamp())), 0)").fetchone()[0]
    if lag > MAX_REPLICA_LAG_S:
        raise ReplicaTooFarBehind(
            f"replica is {lag:.0f}s behind (limit {MAX_REPLICA_LAG_S}s). "
            f"Refusing to extract: a watermark recorded now would skip rows.")
    return lag


def extract(table: str, primary_dsn: str, replica_dsn: str) -> dict:
    run_id = uuid.uuid4()
    if not acquire_lease(table, run_id):        # step 2: am I the only one?
        raise AlreadyRunning(table)

    try:
        check_replica_lag(replica_dsn)
        upper = safe_upper_bound(primary_dsn)
        lower = read_watermark(table) - OVERLAP

        chunks = plan_chunks(lower, upper, CHUNK)
        record_chunks(run_id, table, chunks)     # step 7: the manifest, up front

        totals = {"read": 0, "written": 0}
        for lo, hi in chunks:
            if chunk_done(run_id, table, lo):    # restartable
                continue
            mark(run_id, table, lo, "running")
            try:
                rows = read_chunk(replica_dsn, table, lo, hi)
                written = land_idempotently(table, rows, lo, hi)  # idempotent
                mark(run_id, table, lo, "done",
                     rows_read=len(rows), rows_written=written)
                totals["read"] += len(rows)
                totals["written"] += written
            except Exception as exc:
                mark(run_id, table, lo, "failed", error=str(exc))
                raise

        verify(table, lower, upper, totals)      # step 5: did I get what I expected?

        # Step 6: the watermark and the data move together. If this ordering is
        # reversed and the write fails, the range is PERMANENTLY skipped and
        # nothing errors.
        write_watermark(table, upper)            # property 3: the RANGE's bound
        return totals
    finally:
        release_lease(table, run_id)

🧱 Kestrel Platform — Increment 13: the batch extractor

(a) Implement extract_postgres.py in full, with all seven steps from §13.1. The helper functions are yours to write; the structure above is the contract.

(b) Add the extract_state and extract_chunks tables, and the lease.

(c) Prove restartability. Run the extractor, kill it at roughly 50% (a SIGKILL, not a clean shutdown), restart it, and confirm from extract_chunks that it skipped completed chunks. Record how many it skipped.

(d) Prove idempotency. Run the full extract twice. Assert that the bronze row count and the revenue sum are identical after both runs. These are two different tests and you need both — §13.9.

(e) Prove the lag guard works. Simulate replica lag (pause replication, or stub check_replica_lag) and confirm the extractor refuses to run and says why. Record the error message; a guard whose message does not explain itself gets disabled by the next person.

Parts (c), (d), and (e) are the exercise. Writing the extractor is the easy half.

13.13 Summary

Every batch extract has seven steps, and the three people omit each have a signature failure: the lease (two runs interleaving), verification (an empty result from a connection reset looking identical to an empty result from no changes), and the manifest (being unable to answer what happened on the night of the 14th). And the ordering that matters most: the watermark and the data must commit together — if the watermark advances and the write fails, the range is permanently skipped and nothing errors.

Default to full loads and graduate to incremental. A full load is idempotent by construction, handles deletes for free, self-heals past bugs, and needs no watermark — which means four entire categories of incident do not exist. Three conditions force incremental: the full load no longer fits the window, it puts unacceptable load on the source, or you need history the source does not keep. Kestrel keeps four of six tables on full loads at 2.4 million orders a year, and alerts when a full load exceeds half its window, which turns a future emergency into scheduled work.

Five change-detection strategies, and the two under-considered ones are worth naming. Ranging over a monotonic id catches inserts and silently misses every update — correct only for genuinely append-only tables, which is a claim to verify rather than assume. And full compare — key plus row hash, both sides — catches everything including deletes, needs nothing but a stable key, and is frequently the right answer for a mid-size table with no reliable updated_at and no CDC.

updated_at lies in four ways, each silent and each losing rows: it is set at transaction start rather than commit; it is not maintained on every write path; its granularity is too coarse; and it moves backwards. A real source frequently exhibits several at once, producing a loss rate — about 0.005% at Kestrel — that is invisible to volume monitoring and fatal to reconciliation. That asymmetry is why the platform's acceptance criterion is reconciliation.

Three properties make a watermark safe: its upper bound comes from an authority rather than from the data read; its lower bound overlaps by more than the worst lag and transaction duration, trading loss for duplicates that an idempotent write absorbs; and it advances whether or not rows were found.

Hard deletes leave no trace, so absence is indistinguishable from not-in-this-batch. Four approaches: soft deletes (the right answer, and a source-system change worth asking for), periodic full key reconciliation (what most batch pipelines actually do, with a detection delay equal to the interval), full loads, and CDC. And the interaction that catches people: an overlap window and a tombstone reconciliation can resurrect a deleted row — two independently correct mechanisms, jointly wrong, invisible in each one's own tests.

File ingestion has five standard problems: knowing a file is complete (never process one you saw appear), knowing what you have already processed (filename alone is insufficient — hash the content), files arriving late or never (alert on absence, which produces no error anywhere), the format changing, and corruption.

Legacy sources share one rule: get the data out once, land it raw, never parse in the extraction step — because legacy formats are exactly where your parser will be wrong. And a pipeline's dependencies include human processes, which appear in no diagram; record them in the catalogue with a named owner and a backup.

Backfills have the worst incident record in this book. A backfill is a different program from an incremental load even when it shares code, must be explicitly idempotent, must be chunked and resumable, and must be rate-limited. Throttling is free in money — node-hours are node-hours — and valuable in risk, which makes it one of the easier decisions here.

Restartable and idempotent are different properties. Restartability decides what to redo; idempotency makes redoing harmless. A chunk manifest gives you both, and you need both.

What's next

Chapter 14 is change data capture: reading the database's own replication log instead of querying its tables. It solves the deletes problem outright, removes the watermark entirely, and captures every intermediate state rather than the value at poll time. It also introduces an operational burden that Chapter 7 §7.8 has already warned about — a replication slot whose consumer stops will fill a disk and stop the storefront — and the chapter is honest about the trade.