pipeverify

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.

Related MCP server: nock-mcp

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: <model> 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 its change was safe. pipeverify ships an MCP server so the agent can ask:

uvx pipeverify mcp        # or: pip install pipeverify && pipeverify mcp

{
  "mcpServers": {
    "pipeverify": { "command": "uvx", "args": ["pipeverify", "mcp"] }
  }
}

That block works in Claude Desktop, Claude Code, Cursor, VS Code and any other MCP client (Python 3.10 or newer). The tools:

Every tool is read-only: snapshots are read, the checks run in an in-memory DuckDB, and nothing is written to your files or warehouse. The agent gets pass, review or block with the evidence, and is told to report every review and block finding to you instead of working around it. Database credentials stay on your machine (environment variables or profiles.yml) and are never tool arguments; give the server a read-only role.

Python API

import pipeverify

result = pipeverify.verify(before_df, after_df, key="order_id", expect_change=["status"])
print(result.verdict)              # "pass" | "review" | "block"
for f in result.findings:
    print(f.severity, f.message)

Useful inside an Airflow task, a notebook, or an AI agent's own loop: let the agent propose a fix, run verify, and only apply the fix on PASS.

Optional AI explanation

export ANTHROPIC_API_KEY=...       # or OPENAI_API_KEY with PIPEVERIFY_LLM_MODEL
pip install "pipeverify[ai]"
pipeverify run --explain

This adds a short plain-English explanation of what probably happened and what to check next. It is off by default. Only the findings are sent (check names, counts, percentages, column names); example values from your data are removed first. The verdict never depends on the LLM.

Who uses pipeverify

See ADOPTERS.md. Using it at work or in a project? Tell us how, even in one line. It helps other teams decide whether to try it and shapes what gets built next.

Roadmap

• Native adapters for Airflow and Spark, and more dbt targets (BigQuery, Postgres, Databricks)

• Sampling and pushdown for very large tables, so checks run in the warehouse

• Rules learned from history: what "normal" change looks like for each table

Ideas and pull requests are welcome; see CONTRIBUTING.md and the good first issues. Questions go in Discussions.

Status

Alpha. The checks and verdicts are stable enough to try in CI; options and output formats may still change before 1.0.

Citation

If you use pipeverify in research, please cite the PipeMend benchmark that motivated it (see CITATION.cff).

License

Apache-2.0