Versioning Database Schemas — Complete Guide
In this tutorial, you'll learn about Versioning Database Schemas. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Database schema versioning manages changes to database structure over time while maintaining backward compatibility with existing API versions and application code.
What You'll Learn
By the end of this lesson, you will implement schema Migration strategies, expandable schemas for multiple API versions, and zero-downtime migration patterns.
Why It Matters
Database schemas are coupled to API versions. A breaking schema change can break all API versions that depend on it. Version-aware schema design prevents this.
Real-World Use
Airbnb uses expandable schemas where new columns are nullable and optional. They run both old and new code paths during migration Windows, a pattern called "expand-migrate-contract."
Schema Migration Flow
flowchart LR
Expand[Add Column (nullable)] --> Migrate[Backfill Data]
Migrate --> Contract[Remove Nullable Constraint]
Contract --> OldCode[Old Code Ignores Column]
Contract --> NewCode[New Code Uses Column]
Expand-Migrate-Contract Pattern
# expand_migrate_contract.py
from typing import Any, Dict, List, Optional
import json
class ExpandMigrateContract:
def __init__(self):
self.schema_version = 1
self.columns: Dict[str, Dict] = {}
self.data: List[Dict] = []
def add_column(self, name: str, col_type: str, nullable: bool = True,
default: Any = None):
self.columns[name] = {
"type": col_type,
"nullable": nullable,
"default": default,
"added_in_version": self.schema_version,
}
# Phase 1: Expand (add nullable column)
def expand(self, name: str, col_type: str, default: Any = None):
self.columns[name] = {
"type": col_type,
"nullable": True,
"default": default,
"added_in_version": self.schema_version,
}
for row in self.data:
if name not in row:
row[name] = default
# Phase 2: Migrate (backfill data)
def migrate(self, name: str, transform: callable):
for row in self.data:
if row.get(name) is None:
row[name] = transform(row)
# Phase 3: Contract (make non-nullable)
def contract(self, name: str):
for row in self.data:
if row.get(name) is None:
raise ValueError(f"Cannot contract: NULL exists in '{name}'")
self.columns[name]["nullable"] = False
def insert(self, **kwargs):
row = {}
for col, props in self.columns.items():
row[col] = kwargs.get(col, props["default"])
self.data.append(row)
return row
def query(self, version: int) -> List[Dict]:
results = []
for row in self.data:
filtered = {}
for col, props in self.columns.items():
if props["added_in_version"] <= version:
filtered[col] = row.get(col)
results.append(filtered)
return results
emc = ExpandMigrateContract()
emc.add_column("id", "int", nullable=False)
emc.add_column("name", "str", nullable=False)
emc.insert(id=1, name="Alice")
emc.insert(id=2, name="Bob")
# Expand phase
emc.expand("email", "str")
# old code still works
old_rows = emc.query(version=1)
new_rows = emc.query(version=2)
print(f"v1 (old) query: {old_rows}")
print(f"v2 (new) query: {new_rows}")
# Migrate phase
emc.migrate("email", lambda r: f"{r['name'].lower()}@example.com")
for row in emc.query(version=2):
print(f" after migration: {row}")
# Contract phase
emc.contract("email")
print(f"Contract succeeded: email is now non-nullable")
Expected output:
v1 (old) query: [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
v2 (new) query: [{'id': 1, 'name': 'Alice', 'email': None}, {'id': 2, 'name': 'Bob', 'email': None}]
after migration: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}
after migration: {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
Contract succeeded: email is now non-nullable
Multi-Version View
# multi_version_view.py
from typing import Any, Dict, List
class MultiVersionView:
def __init__(self):
self.schemas: Dict[int, List[str]] = {}
def register(self, version: int, fields: List[str]):
self.schemas[version] = fields
def project(self, data: Dict, target_version: int) -> Dict:
schema = self.schemas.get(target_version)
if not schema:
return {"error": f"Version {target_version} not found"}
result = {}
for field in schema:
if field in data:
result[field] = data[field]
return result
def project_all(self, all_data: List[Dict], target_version: int) -> List[Dict]:
return [self.project(row, target_version) for row in all_data]
view = MultiVersionView()
view.register(1, ["id", "name"])
view.register(2, ["id", "name", "email"])
view.register(3, ["id", "name", "email", "role"])
user_row = {"id": 1, "name": "Alice", "email": "alice@x.com", "role": "admin"}
for v in [1, 2, 3]:
projected = view.project(user_row, v)
print(f"v{v} view: {projected}")
Expected output:
v1 view: {'id': 1, 'name': 'Alice'}
v2 view: {'id': 1, 'name': 'Alice', 'email': 'alice@x.com'}
v3 view: {'id': 1, 'name': 'Alice', 'email': 'alice@x.com', 'role': 'admin'}
Zero-Downtime Migration
# zero_downtime.py
from typing import Any, Dict, List, Optional
class ZeroDowntimeMigration:
def __init__(self):
self._data: List[Dict] = []
self._old_schema_fields = ["id", "name"]
self._new_schema_fields = ["id", "name", "email"]
def read_old(self, row: Dict) -> Dict:
return {k: v for k, v in row.items() if k in self._old_schema_fields}
def read_new(self, row: Dict) -> Dict:
result = {}
for f in self._new_schema_fields:
result[f] = row.get(f)
return result
def write_dual(self, **kwargs) -> int:
row = {}
for f in self._new_schema_fields:
row[f] = kwargs.get(f)
self._data.append(row)
return row["id"]
def migrate_in_background(self, transform: callable):
migrated = 0
for row in self._data:
if row.get("email") is None:
row["email"] = transform(row)
migrated += 1
return migrated
def switch_to_new(self):
self._old_schema_fields = self._new_schema_fields[:]
def remove_old_code(self):
pass
migrator = ZeroDowntimeMigration()
migrator.write_dual(id=1, name="Alice", email=None)
migrator.write_dual(id=2, name="Bob", email=None)
print(f"Old reads work: {migrator.read_old(migrator._data[0])}")
migrated = migrator.migrate_in_background(lambda r: f"{r['name'].lower()}@example.com")
print(f"Migrated {migrated} rows in background")
migrator.switch_to_new()
print(f"New reads work: {migrator.read_new(migrator._data[0])}")
Expected output:
Old reads work: {'id': 1, 'name': 'Alice'}
Migrated 2 rows in background
New reads work: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}
Common Mistakes
1. Removing Columns Immediately
Dropping a column that old API versions still reference breaks those versions. Keep columns until all API versions are migrated.
2. Adding NOT NULL Columns
Adding a non-nullable column to an existing table with rows fails. Always add as nullable first, backfill, then add constraint.
3. Renaming Columns
Renaming breaks all queries using the old name. Add the new column, dual-write, migrate reads, then drop the old column.
4. Ignoring Read Replicas
Schema migrations on the primary may not immediately replicate. Ensure read replicas are synchronized before switching reads.
5. No Rollback Plan
Every migration needs a rollback script. Test rollbacks in staging before running in production.
Practice Questions
1. What is the expand-migrate-contract pattern?
Add nullable column (expand), backfill data (migrate), then add NOT NULL constraint (contract). Allows zero-downtime schema changes.
2. How do you handle column renames in a backward-compatible way?
Add the new column, write to both columns, migrate reads from old to new, then deprecate the old column.
3. Why should new columns be nullable?
Existing rows do not have a value for the new column. Making it nullable allows the migration to complete without breaking existing queries.
4. What is a multi-version view?
A database view that projects only the fields relevant to a specific API version, allowing different versions to read the same underlying table.
Challenge
Design and implement a schema migration system for a three-version API where v1 has {id, name}, v2 adds {email}, and v3 renames {name} to {full_name}, with full backward compatibility.
FAQ
Mini Project: Schema Version Manager
# schema_version_mgr.py
from typing import Any, Dict, List, Optional
class SchemaVersionManager:
def __init__(self):
self._schema: Dict[str, Dict] = {}
def migrate_v1_to_v2(self, row: Dict) -> Dict:
row["email"] = None
return row
def migrate_v2_to_v3(self, row: Dict) -> Dict:
row["full_name"] = row.pop("name")
return row
def run_migration(self, data: List[Dict], from_v: int, to_v: int) -> List[Dict]:
result = []
for row in data:
if from_v == 1 and to_v >= 2:
row = self.migrate_v1_to_v2(dict(row))
if from_v <= 2 and to_v >= 3:
row = self.migrate_v2_to_v3(dict(row))
result.append(row)
return result
mgr = SchemaVersionManager()
data = [{"id": 1, "name": "Alice"}]
migrated = mgr.run_migration(data, from_v=1, to_v=3)
print(f"v1 -> v3: {migrated}")
Expected output:
v1 -> v3: [{'id': 1, 'email': None, 'full_name': 'Alice'}]
What's Next
You understand database schema versioning. Next, learn about versioning microservices, then explore versioning testing strategies.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro