CDC Consumer Correctness Harness

Introduction

30 Second Summary

When a report shows the right number of rows but the wrong values, nobody notices until the damage is done. The most dangerous data errors are the ones that look correct.

In this project, you will build a correctness harness for a Change Data Capture consumer. You will run two deliberately wrong consumers first, measure their damage by count and by checksum, then build the correct one and prove it safe to replay.

What You'll Build

You will run the harness, watch the wrong consumer pass a row-count check while carrying stale values, then pull up the SHA-256 checksum that catches it.

By the end of this project, you'll have:

• A seeded generator that prints exact counts for all four change types before any consumer touches the data, so you always know the baseline.
• Two deliberately wrong consumers whose failures you can run and measure: one lands 2120 rows for 80 expected records, the other passes a count check but fails on checksum because stale values survived.
• A correct consumer whose target matches the source in both count and checksum, with every overwrite visible via PostgreSQL 18 RETURNING OLD/NEW, and a replay proving the write idempotent.
• Secret Mission: Write a readout for two audiences separating structural claims from regression checks, closing with lessons learned and honest limits.

Are there any prerequisites?

You will need Python 3.10 or later, PostgreSQL 18 installed and running, a code editor (VS Code or Cursor), a GitHub account, and a Linear account. Step 1 walks you through verifying and setting up each one.

Before We Start

Before any tool is opened, let's lock in what you're about to build and why it matters. This project proves that a wrong consumer can pass a count check while silently carrying stale data.

Set Up the Workstation and Delivery Tooling

Every tool in this project pins to a specific version, and the versions are load-bearing. PostgreSQL 18 introduces RETURNING OLD/NEW, a syntax that does not exist in earlier versions. The delta-rs load_cdf API stabilized in deltalake 1.x. A wrong version of any tool produces silent failures that look like your code is broken when it is the toolchain.

In this step, you will verify your Python and PostgreSQL installs, install three Python packages, create the GitHub repo and Linear project that deliver every subsequent step, and scaffold the project directory.

In this step, get ready to:

• Install the Python packages and verify PostgreSQL 18.
• Create the GitHub repo and Linear project with five stories.
• Create the project directory structure and source files.

Install the Python packages and verify PostgreSQL

Before installing anything, you need to confirm that Python and PostgreSQL 18 are present and running the right versions.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
• Check your Python version by running this command:

python --version

macOS users

If python is not found or shows Python 2.x, use python3 --version instead. On macOS, the Python 3 binary is often named python3. Use whichever command returns a 3.10+ version for every python command in this project.

✔️ I see Python 3.10 or higher

Python is ready. Move on to the PostgreSQL check below.

ⓧ I see an older version

This project requires Python 3.10 or higher.

• Download the latest Python installer from python.org/downloads.
• Run the installer. On Windows, check Add Python to PATH before clicking Install Now.
• Close and reopen your terminal, then run python --version again to confirm 3.10 or higher.

ⓧ Command not found

Python is not installed yet.

• Download the latest Python installer from python.org/downloads.
• Run the installer. On Windows, check Add Python to PATH before clicking Install Now.
• Close and reopen your terminal, then run python --version again to confirm it prints a version number.

• Check your PostgreSQL version by running this command:

psql --version

You should see output like psql (PostgreSQL) 18.x.

✔️ I see PostgreSQL 18

PostgreSQL 18 is installed and ready. Continue to the package install below.

ⓧ I see an older version

This project requires PostgreSQL 18 specifically. The RETURNING OLD/NEW syntax used throughout this project does not exist in PostgreSQL 17 or earlier.

• Download the PostgreSQL 18 installer from the official PostgreSQL downloads page.
• Run the installer and follow the prompts. On Windows, use the interactive installer by EDB.
• After installation, close and reopen your terminal.
• Run psql --version again and confirm the output begins with psql (PostgreSQL) 18.

What about my existing PostgreSQL install?

PostgreSQL 18 can run alongside an older version on a different port. The installer will let you choose the port number. Keep the default 5432 if you do not have another version running, or pick a different port if you do.

ⓧ Command not found

PostgreSQL is not installed yet.

• Download the PostgreSQL 18 installer from the official PostgreSQL downloads page.
• On Windows, click Windows on the download page and use the interactive installer by EDB. On macOS, click macOS and choose the Postgres.app or the EDB installer.
• Run the installer and follow the prompts to completion.
• Close and reopen your terminal.
• Run psql --version again and confirm the output begins with psql (PostgreSQL) 18.

• Install the three Python packages by running this command:

pip install "psycopg[binary]==3.3.4" duckdb==1.5.5 deltalake==1.6.2

What does [binary] do?

The [binary] extra for psycopg bundles the libpq C library inside the package. Without it, you would need to install libpq separately on your system. With it, the package works out of the box on Windows, macOS, and Linux.

🙋‍♀️ Seeing a permissions error?

If pip install fails with a permissions error, try adding --user to the end of the command. If pip is not found, try pip3 instead.

Why isn't pip installing my packages?

• Create the PostgreSQL database by running this command:

createdb cdc_harness

• Verify the database was created and is running PostgreSQL 18 by running:

psql -d cdc_harness -c "SELECT version();"

You should see output containing PostgreSQL 18 in the version string.

createdb: command not found?

The createdb utility ships with PostgreSQL. If the command is not found, the PostgreSQL bin directory is not on your system PATH. Find the PostgreSQL install directory and add its bin folder to your PATH.

Help me add PostgreSQL to my PATH

Create the GitHub repo and Linear project

The delivery tooling determines how every subsequent step ships. Each of the five stories maps to a branch, a pull request, and a merged result.

• Open github.com in your browser and sign in.
• Click the + icon in the top right corner.
• Click New repository.
• Enter cdc-harness as the Repository name.
• Select Private.
• Check Add a README file.
• Click Create repository.

GitHub creates the repo with a README.md on the main branch.

• Move to your Desktop by running this command:

cd ~/Desktop

• Clone the repo by running this command:

git clone https://github.com/[[GITHUB_USERNAME="your-github-username"]]/cdc-harness.git

• Move into the cloned repo by running:

cd cdc-harness

• Confirm you see README.md by running:

ls

You should see README.md listed in the output. Now update it with the project name.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code (or Cursor) and press Enter to open your code editor.
• Click File in the top menu bar.
• Click Open Folder.
• Select the cdc-harness folder on your Desktop, then click Open.
• Open README.md in the Explorer sidebar.
• Replace the entire contents of README.md with the following:

# CDC Consumer Correctness Harness

## Optional Extras (Require a Paid GitHub Plan)

- Branch protection rules requiring reviewers
- GitHub Advanced Security code scanning

Why note the optional extras?

On the free tier, you get branch-per-story, pull requests, and one GitHub Actions workflow. Branch protection rules that require reviewers and GitHub Advanced Security code scanning need a paid plan. Naming them here makes the repo honest about what is and is not in place.

• Save README.md.
• Switch back to your terminal and commit the updated README by running these commands:

git add README.md
git commit -m "Update README with project name and optional extras"
git push origin main

You should see Git push the commit to main successfully.

Git push rejected or asking for credentials?

If Git asks for a username and password, you may need to set up a personal access token or SSH key for GitHub authentication. GitHub no longer accepts plain passwords for git push.

Help me set up GitHub authentication for git push

Now set up the Linear project to track the five stories that deliver the build.

• Open linear.app in your browser and sign in.
• In the left sidebar, hover over Projects and click the + icon to create a new project.
• Enter CDC Consumer Correctness Harness as the project name.
• Click Create project.

Inside the new project, create one epic and five stories.

• Press C to open the issue creation dialog.
• Enter CDH-1 Design artifacts: interrogation checklist, decision records, topology as the title.
• Click Create issue.
• Repeat for each of the following four stories:

• CDH-2 Seeded generator with guard assertions.
• CDH-3 Wrong consumers and damage report.
• CDH-4 Correct consumer, checksum proof, idempotency replay.
• CDH-5 Real CDF, documentation suite, tagged release.

What about the epic?

In Linear, you can group stories under a project milestone or label them manually. If your workspace supports epics, create one called CDC Harness v1.0 and assign all five stories to it. If your workspace does not show epics, the five stories inside the project serve the same purpose.

You should now see five stories listed in the project, all in Backlog status.

Create the project directory structure and source files

The project uses six Python source files, a docs/ folder for documentation, and a .github/workflows/ folder for the CI workflow. Content ships across the five story branches in later steps.

• In your terminal, confirm you are inside the cdc-harness folder on your Desktop.
• Create the six source files and two folders by running these commands:

macOS

touch config.py feed.py consumers.py verify.py compare_real_cdf.py main.py
mkdir -p docs .github/workflows

Windows

New-Item -ItemType File config.py, feed.py, consumers.py, verify.py, compare_real_cdf.py, main.py
mkdir docs
mkdir .github
mkdir .github\workflows

• Confirm the files and folders exist by running:

ls

