Database Ingestion and Change Data Capture

In plain terms

An operational database is often an important source for analytics, while its first responsibility is serving the application. It holds the orders, the customers, and the subscriptions as they are right now, it is busy serving the application, and a badly written extraction can slow the checkout for every customer at once. Copying from it is therefore a negotiation: how much of its attention the platform may take, how far behind the copy may be, and what the copy is allowed to miss.

Three common approaches are full snapshots, timestamp-based increments, and log-based CDC. Read the whole table and replace the copy. Ask for rows that changed since last time. Or consume the database’s supported change stream and apply the captured changes. Recovery logs need engine-specific decoding or a separate replication log; their raw contents are not a ready-made table of changes. The pipeline patterns article named these as full snapshot, watermark, and change data capture and gave the table that chooses between them. This article is about how each one actually works, what each one silently misses, and what has to be built around it so that the copy can be trusted. The lab builds all three against a small database with a change log of its own and measures the misses.

What a source database gives you to work with

Four capabilities to inspect before choosing an ingestion design.

• A stable row identity. A primary key or another enforced unique key lets updates and deletes find the same row. Capture without a primary key is sometimes possible with an engine-specific replica identity or full row image, but duplicate rows and changing keys make application harder. Define how key changes remove the old identity.
• A modification timestamp, usually updated_at, maintained by the application. It is what a watermark extraction filters on, and its reliability is exactly the application’s discipline in setting it.
• Consistent reads. A transaction-wide snapshot requires suitable isolation and read semantics. Under READ COMMITTED, successive statements can see different committed states even within one transaction.
• A supported change stream. PostgreSQL logical decoding derives changes from WAL. MySQL CDC commonly reads the binary log, which is distinct from InnoDB’s recovery redo log. Configure the required logging format, captured tables, and row images; a physical recovery log is not already a universal stream of complete committed rows. See the MySQL documentation for the binary log and redo log.

A read replica can move scan CPU and I/O away from the primary, but replication and long reads still have costs. Record observed replay lag and source identity; lag has no universal seconds-to-minutes bound. Logical CDC from a replica depends on engine version, topology, connector support, and failover configuration. Choose the source with the database operator rather than assuming every extraction can use the same replica.

Full snapshots, done properly

A snapshot reads the whole table and replaces the copy. It can suit affordable full reads of small or reference tables, initialize a copy, and provide an aligned reference for reconciliation. Done carelessly it is also the mechanism most likely to hurt the source, so three details matter.

One shared snapshot. Use the engine’s snapshot semantics for all chunks. In PostgreSQL REPEATABLE READ, the first non-transaction-control statement establishes the snapshot, not simply BEGIN; READ COMMITTED takes a new snapshot per statement. In SQLite WAL mode, a deferred read transaction pins its snapshot on the first read. Record snapshot identity or source position and capture start/end times; a wall-clock label alone is not an exact database boundary. See PostgreSQL isolation and SQLite isolation.

Parallel readers over a split column. Disjoint ranges must cover the snapshot’s full key domain, including a separate rule for NULL values. Use indexed, stable columns and measured row distributions; quantile-based ranges or hash buckets can help with skew. Independent worker transactions do not automatically share a snapshot. PostgreSQL can export a snapshot for workers to import before reading, while the exporting transaction remains open. Without engine-supported coordination, describe the output as a moving extract, not a point-in-time snapshot. The lab only counts proposed ranges and reads them sequentially on one connection. PostgreSQL documents snapshot synchronization. Sqoop’s boundary-query option supplies two outer bounds, not an arbitrary list of quantile edges.

A source load budget. Agree connection limits, query timeouts, scan rates, and abort thresholds using measurements on the actual source. Four workers and a ten-minute timeout are example settings, not universal safe values. Long snapshots can delay version cleanup or WAL checkpoint progress and increase storage pressure; monitor this along with application latency and replica lag.

Watermarks, and the rows they miss

A watermark extraction reads rows whose updated_at is later than the highest value seen last time. With a selective indexed timestamp and stable key, it can reduce scan cost; it also requires a reliable timestamp contract. Three failure modes need explicit handling.

Commit lag. A row’s updated_at is set by the application when it decides to write, and the row becomes visible when the transaction commits. Between the two can lie seconds, or for a batch job that opened its transaction at 23:00 and committed at 00:02, an hour. An extraction that ran at 23:59 could not see the row and set its watermark past 23:00. The next run asks for updated_at later than the watermark, and the row, whose timestamp is 23:00, is never returned again. That update remains missed until another qualifying write or reconciliation repairs the row. The same happens on a smaller scale whenever many rows share one timestamp at the watermark boundary and some of them commit after the read.

