pipeverify

The open-source verification step for data pipeline changes.

Check that a change left the data right, not just the job green. pipeverify compares tables before and after a change, and against the raw source, and returns PASS , REVIEW or BLOCK . Works with SQL, Spark, dbt, Airflow, or a fix written by an AI agent.

---

A pipeline change can pass every test, finish on time and keep the schema intact while quietly dropping rows, duplicating keys or changing what a column means. More of these changes are now written by AI coding agents, which makes "the job succeeded" an even weaker signal.

pipeverify compares your tables before and after a change and gives a verdict:

It works with any tool that produces tables: SQL scripts, Spark jobs, dbt, Airflow tasks or an AI agent's fix. It needs two snapshots, nothing else. It reads dbt models and Snowflake tables directly, runs as a GitHub Action, and AI agents can call it through MCP.

Why

In the PipeMend benchmark (DOI 10.5281/zenodo.23510784), LLM agents repaired 64 data pipeline incidents. Production-style data quality monitors approved every harmful fix the agents made. What caught them was comparing the data itself. pipeverify packages that check so any team can run it on every change.

Quick start

No install needed to try it: open the 5-minute notebook in Colab.

pip install pipeverify

compare two snapshots of one table

pipeverify compare before/orders.parquet after/orders.parquet --key order_id

BLOCK orders.parquet rows 1,200 -> 1,000, matched 1,000 BLOCK Table lost 200 rows (16.67%) BLOCK 200 key(s) present before are gone after (16.67%) BLOCK Total of 'revenue' moved 16.71% (1,383,131.25 -> 1,152,000.00)

Verdict: BLOCK This change damages the data. Do not merge as is.

Try the bundled demo, five realistic changes to a retail sales table:

git clone https://github.com/akshaysharma1088/pipeverify && cd pipeverify

pip install -e . python examples/retail/make_demo.py for s in examples/retail/scenarios/*/; do pipeverify run -c "$s/pipeverify.yml"; done

Every one of these runs successfully and keeps the same schema.

Sources

before and after can each be:

SQL databases need pip install "pipeverify[sql]" plus the driver for your database. Environment variables in URLs are expanded, so credentials stay out of the config file.

See SQL source examples for Postgres, Snowflake and BigQuery.

dbt: check a pull request against production

# pipeverify.yml, next to dbt_project.yml

tables:

• name: fct_daily_sales before: {dbt: fct_daily_sales, target: prod, manifest: prod-state/manifest.json} after: {dbt: fct_daily_sales, target: ci} key: [store_id, sale_date]

dbt: finds the model's table in dbt's manifest.json and connects the way profiles.yml says for that target, so there is no second set of credentials to keep. manifest is the manifest of the run you are comparing with (the same file --state and --defer use); leave it out for target/manifest.json. pipeverify does not run dbt and writes nothing. DuckDB and Snowflake targets work today. A runnable project, with a join that fans out while every dbt test stays green, is in examples/dbt_pr.

Snowflake

pip install "pipeverify[snowflake]"

export SNOWFLAKE_ACCOUNT=myorg-myaccount SNOWFLAKE_USER=ci_user SNOWFLAKE_PRIVATE_KEY_PATH=~/.ssh/snowflake.p8 export SNOWFLAKE_ROLE=REPORTER SNOWFLAKE_WAREHOUSE=COMPUTE_WH pipeverify compare snowflake:PROD.GOLD.ORDERS snowflake:DEV.GOLD.ORDERS --key order_id

It reads the same SNOWFLAKE_* variables as the Snowflake CLI and DataRanger (password, key pair, token or SNOWFLAKE_AUTHENTICATOR), or a named connection from ~/.snowflake/connections.toml with {snowflake: ..., connection: name}. It runs one SELECT per snapshot, tagged pipeverify in query history, and nothing else, so a read-only role is enough. Each snapshot is loaded into memory: for a big table add where: or pass a query, and compare the slice your change touched.

Config

Check many tables at once with pipeverify.yml (pipeverify init writes a starter file):

thresholds:              # optional, overrides the defaults below for every table

row_change_block_pct: 10 #skip_id_like_totals: true # set to false to also check totals of *_id columns tables:

• name: gold.daily_sales before: snapshots/main/daily_sales.parquet after: snapshots/pr/daily_sales.parquet key: [store_id, sale_date] expect_change: [revenue] # columns this change is meant to modify ignore: [loaded_at] # columns that always change
• name: gold.customers before: {url: "${WAREHOUSE_URL}", table: prod.customers} after: {url: "${WAREHOUSE_URL}", table: pr_123.customers} key: [customer_id] expect_row_change: true # this change is meant to add or remove rows

pipeverify run                     # uses ./pipeverify.yml

pipeverify run --fail-on review # stricter: fail CI on REVIEW too pipeverify run --format markdown -o report.md pipeverify run --json-out report.json

Exit code is 1 when the verdict reaches --fail-on (default block), otherwise 0.

What it checks

Columns listed in expect_change are reported but never raise the verdict. All thresholds can be changed globally or per table; see pipeverify/checks.py for every option.

Source checks

A before/after diff only sees damage to rows that already existed. Errors in new data need evidence the pipeline does not control: the raw source. Add a source block to a table:

tables:

• name: silver.transactions after: snapshots/pr/silver_transactions.parquet before: snapshots/main/silver_transactions.parquet # optional key: [txn_id] source: data: landing/raw_transactions.parquet key: [txn_id] columns: {amount: amount} # output column: source column checks: [conflict] # default; also mismatch, missing, orphan where: "status = 'POSTED'" # source rows that should reach the output

From the command line:

pipeverify compare - silver.parquet -k txn_id --source raw.parquet --source-column amount

In a pilot on the 209 executed fixes in the PipeMend benchmark (25 harmful), production-style monitors caught 16 harmful fixes, and the source conflict check caught the 9 that the monitors approved. Together they caught all 25, at the cost of a REVIEW on 21 of 184 correct fixes, all on incidents where the source data really was ambiguous. This is a small pilot on synthetic data; a larger evaluation is in progress.

GitHub Action

# .github/workflows/pipeverify.yml

on: pull_request permissions: contents: read pull-requests: write jobs: verify: runs-on: ubuntu-latest steps:

• uses: actions/checkout@v7

... build the "after" snapshots for this PR here (for example: dbt build --target ci) ...

• uses: akshaysharma1088/pipeverify@v0.3.0 with: config: pipeverify.yml fail-on: block # or review, or never extras: snowflake # optional drivers: snowflake, sql env: SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }} # only if your sources need them

The action posts the report as a PR comment and in the job summary, fails the job at the verdict you choose, and exposes verdict, report and json as outputs.

Use it from an AI agent (MCP)

An agent that edits a query, a dbt model or a load should not be the one to decide it