Motherduck Migrate To Motherduck

Name: Motherduck Migrate To Motherduck
Author: motherduckdb

motherduckdb/agent-skills

250 installs
53 repo stars
Updated July 31, 2026
motherduckdb/agent-skills

Plan and execute migration of analytical workloads from local DuckDB, warehouses, or files into MotherDuck with schema, connection, and cutover guidance.

About

Guides agents through migrating analytical databases and DuckDB workloads to MotherDuck cloud, covering connection setup, schema transfer, data loading, and validation before decommissioning legacy stores.

MotherDuck cutover
schema parity
bulk load paths
connection setup

Motherduck Migrate To Motherduck by the numbers

250 all-time installs (skills.sh)
+16 installs in the week ending Aug 4, 2026 (Skillselion tracking)
Ranked #202 of 911 Databases skills by installs in the Skillselion catalog
Data as of Aug 4, 2026 (Skillselion catalog sync)

npx skills add https://github.com/motherduckdb/agent-skills --skill motherduck-migrate-to-motherduck

Add your badge

Show developers this skill is listed on Skillselion. Paste this into your README.

[![Listed on Skillselion](https://skillselion.com/badge/skills/motherduckdb/agent-skills/motherduck-migrate-to-motherduck.svg)](https://skillselion.com/skills/motherduckdb/agent-skills/motherduck-migrate-to-motherduck)

Installs	250
repo stars	★ 53
Last updated	July 31, 2026
Repository	motherduckdb/agent-skills ↗

What it does

Plan and execute migration of analytical workloads from local DuckDB, warehouses, or files into MotherDuck with schema, connection, and cutover guidance.

Files

SKILL.mdMarkdownGitHub ↗

Migrate to MotherDuck

Use this skill when the user needs a migration plan from another warehouse, PostgreSQL estate, or mixed analytics stack onto MotherDuck.

This is a use-case skill. It orchestrates motherduck-connect, motherduck-explore, motherduck-load-data, motherduck-model-data, motherduck-query, and motherduck-ducklake.

Start Here: Is a MotherDuck Server Active?

Always determine this before writing a migration plan.

If a remote MotherDuck MCP server or local MotherDuck server is active, use it.
Ask which MotherDuck database or workspace will receive the migration if the user has not specified it.
Explore the live target side first when available:
existing databases and schemas
current landing zones
current analytical tables
naming conventions
any partial migration already in place

Also capture the source-side shape:

source platform
source table grain
key metrics
validation keys
serving workloads after cutover

If no server is active, ask for representative source and target schemas before finalizing the migration plan.

Use This Skill When

The user is moving from Snowflake, Redshift, Postgres, or similar.
The user needs cutover sequencing and validation.
The user needs to decide between native MotherDuck, pg_duckdb, or DuckLake.
The migration plan needs rollback, not just a list of copy commands.

Migration Defaults

native MotherDuck storage first
pg_duckdb when extending an existing PostgreSQL estate is the least disruptive path
validate before cutover
port SQL dialect and data types deliberately before performance tuning
phased cutover over big-bang replacement

Workflow

1. Confirm whether live MotherDuck discovery is available. 2. Classify the source system and the target serving pattern. 3. Inspect the target-side MotherDuck layout if available. 4. Pick the connection and ingestion path. 5. Inventory incompatible SQL, functions, data types, and operational assumptions. 6. Rebuild the analytical model in DuckDB SQL. 7. Run source-vs-target validation. 8. Cut over one workload at a time.

When this skill produces a native DuckDB (md:) connection, watermark it with custom_user_agent=agent-skills/2.3.0(harness-<harness>;llm-<llm>). If metadata is missing, fall back to harness-unknown and llm-unknown.

Output

The output of this skill should be:

the target pattern
the migration sequence
the validation plan
the rollback path
the first cutover slice

If the caller explicitly asks for structured JSON, return raw JSON only with no Markdown fences or prose before/after it. This is mainly for automated tests, regression checks, or downstream tooling that needs a stable machine-readable shape. Normal human-facing use of the skill can stay in prose unless JSON is explicitly requested.

Use this exact top-level shape when JSON is requested:

{
  "summary": {},
  "assumptions": [],
  "implementation_plan": [],
  "validation_plan": [],
  "risks": []
}

References

Read these as reference, not as scripts to execute:

references/MIGRATION_PLAYBOOK.md -- target-pattern selection, migration decision matrix, DuckLake posture, and source-specific questions (Snowflake, Redshift, Postgres, dbt, lakehouse)
references/MIGRATION_VALIDATION.md -- copy-adaptable validation SQL (row counts, metrics with pct_variance, new/deleted/changed records) and a Python orchestrator

Runnable Artifact

artifacts/migration_validation_example.py -- MotherDuck-backed Python example for source-vs-target validation and variance reporting
artifacts/migration_validation_example.ts -- TypeScript companion artifact with the same validation output contract

Run it with:

uv run --with duckdb python skills/motherduck-migrate-to-motherduck/artifacts/migration_validation_example.py

Run the same validation flow against temporary MotherDuck databases:

MOTHERDUCK_ARTIFACT_USE_MOTHERDUCK=1 \
uv run --with duckdb python skills/motherduck-migrate-to-motherduck/artifacts/migration_validation_example.py

Validate the TypeScript companion artifact:

uv run scripts/test_typescript_artifacts.py

Related Skills

motherduck-connect -- choose the connection path for the target system
motherduck-explore -- inspect the target-side MotherDuck workspace
motherduck-load-data -- bulk movement and raw landing patterns
motherduck-model-data -- shape the target analytical model
motherduck-query -- port and validate critical SQL
motherduck-ducklake -- only when open-table-format requirements are explicit

import json
import sys
from pathlib import Path

import duckdb

sys.path.append(str(Path(__file__).resolve().parents[3]))

from scripts._lib.motherduck_artifact_utils import artifact_session


def compare_metrics(conn: duckdb.DuckDBPyConnection, source_table: str, target_table: str, column: str) -> dict:
    results = {}
    for agg in ["count(*)", f"SUM({column})", f"AVG({column})", f"MIN({column})", f"MAX({column})"]:
        src = conn.execute(f"SELECT CAST({agg} AS DOUBLE) FROM {source_table}").fetchone()[0]
        tgt = conn.execute(f"SELECT CAST({agg} AS DOUBLE) FROM {target_table}").fetchone()[0]
        pct = round(100.0 * (tgt - src) / src, 4) if src else None
        results[agg] = {"source": src, "target": tgt, "pct_variance": pct}
    return results


def main() -> None:
    with artifact_session(
        slug="motherduck-migrate-to-motherduck",
        database_keys=["legacy_source", "motherduck_target"],
    ) as session:
        conn = session.conn
        source_table = session.table("legacy_source", "main", "orders")
        target_table = session.table("motherduck_target", "main", "orders")
        conn.execute(f"CREATE TABLE {source_table} (order_id INTEGER, total_amount DOUBLE)")
        conn.execute(f"CREATE TABLE {target_table} (order_id INTEGER, total_amount DOUBLE)")
        conn.executemany(
            f"INSERT INTO {source_table} VALUES (?, ?)",
            [(1, 100.0), (2, 150.0), (3, 200.0)],
        )
        conn.executemany(
            f"INSERT INTO {target_table} VALUES (?, ?)",
            [(1, 100.0), (2, 150.0), (4, 210.0)],
        )

        result = {
            "backend": session.describe(),
            "metric_comparison": compare_metrics(conn, source_table, target_table, "total_amount"),
            "new_records": conn.execute(
                f"SELECT order_id FROM {target_table} EXCEPT SELECT order_id FROM {source_table}"
            ).fetchall(),
            "deleted_records": conn.execute(
                f"SELECT order_id FROM {source_table} EXCEPT SELECT order_id FROM {target_table}"
            ).fetchall(),
        }
        print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

export {};
declare const process: { env: Record<string, string | undefined> };

type OrderRow = { order_id: number; total_amount: number };

function normalizeMetadataValue(value: string | undefined, fallback: string): string {
  const raw = (value ?? "").trim();
  if (!raw) return fallback;
  const normalized = raw.replace(/[^A-Za-z0-9._-]+/g, "-").replace(/^[-._]+|[-._]+$/g, "");
  return normalized || fallback;
}

function buildUseCaseUserAgent(): string {
  const harness = normalizeMetadataValue(process.env.MOTHERDUCK_AGENT_HARNESS, "unknown");
  const llm = normalizeMetadataValue(process.env.MOTHERDUCK_AGENT_LLM, "unknown");
  return `agent-skills/2.3.0(harness-${harness};llm-${llm})`;
}

function aggregate(rows: OrderRow[], kind: "count" | "sum" | "avg" | "min" | "max"): number {
  if (kind === "count") return rows.length;
  const values = rows.map((row) => row.total_amount);
  if (kind === "sum") return values.reduce((sum, value) => sum + value, 0);
  if (kind === "avg") return values.reduce((sum, value) => sum + value, 0) / values.length;
  if (kind === "min") return Math.min(...values);
  return Math.max(...values);
}

function compareMetric(source: OrderRow[], target: OrderRow[], kind: "count" | "sum" | "avg" | "min" | "max") {
  const sourceValue = aggregate(source, kind);
  const targetValue = aggregate(target, kind);
  return {
    source: sourceValue,
    target: targetValue,
    pct_variance: sourceValue ? Number((((targetValue - sourceValue) / sourceValue) * 100).toFixed(4)) : null,
  };
}

const sourceRows: OrderRow[] = [
  { order_id: 1, total_amount: 100.0 },
  { order_id: 2, total_amount: 150.0 },
  { order_id: 3, total_amount: 200.0 },
];

const targetRows: OrderRow[] = [
  { order_id: 1, total_amount: 100.0 },
  { order_id: 2, total_amount: 150.0 },
  { order_id: 4, total_amount: 210.0 },
];

const sourceIds = new Set(sourceRows.map((row) => row.order_id));
const targetIds = new Set(targetRows.map((row) => row.order_id));

const result = {
  backend: {
    mode: "typescript-companion",
    databases: { legacy_source: "legacy_source", motherduck_target: "motherduck_target" },
    user_agent: buildUseCaseUserAgent(),
  },
  metric_comparison: {
    "count(*)": compareMetric(sourceRows, targetRows, "count"),
    "SUM(total_amount)": compareMetric(sourceRows, targetRows, "sum"),
    "AVG(total_amount)": compareMetric(sourceRows, targetRows, "avg"),
    "MIN(total_amount)": compareMetric(sourceRows, targetRows, "min"),
    "MAX(total_amount)": compareMetric(sourceRows, targetRows, "max"),
  },
  new_records: targetRows.filter((row) => !sourceIds.has(row.order_id)).map((row) => [row.order_id]),
  deleted_records: sourceRows.filter((row) => !targetIds.has(row.order_id)).map((row) => [row.order_id]),
};

console.log(JSON.stringify(result, null, 2));

Migrate to MotherDuck

Use this skill when the user needs a migration plan from an existing warehouse, database, or analytics stack onto MotherDuck. This is a use-case skill: it combines connection strategy, ingestion, modeling, query migration, and rollout sequencing into one plan.

Source of truth and verified delivery defaults
Validation Signals (maintainer/reviewer checks)
Language focus and starter snippets (TypeScript cutover, Python validation)
Official product anchors (pg_duckdb, Hypertenancy, read scaling, DuckLake)
Step 1-6: classify, pick target pattern, move data, rebuild model, validate, cut over
Migration decision matrix
DuckLake guidance
Source-specific questions (Snowflake, Redshift, Postgres, dbt, lakehouse)

Source Of Truth

Prefer current MotherDuck public documentation and product pages first.
If the MotherDuck MCP ask_docs_question feature is available, use it before falling back to general web search.
For migration decisions, verify current guidance on:
connection paths
pg_duckdb
Hypertenancy and read scaling
DuckLake
If ask_docs_question is unavailable, use public pages on motherduck.com and motherduck.com/docs.

Verified Delivery Defaults

Defaults that hold across migrations:

decide the target MotherDuck pattern before arguing about tooling
migrate in slices with source-vs-target validation at each step
treat metric comparison and missing-key checks as mandatory, not optional
keep rollback and cutover posture explicit in the first migration plan

Validation Signals

Use these signals for testing, review, and regression checks. They are not an instruction to include a separate "Validation Signals" section in normal user-facing replies.

run artifacts/migration_validation_example.py against temporary MotherDuck databases
verify the output contains metric comparison plus new_records and deleted_records
require an explicit acceptable variance posture for every migration slice
treat plans without rollback and cutover checkpoints as incomplete

Language Focus: TypeScript/Javascript and Python

Prefer Python for migration execution examples:
extract/load scripts
validation checks
data comparison jobs
migration notebooks and cutover helpers
Prefer TypeScript/Javascript when the migration is really about:
moving a product backend to MotherDuck
preserving Node.js service interfaces
re-pointing app-side query paths
If the task includes both product and data movement, show Python for migration mechanics and TypeScript/Javascript for the app cutover path.

TypeScript/Javascript Cutover Starter

type CustomerQueryTarget = {
  mode: "legacy-postgres" | "motherduck-pg" | "pg-duckdb";
  database: string;
};

const rolloutMap: Record<string, CustomerQueryTarget> = {
  acme: { mode: "motherduck-pg", database: "customer_acme" },
  globex: { mode: "legacy-postgres", database: "globex_prod" },
};

Python Validation Starter

import duckdb

def compare_metrics(conn, source_table: str, target_table: str, column: str) -> dict:
    """Compare a numeric column between source and target with % variance."""
    results = {}
    for agg in ["count(*)", f"SUM({column})", f"AVG({column})", f"MIN({column})", f"MAX({column})"]:
        src = conn.sql(f"SELECT {agg}::DOUBLE FROM {source_table}").fetchone()[0]
        tgt = conn.sql(f"SELECT {agg}::DOUBLE FROM {target_table}").fetchone()[0]
        pct = round(100.0 * (tgt - src) / src, 4) if src else None
        results[agg] = {"source": src, "target": tgt, "pct_variance": pct}
    return results

def find_missing_keys(conn, source_table: str, target_table: str, key_col: str) -> dict:
    """Find new, deleted, and changed records between source and target."""
    new = conn.sql(
        f"SELECT {key_col} FROM {target_table} EXCEPT SELECT {key_col} FROM {source_table}"
    ).fetchall()
    deleted = conn.sql(
        f"SELECT {key_col} FROM {source_table} EXCEPT SELECT {key_col} FROM {target_table}"
    ).fetchall()
    return {"new_records": len(new), "deleted_records": len(deleted)}

See references/MIGRATION_VALIDATION.md for the full validation suite: row counts, metric comparisons with % variance, uniqueness checks, new/deleted/changed record tracking, and a Python orchestrator.

Official Product Anchors To Use

pg_duckdb is the official path for adding analytical power to an existing PostgreSQL estate. MotherDuck describes it as a way to keep OLTP fast while handling OLAP through DuckDB, with support for joining PostgreSQL and cloud data and even zero-data-movement analytics on existing PostgreSQL data.
Hypertenancy is the official pattern for giving each customer or user isolated compute. MotherDuck documents one Duckling per user or customer, provisioned automatically per service account.
Read Scaling is the official answer for read-heavy workloads like BI dashboards or high-concurrency read-only apps.
DuckLake is explicitly opt-in. MotherDuck positions it for open-table-format and large lakehouse-style needs, while native MotherDuck storage remains the simpler default for many migrations.

Step 1: Classify the Starting Point

Workload classes:

warehouse replacement
Postgres extension or hybrid analytics
app-serving migration
dashboard and BI migration
lakehouse or open-table-format migration

Do not design the destination before classifying the current stack.

Step 2: Pick the Target Pattern

Use the PG endpoint when the environment already assumes PostgreSQL wire compatibility.
Use the native DuckDB API when local files, hybrid queries, or rich DuckDB control matter.
Use pg_duckdb when extending an existing PostgreSQL estate is the least disruptive path. MotherDuck's public Postgres Integration guidance emphasizes:
analytical acceleration inside PostgreSQL
joins across PostgreSQL, MotherDuck, and object storage
zero-data-movement analytics on existing PostgreSQL data
hybrid workload optimization so OLTP stays in PostgreSQL while OLAP moves to DuckDB
Use DuckLake only when open-table-format requirements are explicit.

Important migration gotcha:

the PG endpoint still runs DuckDB SQL, not PostgreSQL SQL
do not assume PostgreSQL-specific syntax, temp-table habits, local-file imports, or extension management will survive unchanged over the PG endpoint
when the migration depends on local DuckDB features, use a native DuckDB client path instead of forcing everything through PostgreSQL drivers

Migration Decision Matrix

Source is PostgreSQL and the team wants minimal disruption:
start with pg_duckdb
keep transactional paths in PostgreSQL
offload analytical paths to MotherDuck only where needed
Source is a warehouse and the team wants a cleaner MotherDuck landing zone:
move data into native MotherDuck storage first
rebuild the analytics model in DuckDB SQL
add Hypertenancy or read scaling later if the serving workload demands it
Source is an Iceberg or data-lake estate:
evaluate DuckLake only if open-table-format interoperability or bring-your-own-bucket requirements are real
do not default to DuckLake just because the source system was lake-based

Step 3: Move Data

Use motherduck-load-data patterns for the raw move:

Parquet for bulk movement when you control extracts
cloud object storage for staged imports
append-only raw landing first, then transform
validate row counts and key aggregates after every load
avoid row-by-row insert loops; prefer bulk paths, Arrow/dataframes, or COPY

Prefer these patterns:

use direct cloud-to-MotherDuck ingest when the source data already lives in object storage
keep raw, staging, and analytics boundaries explicit during cutover
preserve rollback by leaving the old source of truth untouched until validations pass

Step 4: Rebuild the Analytical Model

Prefer wide analytical tables for BI and dashboard workloads.
Keep raw, staging, and analytics boundaries explicit.
Rework source-specific SQL into DuckDB SQL where needed.
Validate every critical join, aggregate, and date transformation.

Specific rewrite checks:

Postgres-specific SQL and extensions
warehouse-specific DDL assumptions
dbt macros that assume another engine
nested JSON and semi-structured data behavior
time travel or snapshot workflows that need a MotherDuck-native equivalent

Step 5: Validate Correctness

Run source-vs-target checks before cutting over. Every check should output a pct_variance so the user can decide what is acceptable.

1. Row counts — compare total rows between source and target. 2. Metric comparison — compare SUM, AVG, MIN, MAX on key numeric columns side by side. 3. Uniqueness — verify the target has no duplicate keys introduced by the migration. 4. New records — identify IDs in the target that do not exist in the source. 5. Deleted records — identify IDs in the source that are missing from the target. 6. Changed records — find records present in both but with different values. Track the specific IDs. 7. % variance — report variance on every metric so the user can set their own threshold for pass/fail.

Whether the migration is a 1:1 port or an intentional refactor determines what variance is acceptable. The skill provides the measurements; the user decides.

Step 6: Cut Over Safely

run old and new outputs side by side
compare row counts and business metrics using the validation patterns above
cut over one workload or consumer at a time
keep rollback simple until confidence is earned

When the target workload is user-facing:

move to Hypertenancy before general availability if strong tenant isolation is a hard requirement
add read scaling only after concurrency is proven to be the bottleneck
keep one service account and token boundary per customer or workload slice rather than sharing a single broad token

DuckLake Guidance

Use DuckLake when the user explicitly needs one of these:

open-table-format storage
bring-your-own-bucket storage ownership
use of their own compute against the same storage
migration from Iceberg-oriented lake workflows

Do not recommend DuckLake by default when:

the workload is mainly warehouse-style analytics
the user wants the simplest managed path
the team does not have a concrete interoperability or storage-ownership requirement

MotherDuck's public DuckLake guidance currently distinguishes:

fully-managed DuckLake for the easiest start
bring-your-own-bucket DuckLake when storage must remain in the user's cloud
use-your-own-compute scenarios only with bring-your-own-bucket setups today

MotherDuck's public DuckLake guidance also says native MotherDuck storage reads are often materially faster than DuckLake for normal analytical reads. That means warehouse migrations should stay native unless the open-table-format requirement is real.

If the migration really needs bring-your-own-bucket DuckLake:

keep the bucket in the same region as the target MotherDuck deployment when possible
plan for explicit maintenance behavior instead of assuming MotherDuck will compact and maintain files automatically

Source-Specific Questions To Answer

Snowflake: what external functions, orchestration, or security assumptions need replacement?
Redshift: what distribution-key or cluster assumptions disappear?
Postgres: is pg_duckdb enough, or should the workload move fully to MotherDuck? Are zero-data-movement analytics or hybrid operational/analytical joins enough for phase one?
dbt-heavy stacks: which models move unchanged, and which need DuckDB-specific rewrites?
Lakehouse source: does the user actually need DuckLake, or would managed MotherDuck storage simplify the migration?

The output of this skill should be a phased migration plan, not just a list of features.

Migration Validation Reference

Concrete SQL patterns and a Python orchestrator for validating that a migration to MotherDuck produced correct results. Every query outputs a pct_variance column so the user can decide what is acceptable.

Row count comparison
Side-by-side metric comparison
Uniqueness check on target
New / deleted records (EXCEPT queries)
Changed records tracking (column-level and hash-based)
Python validation orchestrator (validate_migration, print_report)
Investigating non-zero variance

---

Row Count Comparison

Compare total row counts between source and target with percentage variance.

WITH counts AS (
    SELECT
        'source' AS side,
        count(*) AS row_count
    FROM "source_db"."main"."orders"
    UNION ALL
    SELECT
        'target' AS side,
        count(*) AS row_count
    FROM "target_db"."main"."orders"
)
SELECT
    MAX(row_count) FILTER (WHERE side = 'source') AS source_rows,
    MAX(row_count) FILTER (WHERE side = 'target') AS target_rows,
    MAX(row_count) FILTER (WHERE side = 'target')
        - MAX(row_count) FILTER (WHERE side = 'source') AS row_diff,
    ROUND(
        100.0
        * (MAX(row_count) FILTER (WHERE side = 'target')
           - MAX(row_count) FILTER (WHERE side = 'source'))
        / NULLIF(MAX(row_count) FILTER (WHERE side = 'source'), 0),
        2
    ) AS pct_variance
FROM counts;

---

Side-by-Side Metric Comparison

Compare key aggregates between source and target. Replace amount and quantity with your numeric columns.

WITH source_metrics AS (
    SELECT
        count(*) AS row_count,
        SUM(amount) AS sum_amount,
        AVG(amount) AS avg_amount,
        MIN(amount) AS min_amount,
        MAX(amount) AS max_amount,
        SUM(quantity) AS sum_quantity,
        AVG(quantity) AS avg_quantity
    FROM "source_db"."main"."orders"
),
target_metrics AS (
    SELECT
        count(*) AS row_count,
        SUM(amount) AS sum_amount,
        AVG(amount) AS avg_amount,
        MIN(amount) AS min_amount,
        MAX(amount) AS max_amount,
        SUM(quantity) AS sum_quantity,
        AVG(quantity) AS avg_quantity
    FROM "target_db"."main"."orders"
),
comparisons AS (
    SELECT unnest([
        {'metric': 'row_count',    'source': s.row_count::DOUBLE,    'target': t.row_count::DOUBLE},
        {'metric': 'sum_amount',   'source': s.sum_amount::DOUBLE,   'target': t.sum_amount::DOUBLE},
        {'metric': 'avg_amount',   'source': s.avg_amount::DOUBLE,   'target': t.avg_amount::DOUBLE},
        {'metric': 'min_amount',   'source': s.min_amount::DOUBLE,   'target': t.min_amount::DOUBLE},
        {'metric': 'max_amount',   'source': s.max_amount::DOUBLE,   'target': t.max_amount::DOUBLE},
        {'metric': 'sum_quantity', 'source': s.sum_quantity::DOUBLE, 'target': t.sum_quantity::DOUBLE},
        {'metric': 'avg_quantity', 'source': s.avg_quantity::DOUBLE, 'target': t.avg_quantity::DOUBLE}
    ]) AS r
    FROM source_metrics s, target_metrics t
)
SELECT
    r.metric AS metric_name,
    r.source AS source_value,
    r.target AS target_value,
    ROUND(r.target - r.source, 4) AS abs_diff,
    ROUND(
        100.0 * (r.target - r.source) / NULLIF(r.source, 0), 2
    ) AS pct_variance
FROM comparisons;

---

Uniqueness Check on Target

Verify the migration did not introduce duplicate records. Replace order_id with your primary key column.

SELECT
    order_id,
    count(*) AS duplicate_count
FROM "target_db"."main"."orders"
GROUP BY order_id
HAVING count(*) > 1
ORDER BY duplicate_count DESC;

An empty result set means no duplicates exist.

---

New Records (In Target, Not In Source)

Identify records that appear in the target but not in the source. These may be expected (if the migration included new data) or a problem.

SELECT order_id
FROM "target_db"."main"."orders"
EXCEPT
SELECT order_id
FROM "source_db"."main"."orders";

Count them:

SELECT count(*) AS new_record_count
FROM (
    SELECT order_id FROM "target_db"."main"."orders"
    EXCEPT
    SELECT order_id FROM "source_db"."main"."orders"
);

---

Deleted Records (In Source, Not In Target)

Identify records that exist in the source but are missing from the target.

SELECT order_id
FROM "source_db"."main"."orders"
EXCEPT
SELECT order_id
FROM "target_db"."main"."orders";

Count them:

SELECT count(*) AS deleted_record_count
FROM (
    SELECT order_id FROM "source_db"."main"."orders"
    EXCEPT
    SELECT order_id FROM "target_db"."main"."orders"
);

---

Changed Records Tracking

Find records that exist in both source and target but have different values. Replace column names with your own.

Column-Level Comparison

Use IS DISTINCT FROM instead of <> to handle NULLs correctly.

SELECT
    s.order_id,
    s.amount AS source_amount,
    t.amount AS target_amount,
    s.status AS source_status,
    t.status AS target_status
FROM "source_db"."main"."orders" s
JOIN "target_db"."main"."orders" t
    ON s.order_id = t.order_id
WHERE s.amount IS DISTINCT FROM t.amount
   OR s.status IS DISTINCT FROM t.status;

Hash-Based Comparison for Wide Tables

When a table has many columns, compare row hashes instead of listing every column.

WITH source_hashed AS (
    SELECT
        order_id,
        md5(COLUMNS(* EXCLUDE (order_id))::VARCHAR) AS row_hash
    FROM "source_db"."main"."orders"
),
target_hashed AS (
    SELECT
        order_id,
        md5(COLUMNS(* EXCLUDE (order_id))::VARCHAR) AS row_hash
    FROM "target_db"."main"."orders"
)
SELECT
    COALESCE(s.order_id, t.order_id) AS order_id,
    s.row_hash AS source_hash,
    t.row_hash AS target_hash
FROM source_hashed s
JOIN target_hashed t
    ON s.order_id = t.order_id
WHERE s.row_hash IS DISTINCT FROM t.row_hash;

Once you identify changed IDs via hashing, use the column-level comparison query filtered to those IDs to see exactly what changed.

Performance Note

For large tables, filter both sides to a date range or partition before comparing:

-- Add a WHERE clause to both source and target CTEs
WHERE order_date >= '2024-01-01' AND order_date < '2024-02-01'

---

Python Validation Orchestrator

Runs all checks and returns a structured report. Uses the DuckDB Python API.

"""
Migration Validation Orchestrator
Runs source-vs-target checks and reports variance.

Install: pip install duckdb
"""

import duckdb


def validate_migration(
    source_conn: duckdb.DuckDBPyConnection,
    target_conn: duckdb.DuckDBPyConnection,
    source_table: str,
    target_table: str,
    key_column: str,
    numeric_columns: list[str],
    variance_threshold_pct: float = 0.0,
) -> dict:
    """
    Run all migration validation checks.

    Args:
        source_conn: Connection to the source database.
        target_conn: Connection to the target database.
        source_table: Fully qualified source table (e.g., '"source_db"."main"."orders"').
        target_table: Fully qualified target table (e.g., '"target_db"."main"."orders"').
        key_column: Primary key column name for record-level comparisons.
        numeric_columns: List of numeric column names for metric comparisons.
        variance_threshold_pct: Acceptable % variance. 0.0 means exact match required.

    Returns:
        Dict with results for each check and an overall pass/fail.
    """
    results = {}

    # --- Row counts ---
    source_count = source_conn.sql(f"SELECT count(*) FROM {source_table}").fetchone()[0]
    target_count = target_conn.sql(f"SELECT count(*) FROM {target_table}").fetchone()[0]
    count_variance = (
        round(100.0 * (target_count - source_count) / source_count, 2)
        if source_count > 0
        else None
    )
    results["row_counts"] = {
        "source": source_count,
        "target": target_count,
        "diff": target_count - source_count,
        "pct_variance": count_variance,
        "pass": abs(count_variance or 0) <= variance_threshold_pct,
    }

    # --- Metric comparison ---
    metrics = {}
    for col in numeric_columns:
        for agg in ["SUM", "AVG", "MIN", "MAX"]:
            source_val = source_conn.sql(
                f"SELECT {agg}({col})::DOUBLE FROM {source_table}"
            ).fetchone()[0]
            target_val = target_conn.sql(
                f"SELECT {agg}({col})::DOUBLE FROM {target_table}"
            ).fetchone()[0]
            pct = (
                round(100.0 * (target_val - source_val) / source_val, 4)
                if source_val
                else None
            )
            metric_key = f"{agg.lower()}_{col}"
            metrics[metric_key] = {
                "source": source_val,
                "target": target_val,
                "pct_variance": pct,
                "pass": abs(pct or 0) <= variance_threshold_pct,
            }
    results["metrics"] = metrics

    # --- Uniqueness ---
    dupes = target_conn.sql(
        f"SELECT {key_column}, count(*) AS cnt FROM {target_table} "
        f"GROUP BY {key_column} HAVING cnt > 1"
    ).fetchall()
    results["uniqueness"] = {
        "duplicate_count": len(dupes),
        "duplicate_keys": [row[0] for row in dupes[:20]],
        "pass": len(dupes) == 0,
    }

    # --- New records (in target, not in source) ---
    new_ids = target_conn.sql(
        f"SELECT {key_column} FROM {target_table} "
        f"EXCEPT SELECT {key_column} FROM {source_table}"
    ).fetchall()
    results["new_records"] = {
        "count": len(new_ids),
        "sample_keys": [row[0] for row in new_ids[:20]],
        "pass": len(new_ids) == 0,
    }

    # --- Deleted records (in source, not in target) ---
    deleted_ids = source_conn.sql(
        f"SELECT {key_column} FROM {source_table} "
        f"EXCEPT SELECT {key_column} FROM {target_table}"
    ).fetchall()
    results["deleted_records"] = {
        "count": len(deleted_ids),
        "sample_keys": [row[0] for row in deleted_ids[:20]],
        "pass": len(deleted_ids) == 0,
    }

    # --- Changed records (hash comparison) ---
    changed = target_conn.sql(f"""
        WITH source_h AS (
            SELECT {key_column},
                   md5(COLUMNS(* EXCLUDE ({key_column}))::VARCHAR) AS rh
            FROM {source_table}
        ),
        target_h AS (
            SELECT {key_column},
                   md5(COLUMNS(* EXCLUDE ({key_column}))::VARCHAR) AS rh
            FROM {target_table}
        )
        SELECT s.{key_column}
        FROM source_h s
        JOIN target_h t ON s.{key_column} = t.{key_column}
        WHERE s.rh IS DISTINCT FROM t.rh
    """).fetchall()
    results["changed_records"] = {
        "count": len(changed),
        "sample_keys": [row[0] for row in changed[:20]],
        "pass": len(changed) == 0,
    }

    # --- Overall ---
    results["overall_pass"] = all(
        v.get("pass", True)
        for v in results.values()
        if isinstance(v, dict) and "pass" in v
    ) and all(
        m.get("pass", True) for m in results.get("metrics", {}).values()
    )

    return results


def print_report(results: dict) -> None:
    """Print a human-readable validation report."""
    print("=== Migration Validation Report ===\n")

    rc = results["row_counts"]
    status = "PASS" if rc["pass"] else "FAIL"
    print(f"Row Counts [{status}]: source={rc['source']}  target={rc['target']}  "
          f"diff={rc['diff']}  variance={rc['pct_variance']}%")

    print("\nMetrics:")
    for name, m in results["metrics"].items():
        status = "PASS" if m["pass"] else "FAIL"
        print(f"  {name} [{status}]: source={m['source']}  target={m['target']}  "
              f"variance={m['pct_variance']}%")

    u = results["uniqueness"]
    status = "PASS" if u["pass"] else "FAIL"
    print(f"\nUniqueness [{status}]: {u['duplicate_count']} duplicate keys found")

    nr = results["new_records"]
    status = "PASS" if nr["pass"] else "FAIL"
    print(f"New Records [{status}]: {nr['count']} records in target not in source")

    dr = results["deleted_records"]
    status = "PASS" if dr["pass"] else "FAIL"
    print(f"Deleted Records [{status}]: {dr['count']} records in source not in target")

    cr = results["changed_records"]
    status = "PASS" if cr["pass"] else "FAIL"
    print(f"Changed Records [{status}]: {cr['count']} records with different values")

    overall = "PASS" if results["overall_pass"] else "FAIL"
    print(f"\n=== Overall: {overall} ===")

---

Investigating Non-Zero Variance

When validation reports non-zero variance, do not treat it as an automatic failure. Investigate in order:

1. Row count differs? Check for new or deleted records first. Use the EXCEPT queries above to find exactly which keys are affected. New records may be expected if the migration included a data refresh.

2. Aggregates differ but row count matches? Run the column-level comparison to find changed rows. Common causes:

floating-point rounding differences between source and DuckDB (usually < 0.01%)
timezone handling differences on timestamps that affect date-based aggregations
NULL handling differences (NULL + 5 returns NULL in DuckDB, some sources treat it as 5)

3. Small variance (< 0.5%)? Document it and decide with the user whether it is acceptable. Many migrations accept small rounding variance on financial aggregates.

4. Large variance (> 1%)? Narrow it down to specific rows. Filter the metric comparison to date ranges, customer segments, or product categories to isolate the affected partition. The usual suspects are:

duplicate rows in either source or target
a WHERE clause in the migration that excluded records
type coercion that changed values (e.g., truncating DECIMAL precision)

5. Hash comparison finds changed rows? Use the column-level comparison filtered to those keys to see exactly which columns changed. This is the fastest path to root cause.

---

Usage Example

import duckdb

# Connect to both databases
USE_CASE_USER_AGENT = "agent-skills/2.3.0(harness-<harness>;llm-<llm>)"
conn = duckdb.connect(f"md:?custom_user_agent={USE_CASE_USER_AGENT}")

# If source is a Postgres database (local DuckDB only):
# conn.sql("ATTACH 'dbname=legacy host=pg.example.com' AS source_db (TYPE POSTGRES, READ_ONLY)")

results = validate_migration(
    source_conn=conn,
    target_conn=conn,
    source_table='"source_db"."main"."orders"',
    target_table='"target_db"."main"."orders"',
    key_column="order_id",
    numeric_columns=["amount", "quantity"],
    variance_threshold_pct=0.5,  # allow 0.5% variance
)

print_report(results)