Overlap re-reads from before the last watermark and merges the selected rows again. Its bound must account for late commits, replica visibility, timestamp granularity, and clock error; without a measured bound it is a mitigation, not a guarantee. It cannot capture an intermediate state that has already been overwritten. In this lab a three-hour overlap repairs 14 stale rows, but timestamps can move backward: the serial loader replaces by extraction order, without an updated_at guard. Such a guard would reject the very backdated correction being recovered. Use a trustworthy version or serialized, ordered runs, and reconcile remaining gaps.

Deletes. A deleted row has no updated_at to filter on because it has nothing at all. Overlap alone cannot detect the removal. Use captured deletes, a reliable soft-delete field, or reconciliation.

Writes that skip the timestamp. A bulk correction run by a DBA in SQL, a column changed by a trigger, an ORM path that forgot the field. The row changed and updated_at did not. These rows are stale in the copy until a snapshot happens to replace them, and there is no way to count them from the copy alone.

Store progress only after durable output, preferably in the same destination transaction. Use a fixed extraction snapshot or upper bound and keyset pagination: ORDER BY (updated_at, primary_key) must be paired with a predicate that resumes after that pair. Sorting alone does not implement a cursor. A tie-break key prevents pagination skips within a stable read; it does not fix a row that commits late behind the cursor. Re-read the overlap on subsequent runs, and leave the watermark unchanged for an empty batch.

Change data capture: reading the log

Log-based CDC decodes a configured source change stream. Debezium is one implementation, commonly deployed with Kafka Connect. The simplified payload below illustrates an update; field availability and ordering metadata depend on the connector and source configuration.

{
  "op": "u",
  "source": {"table": "orders", "lsn": 8813420, "txId": 55021, "ts_ms": 1772937720000},
  "before": {"order_id": 1746, "status": "paid",    "amount_cents": 8522, "updated_at": "2026-03-06T23:40:16Z"},
  "after":  {"order_id": 1746, "status": "shipped", "amount_cents": 8522, "updated_at": "2026-03-06T23:00:00Z"}
}

Operations distinguish creation, update, deletion, and snapshot reads. Do not assume one LSN uniquely identifies every row event or orders independent databases and Kafka partitions. Follow the connector’s source-offset and transaction-order contract; preserve per-key order and buffer transaction boundaries when atomic multi-row visibility is required. Before-images can be partial. PostgreSQL replica identity settings affect what is available. The lab’s lsn is instead a unique integer sequence in one serialized trigger log.

CDC can retain captured deletes and intermediate changes that periodic row reads miss. It still consumes decoding CPU, network bandwidth, and log storage, and usually needs an initial snapshot. Latency and completeness depend on capture scope, retained logs, and operation. Monitor both connector progress and downstream consumer progress: when a connector writes to a broker, a database slot can advance before the analytics table catches up. Retention exhaustion requires an explicit recovery path rather than silently skipping ahead.

A trigger log writes capture records in the source transaction, adding write cost and retention work. Its sequence is not automatically commit order on a database with concurrent writers: transaction A may allocate ID 10, B commit ID 11, and A commit later. Polling above 11 would miss A. This SQLite lab has one writer and transactional trigger rows, so a visible maximum sequence is safe under its restricted assumptions. Triggers record statement-time information, not automatically a database commit timestamp.

Joining snapshot and stream. Establish a consistent source snapshot and a corresponding change-stream boundary, retain changes while scanning, publish the complete snapshot, then apply the retained suffix without gaps. A snapshot and a log position sampled independently can miss or double changes. Use the connector’s supported protocol, not a wall-clock guess. PostgreSQL documents exported snapshots tied to a logical slot. Exercise 1 can use a visible maximum trigger ID only because its single-writer log is transactional. Do not generalize that shortcut to concurrent sequence allocation.

Do not assume every after-image is complete. With some PostgreSQL replica-identity settings, Debezium uses an unavailable-value placeholder for an unchanged TOAST value. That placeholder is not SQL NULL: preserve the prior value under the connector’s rules or stop if the required prior state is unavailable. Blind whole-row replacement can corrupt a previously correct copy. The lab’s small trigger payloads deliberately contain complete after-images.

A logical replication slot retains PostgreSQL decoding state independently of a client connection. Its progress does not establish that the final analytical destination is caught up.

Applying changes at the destination

A stream of change records is not a table. Turning it into one is a merge, in position order, with three decisions made in advance.

Current state. Creates and updates replace a key with the captured after-image; deletes remove it. Replaying an old segment over newer state can temporarily or permanently regress rows or resurrect deleted ones. Commit ordered batches and their checkpoint atomically, or use source-version guards with retained delete tombstones. A checkpoint is the largest fully applied contiguous source prefix, not just the largest position observed. Keep only the last event per key for a current-state batch when full after-images are available; retain earlier events separately when history is required.

