Schema Evolution

This guide covers the comprehensive schema evolution system for detecting, tracking, and managing schema changes over time.

Overview

The schema evolution module provides:

• Change Detection: Identify schema differences (columns added/removed, type changes, renames)
• History Management: Version schemas with semantic versioning, diff, and rollback
• Continuous Monitoring: Watch schemas for changes with real-time alerts
• Impact Analysis: Assess the impact of breaking changes on downstream systems
• Advanced Rename Detection: Multiple similarity algorithms for detecting column renames

Installation

The schema evolution module is included in the core Truthound package:

pip install truthound

Quick Start

Detect Schema Changes

from truthound.profiler.evolution import SchemaEvolutionDetector

# Create detector
detector = SchemaEvolutionDetector()

# Define schemas
old_schema = {"id": "Int64", "name": "Utf8", "email": "Utf8"}
new_schema = {"id": "Int64", "name": "Utf8", "user_email": "Utf8", "age": "Int32"}

# Detect changes
changes = detector.detect_changes(new_schema, old_schema)

for change in changes:
    print(f"[{change.severity.value}] {change.description}")
    if change.breaking:
        print(f"  ⚠️ BREAKING CHANGE")
    if change.migration_hint:
        print(f"  💡 Hint: {change.migration_hint}")

Output:

[critical] Column 'email' removed (was type Utf8)
  ⚠️ BREAKING CHANGE
  💡 Hint: Column removed - update consumers to not depend on this column
[info] Column 'user_email' added with type Utf8
  💡 Hint: New column - ensure consumers handle missing data for older records
[info] Column 'age' added with type Int32
  💡 Hint: New column - ensure consumers handle missing data for older records

Manage Schema History

from truthound.profiler.evolution import SchemaHistory

# Create history with semantic versioning
history = SchemaHistory.create(
    storage_type="file",
    path="./schema_history",
    version_strategy="semantic",
)

# Save initial schema
v1 = history.save(
    {"id": "Int64", "name": "Utf8"},
    metadata={"author": "data-team"},
)
print(f"Version: {v1.version}")  # 1.0.0

# Save updated schema (non-breaking addition)
v2 = history.save({"id": "Int64", "name": "Utf8", "email": "Utf8"})
print(f"Version: {v2.version}")  # 1.1.0 (minor bump)

# Get diff between versions
diff = history.diff(v1, v2)
print(diff.format_text())

# Rollback if needed
v3 = history.rollback(v1, reason="Incompatible with legacy systems")

Watch for Changes

from truthound.profiler.evolution import SchemaWatcher, FileSchemaSource

# Create watcher
watcher = SchemaWatcher()

# Add schema files to watch
watcher.add_source(FileSchemaSource("./schemas/users.json"))
watcher.add_source(FileSchemaSource("./schemas/orders.json"))

# Add event handler
def on_change(event):
    if event.has_breaking_changes():
        print(f"🚨 Breaking changes in {event.source}!")
        for change in event.changes:
            if change.breaking:
                print(f"  - {change.description}")

from truthound.profiler.evolution import CallbackEventHandler
watcher.add_handler(CallbackEventHandler(on_change))

# Start watching (polls every 60 seconds)
watcher.start(poll_interval=60)

# ... later ...
watcher.stop()

CLI Commands

Schema Check

Compare schemas and detect changes:

# Compare current schema to baseline
th schema-check current.json --baseline baseline.json

# Output as JSON
th schema-check current.json -b baseline.json --format json

# Output as Markdown (for reports)
th schema-check current.json -b baseline.json --format markdown

# Fail CI/CD on breaking changes
th schema-check current.json -b baseline.json --fail-on-breaking

# Disable rename detection
th schema-check current.json -b baseline.json --no-detect-renames

# Adjust similarity threshold for rename detection
th schema-check current.json -b baseline.json --similarity 0.7

Schema History

Manage schema version history:

# Initialize history storage
th schema-history init ./my_history
th schema-history init ./my_history --strategy timestamp

# Save a schema version
th schema-history save schema.json
th schema-history save schema.json --version 2.0.0
th schema-history save schema.json -m "Added email column"

# List versions
th schema-history list
th schema-history list --limit 20
th schema-history list --format json

