Schema Evolution and Data Contracts

Every dataset has a schema, whether it is declared or not. The question is only whether a change to it is caught in a pull request or discovered by a dashboard at 9am.

Which changes are safe

• Backward compatible — new consumer reads old data. Needed when consumers upgrade first.
• Forward compatible — old consumer reads new data. Needed when producers upgrade first, which is the normal case for a shared dataset with many readers.
• Full — both. What a widely consumed dataset should require.

The last row is the dangerous one. Redefining amount from gross to net passes every structural check ever written and silently changes every number downstream.

Evolution in practice

import duckdb
con = duckdb.connect()

con.execute("""
    copy (select * from (values
        (1001, 1, date '2026-01-04', 'completed', 25.50),
        (1002, 2, date '2026-01-04', 'completed', 12.00)
    ) t(order_id, customer_id, ordered_at, status, amount))
    to 'orders_v1.parquet' (format parquet)
""")

con.execute("""
    copy (select * from (values
        (1003, 1, date '2026-01-05', 'completed', 40.00, 'web'),
        (1004, 3, date '2026-01-05', 'refunded',   8.75, 'app')
    ) t(order_id, customer_id, ordered_at, status, amount, channel))
    to 'orders_v2.parquet' (format parquet)
""")

print(con.execute("""
    select * from read_parquet(['orders_v1.parquet','orders_v2.parquet'], union_by_name = true)
    order by order_id
""").df().to_string(index=False))

 order_id  customer_id ordered_at    status  amount channel
     1001            1 2026-01-04 completed   25.50    None
     1002            2 2026-01-04 completed   12.00    None
     1003            1 2026-01-05 completed   40.00     web
     1004            3 2026-01-05  refunded    8.75     app

An added nullable column reads cleanly across both files — old rows get NULL. That is the one change you can make freely, and the reason “add a column” is the safe answer to almost every schema request.

Without union_by_name the same read fails:

con.execute("select * from read_parquet(['orders_v1.parquet','orders_v2.parquet'])").df()

duckdb.duckdb.InvalidInputException: Invalid Input Error: Schema mismatch between files.
Column count mismatch: expected 5 columns but found 6.
Use union_by_name = true to read files with different schemas.

The rename

con.execute("""
    copy (select * from (values
        (1005, 2, date '2026-01-06', 'completed', 63.20, 'web')
    ) t(order_id, customer_id, ordered_at, status, amount_gbp, channel))
    to 'orders_v3.parquet' (format parquet)
""")

print(con.execute("""
    select order_id, amount, amount_gbp
    from read_parquet(['orders_v2.parquet','orders_v3.parquet'], union_by_name = true)
    order by order_id
""").df().to_string(index=False))

 order_id  amount  amount_gbp
     1003   40.00        None
     1004    8.75        None
     1005    None       63.20

Two half-empty columns, and every existing query — sum(amount) — silently starts excluding new rows. No error anywhere.

The safe version is an expand-and-contract migration:

# phase 1: write both, keep the old one authoritative
con.execute("""
    copy (select order_id, amount, amount as amount_gbp from read_parquet('orders_v2.parquet'))
    to 'orders_dual.parquet' (format parquet)
""")
print(con.execute("select * from 'orders_dual.parquet'").df().to_string(index=False))

 order_id  amount  amount_gbp
     1003   40.00       40.00
     1004    8.75        8.75

Then migrate consumers, verify nobody reads the old name, and only then drop it. Three deploys instead of one — which is the actual cost of a rename on a shared dataset, and the reason to get names right the first time.

A contract

# contracts/orders.yml
dataset: bookshop.silver.orders
version: 2.1.0
owner: [email protected]
description: One row per non-pending order. Amounts are gross, in GBP, excluding shipping.

schema:
  - name: order_id
    type: bigint
    required: true
    unique: true
  - name: customer_id
    type: bigint
    required: true
    description: References bookshop.silver.customers.customer_id
  - name: ordered_at
    type: date
    required: true
  - name: status
    type: string
    required: true
    accepted_values: [completed, returned, refunded]
  - name: amount
    type: "decimal(10,2)"
    required: true
    description: Gross order total in GBP, excluding shipping. Refunds are negative.
  - name: channel
    type: string
    required: false        # added in 2.1.0
    accepted_values: [web, app, phone]

guarantees:
  freshness: 6h
  compatibility: full
  volume_change_threshold: 0.5

The parts that matter are not the column list. compatibility: full says what changes are permitted, freshness and volume_change_threshold say what the consumer can rely on operationally, and the amount description pins down the semantics that no type can express.

Enforcing it in CI

# check_contract.py
import duckdb, sys, yaml

TYPE_MAP = {"bigint": "BIGINT", "date": "DATE", "string": "VARCHAR", "decimal(10,2)": "DECIMAL(10,2)"}