Deletes. Choose physical deletion or a filtered deleted flag per table and consumer. A technical delete is not automatically a refund or permission to retain personal history forever. Define retention and deletion propagation for raw logs, replicas, and history. If per-key version guards are used, keep enough deletion metadata to reject older replays without retaining unnecessary payloads.

History. First decide whether the question is about source observation time or business effective time. CDC is useful for captured state transitions, but a backdated business change may take effect before its commit. Use effective dates for that question, and source positions for deterministic capture order. The lab builds half-open [valid_from, valid_to) intervals using a simulated recorded_at, not true commit or business-effective timestamps. Same-time changes need an explicit position tie-break; writes within one transaction need a policy about whether intermediate states should be visible. Deduplicate captured events before maintaining persistent history.

Reconciliation: checking the copy

Every mechanism above can drift from the source without any error being raised: a watermark that skipped a transaction, a connector that restarted from the wrong position, a delete nobody captured. The copy needs scheduled reconciliation against the source, and the check has to be cheap enough to run often.

Chunked checksums compare counts and hashes over the same key ranges, columns, and source boundary. Matching hashes provide evidence, not mathematical proof of equality. Computing hashes still reads the covered data unless maintained summaries already exist; the main saving can be network transfer and targeted repair. Include the union of source and destination key domains so a deleted highest key or an empty source does not hide stale destination rows. Align snapshots or reconcile at a known applied CDC boundary; comparing a live source with a lagging target confuses normal lag with drift.

Canonicalize column order, NULL markers, decimal scale, timestamp zone, and row ordering before hashing. The lab uses a full SHA-256 digest over deterministic JSON for a fixed integer/text schema and confirms repaired dictionaries with exact equality. A Python serialization recipe is not automatically identical to a warehouse SQL hash expression; test both implementations with shared fixtures.

Schema changes

Schema evolution needs an explicit contract for each ingestion path. Do not assume CDC sends DDL events before changed rows: the Debezium PostgreSQL connector does not report DDL change events, and before-image availability depends on replica identity. Capture permitted additions only when storage and readers support them; a new field may also contain sensitive data. Detect incompatible changes, preserve recoverable input, stop affected publication and progress, and coordinate migration. For documents, profile missing fields and types; flattening nested arrays needs an explicit grain and stable child key.

Failure and recovery

Recovery depends on durable output, a checkpoint that never skips unapplied input, and replay behavior suited to the destination. Snapshot replacement, current-state merge, and history append need different implementations. With separate data and progress stores, failed acknowledgements can cause retries; make those retries safe before advancing the checkpoint. Exercise 6 demonstrates local transactional recovery, not a universal exactly-once guarantee.

Choosing per table

The subscription commerce company’s Postgres database has a few hundred tables, and three of them show the whole decision.

Choose a watermark when current-state copies, timestamp behavior, visibility delay, and deletion lag meet the consumer’s requirements. Choose CDC when the supported capture stream is needed for deletes or intermediate history and its operational cost is acceptable. Choose snapshots when a complete consistent read is affordable or for initialization. Design for known failure modes before launch, then adjust bounds and cadence from measurements.

Anti-patterns

• Extracting from the primary at peak. The nightly job that became an hourly job that now runs at 09:00 against the checkout database.
• Chunks outside a transaction. A “snapshot” whose first and last chunks are forty minutes apart, with no as_of anyone could state.
• Splitting on a skewed column. Four workers, one of them doing half the work, the job timed by the slowest.
• The strict watermark. updated_at > last_seen with no overlap, missing writes that become visible behind the saved cursor.
• Ordering CDC by timestamp. Records applied in updated_at order because it “looked right”, leaving rows in stale states.
• An unwatched replication slot. The connector stopped in March; the source disk filled in April.
• Retrying by appending. Chunk three failed, the retry ran all four again on top of the two that succeeded.
• No reconciliation. The copy has drifted for a year and the first sign is a finance number that does not match.

Lab

The lab uses SQLite WAL mode, one writer, and a trigger log with partial before-images, complete after-images, recorded_at, and a unique integer lsn. It does not decode WAL or simulate concurrent committing transactions. Each day draws 345 random operations, then schedules three backdated updates for the next day. recorded_at is a monotonically ordered simulated statement clock; updated_at can move backward. All formatted timestamps represent UTC. This models delayed visibility by deferring writes, not by holding a real transaction open across midnight. A seven-day run still has three updates pending for day eight. Exercise 1 demonstrates a snapshot/log handoff; exercise 6 uses a file-backed target for rollback and reopen recovery. Run setup once; each exercise uses src or creates its own source.

import hashlib, json, random, sqlite3, tempfile
from pathlib import Path
from datetime import datetime, timedelta