You should see config.py, feed.py, consumers.py, verify.py, compare_real_cdf.py, main.py, docs, and README.md in the output.

The .gitignore file keeps generated artifacts out of version control.

• In your code editor, right-click in the Explorer sidebar and select New File.
• Name the file .gitignore.
• Paste the following content into .gitignore:

__pycache__/
*.pyc
delta_cdf_check/

What does this .gitignore do?

__pycache__/ and *.pyc keep Python's compiled bytecode out of the repo. delta_cdf_check/ is the temporary Delta table directory that the real CDF comparison script creates and cleans up. If the cleanup fails, Git still ignores it.

• Save .gitignore.

✔️ Awesome, I've got everything!

Great. Double check you have saved .gitignore in your editor.

ⓧ I'd like to double check the full code

Your .gitignore should contain exactly this:

__pycache__/
*.pyc
delta_cdf_check/

Do not commit these files yet. They ship across the five story branches in the steps ahead.

Before you run the next command, what three package version numbers do you expect to see?

• Verify all three Python packages by running this command:

python -c "import psycopg; import duckdb; import deltalake; print('psycopg', psycopg.__version__); print('duckdb', duckdb.__version__); print('deltalake', deltalake.__version__)"

You should see psycopg 3.3.4, duckdb 1.5.5, and deltalake 1.6.2.

Seeing import errors or wrong versions?

If the python -c command fails with ModuleNotFoundError, the packages were installed for a different Python than the one you are running. Make sure pip and python point to the same installation.

On macOS, you may need to use pip3 and python3 consistently.

Help me fix my Python package imports

Your workstation is ready and every tool is version-verified. Next up, you will write the design artifacts and the seeded generator that produces the change feed with exact, predictable counts.

Design Artifacts and Seeded Generator

Your workstation is set up, your GitHub repo is live, and your Linear project has five stories waiting for work.

The interrogation checklist and decision records ship before a single line of consumer code exists. The questions they contain are the deliverable another organization answers, and those answers constrain every line of code that follows. After the documents land, the seeded generator runs and prints its own counts. That gives every downstream check a known-good baseline to measure against.

In this step, get ready to:

• Write the interrogation checklist, decision records, and topology as design artifacts in the docs/ folder.
• Implement the seeded generator in config.py and feed.py with deterministic counts for all four change types.
• Wire main.py to run the generator, verify the output, and merge both story branches.

Write the design artifacts

The design artifacts answer one question: what did you decide, and what would cause you to revisit? Each document ships in the same pull request as the code it constrains, not after it.

• Check out a new branch for the first Linear story by running this command:

git checkout -b cdh-1-design-artifacts

The interrogation checklist is the technical annex of an interface agreement under NIST SP 800-47 Rev 1. Each side keeps its own boundary and its own Authorizing Official. Each question carries the consequence of its answer beside it, which is what turns it from an interrogation into a shared risk.

• Create a new file called interrogation_checklist.md inside the docs/ folder, then paste in the following content:

# Interrogation Checklist: Change Data Feed Interface Agreement

Technical Annex under NIST SP 800-47 Rev 1
Reviewed: [date]

## Boundary statement

This document is the technical annex of an interface agreement between two organizations. Each side keeps its own boundary and its own Authorizing Official.

## Questions

### 1. Are the tables managed by the platform catalog or external to it?

Consequence: Catalog-managed tables carry lineage. External tables require the consumer to infer provenance.

### 2. Does the transaction log travel with the data?

Consequence: Without it, replay after a gap requires a full re-snapshot.

### 3. What is the retention window on the feed?

Consequence: A window shorter than the consumer batch interval means silent data loss.

### 4. How are schema changes announced, and how far ahead?

Consequence: An unannounced column rename breaks the tag contract silently.

Why consequences beside each question?

A question without a consequence is an interrogation. A question with a consequence is a shared risk. The upstream analytics owner can read each consequence and decide whether it changes their serving tier date.

The checklist is designed to be answerable in ten minutes, not to be impressive. A request that respects the other team's boundary gets answered faster than one that ignores it.

• Create a new file called decision_records.md inside the docs/ folder, then paste in the following content:

# Decision Records

Reviewed: [date]

## DR-1: Synthetic seeded feed over a live upstream

Alternative: Connect to staging. Tradeoff: No network dependency. Consequence: Shape must be validated against a real CDF once. Reversal: When staging with representative data exists.

## DR-2: Measure damage before writing the fix

Alternative: Build correct consumer first. Tradeoff: Learner sees failure before fix. Consequence: Wrong answers captured as numbers. Reversal: Never.

## DR-3: Change type as exact enumerated contract

Alternative: Pass through unknown types. Tradeoff: Fifth type breaks pipeline instead of corrupting target. Consequence: Requires code change for new type. Reversal: When upstream publishes schema registry.

## DR-4: Overwrite in evidence via RETURNING OLD/NEW

Alternative: Before-and-after query. Tradeoff: Requires PostgreSQL 18. Consequence: Overwrite visible in the write itself. Reversal: When target moves to database without RETURNING OLD/NEW.

What makes a decision record useful?

Each record names four things: the alternative that was considered, the tradeoff, the consequence of the chosen approach, and the reversal trigger. The reversal trigger is the most important part. It tells a future engineer when to revisit this decision instead of arguing about it now.

• Confirm the two files exist by running:

ls docs/

You should see interrogation_checklist.md and decision_records.md listed in the output.

• Create one more file called topology.md inside docs/ with this content:

# Topology

Reviewed: [date]

Data flow: generator -> three consumers -> PostgreSQL tables -> DuckDB verification. Real CDF comparison is a sidecar.

• Verify all three design files exist by running:

ls docs/

You should now see decision_records.md, interrogation_checklist.md, and topology.md in the output.

• Stage, commit, and push the design artifacts by running these commands:

git add docs/
git commit -m "CDH-1: design artifacts"
git push -u origin cdh-1-design-artifacts

• Open your cdc-harness repo on GitHub in your browser.
• Click the banner that says the branch was recently pushed, then click Compare & pull request.
• In the pull request title, type CDH-1: Design artifacts.
• Click Create pull request, then click Merge pull request, then Confirm merge.
• In Linear, open story CDH-1 and move it to Done.

Don't see the compare & pull request banner?

Click the Pull requests tab at the top of the repo page, then click New pull request. Select cdh-1-design-artifacts as the compare branch and main as the base branch.

Help me push and open a pull request for my design artifacts branch.

• Switch back to main and pull the merge by running:

git checkout main
git pull

Implement the config and generator

The generator uses a fixed seed so every run produces identical output. That means every count printed by the generator is an exact number you can assert against later. If a downstream check disagrees, the consumer is wrong.

• Check out a new branch for story CDH-2 by running:

git checkout -b cdh-2-generator

• Open config.py in your editor and paste in these constants:

SEED = 20260814
NUM_INITIAL_RECORDS = 100
NUM_UPDATES = 1000
NUM_DELETES = 20
PG_CONNINFO = "dbname=cdc_harness"
DELTA_TABLE_PATH = "delta_cdf_check"

What do these constants control?

• SEED locks the random number generator so every run produces the same feed. Change it, and every downstream count changes with it.
• NUM_INITIAL_RECORDS, NUM_UPDATES, and NUM_DELETES set the exact shape of the synthetic change data feed.
• PG_CONNINFO and DELTA_TABLE_PATH point at the PostgreSQL database and the local Delta table path used later for the real CDF comparison.

• Save config.py.
• Open feed.py in your editor and paste in the imports and module-level constants:

import random
import hashlib

from config import SEED, NUM_INITIAL_RECORDS, NUM_UPDATES, NUM_DELETES

VALID_CHANGE_TYPES = frozenset({
    "insert",
    "update_preimage",
    "update_postimage",
    "delete",
})

REGIONS = ["us-east-1", "us-west-2", "eu-west-1", "ap-southeast-1"]

Why a frozenset for change types?

VALID_CHANGE_TYPES lists the four exact _change_type values a Delta Lake change data feed emits. They use underscores: insert, update_preimage, update_postimage, delete. A filter written against a hyphenated spelling matches nothing and reproduces the bug being taught.

A frozenset is immutable, so no code path can accidentally modify the contract at runtime.

• Save feed.py and verify both files load together by running:

python -c "from feed import VALID_CHANGE_TYPES; print(VALID_CHANGE_TYPES)"

You should see the four change type strings printed as a frozenset.

Seeing a ModuleNotFoundError?

Make sure your terminal is in the cdc-harness project folder. Both config.py and feed.py must be in the same directory.

Help me fix a ModuleNotFoundError when importing from feed.py.

Now build the generator function. It creates the synthetic change feed in three phases: inserts, updates, and deletes. Each phase writes rows with a specific _change_type and an incrementing _commit_version.

• In feed.py, below the REGIONS list, add the first part of the generate_feed() function with the insert phase:

def generate_feed():
    rng = random.Random(SEED)
    feed = []
    source_state = {}
    commit_version = 0

    # Phase 1: initial inserts
    commit_version += 1
    for i in range(NUM_INITIAL_RECORDS):
        key = f"record-{i:04d}"
        region = rng.choice(REGIONS)
        amount = rng.randint(100, 9999)
        source_state[key] = {"region": region, "amount": amount}
        feed.append({
            "natural_key": key,
            "region": region,
            "amount": amount,
            "_change_type": "insert",
            "_commit_version": commit_version,
        })

What does Phase 1 do?

The seeded random.Random(SEED) ensures every run produces identical data. Phase 1 creates 100 initial records as insert rows at commit version 1. Each record also goes into source_state, which tracks what the target table should look like after all changes are applied.

• Continuing inside the generate_feed() function, add the update phase directly below the insert loop:

    # Phase 2: updates (each produces preimage + postimage)
    keys = list(source_state.keys())
    for _ in range(NUM_UPDATES):
        commit_version += 1
        key = rng.choice(keys)
        old = source_state[key].copy()
        feed.append({
            "natural_key": key,
            "region": old["region"],
            "amount": old["amount"],
            "_change_type": "update_preimage",
            "_commit_version": commit_version,
        })
        new_region = rng.choice(REGIONS)
        new_amount = rng.randint(100, 9999)
        source_state[key] = {"region": new_region, "amount": new_amount}
        feed.append({
            "natural_key": key,
            "region": new_region,
            "amount": new_amount,
            "_change_type": "update_postimage",
            "_commit_version": commit_version,
        })

Why does one update produce two rows?

A change data feed represents one logical update as two rows: an update_preimage (the row as it was) and an update_postimage (the row as it is now). A thousand logical updates arrive as two thousand change rows. This is a property of the representation. A consumer that does not filter preimages will double-apply every update.

• Save feed.py and confirm the module still loads by running:

python -c "import feed; print('feed.py OK')"

You should see feed.py OK printed with no errors. The function is not complete yet, but the syntax is valid.

• Still inside generate_feed(), add the delete phase and the return statement below the update loop. Then add the print_feed_counts() function below generate_feed():

    # Phase 3: deletes
    commit_version += 1
    delete_keys = rng.sample(list(source_state.keys()), NUM_DELETES)
    for key in delete_keys:
        old = source_state[key]
        feed.append({
            "natural_key": key,
            "region": old["region"],
            "amount": old["amount"],
            "_change_type": "delete",
            "_commit_version": commit_version,
        })
        del source_state[key]

    return feed, source_state


def print_feed_counts(feed):
    counts = {}
    for row in feed:
        ct = row["_change_type"]
        counts[ct] = counts.get(ct, 0) + 1
    print("--- Generator counts ---")
    for ct in ["insert", "update_preimage", "update_postimage", "delete"]:
        print(f"  {ct}: {counts.get(ct, 0)}")
    print(f"  total rows: {len(feed)}")
    return counts

What does this code do?

Phase 3 uses rng.sample to pick 20 unique keys to delete. Because sample never picks the same key twice, the source truth after all changes is exactly 100 - 20 = 80 records.

print_feed_counts() prints the count per change type and the total row count. The generator prints its own counts before any consumer runs, so a fixture that emitted no updates is caught at the source.

• Save feed.py and run the generator to see counts by running:

python -c "from feed import generate_feed, print_feed_counts; feed, state = generate_feed(); print_feed_counts(feed); print(f'{len(state)} records')"

You should see: insert: 100, update_preimage: 1000, update_postimage: 1000, delete: 20, total rows: 2120, and 80 records.

Seeing an IndentationError?

Each phase block must be indented inside the generate_feed() function body. The return feed, source_state line must be at the same indentation level as the three phase blocks (one level inside the function).

Help me fix an IndentationError in my generate_feed function.

• Add the last two helper functions below print_feed_counts() in feed.py:

def compute_checksum(rows):
    if not rows:
        return hashlib.sha256(b"empty").hexdigest()
    sorted_rows = sorted(rows, key=lambda r: r["natural_key"])
    parts = []
    for row in sorted_rows:
        canonical = f"{row['natural_key']}|{row['region']}|{row['amount']}"
        parts.append(hashlib.sha256(canonical.encode("utf-8")).hexdigest())
    combined = "|".join(parts)
    return hashlib.sha256(combined.encode("utf-8")).hexdigest()


def source_to_rows(source_state):
    return [
        {"natural_key": k, "region": v["region"], "amount": v["amount"]}
        for k, v in source_state.items()
    ]

How does the checksum work?

compute_checksum() sorts rows by natural_key, hashes each row with SHA-256, then hashes the combined result. The sort makes the checksum independent of insertion order. Two tables with the same rows in different order produce the same checksum.

source_to_rows() flattens the source_state dict into a list of row dicts so it can be passed to compute_checksum().

• Save feed.py.

✔️ Awesome, I've got everything!

Great. Confirm you have saved feed.py with all six constructs: imports, VALID_CHANGE_TYPES, REGIONS, generate_feed(), print_feed_counts(), compute_checksum(), and source_to_rows().

ⓧ I'd like to double check the full code

Scroll up and confirm each of the following chunks is present in feed.py, in this order:

• Imports (random, hashlib, config imports).
• VALID_CHANGE_TYPES frozenset and REGIONS list.
• generate_feed() with Phase 1 (inserts), Phase 2 (preimage/postimage pairs), Phase 3 (deletes), and return feed, source_state.
• print_feed_counts() function.
• compute_checksum() and source_to_rows() functions.

Wire the entry point and run the generator

The entry point calls the generator, prints its counts, and computes the source truth checksum. A guard assertion catches an empty fixture before any consumer touches it.

• Open main.py in your editor and paste in the following code:

from feed import generate_feed, print_feed_counts, compute_checksum, source_to_rows


def main():
    feed, source_state = generate_feed()
    counts = print_feed_counts(feed)

    assert counts.get("update_preimage", 0) > 0, (
        "Generator produced no updates. The fixture is empty."
    )

    source_rows = source_to_rows(source_state)
    source_checksum = compute_checksum(source_rows)
    print(f"\n  Source truth: {len(source_state)} records")
    print(f"  Source checksum: {source_checksum[:16]}...")


if __name__ == "__main__":
    main()

What does the guard assertion do?

The assert counts.get("update_preimage", 0) > 0 line catches a fixture that emitted no updates before any consumer runs. If someone changes the config and accidentally sets NUM_UPDATES to zero, the assertion fails immediately instead of letting the consumers produce meaningless results.

• Save main.py.

✔️ Awesome, I've got everything!

Great. Confirm you have saved main.py with the import line, the main() function, and the if __name__ guard.

ⓧ I'd like to double check the full code

from feed import generate_feed, print_feed_counts, compute_checksum, source_to_rows


def main():
    feed, source_state = generate_feed()
    counts = print_feed_counts(feed)

    assert counts.get("update_preimage", 0) > 0, (
        "Generator produced no updates. The fixture is empty."
    )

    source_rows = source_to_rows(source_state)
    source_checksum = compute_checksum(source_rows)
    print(f"\n  Source truth: {len(source_state)} records")
    print(f"  Source checksum: {source_checksum[:16]}...")


if __name__ == "__main__":
    main()

Before you run the generator, take a moment. Do you expect the total row count to be 100 + 1000 + 20 = 1120? Or something different?

• Run the generator by executing:

python main.py

The total is 2120, not 1120. Each of the 1000 updates produces two rows: a preimage and a postimage. That is 100 inserts + 1000 preimages + 1000 postimages + 20 deletes = 2120 total rows. The source truth is 80 records, and a SHA-256 checksum prefix is printed.

Seeing an AssertionError about no updates?

Check that NUM_UPDATES in config.py is set to 1000, not 0. The guard assertion is working correctly by catching an empty fixture.

Help me debug why my generator assertion is failing.

• Stage, commit, and push the generator code by running:

git add config.py feed.py main.py
git commit -m "CDH-2: seeded generator with guard assertions"
git push -u origin cdh-2-generator

• Return to your cdc-harness repo on GitHub.
• Click Compare & pull request for the cdh-2-generator branch.
• Set the title to CDH-2: Seeded generator with guard assertions.
• Click Create pull request, then Merge pull request, then Confirm merge.
• In Linear, open story CDH-2 and move it to Done.

• Switch back to main and pull the merge by running:

git checkout main
git pull

Two stories down, three to go. Your generator prints exact counts that every downstream check will measure against. Next up, you will run two deliberately wrong consumers against this feed and measure the damage each one does.

Two Wrong Consumers and the Damage Report

Your generator proved the feed has the right shape: 2120 rows representing 80 final records. The question now is what happens when a consumer reads that feed without understanding its structure.

Two consumers will each handle the feed differently. One appends every row blindly. The other upserts by natural key without filtering change types. Both are wrong, but only one failure is visible when you check the row count alone.

In this step, get ready to:

• Build two wrong consumers and create the PostgreSQL target tables.
• Build a verification harness that measures counts and checksums with DuckDB.
• Run the full pipeline and read the damage report.

Check out the branch and build the two wrong consumers

Each story in Linear gets its own branch. The consumers and their damage report ship together as one pull request.

• Create and switch to the cdh-3-wrong-consumers branch by running:

git checkout -b cdh-3-wrong-consumers

The three PostgreSQL target tables need a schema before any consumer can write to them. Each table serves a different consumer strategy.

• Open consumers.py and paste the imports and table schema:

import psycopg

from config import PG_CONNINFO

# DDL for the three target tables
SCHEMA_DDL = """
DROP TABLE IF EXISTS target_append CASCADE;
DROP TABLE IF EXISTS target_upsert CASCADE;
DROP TABLE IF EXISTS target_filtered CASCADE;

CREATE TABLE target_append (
    row_id SERIAL PRIMARY KEY,
    natural_key TEXT NOT NULL,
    region TEXT NOT NULL,
    amount INTEGER NOT NULL,
    _change_type TEXT NOT NULL,
    _commit_version BIGINT NOT NULL
);

CREATE TABLE target_upsert (
    natural_key TEXT PRIMARY KEY,
    region TEXT NOT NULL,
    amount INTEGER NOT NULL
);

CREATE TABLE target_filtered (
    natural_key TEXT PRIMARY KEY,
    region TEXT NOT NULL,
    amount INTEGER NOT NULL
);
"""

What does this schema do?

Three tables represent three consumer strategies. target_append keeps every row including metadata columns like _change_type. It has no primary key constraint on natural_key, so duplicates are allowed.

target_upsert and target_filtered both have a primary key on natural_key, so each key appears at most once. The target_filtered table is created now but won't be used until the next step.

• Still in consumers.py, add the setup_tables() function and the first wrong consumer below the SCHEMA_DDL string:

def setup_tables():
    # Drop and recreate all three target tables
    with psycopg.connect(PG_CONNINFO) as conn:
        conn.execute(SCHEMA_DDL)
        conn.commit()


def consume_append(feed):
    # Wrong consumer 1: blindly inserts every row
    with psycopg.connect(PG_CONNINFO) as conn:
        for row in feed:
            conn.execute(
                "INSERT INTO target_append "
                "(natural_key, region, amount, _change_type, _commit_version) "
                "VALUES (%s, %s, %s, %s, %s)",
                (row["natural_key"], row["region"], row["amount"],
                 row["_change_type"], row["_commit_version"]),
            )
        conn.commit()
    print(f"  Append consumer: inserted {len(feed)} rows")

What does the append consumer do?

consume_append() inserts every single row from the feed into target_append with no filtering. Preimages, postimages, inserts, deletes. All 2120 rows land in the table.

• Save consumers.py.
• Verify the tables get created in PostgreSQL by running:

python -c "from consumers import setup_tables; setup_tables(); print('Tables created')"

• Confirm the tables exist by running:

psql -d cdc_harness -c "\dt target_*"

You should see three tables listed: target_append, target_filtered, and target_upsert.

Not seeing the tables?

• Make sure the cdc_harness database exists. You created it in the previous step with createdb cdc_harness.
• Check that PostgreSQL 18 is running. If psql cannot connect, restart the PostgreSQL service.
• Help me troubleshoot my PostgreSQL connection.

The second wrong consumer is more interesting. It upserts every row by natural key, but it processes them in alphabetical order of _change_type within each commit version.

• Add the second wrong consumer below consume_append() in consumers.py:

def consume_upsert(feed):
    # Wrong consumer 2: upserts all rows sorted by version and change type
    # "update_postimage" < "update_preimage" alphabetically,
    # so postimage applies first and preimage overwrites it
    sorted_feed = sorted(
        feed, key=lambda r: (r["_commit_version"], r["_change_type"])
    )
    with psycopg.connect(PG_CONNINFO) as conn:
        for row in sorted_feed:
            if row["_change_type"] == "delete":
                conn.execute(
                    "DELETE FROM target_upsert WHERE natural_key = %s",
                    (row["natural_key"],),
                )
            else:
                conn.execute(
                    "INSERT INTO target_upsert (natural_key, region, amount) "
                    "VALUES (%s, %s, %s) "
                    "ON CONFLICT (natural_key) DO UPDATE "
                    "SET region = EXCLUDED.region, amount = EXCLUDED.amount",
                    (row["natural_key"], row["region"], row["amount"]),
                )
        conn.commit()
    print(f"  Upsert consumer: processed {len(sorted_feed)} rows")

What does the sort key do here?

The feed is sorted by (_commit_version, _change_type). Within the same commit version, Python sorts _change_type alphabetically. The letter o in update_postimage comes before r in update_preimage. Think about what that means for which value ends up in the table.

• Save consumers.py.
• Verify the file parses correctly by running:

python -c "from consumers import setup_tables, consume_append, consume_upsert; print('All three functions loaded')"

You should see All three functions loaded printed with no errors.

✔️ Awesome, I've got everything!

Double check you've saved consumers.py.

ⓧ I'd like to double check the full code

consumers.py is built from the three chunks above. Scroll up and confirm each one is present, in this order:

• Imports (import psycopg, from config import PG_CONNINFO) and the SCHEMA_DDL string with three CREATE TABLE statements.
• setup_tables() that executes the DDL and commits.
• consume_append(feed) that inserts every row into target_append.
• consume_upsert(feed) that sorts by version and change type, then upserts into target_upsert.

Build the verification harness

Counting rows is not enough to prove a consumer is correct. A table with the right count can still hold the wrong values. The verification harness checks both count and SHA-256 checksum.

• Open verify.py and paste the imports and table reader:

import duckdb
import pyarrow as pa
import psycopg

from config import PG_CONNINFO
from feed import compute_checksum


def read_pg_table(table_name):
    # Read a PostgreSQL table into a list of dicts
    with psycopg.connect(PG_CONNINFO) as conn:
        rows = conn.execute(
            f"SELECT natural_key, region, amount "
            f"FROM {table_name} ORDER BY natural_key"
        ).fetchall()
    return [
        {"natural_key": r[0], "region": r[1], "amount": r[2]}
        for r in rows
    ]

What does this code do?

read_pg_table() reads only the data columns (natural_key, region, amount) from any target table, ordered by key. This gives a consistent shape for checksum comparison against the source truth.

pyarrow is a transitive dependency of deltalake, so it was installed automatically. DuckDB uses it to register in-memory tables for analytical queries.

• Below read_pg_table(), start the verify_all() function with the source truth baseline and both sink comparisons:

def verify_all(source_rows, source_checksum, feed):
    source_count = len(source_rows)

    print(f"\n--- Source truth ---")
    print(f"  Row count: {source_count}")
    print(f"  Checksum:  {source_checksum[:16]}...")

    # Check append sink
    with psycopg.connect(PG_CONNINFO) as conn:
        append_count = conn.execute(
            "SELECT count(*) FROM target_append"
        ).fetchone()[0]

    print(f"\n--- Append sink (wrong consumer 1) ---")
    print(f"  Row count: {append_count} (expected {source_count})")
    print(f"  WRONG: {append_count} rows for {source_count} expected records")

    # Check upsert sink
    upsert_rows = read_pg_table("target_upsert")
    upsert_count = len(upsert_rows)
    upsert_checksum = compute_checksum(upsert_rows)

    print(f"\n--- Upsert sink (wrong consumer 2) ---")
    print(f"  Row count: {upsert_count} (expected {source_count})")
    u_count_match = "MATCH" if upsert_count == source_count else "MISMATCH"
    print(f"  Count:    {u_count_match}")
    u_cksum_match = "MATCH" if upsert_checksum == source_checksum else "MISMATCH"
    print(f"  Checksum: {u_cksum_match}")
    if upsert_checksum != source_checksum:
        print(f"  WRONG: count looks right but checksum proves stale values")

What does this code do?

verify_all() prints the source truth (count and checksum) as the baseline. Then it checks each sink against that baseline.

For the append sink, a count check alone is enough to catch the failure. For the upsert sink, the function checks both count AND checksum, because one of those will match and the other won't.

• Save verify.py.
• Confirm the file parses and connects to PostgreSQL by running:

python -c "from verify import read_pg_table; print(f'target_append has {len(read_pg_table(chr(39) + chr(39)  or  chr(116)+chr(97)+chr(114)+chr(103)+chr(101)+chr(116)+chr(95)+chr(97)+chr(112)+chr(112)+chr(101)+chr(110)+chr(100))} rows')"

Actually, let's use a simpler check. Confirm the import works by running:

• Verify the file parses and connects by running:

python -c "from verify import read_pg_table; print('verify.py loaded')"

You should see verify.py loaded with no import errors.

The verification harness also needs DuckDB to run an analytical query over the raw feed. This confirms the feed composition independently of PostgreSQL.

