0Pricing
AI Engineering Academy · Lesson

Schema Evolution and Backward Compatibility

Manage breaking schema changes in long-running extraction pipelines by versioning schemas, migrating historical extractions, and running parallel validation during transitions.

Schema Evolution and Backward Compatibility is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Schemas Change Over Time

Extraction schemas are not static. Business requirements evolve, new document types appear, and you discover fields you should have captured from the start. Changing a schema in a live pipeline creates a backward compatibility problem: existing extracted records use the old schema, while new records use the new one. Managing this transition safely is what schema evolution is about.

Versioning Your Schemas

Assign a version number to each schema and store it alongside every extracted record. When you change the schema, increment the version. This lets you query records by schema version, run migrations on old records, and maintain separate validation logic for each version. A simple string field schema_version in every output model is sufficient.

from pydantic import BaseModel
from typing import Literal

class InvoiceV1(BaseModel):
    schema_version: Literal['1.0'] = '1.0'
    vendor: str
    total_amount: float

class InvoiceV2(BaseModel):
    schema_version: Literal['2.0'] = '2.0'
    vendor: str
    vendor_tax_id: str | None = None  # new field
    total_amount: float
    currency: str = 'USD'  # new field with default

Additive vs Breaking Changes

Additive changes are safe: adding an Optional field or a field with a default does not break old extraction code or old records. Breaking changes are risky: renaming a field, changing a type from string to int, or removing a field will break downstream consumers. Always prefer additive changes. When a breaking change is unavoidable, create a new major schema version and migrate in a controlled way.

# Safe: additive change - add optional field
class ProductV2(BaseModel):
    name: str
    price: float
    sku: str | None = None   # NEW optional field - backward safe
    category: str = 'general'  # NEW with default - backward safe

# Risky: breaking change - rename or retype
# class ProductV2(BaseModel):
#     product_name: str  # RENAMED from name - breaks consumers
#     price_cents: int   # RETYPED from float - breaks data

Storing Schema Version in the Database

Include the schema version in your extraction results table so you always know which version produced each record. A jsonb column storing the full extracted data plus a schema_version text column is a common pattern. This lets you write version-aware queries and selectively migrate older records during low-traffic windows.

-- PostgreSQL table design
CREATE TABLE extractions (
    doc_id TEXT PRIMARY KEY,
    schema_version TEXT NOT NULL,
    extracted_data JSONB NOT NULL,
    extracted_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_schema_version ON extractions(schema_version);

-- Query old records needing migration
SELECT doc_id, extracted_data
FROM extractions
WHERE schema_version = '1.0'
LIMIT 1000;

Writing Migration Scripts

Write a migration script for each schema version transition that reads old records, transforms them to the new format, and writes them back with the new version. Run migrations in small batches with transactions so a failure does not leave the database in a half-migrated state. Always keep the old schema available until migration is verified complete.

import asyncpg
import json

async def migrate_v1_to_v2(pool, batch_size=100):
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            'SELECT doc_id, extracted_data FROM extractions WHERE schema_version=$1 LIMIT $2',
            '1.0', batch_size
        )
        for row in rows:
            old = row['extracted_data']
            new_data = {
                'schema_version': '2.0',
                'vendor': old['vendor'],
                'vendor_tax_id': None,  # unknown for old records
                'total_amount': old['total_amount'],
                'currency': 'USD'  # assume USD for old records
            }
            await conn.execute(
                'UPDATE extractions SET extracted_data=$1, schema_version=$2 WHERE doc_id=$3',
                json.dumps(new_data), '2.0', row['doc_id']
            )

Parallel Validation During Transitions

During a schema migration, run parallel validation: extract with both the old and new schema simultaneously for a sample of incoming documents. Compare the results to verify the new schema captures everything the old one did plus the new fields. Only retire the old schema after parallel validation shows stable parity on a statistically significant sample.

async def parallel_validate(text: str) -> dict:
    v1_result, v2_result = await asyncio.gather(
        extract_with_schema(text, InvoiceV1),
        extract_with_schema(text, InvoiceV2)
    )
    discrepancy = (
        v1_result.vendor != v2_result.vendor or
        abs(v1_result.total_amount - v2_result.total_amount) > 0.01
    )
    if discrepancy:
        log_discrepancy(text, v1_result, v2_result)
    return {'v1': v1_result, 'v2': v2_result, 'discrepancy': discrepancy}

Feature Flags for Schema Rollout