SCHEMA = """
CREATE TABLE orders (order_id INTEGER PRIMARY KEY, customer_id INTEGER, status TEXT, amount_cents INTEGER, updated_at TEXT);
CREATE TABLE clock (now TEXT);
INSERT INTO clock VALUES ('2026-03-01 00:00:00');
CREATE TABLE change_log (lsn INTEGER PRIMARY KEY AUTOINCREMENT, op TEXT, order_id INTEGER, before TEXT, after TEXT, recorded_at TEXT);
CREATE TRIGGER log_insert AFTER INSERT ON orders BEGIN
  INSERT INTO change_log (op, order_id, before, after, recorded_at)
  VALUES ('insert', NEW.order_id, NULL, json_object('customer_id', NEW.customer_id, 'status', NEW.status,
          'amount_cents', NEW.amount_cents, 'updated_at', NEW.updated_at), (SELECT now FROM clock));
END;
CREATE TRIGGER log_update AFTER UPDATE ON orders BEGIN
  INSERT INTO change_log (op, order_id, before, after, recorded_at)
  VALUES ('update', NEW.order_id, json_object('status', OLD.status, 'updated_at', OLD.updated_at),
          json_object('customer_id', NEW.customer_id, 'status', NEW.status, 'amount_cents', NEW.amount_cents,
          'updated_at', NEW.updated_at), (SELECT now FROM clock));
END;
CREATE TRIGGER log_delete AFTER DELETE ON orders BEGIN
  INSERT INTO change_log (op, order_id, before, after, recorded_at)
  VALUES ('delete', OLD.order_id, json_object('status', OLD.status, 'updated_at', OLD.updated_at), NULL, (SELECT now FROM clock));
END;
"""

def stamp(t):
    return t.strftime("%Y-%m-%d %H:%M:%S")

class Source:
    """The shop's orders database, with a trigger-maintained change log standing in for the transaction log."""

    def __init__(self):
        self.directory = tempfile.TemporaryDirectory()
        self.path = self.directory.name + "/shop.db"
        self.con = sqlite3.connect(self.path, isolation_level=None)
        self.con.execute("PRAGMA journal_mode=WAL")
        self.con.executescript(SCHEMA)
        self.rng, self.next_id, self.slow_commits = random.Random(0), 1, []

    def close(self):
        self.con.close()
        self.directory.cleanup()

    def query(self, sql, params=()):
        return self.con.execute(sql, params).fetchall()

    def run_day(self, day):
        """345 random operations plus deferred backdated updates; one writer only."""
        rng, con = self.rng, self.con
        pending, self.slow_commits = self.slow_commits, []
        def apply_pending():
            for order_id, status, updated_at in pending:
                con.execute("UPDATE clock SET now = ?", (stamp(day + timedelta(minutes=2)),))
                con.execute("UPDATE orders SET status = ?, updated_at = ? WHERE order_id = ?",
                            (status, updated_at, order_id))
            pending.clear()
        for second in sorted(rng.sample(range(86400), 345)):
            if second >= 120:
                apply_pending()
            now = day + timedelta(seconds=second)
            con.execute("UPDATE clock SET now = ?", (stamp(now),))
            roll = rng.random()
            if roll < 0.87 or self.next_id == 1:
                customer = 1 if rng.random() < 0.3 else rng.randint(2, 200)
                con.execute("INSERT INTO orders VALUES (?, ?, 'paid', ?, ?)",
                            (self.next_id, customer, rng.randint(500, 9900), stamp(now)))
                self.next_id += 1
            elif roll < 0.985:
                target = rng.randint(max(1, self.next_id - 600), self.next_id - 1)
                con.execute("UPDATE orders SET status = ?, updated_at = ? WHERE order_id = ?",
                            (rng.choice(["shipped", "refunded"]), stamp(now), target))
            else:
                con.execute("DELETE FROM orders WHERE order_id = ?", (rng.randint(max(1, self.next_id - 600), self.next_id - 1),))
        apply_pending()
        for _ in range(3):
            target = rng.randint(max(1, self.next_id - 300), self.next_id - 1)
            self.slow_commits.append((target, "shipped", stamp(day + timedelta(hours=23))))

def build_source(days):
    source = Source()
    for k in range(days):
        source.run_day(first_day + timedelta(days=k))
    return source

def checksum(rows):
    ordered = sorted(rows, key=lambda row: row[0])
    payload = json.dumps(ordered, ensure_ascii=False, separators=(",", ":"))
    return len(rows), hashlib.sha256(payload.encode()).hexdigest()

COLUMNS = "order_id, customer_id, status, amount_cents, updated_at"
first_day = datetime(2026, 3, 1)
src = build_source(7)
print(src.query("SELECT count(*), min(order_id), max(order_id) FROM orders")[0])
print(src.query("SELECT op, count(*) FROM change_log GROUP BY op ORDER BY op"))
print("largest customer's share of orders: {:.0%}".format(src.query("SELECT 1.0 * sum(customer_id = 1) / count(*) FROM orders")[0][0]))
# (2010, 2, 2052)
# [('delete', 42), ('insert', 2052), ('update', 334)]
# largest customer's share of orders: 29%