• Still in verify.py, add the DuckDB verification and return statement at the end of verify_all(), directly below the last print line you just wrote:

    # DuckDB analytical verification of feed composition
    feed_table = pa.table({
        "natural_key": [r["natural_key"] for r in feed],
        "region": [r["region"] for r in feed],
        "amount": [r["amount"] for r in feed],
        "_change_type": [r["_change_type"] for r in feed],
        "_commit_version": [r["_commit_version"] for r in feed],
    })
    con = duckdb.connect()
    con.register("feed", feed_table)
    composition = con.sql(
        "SELECT _change_type, count(*) AS cnt "
        "FROM feed GROUP BY _change_type ORDER BY _change_type"
    ).fetchall()
    print("\n--- DuckDB verification ---")
    print("  Feed composition:")
    for row in composition:
        print(f"    {row[0]}: {row[1]}")
    con.close()

    return {
        "source_count": source_count,
        "source_checksum": source_checksum,
        "append_count": append_count,
        "upsert_count": upsert_count,
        "upsert_checksum": upsert_checksum,
    }

What does the DuckDB verification do?

The feed is converted to a PyArrow table and registered in DuckDB as a virtual table named feed. The SQL query groups rows by _change_type and counts them, giving an independent confirmation that the generator output has the expected composition.

• Save verify.py.

✔️ Awesome, I've got everything!

Double check you've saved verify.py.

ⓧ I'd like to double check the full code

verify.py is built from the three chunks above. Scroll up and confirm each one is present, in this order:

• Imports: duckdb, pyarrow, psycopg, plus PG_CONNINFO and compute_checksum.
• read_pg_table() returning a list of dicts.
• verify_all() with source truth printout, append sink check, upsert sink check (count + checksum), DuckDB composition query, and the return dict.

Wire the pipeline and read the damage report

main.py currently calls only the generator. It needs to set up the tables, run both wrong consumers, and then call the verification harness.

• In main.py, find the existing import line at the top of the file:

from feed import generate_feed, print_feed_counts, compute_checksum, source_to_rows

• Add these two import lines directly below it:

from consumers import setup_tables, consume_append, consume_upsert
from verify import verify_all

• Now find this line inside the main() function:

    print(f"  Source checksum: {source_checksum[:16]}...")

• Add the consumer and verification calls directly below that line, before the if __name__ block:

    # Set up PostgreSQL tables
    setup_tables()

    # Run wrong consumer 1: append-only
    print("\n--- Running append consumer (wrong) ---")
    consume_append(feed)

    # Run wrong consumer 2: upsert without filtering
    print("\n--- Running upsert consumer (wrong) ---")
    consume_upsert(feed)

    # Verify both sinks against source truth
    verify_all(source_rows, source_checksum, feed)

What does this code do?

setup_tables() drops and recreates all three target tables, giving a clean slate. Then both wrong consumers run against the same feed. Finally, verify_all() measures each sink against the source truth and prints the damage report.

• Save main.py.

✔️ Awesome, I've got everything!

Double check you've saved main.py.

ⓧ I'd like to double check the full code

main.py should now have three import lines at the top, and the body of main() should contain, in order:

• The generator call, count print, guard assertion, source rows, and checksum print from the previous step.
• setup_tables() call.
• consume_append(feed) and consume_upsert(feed) calls, each preceded by a print statement.
• verify_all(source_rows, source_checksum, feed) call at the end.

Before you run this, think about what you expect to see. The feed has 2120 total rows but only 80 final records. The append consumer inserts every row. The upsert consumer upserts by natural key. Will both consumers land 80 rows? Will both checksums match the source?

• Run the full pipeline by running:

python main.py

Read the damage report. You should see two distinctly different failures.

What should I see?

The append sink reports 2120 rows for 80 expected records, printed with the word WRONG. That failure is obvious. You can reject it on sight.

The upsert sink reports 80 rows. Count says MATCH. But the checksum says MISMATCH, printed with WRONG. The row count stayed correct while the values went stale. A count-based check would have reported success.

The DuckDB verification confirms the feed composition: 100 inserts, 1000 update_postimage, 1000 update_preimage, 20 deletes.

💡 Why is the second failure harder?

Within each commit version, the sort key (_commit_version, _change_type) places update_postimage before update_preimage alphabetically. The postimage (new value) is applied first, then the preimage (old value) overwrites it. The table ends up holding the stale value for every updated key.

The dangerous part is that the row count stays right. If your monitoring only checks counts, this failure passes every check. Only the checksum catches it.

🙋‍♀️ Seeing an import error or connection failure?

• If you see ModuleNotFoundError for psycopg or duckdb, confirm the packages were installed in the previous step.
• If you see a connection refused error, make sure PostgreSQL 18 is running and the cdc_harness database exists.
• If you see a syntax error, check that your verify_all() function is indented consistently. The DuckDB block must be inside the same function body.
• Help me debug my pipeline run.

Both wrong answers are now captured as numbers. Time to ship them.

• Stage and commit your changes by running:

git add consumers.py verify.py main.py
git commit -m "CDH-3: wrong consumers and damage report"

• Push the branch by running:

git push -u origin cdh-3-wrong-consumers

• Open a pull request on GitHub referencing CDH-3.
• Merge the pull request into main.
• In Linear, close story CDH-3.
• Switch back to main and pull the merged changes by running:

git checkout main
git pull

The damage report is delivered. You have two wrong answers captured as numbers, not assertions. The append sink's 2120 rows is the failure anyone would catch. The upsert sink's 80 rows with a wrong checksum is the one that costs money. Next up, you'll build the consumer that gets it right.

The Correct Consumer and the Checksum Proof

The previous step proved that two consumer designs fail against the same feed. The append sink fails on count, and the upsert sink fails on checksum while the count looks right. The upsert failure is the harder one because a count-based check reports success.

The fix is a filtered consumer that drops update_preimage rows before applying the feed. Two guards prevent a false pass: one asserts the filter actually removed the expected number of preimages, and the other rejects any unknown change type. Every overwrite is visible via PostgreSQL 18's RETURNING OLD/NEW, putting the evidence in the write itself rather than in a before-and-after query.

In this step, get ready to:

• Implement the filtered consumer with preimage filtering, guard assertions, and RETURNING OLD/NEW.
• Wire the consumer into the harness and extend the verification to cover the filtered sink.
• Prove the write idempotent by replaying the same batch and comparing checksums.

Check out the branch and implement the filtered consumer

The filtered consumer does three things the wrong consumers did not. It drops the before-value rows by matching on the exact string update_preimage with underscores. It rejects any unknown _change_type value with a loud error. It applies the remaining rows in commit-version order so the latest value wins.

• Switch to your terminal and check out a new branch by running these commands:

git checkout main
git pull origin main
git checkout -b cdh-4-correct-consumer

The function is built in two parts. The first part filters the feed and enforces the guards. The second part applies the filtered rows to PostgreSQL using RETURNING OLD/NEW.

• In consumers.py, scroll to the bottom of the file after the consume_upsert() function. Add the first part of consume_filtered() by copying the following code:

def consume_filtered(feed, expected_preimage_count):
    # Only these three types make it to the target
    ALLOWED = {"insert", "update_postimage", "delete"}
    removed = 0
    applied = []

    # Filter out preimage rows, counting each removal
    for row in feed:
        if row["_change_type"] == "update_preimage":
            removed += 1
            continue
        # Reject any unknown change type loudly
        if row["_change_type"] not in ALLOWED:
            raise ValueError(
                f"Unknown _change_type: {row['_change_type']!r}. "
                f"The contract is {ALLOWED | {'update_preimage'}}."
            )
        applied.append(row)

    # Guard: removed count must match expected preimages
    assert removed == expected_preimage_count, (
        f"Filter removed {removed} rows, expected {expected_preimage_count}. "
        f"A filter that removed zero rows matched against a spelling "
        f"the feed never uses."
    )

    # Sort remaining rows by commit version
    applied.sort(key=lambda r: r["_commit_version"])

What does this code do?

The ALLOWED set defines the three change types the consumer accepts. The filter loop skips update_preimage rows while counting them. Any value not in the full set of four raises a ValueError.

The assert is the filter spelling guard. A filter written against a hyphenated spelling (update-preimage) matches nothing, removes zero rows, and silently reproduces the bug being taught. The assert catches that before the consumer touches the database.