def check(contract_path, table):
    contract = yaml.safe_load(open(contract_path))
    con = duckdb.connect("bookshop.duckdb")
    actual = {r[0]: r[1] for r in con.execute(f"describe {table}").fetchall()}
    expected = {f["name"]: f for f in contract["schema"]}

    breaking, additive = [], []

    for name, field in expected.items():
        if name not in actual:
            breaking.append(f"missing column '{name}'")
        elif actual[name] != TYPE_MAP[field["type"]]:
            breaking.append(f"'{name}': contract says {TYPE_MAP[field['type']]}, table has {actual[name]}")

    for name in actual:
        if name not in expected:
            additive.append(f"undeclared column '{name}'")

    for f in contract["schema"]:
        if f.get("required") and con.execute(f"select count(*) from {table} where {f['name']} is null").fetchone()[0]:
            breaking.append(f"'{f['name']}' is required but contains nulls")
        if f.get("unique") and con.execute(
            f"select count(*) - count(distinct {f['name']}) from {table}").fetchone()[0]:
            breaking.append(f"'{f['name']}' is not unique")
        if f.get("accepted_values"):
            vals = "', '".join(f["accepted_values"])
            bad = con.execute(f"select count(*) from {table} where {f['name']} not in ('{vals}')").fetchone()[0]
            if bad:
                breaking.append(f"'{f['name']}': {bad} rows outside accepted values")

    for a in additive:
        print(f"ADDITIVE  {a}")
    for b in breaking:
        print(f"BREAKING  {b}")
    if breaking:
        sys.exit(f"\ncontract {contract['dataset']} v{contract['version']} violated: {len(breaking)} breaking change(s)")
    print(f"\ncontract {contract['dataset']} v{contract['version']} satisfied")

check("contracts/orders.yml", "bookshop.silver.orders")

python check_contract.py

contract bookshop.silver.orders v2.1.0 satisfied

Now rename the column at the source and re-run:

BREAKING  missing column 'amount'
ADDITIVE  undeclared column 'amount_gbp'

contract bookshop.silver.orders v2.1.0 violated: 1 breaking change(s)

# .github/workflows/contracts.yml
- run: python check_contract.py

Run python check_contract.py
BREAKING  missing column 'amount'
Error: Process completed with exit code 1

The producer’s pull request is red. That is the entire point — a contract that is only checked at read time tells the consumer they are broken, which they already knew.

Handling what you did not expect

Contracts work when both sides participate. For a feed you do not control, keep unknown fields instead of failing on them:

con.execute("""
    create or replace table bronze_orders as
    select
        try_cast(json_extract_string(payload, '$.order_id')  as bigint)        as order_id,
        try_cast(json_extract_string(payload, '$.amount')    as decimal(10,2)) as amount,
        json_extract_string(payload, '$.status')                               as status,
        payload                                                                as _raw,
        now()                                                                  as _ingested_at
    from (values
        ('{"order_id":1001,"amount":"25.50","status":"completed"}'),
        ('{"order_id":1002,"amount":"12.00","status":"completed","channel":"web","promo":"NEWYEAR"}')
    ) t(payload)
""")

print(con.execute("""
    select order_id, amount,
           json_keys(_raw) as keys_present
    from bronze_orders order by order_id
""").df().to_string(index=False))

 order_id  amount                                       keys_present
     1001   25.50                  [order_id, amount, status]
     1002   12.00 [order_id, amount, status, channel, promo]

Two new fields appeared and nothing broke, because the raw payload is retained. Detect the change rather than discovering it:

print(con.execute("""
    select unnest(json_keys(_raw)) as field, count(*) as rows
    from bronze_orders group by 1 order by rows
""").df().to_string(index=False))

     field  rows
     promo     1
   channel     1
    status     2
    amount     2
  order_id     2

A field present in some rows and not others is either a new addition or a removal in progress. Alerting on a change to that key set gives you a day’s notice instead of an incident.

Versioning a dataset

bookshop.silver.orders_v1     deprecated 2026-06-30, read-only
bookshop.silver.orders_v2     current
bookshop.silver.orders        view → orders_v2

Publish both during the migration, point the unversioned name at the current version, and give consumers a deprecation date. Track who is still reading the old one before removing it:

print(con.execute("""
    select object_name, count(*) as reads, max(query_time) as last_read
    from query_log
    where object_name like '%orders_v1%' and query_time >= current_date - 30
    group by 1
""").df().to_string(index=False))

        object_name  reads           last_read
 silver.orders_v1       412 2026-09-08 14:22:00

412 reads last month means the deprecation is not done, whatever the date on the announcement. Every warehouse has this log — Snowflake’s query_history, Databricks’ system.access.audit, Postgres with pg_stat_statements — and checking it is the difference between a clean removal and an outage.

Practice

 order_id  amount channel
     1001   25.50    None
     1003   40.00     web

 order_id  amount  amount_gbp
     1003   40.00        None
     1005    None       63.20

BREAKING  missing column 'amount'
Error: Process completed with exit code 1

     field  rows
     promo     1
   channel     1
  order_id     2

That closes the data engineering track. The thread through all ten lessons: a pipeline’s correctness is decided by what happens on the second run, the late row, and the change nobody told you about — not by the transformation in the middle.