# Show version details
th schema-history show 1.0.0
th schema-history show abc12345 --format json

# Rollback to a version
th schema-history rollback 1.0.0
th schema-history rollback 1.0.0 --reason "Incompatible change" --yes

Schema Diff

Show diff between schema versions:

# Diff between two versions
th schema-diff 1.0.0 2.0.0

# Diff to latest
th schema-diff 1.0.0

# Output formats
th schema-diff 1.0.0 2.0.0 --format json
th schema-diff 1.0.0 2.0.0 --format markdown

Schema Watch

Monitor schemas for changes:

# Watch a single file
th schema-watch schema.json

# Watch multiple files
th schema-watch schema1.json schema2.json

# Custom poll interval
th schema-watch schema.json --interval 30

# Track history
th schema-watch schema.json --history ./schema_history

# Write alerts to file
th schema-watch schema.json --alert-file alerts.jsonl

# Only alert on breaking changes
th schema-watch schema.json --only-breaking

# Single check (for CI/CD)
th schema-watch schema.json --once --only-breaking

Python API

SchemaEvolutionDetector

The core class for detecting schema changes:

from truthound.profiler.evolution import SchemaEvolutionDetector

detector = SchemaEvolutionDetector(
    detect_renames=True,           # Enable rename detection
    rename_similarity_threshold=0.8,  # Threshold for considering a rename
)

# Can accept various schema formats
# - dict: {"col": "Type"}
# - polars.Schema
# - polars.DataFrame
# - polars.LazyFrame

changes = detector.detect_changes(current_schema, baseline_schema)
summary = detector.get_change_summary(changes)

print(f"Total changes: {summary.total_changes}")
print(f"Breaking changes: {summary.breaking_changes}")
print(f"Compatibility: {summary.compatibility_level.value}")

Change Types

The system detects these change types:

Change Type	Breaking?	Severity
COLUMN_ADDED	No	INFO
COLUMN_REMOVED	Yes	CRITICAL
COLUMN_RENAMED	Yes	CRITICAL
TYPE_CHANGED	Depends*	WARNING/CRITICAL
NULLABLE_CHANGED	Depends*	INFO/CRITICAL

*Type changes are non-breaking for compatible upgrades (e.g., Int32 → Int64). Nullable→NonNullable is breaking.

SchemaHistory

Manage schema versions with history:

from truthound.profiler.evolution import SchemaHistory

# Create with different strategies
history = SchemaHistory.create(
    storage_type="file",       # "file" or "memory"
    path="./schema_history",   # Required for "file"
    max_versions=100,          # Max versions to keep
    version_strategy="semantic",  # "semantic", "incremental", "timestamp", "git"
    compress=True,             # Compress stored files
)

# Save versions
v1 = history.save(schema, version="1.0.0", metadata={"author": "team"})

# Get versions
latest = history.latest
baseline = history.baseline
specific = history.get("version-id")
by_string = history.get_by_version("1.0.0")

# List and filter
versions = history.list(limit=10)
recent = history.list(since=datetime.now() - timedelta(days=7))

# Diff
diff = history.diff("1.0.0", "2.0.0")
print(diff.format_text())

# Check breaking changes
if history.has_breaking_changes_since("1.0.0"):
    print("Breaking changes detected!")

# Rollback
new_version = history.rollback("1.0.0", reason="Revert breaking change")

Version Strategies

Strategy	Example	Use Case
semantic	1.2.3	Standard versioning, auto-bumps based on change type
incremental	1, 2, 3	Simple incrementing numbers
timestamp	20260128.143052	Time-based versions
git	a1b2c3d4	Git-like short hashes

ColumnRenameDetector

Advanced rename detection with multiple algorithms:

from truthound.profiler.evolution import (
    ColumnRenameDetector,
    CompositeSimilarity,
    LevenshteinSimilarity,
    JaroWinklerSimilarity,
    TokenSimilarity,
)

# Use composite similarity (default)
detector = ColumnRenameDetector(
    similarity_threshold=0.8,
    require_type_match=True,
    allow_compatible_types=True,
)

result = detector.detect(
    added_columns={"user_email": "Utf8", "customer_name": "Utf8"},
    removed_columns={"email": "Utf8", "cust_name": "Utf8"},
)