• Directly below the applied.sort() line you just added, copy the second part of the function:

    # Apply filtered rows to target_filtered
    overwrites = []
    with psycopg.connect(PG_CONNINFO) as conn:
        for row in applied:
            if row["_change_type"] == "delete":
                conn.execute(
                    "DELETE FROM target_filtered WHERE natural_key = %s",
                    (row["natural_key"],),
                )
            else:
                # RETURNING OLD/NEW puts the overwrite in evidence
                result = conn.execute(
                    "INSERT INTO target_filtered (natural_key, region, amount) "
                    "VALUES (%s, %s, %s) "
                    "ON CONFLICT (natural_key) DO UPDATE "
                    "SET region = EXCLUDED.region, amount = EXCLUDED.amount "
                    "RETURNING old.natural_key AS old_key, "
                    "old.amount AS old_amount, "
                    "new.amount AS new_amount",
                    (row["natural_key"], row["region"], row["amount"]),
                ).fetchone()
                if result and result[0] is not None:
                    overwrites.append({
                        "key": result[0],
                        "old_amount": result[1],
                        "new_amount": result[2],
                    })
        conn.commit()

    print(f"  Filtered consumer: {removed} preimage rows removed")
    print(f"  Filtered consumer: {len(applied)} rows applied")
    print(f"  Filtered consumer: {len(overwrites)} overwrites via RETURNING OLD/NEW")
    return removed, overwrites

What does RETURNING OLD/NEW do here?

PostgreSQL 18 introduced the ability to return both the old and new values from a single write. The INSERT ... ON CONFLICT DO UPDATE ... RETURNING old.natural_key AS old_key, old.amount AS old_amount, new.amount AS new_amount clause gives back both values in the same round trip.

For a fresh insert, old_key is None because there is no previous row. For a conflict (overwrite), the old values are non-null, and the function collects them into the overwrites list. This puts the overwrite in evidence at write time, not via inference from a before-and-after query.

• Save consumers.py.
• Confirm the function parses correctly by running:

python -c "from consumers import consume_filtered; print('consume_filtered loaded OK')"

You should see consume_filtered loaded OK printed. If you see a SyntaxError, the indentation of the second part does not line up with the first.

Seeing an IndentationError or SyntaxError?

• The second part of the function (starting at overwrites = []) must be indented one level inside consume_filtered(). Check that it aligns with applied.sort() above it.
• Make sure the triple-quoted strings inside the assert and ValueError are closed properly. An unclosed f-string breaks the rest of the file.
• Help me debug my consume_filtered function

✔️ Awesome, I've got everything!

Great. Double check you have saved consumers.py.

ⓧ I'd like to double check the full code

The full consumers.py file is long, so scroll through it and confirm these sections appear in order:

• The import psycopg and from config import PG_CONNINFO at the top.
• The SCHEMA_DDL string with all three tables (target_append, target_upsert, target_filtered).
• The setup_tables(), consume_append(), and consume_upsert() functions from the previous step.
• The new consume_filtered() function at the bottom, starting with ALLOWED = {"insert", "update_postimage", "delete"} and ending with return removed, overwrites.

Why does the tag contract reject unknown types?

The consumer matches four exact _change_type values and raises an error on anything else. If the upstream adds a fifth type, the pipeline breaks instead of silently corrupting the target. Loudly wrong beats quietly wrong.

This is the tag contract from decision record DR-3. It requires a code change to accept a new type, which is the deliberate tradeoff.

Wire the consumer and extend verification

The consumer function exists but nothing calls it yet. The verification function also needs a new section to check the filtered sink's count and checksum against the source truth.

• In verify.py, find the line that says WRONG: count looks right but checksum proves stale values. Add the following filtered sink section directly below it:

    # Filtered sink
    filtered_rows = read_pg_table("target_filtered")
    filtered_count = len(filtered_rows)
    filtered_checksum = compute_checksum(filtered_rows)

    print(f"\n--- Filtered sink (correct consumer) ---")
    print(f"  Row count: {filtered_count} (expected {source_count})")
    f_count_match = "MATCH" if filtered_count == source_count else "MISMATCH"
    print(f"  Count:    {f_count_match}")
    f_cksum_match = "MATCH" if filtered_checksum == source_checksum else "MISMATCH"
    print(f"  Checksum: {f_cksum_match}")

• Still in verify.py, scroll down to the DuckDB verification section. Find the line con.close() and add this query directly above it:

    # Count distinct keys that were deleted
    delete_keys = con.sql(
        "SELECT count(DISTINCT natural_key) FROM feed WHERE _change_type = 'delete'"
    ).fetchone()[0]
    print(f"  Distinct keys deleted: {delete_keys}")

• Scroll to the return dict at the bottom of verify_all(). Find this line:

        "upsert_checksum": upsert_checksum,
    }

• Add the two filtered entries so the return dict now looks like this:

        "upsert_checksum": upsert_checksum,
        "filtered_count": filtered_count,
        "filtered_checksum": filtered_checksum,
    }

• Save verify.py.

What does the filtered sink check do?

It reads the target_filtered table, computes its SHA-256 checksum, and compares both count and checksum against the source truth. The DuckDB query adds a distinct-key count for deletes, confirming all 20 deleted keys are unique.

Now update main.py to call the new consumer and capture the verification results.

• In main.py, find the import line that starts with from consumers import. Find this line:

from consumers import setup_tables, consume_append, consume_upsert

• Add consume_filtered to the end of that import, so the line now looks like this:

from consumers import setup_tables, consume_append, consume_upsert, consume_filtered

• Scroll down into the main() function body. Find the line verify_all(source_rows, source_checksum, feed) near the bottom. Delete that line and replace it with the following code:

    # --- Correct consumer: filtered ---
    print("\n--- Running filtered consumer (correct) ---")
    removed, overwrites = consume_filtered(feed, counts["update_preimage"])

    if overwrites:
        print("  Sample overwrites (RETURNING OLD/NEW):")
        for ow in overwrites[:5]:
            print(f"    {ow['key']}: {ow['old_amount']} -> {ow['new_amount']}")

    # --- Verification ---
    results = verify_all(source_rows, source_checksum, feed)

What does this code do?

The call to consume_filtered(feed, counts["update_preimage"]) passes the generator's own preimage count as the expected value. The filter guard inside consume_filtered() checks the actual removed count against this number.

The first five overwrites are printed so you see the PostgreSQL 18 RETURNING OLD/NEW evidence on screen. The result of verify_all() is captured in results because the idempotency replay needs it next.

• Save main.py.

Before you run this, what do you think the filtered sink's count and checksum will show? Will both match the source truth, or will one still fail?

• Run the harness by running:

python main.py

In the output, look for the filtered sink section. You should see count MATCH (80) and checksum MATCH. The sample overwrites print old and new amounts for the first five keys, proving the overwrite happened and is visible.

Seeing a checksum MISMATCH on the filtered sink?

• Check that the filter matches the exact string update_preimage with underscores. A hyphenated spelling (update-preimage) matches nothing and reproduces the upsert bug.
• Confirm the applied.sort(key=lambda r: r["_commit_version"]) line is present. Without it, rows arrive in feed order rather than commit order.
• Help me debug my filtered consumer checksum mismatch

✔️ Awesome, I've got everything!

Great. Double check you have saved both verify.py and main.py.

ⓧ I'd like to double check the full code

Both files are long. Scroll through each and confirm these sections appear in order.

verify.py:

• Imports: duckdb, pyarrow, psycopg, compute_checksum.
• read_pg_table() function.
• verify_all() with source truth, append sink, upsert sink, then the new filtered sink section, then DuckDB verification including the delete_keys query.
• Return dict with filtered_count and filtered_checksum at the end.

main.py:

• Import line includes consume_filtered.
• After the upsert consumer call: the filtered consumer section calling consume_filtered(feed, counts["update_preimage"]) and printing overwrites.
• results = verify_all(source_rows, source_checksum, feed) capturing the return value.

Add the idempotency replay and ship the branch

The consumer produced a correct target once. A correct consumer should also be safe to replay. If you truncate the target and run the same batch again, the checksum should be identical.

• In main.py, add three new imports at the very top of the file. Find the existing import lines and update them so the top of the file looks like this:

import psycopg

from config import PG_CONNINFO
from feed import generate_feed, print_feed_counts, compute_checksum, source_to_rows
from consumers import setup_tables, consume_append, consume_upsert, consume_filtered
from verify import verify_all, read_pg_table

What are the new imports for?

The replay needs psycopg and PG_CONNINFO to connect directly to PostgreSQL and truncate the table. The read_pg_table() function reads the table back so the replay checksum can be computed.

• Scroll to the bottom of the main() function, after the results = verify_all(...) line. Add the idempotency replay by copying this code:

    # --- Idempotency replay ---
    print("\n--- Idempotency replay ---")
    with psycopg.connect(PG_CONNINFO) as conn:
        conn.execute("DELETE FROM target_filtered")
        conn.commit()

    # Replay the same feed through the same consumer
    consume_filtered(feed, counts["update_preimage"])

    # Read back and compare checksums
    replay_rows = read_pg_table("target_filtered")
    replay_checksum = compute_checksum(replay_rows)

    if replay_checksum == results["filtered_checksum"]:
        print(f"  Replay checksum: MATCH. Write is idempotent.")
    else:
        print(f"  Replay checksum: MISMATCH. Write is NOT idempotent.")

What does the replay prove?

The replay clears target_filtered and runs the exact same batch through consume_filtered() again. If the checksum matches the first run, the write is idempotent. The same batch can be applied twice without changing the result.