Seven days leave 2,010 live orders out of 2,052 created, and the change log holds every one of the 2,428 writes that produced them, including the 42 deletes that the orders table itself no longer shows.

1. Plan splits and read a consistent snapshot. Split the table into four ranges on order_id and again on customer_id, and count the rows each reader would get. Then take a snapshot inside one read transaction while another day of business runs against the same file, and compare the snapshot’s row count with the source’s afterwards.

def split_ranges(query, column, parts):
    if column not in {"order_id", "customer_id"}:
        raise ValueError("unsupported split column")
    if type(parts) is not int or parts < 1:
        raise ValueError("parts must be a positive integer")
    if query(f"SELECT count(*) FROM orders WHERE {column} IS NULL")[0][0]:
        raise ValueError("NULL split keys require a separate partition")
    lo, hi = query(f"SELECT min({column}), max({column}) FROM orders")[0]
    if lo is None:
        return []
    count = min(parts, hi - lo + 1)
    edges = [lo + i * (hi - lo + 1) // count for i in range(count + 1)]
    return [(a, b - 1) for a, b in zip(edges, edges[1:])]

for column in ("order_id", "customer_id"):
    sizes = [src.query(f"SELECT count(*) FROM orders WHERE {column} BETWEEN ? AND ?", r)[0][0]
             for r in split_ranges(src.query, column, 4)]
    print(f"split on {column:11}: chunks of {sizes}, largest share {max(sizes) / sum(sizes):.0%}")

busy = build_source(7)
reader = sqlite3.connect(busy.path, isolation_level=None)
reader.execute("BEGIN")
# First read pins SQLite's snapshot; this label is a simulated source clock.
snapshot_label = reader.execute("SELECT now FROM clock").fetchone()[0]
boundary = reader.execute("SELECT coalesce(max(lsn), 0) FROM change_log").fetchone()[0]
query = lambda sql, params=(): reader.execute(sql, params).fetchall()
ranges = split_ranges(query, "order_id", 4)
busy.run_day(first_day + timedelta(days=7))
snapshot = [reader.execute(f"SELECT {COLUMNS} FROM orders WHERE order_id BETWEEN ? AND ?", r).fetchall()
            for r in ranges]
reader.execute("COMMIT")
reader.close()
print("snapshot label", snapshot_label, "holds", sum(map(len, snapshot)), "rows; current source holds",
      busy.query("SELECT count(*) FROM orders")[0][0])
copy_state = {row[0]: row for chunk in snapshot for row in chunk}
for lsn, op, order_id, after in busy.query(
        "SELECT lsn, op, order_id, after FROM change_log WHERE lsn > ? ORDER BY lsn", (boundary,)):
    if op == "delete":
        copy_state.pop(order_id, None)
    else:
        a = json.loads(after)
        copy_state[order_id] = (order_id, a["customer_id"], a["status"], a["amount_cents"], a["updated_at"])
print("snapshot plus changes after boundary matches source:",
      copy_state == {r[0]: r for r in busy.query(f"SELECT {COLUMNS} FROM orders")})
busy.close()
# split on order_id   : chunks of [499, 501, 501, 509], largest share 25%
# split on customer_id: chunks of [940, 371, 333, 366], largest share 47%
# snapshot label 2026-03-07 23:55:35 holds 2010 rows; current source holds 2307
# snapshot plus changes after boundary matches source: True

2. Watermark extraction and the transactions it misses. Build a fresh source one day at a time, and after each day extract rows with updated_at later than the highest value seen so far and merge them into a copy. After seven days compare the copy with the source: rows missing, rows present but stale, rows deleted at the source but kept. Then repeat with the extraction starting three hours before the watermark.

def load_by_watermark(days, overlap):
    if type(days) is not int or days < 0 or not isinstance(overlap, timedelta) or overlap < timedelta(0):
        raise ValueError("days and overlap must be non-negative")
    source, dest, watermark = Source(), {}, "0000-00-00"
    for k in range(days):
        source.run_day(first_day + timedelta(days=k))
        since = stamp(datetime.strptime(watermark, "%Y-%m-%d %H:%M:%S") - overlap) if watermark > "1" else watermark
        rows = source.query(f"SELECT {COLUMNS} FROM orders WHERE updated_at > ? ORDER BY updated_at, order_id", (since,))
        for row in rows:
            dest[row[0]] = row
        watermark = max([watermark] + [row[4] for row in rows])
    truth = {row[0]: row for row in source.query(f"SELECT {COLUMNS} FROM orders")}
    stale = sum(1 for k in truth if k in dest and dest[k] != truth[k])
    source.close()
    return {"missing": len(truth.keys() - dest.keys()), "stale": stale, "deleted but kept": len(dest.keys() - truth.keys())}

print("strict watermark:", load_by_watermark(7, timedelta(0)))
print("3-hour overlap:  ", load_by_watermark(7, timedelta(hours=3)))
# strict watermark: {'missing': 0, 'stale': 14, 'deleted but kept': 22}
# 3-hour overlap:   {'missing': 0, 'stale': 0, 'deleted but kept': 22}

3. Apply the change log in order. Read the change log and apply every record to an empty dictionary keyed by order_id, in lsn order, and compare the result with the source by checksum. Then apply the same records ordered by the updated_at inside each record instead, and count the orders that end up in a different state.

def apply(changes):
    state = {}
    for op, order_id, after in changes:
        if op == "delete":
            state.pop(order_id, None)
        elif op in {"insert", "update"}:
            a = json.loads(after)
            state[order_id] = (order_id, a["customer_id"], a["status"], a["amount_cents"], a["updated_at"])
        else:
            raise ValueError(f"unknown operation: {op}")
    return state

log = src.query("SELECT op, order_id, after, lsn FROM change_log ORDER BY lsn")
truth = checksum(src.query(f"SELECT {COLUMNS} FROM orders"))
by_lsn = apply( for c in log])
print("applied in log order:      ", checksum(list(by_lsn.values())), "source:", truth)
by_app_time = sorted(log, key=lambda c: (json.loads(c[2])["updated_at"] if c[2] else "9999", c[3]))
by_time = apply( for c in by_app_time])
wrong = sorted(k for k in by_lsn.keys() | by_time.keys() if by_time.get(k) != by_lsn.get(k))
print("applied in updated_at order:", checksum(list(by_time.values())), "orders in the wrong state:", len(wrong))
if wrong:
    print("example:", by_lsn.get(wrong[0]), "became", by_time.get(wrong[0]))