Use feature flags to control when your pipeline switches from the old schema to the new one. This lets you gradually roll out the new schema to a percentage of traffic, monitor error rates, and instantly roll back if something goes wrong — without redeploying code. Feature flag services like LaunchDarkly or a simple database row both work.

import os

def get_active_schema():
    version = os.environ.get('EXTRACTION_SCHEMA_VERSION', '1.0')
    schemas = {
        '1.0': InvoiceV1,
        '2.0': InvoiceV2,
    }
    return schemas.get(version, InvoiceV1)

async def extract_document(text: str):
    SchemaClass = get_active_schema()
    return await extract_with_schema(text, SchemaClass)

Consumer Compatibility with Union Types

Downstream consumers that read extracted data need to handle multiple schema versions gracefully. Use a discriminated union in your consumer code that selects the correct parsing logic based on the schema_version field. This is more robust than writing conditional if-else chains and easier to extend when version 3 arrives.

from pydantic import BaseModel
from typing import Union, Annotated
from typing import Literal

def parse_extraction(raw: dict) -> Union[InvoiceV1, InvoiceV2]:
    version = raw.get('schema_version', '1.0')
    if version == '1.0':
        return InvoiceV1(**raw)
    elif version == '2.0':
        return InvoiceV2(**raw)
    else:
        raise ValueError(f'Unknown schema version: {version}')

Testing Schema Changes Before Deployment

Before deploying a new schema, run it against your entire regression test set: a curated collection of representative documents with known expected outputs. Compare F1 scores for each field between the old and new schema. A regression in F1 for any field means the new schema description confused the model — fix the field description before shipping.

def eval_schema_on_test_set(test_cases: list, SchemaClass) -> dict:
    field_f1 = {}
    for case in test_cases:
        result = extract_with_schema(case['text'], SchemaClass)
        for field in case['expected']:
            expected = case['expected'][field]
            actual = getattr(result, field, None)
            # Update precision/recall counters
            update_metrics(field_f1, field, expected, actual)
    return {k: compute_f1(v) for k, v in field_f1.items()}

Handling Schema Deprecation

Once a schema version is no longer used for new extractions, you can deprecate it. Deprecation means: stop accepting new records in that version, keep old records readable, and schedule a sunset date when old records will be migrated or archived. Document the deprecation in a changelog so all consumers know to upgrade their parsing code.

DEPRECATED_VERSIONS = {'1.0'}
SUNSET_DATE = '2026-09-01'

def warn_if_deprecated(version: str):
    if version in DEPRECATED_VERSIONS:
        import warnings
        warnings.warn(
            f'Schema version {version} is deprecated. '
            f'It will be removed after {SUNSET_DATE}. '
            'Migrate consumers to version 2.0.',
            DeprecationWarning,
            stacklevel=2
        )

Changelog and Communication

Every schema change must be accompanied by a changelog entry that describes what changed, why, migration instructions, and the expected impact. Share changelog entries with all teams that consume extracted data before deploying the change. Many schema migration disasters happen not from technical failures but from consumers that were not informed a change was coming.

# CHANGELOG.md entry format:
# ## Schema v2.0 (2026-07-01)
# ### Changes
# - ADDED: vendor_tax_id (Optional[str]) - VAT/EIN extracted from header
# - ADDED: currency (str, default='USD') - detected from symbol/code
# ### Migration
# Run: python scripts/migrate_v1_to_v2.py --batch-size=500
# ### Consumers
# - billing-service: update parse_extraction() to handle v2
# - audit-service: query now supports currency filter

Quick Check

Test your understanding of schema evolution and backward compatibility in extraction pipelines.

Lesson Recap

In this lesson you learned: schema versioning stores a version identifier alongside every extracted record so you can migrate selectively, additive changes are safe while renaming or retyping fields requires careful migration, and parallel validation lets you verify the new schema before retiring the old one. Next up we measure LLM latency with TTFT and TPOT metrics.

Frequently asked questions

Is the “Schema Evolution and Backward Compatibility” lesson free?

Yes — the full text of “Schema Evolution and Backward Compatibility” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Schema Evolution and Backward Compatibility”?

Manage breaking schema changes in long-running extraction pipelines by versioning schemas, migrating historical extractions, and running parallel validation during transitions. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Schema Evolution and Backward Compatibility” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Instructor: Typed Extraction with Pydantic
  2. Handling Partial and Missing Data
  3. Batch Processing with Async and Queues
  4. Schema Evolution and Backward Compatibility
← Back to AI Engineering Academy