• Save main.py.

Do you think replaying the same batch through the same consumer will produce the same checksum, or could something differ on the second run?

• Run the full harness by running:

python main.py

Look for four things in the output. The filtered consumer removed exactly 1000 preimages. The filtered sink shows count MATCH (80) and checksum MATCH. The sample overwrites show old and new amounts via RETURNING OLD/NEW. The replay checksum shows MATCH with the word "idempotent."

Seeing an AssertionError about removed rows?

• If the assertion says it removed 0 rows, the filter string does not match the feed's spelling. Check that the filter line uses update_preimage with underscores, not hyphens.
• If the assertion says it removed a number other than 1000, confirm the generator's NUM_UPDATES in config.py is set to 1000.
• Help me fix my consume_filtered assertion error

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py.

ⓧ I'd like to double check the full code

main.py is long. Scroll through and confirm these sections appear in order:

• import psycopg at the very top.
• from config import PG_CONNINFO on the next line.
• Imports of consume_filtered and read_pg_table.
• Generator, setup, append consumer, upsert consumer calls.
• Filtered consumer call with counts["update_preimage"] and overwrites printing.
• results = verify_all(...) capturing the return value.
• Idempotency replay section: truncate, replay, compare checksums.

• Push the branch and open a pull request by running:

git add -A
git commit -m "CDH-4: correct consumer, checksum proof, idempotency replay"
git push -u origin cdh-4-correct-consumer

• Switch to your browser and open your cdc-harness repo on GitHub.
• Open a pull request from cdh-4-correct-consumer to main. Reference CDH-4 in the PR description.
• Merge the pull request.
• In Linear, move story CDH-4 to Done.

The correct consumer matches the source in both count and checksum. The replay proves the write is safe to rerun. Next up, you will validate the synthetic feed's shape against a real delta-rs change data feed, write the documentation suite, and tag the release.

Real CDF, Documentation, and Tagged Release

Your correct consumer matches the source in both count and SHA-256 checksum. The replay proved the write idempotent. Every overwrite is visible via PostgreSQL 18 RETURNING OLD/NEW.

The synthetic feed is useful only if its shape matches what a real upstream emits. A real delta-rs change data feed confirms the same metadata columns and the same underscore-spelled values. This step validates that shape, writes the full documentation suite, creates a GitHub Actions workflow to run the checksum harness on every push, and tags the release.

In this step, get ready to:

• Implement the real CDF shape comparison and wire it into the harness.
• Write the documentation suite and create the CI workflow.
• Tag the release and close the final Linear story.

Check out the release branch and implement the real CDF comparison

The DuckDB Delta extension does not expose a change data feed. The deltalake Python package (delta-rs under Apache 2.0) does, through its load_cdf() method. The comparison creates a small Delta table with CDF enabled, merges updated rows, reads the CDF, and checks that the metadata columns and change type values match the synthetic feed.

• Create the release branch by running:

git checkout -b cdh-5-cdf-docs-release main

• Open compare_real_cdf.py and add the imports and the first part of the compare_real_cdf() function:

import os
import shutil

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

from config import DELTA_TABLE_PATH


def compare_real_cdf():
    # Clean up any leftover Delta table from a previous run
    if os.path.exists(DELTA_TABLE_PATH):
        shutil.rmtree(DELTA_TABLE_PATH)

    # Three rows of sample data
    initial = pa.table({
        "natural_key": ["alpha", "beta", "gamma"],
        "region": ["us-east-1", "us-west-2", "eu-west-1"],
        "amount": [100, 200, 300],
    })

    # Write with change data feed enabled
    write_deltalake(
        DELTA_TABLE_PATH,
        initial,
        configuration={"delta.enableChangeDataFeed": "true"},
    )

What does this code do?

The function starts by removing any leftover Delta table directory so each run is clean. It then creates a three-row PyArrow table and writes it to the local path defined in config.py (delta_cdf_check).

The configuration={"delta.enableChangeDataFeed": "true"} flag is what tells Delta Lake to record change rows. Without it, load_cdf() has nothing to read.

• Below the write_deltalake call, still inside compare_real_cdf(), add the merge operation and CDF read:

    # Merge two updated rows into the Delta table
    dt = DeltaTable(DELTA_TABLE_PATH)
    updated = pa.table({
        "natural_key": ["alpha", "beta"],
        "region": ["us-east-1", "us-west-2"],
        "amount": [150, 250],
    })

    (
        dt.merge(
            source=updated,
            predicate="target.natural_key = source.natural_key",
            source_alias="source",
            target_alias="target",
        )
        .when_matched_update_all()
        .when_not_matched_insert_all()
        .execute()
    )

    # Read the full change data feed from version 0
    dt = DeltaTable(DELTA_TABLE_PATH)
    cdf_table = dt.load_cdf(starting_version=0).read_all()

What does this code do?

The merge updates alpha and beta with new amounts while leaving gamma untouched. Delta Lake records the changes internally. The load_cdf(starting_version=0) call then reads every change row from the table's creation onward, returning an Arrow table with the same three metadata columns the synthetic feed uses: _change_type, _commit_version, and _commit_timestamp.

• Save compare_real_cdf.py and confirm the module loads by running:

python -c "from compare_real_cdf import compare_real_cdf; print('import OK')"

You should see import OK with no errors.

Seeing an ImportError?

Check that you saved compare_real_cdf.py in the same folder as config.py. The from config import DELTA_TABLE_PATH line needs config.py to be importable from the same directory.

help me debug an ImportError in compare_real_cdf.py

• Complete the function by adding the comparison logic and cleanup below the cdf_table line:

    print("\n--- Real CDF shape comparison ---")
    print(f"  Columns: {cdf_table.column_names}")

    # Check metadata columns match synthetic feed
    expected_meta = {"_change_type", "_commit_version", "_commit_timestamp"}
    actual_meta = set(cdf_table.column_names) & expected_meta
    missing = expected_meta - actual_meta

    if not missing:
        print("  Metadata columns: MATCH")
    else:
        print(f"  Metadata columns: MISMATCH (missing {missing})")

    # Check _change_type values use underscore spelling
    change_types = set(cdf_table.column("_change_type").to_pylist())
    print(f"  _change_type values found: {sorted(change_types)}")
    uses_underscores = all("_" in ct for ct in change_types if "update" in ct)
    print(f"  Underscore spelling confirmed: {uses_underscores}")

    if "update_preimage" not in change_types:
        print("  Note: update_preimage not observed in this small fixture.")
        print("  Leg not performed, recorded as not performed.")

    # Clean up the Delta table directory
    shutil.rmtree(DELTA_TABLE_PATH, ignore_errors=True)
    return cdf_table

What does this code do?

The comparison checks two things against the synthetic feed. First, that the real CDF carries the same three metadata columns (_change_type, _commit_version, _commit_timestamp). Second, that _change_type values use underscore spelling.

If update_preimage is not observed in this small three-row fixture, the code records the leg as not performed. A leg that could not run is recorded as not performed, never as a match.

• Save compare_real_cdf.py.

✔️ Awesome, I've got everything!

Great. Double check you saved compare_real_cdf.py.

ⓧ I'd like to double check the full code

Scroll up through your compare_real_cdf.py and confirm these three sections appear in order:

• Imports (os, shutil, pyarrow, DeltaTable, write_deltalake, DELTA_TABLE_PATH) and function definition opening with cleanup and write_deltalake() call.
• Merge block creating updated table, calling dt.merge(), and reading CDF with load_cdf(starting_version=0).read_all().
• Print comparison checking metadata columns, _change_type values, underscore spelling, and cleanup with shutil.rmtree().

Now wire the comparison and the honest limits block into main.py.

• In main.py, find the last import line:

from verify import verify_all, read_pg_table

• Add the new import directly below it so the two lines read:

from verify import verify_all, read_pg_table
from compare_real_cdf import compare_real_cdf

• Scroll to the end of the main() function. Find the idempotency replay block that ends with:

    if replay_checksum == results["filtered_checksum"]:
        print(f"  Replay checksum: MATCH. Write is idempotent.")
    else:
        print(f"  Replay checksum: MISMATCH. Write is NOT idempotent.")

• Add the following code below that block, before the if __name__ line:

    # Run real CDF shape comparison
    compare_real_cdf()

    # State what this harness proves and what it does not
    print("\n--- Honest limits ---")
    print("  This proves a consumer correct against a synthetic feed")
    print("  whose shape was checked once against a real one.")
    print("  It proves nothing about the upstream's retention window,")
    print("  how that upstream announces a schema change, or its")
    print("  behavior under concurrent writers at volume.")
    print("  A leg that could not run is recorded as not performed,")
    print("  never as a match.")

Why print honest limits?

Every claim this harness makes is scoped. It proves correctness against a synthetic feed whose shape was checked once against a real one. It says nothing about the upstream's retention window, schema change behavior, or concurrent writer behavior.

Printing the limits in every run makes the scope visible. A reader never has to wonder what was proved.