# applied in log order:       (2010, 'daa003ca10da4ad30736d9b08e261fd9a97655c21134b57afb71127e6923ef2e') source: (2010, 'daa003ca10da4ad30736d9b08e261fd9a97655c21134b57afb71127e6923ef2e')
# applied in updated_at order: (2010, '3bfdec452477d0c08fd038109b0a4dc9220d4f2e1a31b64d50f505ff88f8fa7d') orders in the wrong state: 1
# example: (1746, 1, 'shipped', 8522, '2026-03-06 23:00:00') became (1746, 1, 'paid', 8522, '2026-03-06 23:40:16')

4. History from the log. Turn the change log into a version table: one row per order per interval of unchanged tracked attributes, with valid_from and valid_to taken from simulated recorded_at, skipping writes that changed nothing tracked. Print the versions of the order with the most, answer what its status was at noon on 3 March, and count orders that were refunded and later deleted.

history, open_version = [], {}
for lsn, op, order_id, after, recorded_at in src.query("SELECT lsn, op, order_id, after, recorded_at FROM change_log ORDER BY lsn"):
    a = json.loads(after) if after else None
    if op not in {"insert", "update", "delete"}:
        raise ValueError(f"unknown operation: {op}")
    if a and order_id in open_version and all(open_version[order_id][k] == a[k] for k in ("status", "amount_cents", "customer_id")):
        continue
    if order_id in open_version:
        open_version.pop(order_id)["valid_to"] = recorded_at
    if op != "delete":
        version = dict(a, order_id=order_id, valid_from=recorded_at, valid_to="9999-12-31 00:00:00", lsn=lsn)
        history.append(version)
        open_version[order_id] = version

assert all(v["valid_from"] <= v["valid_to"] for v in history)
print(len(history), "versions for", len({v["order_id"] for v in history}), "orders;",
      sum(v["valid_to"] == "9999-12-31 00:00:00" for v in history), "currently open")
busiest = max(sorted({v["order_id"] for v in history}), key=lambda k: sum(v["order_id"] == k for v in history))
for v in [v for v in history if v["order_id"] == busiest]:
    print(busiest, v["status"], "from", v["valid_from"], "to", v["valid_to"])

def status_as_of(order_id, when):
    rows = [v for v in history if v["order_id"] == order_id and v["valid_from"] <= when < v["valid_to"]]
    if len(rows) > 1:
        raise ValueError("overlapping history intervals")
    return rows[0]["status"] if rows else "not present in captured history"

print("order", busiest, "as of 2026-03-03 12:00:00:", status_as_of(busiest, "2026-03-03 12:00:00"))
print("refunded orders that were later deleted:", len({v["order_id"] for v in history if v["status"] == "refunded"}
      & {r[0] for r in src.query("SELECT order_id FROM change_log WHERE op = 'delete'")}))