# Confirmed renames (high confidence)
for rename in result.confirmed_renames:
    print(f"✓ {rename.old_name} → {rename.new_name}")
    print(f"  Similarity: {rename.similarity:.0%}")
    print(f"  Confidence: {rename.confidence.value}")

# Possible renames (need review)
for rename in result.possible_renames:
    print(f"? {rename.old_name} → {rename.new_name}")
    print(f"  Reasons: {', '.join(rename.reasons)}")

# Unmatched columns
print(f"Unmatched added: {result.unmatched_added}")
print(f"Unmatched removed: {result.unmatched_removed}")

Similarity algorithms:

Algorithm	Best For
LevenshteinSimilarity	General edit distance
JaroWinklerSimilarity	Short strings, common prefixes
NgramSimilarity	Partial matches, abbreviations
TokenSimilarity	snake_case/camelCase names
CompositeSimilarity	Weighted combination (default)

SchemaWatcher

Continuous monitoring:

from truthound.profiler.evolution import (
    SchemaWatcher,
    FileSchemaSource,
    PolarsSchemaSource,
    DictSchemaSource,
    LoggingEventHandler,
    AlertingEventHandler,
    HistoryEventHandler,
)

# Create watcher
watcher = SchemaWatcher()

# Add sources
watcher.add_source(FileSchemaSource("schema.json"))
watcher.add_source(PolarsSchemaSource(lambda: pl.read_csv("data.csv"), "data"))
watcher.add_source(DictSchemaSource({"id": "Int64"}, "manual"))

# Add handlers
watcher.add_handler(LoggingEventHandler())
watcher.add_handler(HistoryEventHandler(history))
watcher.add_handler(AlertingEventHandler(alert_manager))

# Lifecycle control
watcher.start(poll_interval=60, daemon=True)
watcher.pause()
watcher.resume()
watcher.stop()

# Check immediately
events = watcher.check_now()

Breaking Change Alerts

Enhanced alerting with impact analysis:

from truthound.profiler.evolution import (
    BreakingChangeAlertManager,
    ImpactAnalyzer,
    ColumnRemovedChange,
)

# Setup impact analyzer with consumer mappings
analyzer = ImpactAnalyzer()
analyzer.register_consumer("dashboard", ["users", "orders"])
analyzer.register_consumer("reports", ["users", "products"])
analyzer.register_query("users", "SELECT email FROM users")

# Create alert manager
manager = BreakingChangeAlertManager(
    impact_analyzer=analyzer,
    alert_storage_path="./alerts.json",
)

# Create alert
changes = [ColumnRemovedChange("email", "Utf8")]
alert = manager.create_alert(changes, source="users")

print(f"Alert: {alert.title}")
print(f"Impact Scope: {alert.impact.scope.value}")
print(f"Affected Consumers: {alert.impact.affected_consumers}")
print(f"Risk Level: {alert.impact.data_risk_level}/5")
print(f"Recommendations:")
for rec in alert.impact.recommendations:
    print(f"  - {rec}")

# Track alerts
history = manager.get_alert_history(status="open")
manager.acknowledge_alert(alert.alert_id)
manager.resolve_alert(alert.alert_id)

# Get statistics
stats = manager.get_stats()
print(f"Total: {stats['total']}, Open: {stats['open']}")

Format alerts for notifications:

# Slack
slack_payload = alert.format_slack_message()

# Email
subject, html_body = alert.format_email()

Factory Functions

Create pre-configured components:

from truthound.profiler.evolution import (
    create_watcher,
    create_rename_detector,
)

# Create watcher with common settings
watcher = create_watcher(
    sources=[FileSchemaSource("schema.json")],
    poll_interval=60,
    enable_logging=True,
    enable_history=True,
    history_path="./history",
    on_change=lambda e: print(f"Change: {e.source}"),
    auto_start=True,
)

# Create rename detector with specific algorithm
detector = create_rename_detector(
    algorithm="jaro_winkler",
    threshold=0.85,
    require_type_match=True,
)

Integration Examples

CI/CD Pipeline

# GitHub Actions
- name: Check Schema Changes
  run: |
    th schema-check current.json -b baseline.json --fail-on-breaking

# Python script for CI
from truthound.profiler.evolution import SchemaEvolutionDetector
import sys