• Save main.py.

Before you run, what new output sections do you expect to see beyond the output you saw in the last step?

• Run the full harness by running:

python main.py

At the end of the output, you should see a new section headed --- Real CDF shape comparison --- showing the column names, Metadata columns: MATCH, the _change_type values found, and Underscore spelling confirmed: True. After that, the --- Honest limits --- block prints.

Seeing a FileNotFoundError or ModuleNotFoundError?

Make sure you added from compare_real_cdf import compare_real_cdf at the top of main.py. Check that compare_real_cdf.py is saved in the same directory as main.py.

help me debug a missing module error in main.py

Write the documentation suite

The documentation ships in the same pull request as the code. Every page carries a reviewed date. Point-in-time numbers are labeled as snapshots, and every claim traces to a file, a command, or a named source.

• Create docs/runbook.md with the four failure modes by adding this content:

# Runbook

Reviewed: [date]

Four failure modes: (a) filtered and unfiltered runs agree - check filter spelling; (b) right count wrong checksum - check apply ordering; (c) feed cannot be read at all; (d) replay changes the target - check the key it was written on.

Why these four failure modes?

These are the four ways this harness fails in practice. Each one maps to a specific diagnostic. If the filtered and unfiltered runs agree, the filter matched nothing. If the count is right but the checksum is wrong, the apply ordering is stale.

• Create docs/config_reference.md listing the six configuration values:

# Config Reference

Reviewed: [date]

SEED, NUM_INITIAL_RECORDS, NUM_UPDATES, NUM_DELETES, PG_CONNINFO, DELTA_TABLE_PATH.

• Create the remaining seven documentation files in docs/. Each file has a title heading and a reviewed date. Create each of these files with the pattern shown below:

# [Title]

Reviewed: [date]

• Create docs/architecture.md with the title # Architecture.
• Create docs/quickstart.md with the title # Quickstart.
• Create docs/operations.md with the title # Operations.
• Create docs/decision_index.md with the title # Decision Index.
• Create docs/agent_architecture.md with the title # Agent Architecture.
• Create docs/limits.md with the title # Limits.
• Create docs/source_map.md with the title # Source Map.

• Confirm all documentation files exist by running:

ls docs/

You should see twelve files: the three from earlier steps (interrogation_checklist.md, decision_records.md, topology.md) plus the nine you just created.

Missing files in the listing?

Double check that each file was saved inside the docs/ folder, not in the project root. The filenames are case-sensitive.

help me find my missing docs files

Create the CI workflow and tag the release

The GitHub Actions workflow is the regression gate that catches a broken harness before it reaches main. It runs the full python main.py harness against a PostgreSQL 18 service container on every push.

• Create the workflow directory by running:

mkdir -p .github/workflows

• Create .github/workflows/harness.yml and add the workflow trigger and PostgreSQL service container:

name: CDC Harness

on:
  push:
    branches: ["main", "cdh-*"]
  pull_request:
    branches: ["main"]

jobs:
  harness:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:18
        env:
          POSTGRES_DB: cdc_harness
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
        ports:
          - 5432:5432
        options: >-
          --health-cmd "pg_isready -U postgres"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

What does this service block do?

The services key tells GitHub Actions to spin up a postgres:18 Docker container alongside the runner. The health check waits until PostgreSQL is ready before any step runs. Port 5432 is mapped to the host so the harness can connect.

• Below the services block, still inside the harness job, add the environment variable and steps:

    env:
      PG_CONNINFO: "dbname=cdc_harness host=localhost user=postgres password=postgres"

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install "psycopg[binary]==3.3.4" duckdb==1.5.5 deltalake==1.6.2

      - name: Run harness
        run: python main.py

Why does PG_CONNINFO include host and user?

Locally, config.py sets PG_CONNINFO = "dbname=cdc_harness" which uses your local socket. In CI, the PostgreSQL service container runs as a separate Docker container. The env block overrides PG_CONNINFO to include host=localhost and the service credentials so psycopg can connect over TCP.

• Save harness.yml.

✔️ Awesome, I've got everything!

Great. Confirm the file is saved at .github/workflows/harness.yml.

ⓧ I'd like to double check the full code

Scroll through your harness.yml and confirm these two sections appear in order:

• The trigger block (on: push and pull_request), then the harness job with services.postgres using image: postgres:18, health checks, and port 5432:5432.
• The env block with PG_CONNINFO, then four steps: checkout, setup Python 3.12, install dependencies, and run harness.

• Add a note to your README.md mentioning that branch protection rules requiring reviewers and GitHub Advanced Security code scanning need a paid plan. These are optional extras.
• Stage, commit, and push the branch by running:

git add -A
git commit -m "CDH-5: real CDF comparison, docs, CI workflow"
git push -u origin cdh-5-cdf-docs-release

• Switch back to GitHub in your browser.
• Open a pull request from cdh-5-cdf-docs-release into main. Reference CDH-5 in the description.
• Merge the pull request.

What about the workflow run?

The push triggers the GitHub Actions workflow. You can check its status under the Actions tab in your repo. The workflow needs a real PostgreSQL 18 service container, so it runs in CI only. Locally, you already confirmed the harness passes.

• Tag the release by running:

git checkout main
git pull
git tag -a v1.0.0 -m "CDC Consumer Correctness Harness v1.0.0"
git push origin v1.0.0

• Switch to Linear.
• Close story CDH-5.
• Mark the project "CDC Consumer Correctness Harness" as Completed.

Confirm the tag exists in GitHub by checking the Tags section on your repo page. You should see v1.0.0 listed.

What is the handover test?

The handover test is a person. Someone who did not build this clones the repo, runs the wrong consumer first, and reads both failures from the runbook alone. Every question they have to ask is a defect in the document.

The question to watch for is whether they can explain why the second failure is worse: the row count stays correct while the values go stale, so a count-based check reports success.

🙋‍♀️ Tag push rejected?

Make sure you pulled the latest main before tagging. If your local main is behind the remote, run git pull first, then tag again.

help me push a git tag to my remote

Secret mission

Readout for Two Audiences

The work is not delivered until it is communicated. Two audiences need different things from the same results. Can you write a readout that separates what holds forever from what holds only for this fixture, and frame adoption around an audience that does not work for you?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine and free-tier services, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• PostgreSQL database cdc_harness with three tables (target_append, target_upsert, target_filtered).
• Private GitHub repo cdc-harness.
• Linear project "CDC Consumer Correctness Harness."
• Local project folder cdc-harness on your Desktop.

Keep everything running

No action needed. Choose this if you want to keep experimenting with the harness, re-running the consumers, or refining the documentation.

• The PostgreSQL database persists locally at no cost.
• The GitHub repo and Linear project remain on their free tiers with no charges.
• The local Delta table directory (delta_cdf_check) is cleaned up automatically by the script each time it runs.

Pause - I'll come back to this later

There are no running services to pause. Everything runs locally or on free tiers, so nothing accumulates cost while you are away.

• The PostgreSQL database, GitHub repo, and Linear project all persist without consuming resources.
• Everything is ready for you to pick up whenever you return.

Delete - I don't want to use this again

Remove all project resources and start fresh.

• Drop the PostgreSQL database by running:

dropdb cdc_harness

• Delete the local project folder.

macOS

rm -rf ~/Desktop/cdc-harness

Windows

Remove-Item -Recurse -Force ~\Desktop\cdc-harness

• Delete the GitHub repo:

• Open the cdc-harness repo on GitHub.
• Click Settings.
• Scroll down to the Danger Zone section.
• Click Delete this repository and follow the confirmation prompts.

• Archive the Linear project:

• Open the "CDC Consumer Correctness Harness" project in Linear.
• Open the project settings and click Archive.

Nice Work!

Nice Work!

Nice work! You built a CDC consumer correctness harness that proves one consumer right and two consumers wrong, with every claim backed by a count or a checksum.

You've learned how to:

• Built a seeded synthetic change data feed generator, ran two deliberately wrong consumers against it, and measured distinct failure modes: the append sink landing 2120 rows for 80 expected records, and the upsert sink passing on count but failing on SHA-256 checksum because preimage rows carry stale values.
• Built a correct consumer that filters preimages by exact string match, asserts the removed count matches the generator, produces a target matching the source in both count and checksum, and proves the write idempotent by replaying the same batch. Every overwrite is visible via PostgreSQL 18 RETURNING OLD/NEW, putting the evidence in the write itself.
• Delivered through a private GitHub repo with branch-per-story closing Linear tickets, a GitHub Actions workflow running the checksum harness on push, a tagged v1.0.0 release, and a documentation suite including an interrogation checklist as a NIST SP 800-47 Rev 1 technical annex, four decision records, a runbook naming four failure modes, and a source map with reviewed dates.
• Secret Mission: Delivered a readout for two audiences separating structural claims from regression checks, framing adoption around an upstream owner who does not work for you, and closing with lessons learned and honest limits.

Ready to quiz yourself?