# 2371 versions for 2052 orders; 2010 currently open
# 17 paid from 2026-03-01 00:58:19 to 2026-03-01 05:19:43
# 17 shipped from 2026-03-01 05:19:43 to 2026-03-01 20:58:11
# 17 refunded from 2026-03-01 20:58:11 to 2026-03-02 08:31:50
# 17 shipped from 2026-03-02 08:31:50 to 9999-12-31 00:00:00
# order 17 as of 2026-03-03 12:00:00: shipped
# refunded orders that were later deleted: 2

5. Reconcile by chunked checksum. Load a fresh seven-day source with a strict watermark, then compare the copy with the source in ranges of 250 keys by count and hash. For the ranges that differ, list missing, stale, and deleted keys, repair only those ranges, and confirm the whole copy now matches.

source, dest, watermark = Source(), {}, "0000-00-00"
for k in range(7):
    source.run_day(first_day + timedelta(days=k))
    rows = source.query(f"SELECT {COLUMNS} FROM orders WHERE updated_at > ?", (watermark,))
    dest.update({row[0]: row for row in rows})
    watermark = max([watermark] + [row[4] for row in rows])

truth = {row[0]: row for row in source.query(f"SELECT {COLUMNS} FROM orders")}
def key_chunks(left, right, width=250):
    if type(width) is not int or width < 1:
        raise ValueError("width must be positive")
    keys = set(left) | set(right)
    return [(lo, lo + width - 1) for lo in range(min(keys), max(keys) + 1, width)] if keys else []

chunks = key_chunks(truth, dest)
differing =  <= k <= c[1]]) != checksum([r for k, r in dest.items() if c[0] <= k <= c[1]])]
print(len(chunks), "chunks of 250 keys compared by checksum;", len(differing), "differ:", differing)
for lo, hi in differing:
    keys = {k for k in truth if lo <= k <= hi} | {k for k in dest if lo <= k <= hi}
    print(f"  {lo}-{hi}: missing {sum(k not in dest for k in keys)}, stale {sum(k in dest and k in truth and dest[k] != truth[k] for k in keys)}, deleted {sum(k not in truth for k in keys)}")

for lo, hi in differing:
    for k in [k for k in dest if lo <= k <= hi and k not in truth]:
        del dest[k]
    dest.update({k: r for k, r in truth.items() if lo <= k <= hi})
print("after repairing only those chunks:", dest == truth)

source.close()
# 9 chunks of 250 keys compared by checksum; 7 differ: [(2, 251), (252, 501), (502, 751), (752, 1001), (1002, 1251), (1252, 1501), (1502, 1751)]
#   2-251: missing 0, stale 2, deleted 3
#   252-501: missing 0, stale 2, deleted 2
#   502-751: missing 0, stale 2, deleted 2
#   752-1001: missing 0, stale 3, deleted 4
#   1002-1251: missing 0, stale 2, deleted 5
#   1252-1501: missing 0, stale 1, deleted 5
#   1502-1751: missing 0, stale 2, deleted 1
# after repairing only those chunks: True

6. Fail and recover. Import the table in four chunks into a staging list and make the third chunk fail. Retry once by running the import again on top of what is there, and once after clearing staging. Then replay the change log as a consumer that stores its position every 500 records, raise an exception after 250 writes in the next transaction, reopen the destination file, and restart from its durable checkpoint.

staging = []

def import_chunks(fail_at=None):
    for n, (lo, hi) in enumerate([(1, 600), (601, 1200), (1201, 1800), (1801, 2400)], start=1):
        if n == fail_at:
            raise RuntimeError(f"chunk {n} failed: connection reset")
        staging.extend(src.query(f"SELECT {COLUMNS} FROM orders WHERE order_id BETWEEN ? AND ?", (lo, hi)))

try:
    import_chunks(fail_at=3)
except RuntimeError as exc:
    print(exc, "with", len(staging), "rows already in staging")
import_chunks()
print("retry that appends:", len(staging), "rows;", len(staging) - len({r[0] for r in staging}), "duplicates")
staging.clear()
import_chunks()
print("retry after clearing staging:", len(staging), "rows; matches source:", checksum(staging) == checksum(src.query(f"SELECT {COLUMNS} FROM orders")))