detector = SchemaEvolutionDetector()
changes = detector.detect_changes(current, baseline)

breaking = [c for c in changes if c.breaking]
if breaking:
    print("❌ Breaking changes detected:")
    for c in breaking:
        print(f"  - {c.description}")
    sys.exit(1)

print("✅ No breaking changes")

Airflow DAG

from airflow import DAG
from airflow.operators.python import PythonOperator

def check_schema():
    from truthound.profiler.evolution import SchemaHistory

    history = SchemaHistory.create(storage_type="file", path="/data/history")

    # Get current schema from database
    current = get_database_schema()

    # Save and check for breaking changes
    version = history.save(current)

    if version.has_breaking_changes():
        raise ValueError(f"Breaking changes: {version.changes_from_parent}")

with DAG("schema_monitor", schedule_interval="@daily") as dag:
    check_schema_task = PythonOperator(
        task_id="check_schema",
        python_callable=check_schema,
    )

Slack Notifications

import requests
from truthound.profiler.evolution import (
    SchemaWatcher,
    BreakingChangeAlertManager,
    AlertingEventHandler,
)

class SlackNotifier:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send(self, alert):
        payload = alert.format_slack_message()
        requests.post(self.webhook_url, json=payload)
        return True

# Setup
manager = BreakingChangeAlertManager(notifiers=[
    SlackNotifier("https://hooks.slack.com/services/...")
])

watcher = SchemaWatcher()
watcher.add_handler(AlertingEventHandler(manager))
watcher.start()

Best Practices

1. Version Strategy Selection

• Semantic: Best for APIs and shared schemas
• Timestamp: Best for audit trails
• Git-like: Best for development environments

2. Breaking Change Policy

# Define what constitutes breaking
BREAKING_TYPES = {
    ChangeType.COLUMN_REMOVED,
    ChangeType.TYPE_CHANGED,  # Only incompatible
    ChangeType.NULLABLE_CHANGED,  # Only nullable→non-nullable
}

# Allow grace period for breaking changes
MIN_DEPRECATION_DAYS = 30

3. Rename Detection Tuning

# Strict (production)
detector = ColumnRenameDetector(
    similarity_threshold=0.9,
    require_type_match=True,
)

# Lenient (development)
detector = ColumnRenameDetector(
    similarity_threshold=0.7,
    require_type_match=False,
)

4. Alert Configuration

# Critical alerts only
watcher.add_handler(AlertingEventHandler(
    alert_manager,
    only_breaking=True,
))

# All changes with logging
watcher.add_handler(LoggingEventHandler(
    min_severity=ChangeSeverity.INFO,
))

Related Guides

• Drift Detection - Data distribution drift
• Quality Scoring - Impact on data quality
• Checkpoint Basics - CI/CD integration
• Notifications - Alert routing

API Reference

Full API documentation available at:

• truthound.profiler.evolution.SchemaEvolutionDetector
• truthound.profiler.evolution.SchemaHistory
• truthound.profiler.evolution.SchemaWatcher
• truthound.profiler.evolution.ColumnRenameDetector
• truthound.profiler.evolution.BreakingChangeAlertManager
• truthound.profiler.evolution.ImpactAnalyzer

Declared Schema Compatibility

Compare declared nullability with NullabilityCompatibilityChecker(old_nullable=..., new_nullable=...). Dtypes alone do not contain nullability, so constraints are not inferred from values or samples. Common-column declarations must be paired actual booleans. Nullable-to-non-nullable tightening is breaking. strict=True rejects every reported issue. Rule exceptions and invalid verdicts raise CompatibilityAnalysisError instead of becoming full compatibility. Execution guards apply around each native rule.

import polars as pl
from truthound.profiler.evolution import NullabilityCompatibilityChecker, SchemaCompatibilityAnalyzer

analyzer = SchemaCompatibilityAnalyzer(
    rules=[NullabilityCompatibilityChecker(
        old_nullable={"x": True}, new_nullable={"x": False},
    )],
    strict=True,
)
report = analyzer.analyze({"x": pl.Int64}, {"x": pl.Int64})
assert not report.is_compatible
for issue in report.issues:
    print(issue.change_type.value, issue.breaking)