def apply_log(con, changes, fail_after=None):
    """Apply an ordered, contiguous source batch and its checkpoint atomically."""
    changes = list(changes)
    if any(not isinstance(c, (tuple, list)) or len(c) != 4 for c in changes):
        raise ValueError("each change must have four fields")
    if any(type(c[2]) is not int or c[2] < 1 for c in changes):
        raise ValueError("toy order IDs must be positive integers")
    if any(type(c[0]) is not int or c[0] < 1 for c in changes):
        raise ValueError("toy source positions must be positive integers")
    if any(a[0] >= b[0] for a, b in zip(changes, changes[1:])):
        raise ValueError("batch must have strictly increasing source positions")
    if con.in_transaction:
        raise ValueError("apply_log requires an idle connection")
    con.execute("BEGIN IMMEDIATE")
    try:
        saved = con.execute("SELECT position FROM progress WHERE id=1").fetchone()
        if saved is None or type(saved[0]) is not int or saved[0] < 0:
            raise ValueError("missing or invalid checkpoint; restore compatible state")
        position = saved[0]
        for n, (lsn, op, order_id, after) in enumerate(changes, start=1):
            if lsn <= position:
                continue
            if lsn != position + 1:
                raise ValueError("gap in this dense toy log; checkpoint unchanged")
            if op == "delete":
                con.execute("DELETE FROM replica WHERE order_id=?", (order_id,))
            elif op in {"insert", "update"}:
                a = json.loads(after)
                con.execute("""INSERT INTO replica VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(order_id) DO UPDATE SET customer_id=excluded.customer_id,
                    status=excluded.status, amount_cents=excluded.amount_cents, updated_at=excluded.updated_at""",
                    (order_id, a["customer_id"], a["status"], a["amount_cents"], a["updated_at"]))
            else:
                raise ValueError(f"unknown operation: {op}")
            position = lsn
            if n == fail_after:
                raise RuntimeError("simulated failure before checkpoint")
        con.execute("UPDATE progress SET position=? WHERE id=1", (position,))
        con.execute("COMMIT")
    except BaseException:
        if con.in_transaction:
            con.execute("ROLLBACK")
        raise

# A file-backed destination; closing and reopening loses Python connection state.
replica_dir = tempfile.TemporaryDirectory()
replica_path = replica_dir.name + "/replica.db"
con = sqlite3.connect(replica_path)
con.executescript("""CREATE TABLE replica (order_id INTEGER PRIMARY KEY, customer_id INTEGER,
    status TEXT, amount_cents INTEGER, updated_at TEXT);
    CREATE TABLE progress (id INTEGER PRIMARY KEY, position INTEGER);
    INSERT INTO progress VALUES (1, 0);""")
log = src.query("SELECT lsn, op, order_id, after FROM change_log ORDER BY lsn")
for start in range(0, 1500, 500):
    apply_log(con, log[start:start + 500])
before_failure = con.execute(f"SELECT {COLUMNS} FROM replica ORDER BY order_id").fetchall()
try:
    apply_log(con, log[1500:2000], fail_after=250)
except RuntimeError as exc:
    print(exc)
con.close()
con = sqlite3.connect(replica_path)
position = con.execute("SELECT position FROM progress WHERE id=1").fetchone()[0]
print("reopened checkpoint:", position, "; partial writes rolled back:",
      con.execute(f"SELECT {COLUMNS} FROM replica ORDER BY order_id").fetchall() == before_failure)
apply_log(con,  > position])
expected = sorted(src.query(f"SELECT {COLUMNS} FROM orders"))
print("restarted from stored position:", con.execute(f"SELECT {COLUMNS} FROM replica ORDER BY order_id").fetchall() == expected)
apply_log(con, log)
print("replaying old records preserves current state:",
      con.execute(f"SELECT {COLUMNS} FROM replica ORDER BY order_id").fetchall() == expected)
con.close()
replica_dir.cleanup()
src.close()
# chunk 3 failed: connection reset with 1172 rows already in staging
# retry that appends: 3182 rows; 1172 duplicates
# retry after clearing staging: 2010 rows; matches source: True
# simulated failure before checkpoint
# reopened checkpoint: 1500 ; partial writes rolled back: True
# restarted from stored position: True
# replaying old records preserves current state: True

The loader rejects missing, boolean, non-integer, or non-positive order IDs before writing. SQLite can allocate an INTEGER PRIMARY KEY for a NULL input, so relying on insertion alone would create a different row and still advance progress. A missing or invalid checkpoint is also an error, not permission to start at zero against an existing target. The loader requires an idle connection and explicitly begins and commits its transaction. It also rejects missing positions in this deliberately dense 1, 2, 3… log, rolling back earlier writes in the batch. Real LSNs and connector offsets are not generally consecutive integers: use the source’s continuity and transaction protocol instead of copying the +1 check. The toy never changes primary keys, truncates tables, filters log records, or purges its log. Those operations require separate capture and recovery policies. Older records are skipped under the assumption that the same source position always denotes the same immutable event.

7. Write the ingestion specification. For the three tables in the choosing section, write one block each: mechanism, key, split or watermark column, overlap, schedule, source (primary or replica) and connection ceiling, how deletes are represented, whether history is kept, the reconciliation method and period, and what a rerun does.

The final executable block closes src and removes its temporary directory. If execution stops early, call src.close() when finished. Temporary sources and the destination created within exercises are closed there.