Your analytics team just discovered that sales reports for the last quarter are wrong. Not subtly wrong — catastrophically wrong. A customer was reclassified from "Enterprise" to "SMB" three months ago, and because your pipeline simply overwrote the old value, every historical order that customer placed now looks like it came from an SMB account. The revenue attribution is off by hundreds of thousands of dollars, and now someone has to explain to the board why the numbers changed overnight.
This is exactly the problem that Slowly Changing Dimensions (SCDs) exist to solve. In any real-world data warehouse, the world changes: customers move, products get repriced, salespeople switch territories, and companies restructure. How your pipeline handles those changes determines whether your historical reporting is trustworthy. Get it wrong, and you either silently corrupt the past or build a system so bloated with history that it becomes unmaintainable.
By the end of this lesson, you'll know how to implement the three most common SCD strategies — Type 1 (overwrite), Type 2 (full history with surrogate keys), and Type 3 (limited history with prior-value columns) — using both Python and SQL. You'll understand when each strategy is appropriate, how to handle the edge cases that trip up most pipelines, and how to build a production-ready implementation you can actually maintain.
What you'll learn:
You should be comfortable with:
Before writing any code, let's be precise about what we're solving. In a data warehouse, a dimension table describes the context of your facts — who, what, where. A dim_customer table tells you that order #8842 was placed by Acme Corp, an Enterprise customer in the Northeast region.
The problem is that "Enterprise" and "Northeast" aren't permanent facts. Acme Corp might get reclassified, move offices, or get acquired. When that happens, you have three fundamental choices:
Each choice has real consequences for storage, query complexity, and analytical fidelity. Let's build all three.
Throughout this lesson, we'll work with a customers dimension. Here's our source system schema — the data that arrives daily from a CRM:
-- Source system table (CRM export)
CREATE TABLE crm_customers (
customer_id INTEGER PRIMARY KEY,
company_name VARCHAR(200),
segment VARCHAR(50), -- 'Enterprise', 'SMB', 'Startup'
region VARCHAR(50),
account_owner VARCHAR(100),
updated_at TIMESTAMP
);
And here's some representative data arriving in today's batch:
import pandas as pd
from datetime import datetime, timezone
# Simulated CRM extract - today's snapshot
crm_data = pd.DataFrame([
{"customer_id": 1001, "company_name": "Acme Corp", "segment": "Enterprise", "region": "Northeast", "account_owner": "Sarah Chen", "updated_at": "2024-03-15 09:00:00"},
{"customer_id": 1002, "company_name": "Bright Ideas LLC","segment": "SMB", "region": "Southeast", "account_owner": "Marcus Webb", "updated_at": "2024-03-15 08:30:00"},
{"customer_id": 1003, "company_name": "Pinnacle Health", "segment": "Enterprise", "region": "Midwest", "account_owner": "Sarah Chen", "updated_at": "2024-03-14 17:00:00"},
{"customer_id": 1004, "company_name": "Dune Analytics", "segment": "Startup", "region": "West", "account_owner": "Priya Nair", "updated_at": "2024-03-15 10:15:00"},
])
Now let's say a few days later, the next CRM extract shows changes:
# Next batch - several records have changed
crm_data_updated = pd.DataFrame([
{"customer_id": 1001, "company_name": "Acme Corp", "segment": "SMB", "region": "Northeast", "account_owner": "Marcus Webb", "updated_at": "2024-03-18 14:00:00"}, # segment changed, owner changed
{"customer_id": 1002, "company_name": "Bright Ideas LLC","segment": "SMB", "region": "Southeast", "account_owner": "Marcus Webb", "updated_at": "2024-03-15 08:30:00"}, # no change
{"customer_id": 1003, "company_name": "Pinnacle Health", "segment": "Enterprise", "region": "West", "account_owner": "Priya Nair", "updated_at": "2024-03-18 11:00:00"}, # region changed, owner changed
{"customer_id": 1004, "company_name": "Dune Analytics", "segment": "SMB", "region": "West", "account_owner": "Priya Nair", "updated_at": "2024-03-18 09:45:00"}, # segment changed
])
This is the data we'll process with each SCD strategy.
Type 1 is the simplest strategy: when a value changes, you update the row and throw away the old value. No history, no versioning — just the current state of the world.
When to use it: For attributes where history genuinely doesn't matter. If a customer's phone number changes, you probably don't need to know what it used to be. Type 1 is also appropriate when you're correcting data quality errors — if "Acme Crp" was a typo for "Acme Corp," you want to overwrite that everywhere.
When to avoid it: Any time business users will ask "what was this value at the time of the transaction?" Type 1 will give them the wrong answer.
The SQL pattern is a standard upsert. In PostgreSQL:
-- Create the Type 1 dimension table
CREATE TABLE dim_customer_t1 (
customer_id INTEGER PRIMARY KEY,
company_name VARCHAR(200),
segment VARCHAR(50),
region VARCHAR(50),
account_owner VARCHAR(100),
updated_at TIMESTAMP,
dw_updated_at TIMESTAMP DEFAULT NOW() -- when WE updated it
);
-- The upsert: insert new rows, update existing ones
INSERT INTO dim_customer_t1 (
customer_id, company_name, segment, region, account_owner, updated_at, dw_updated_at
)
SELECT
customer_id,
company_name,
segment,
region,
account_owner,
updated_at,
NOW() AS dw_updated_at
FROM staging_crm_customers
ON CONFLICT (customer_id) DO UPDATE SET
company_name = EXCLUDED.company_name,
segment = EXCLUDED.segment,
region = EXCLUDED.region,
account_owner = EXCLUDED.account_owner,
updated_at = EXCLUDED.updated_at,
dw_updated_at = NOW()
-- Only update if something actually changed (avoid unnecessary writes)
WHERE (
dim_customer_t1.segment != EXCLUDED.segment OR
dim_customer_t1.region != EXCLUDED.region OR
dim_customer_t1.account_owner != EXCLUDED.account_owner OR
dim_customer_t1.company_name != EXCLUDED.company_name
);
That WHERE clause on the update is important. Without it, you'd rewrite every row on every pipeline run, which murders performance and makes your dw_updated_at timestamp useless as an audit field.
import pandas as pd
import sqlalchemy as sa
from sqlalchemy import text
def apply_scd_type1(engine, new_data: pd.DataFrame, target_table: str, natural_key: str):
"""
Apply SCD Type 1 logic: upsert into target table.
new_data: DataFrame with incoming records
natural_key: the business key column (e.g., 'customer_id')
"""
with engine.begin() as conn:
# Load current dimension state
existing = pd.read_sql(f"SELECT * FROM {target_table}", conn)
# Identify new records vs. updates
merged = new_data.merge(
existing[[natural_key]],
on=natural_key,
how='left',
indicator=True
)
new_records = merged[merged['_merge'] == 'left_only'].drop(columns=['_merge'])
updates = merged[merged['_merge'] == 'both'].drop(columns=['_merge'])
print(f"New records: {len(new_records)}, Potential updates: {len(updates)}")
# For new records, just insert
if not new_records.empty:
new_records['dw_updated_at'] = pd.Timestamp.now()
new_records.to_sql(target_table, conn, if_exists='append', index=False)
# For updates, check if anything actually changed
if not updates.empty:
existing_subset = existing[existing[natural_key].isin(updates[natural_key])]
cols_to_compare = ['company_name', 'segment', 'region', 'account_owner']
comparison = updates.merge(
existing_subset[[natural_key] + cols_to_compare],
on=natural_key,
suffixes=('_new', '_old')
)
# Find rows where at least one column changed
changed_mask = False
for col in cols_to_compare:
changed_mask = changed_mask | (comparison[f'{col}_new'] != comparison[f'{col}_old'])
changed_records = updates[updates[natural_key].isin(
comparison[changed_mask][natural_key]
)]
print(f"Actually changed: {len(changed_records)}")
# Execute updates
for _, row in changed_records.iterrows():
conn.execute(text(f"""
UPDATE {target_table}
SET company_name = :company_name,
segment = :segment,
region = :region,
account_owner = :account_owner,
updated_at = :updated_at,
dw_updated_at = NOW()
WHERE {natural_key} = :{natural_key}
"""), row.to_dict())
Performance note: The row-by-row update loop above is fine for thousands of records, but it'll crawl at millions. For large volumes, use a staging table +
UPDATE ... FROMjoin pattern instead. We'll cover that in the Type 2 section.
Type 2 is the heavyweight. Instead of overwriting, you "close out" the old record and insert a new one. Every version of a dimension member lives in the table, and each version is valid for a specific time range.
This is the strategy you want when someone asks: "What segment was Acme Corp in when they placed that order in February?" With Type 2, you can answer that precisely.
The tradeoff is complexity. Your dimension table grows over time, queries need to filter on the effective date range, and your fact tables must join on a surrogate key (not the natural business key) to link to the correct version.
CREATE TABLE dim_customer_t2 (
-- Surrogate key: unique per row, not per customer
customer_sk BIGSERIAL PRIMARY KEY,
-- Natural key: the business identifier, shared across versions
customer_id INTEGER NOT NULL,
-- The actual attributes
company_name VARCHAR(200),
segment VARCHAR(50),
region VARCHAR(50),
account_owner VARCHAR(100),
-- Version control columns
effective_from DATE NOT NULL,
effective_to DATE, -- NULL means "currently active"
is_current BOOLEAN NOT NULL DEFAULT TRUE,
-- Audit columns
source_updated_at TIMESTAMP,
dw_created_at TIMESTAMP DEFAULT NOW()
);
-- Critical indexes for query performance
CREATE INDEX idx_dim_customer_t2_natural_key ON dim_customer_t2 (customer_id);
CREATE INDEX idx_dim_customer_t2_current ON dim_customer_t2 (customer_id, is_current);
CREATE INDEX idx_dim_customer_t2_dates ON dim_customer_t2 (customer_id, effective_from, effective_to);
A few design choices here worth explaining:
Surrogate key (customer_sk): This is what your fact table's foreign key references. Every order placed by Acme Corp while they were "Enterprise" in "Northeast" links to the surrogate key that represents that specific version. That's what makes the history queries work.
effective_to = NULL: There are two schools of thought here. Some teams use a sentinel date like 9999-12-31 to represent "still current," which simplifies some queries by avoiding IS NULL checks. I prefer NULL because it's semantically honest and avoids the Y10K problem, but either works — just be consistent.
is_current: A redundant but useful flag. It makes "give me the current state of all customers" a simple filter instead of a MAX(effective_from) aggregation.
Here's a complete Type 2 load pattern using a staging table:
-- Step 1: Load new/changed data into a staging table
-- (Assume staging_crm_customers is already populated)
-- Step 2: Close out records that have changed
UPDATE dim_customer_t2 AS dim
SET
effective_to = CURRENT_DATE - INTERVAL '1 day',
is_current = FALSE
FROM staging_crm_customers AS stg
WHERE
dim.customer_id = stg.customer_id
AND dim.is_current = TRUE
AND (
dim.segment != stg.segment OR
dim.region != stg.region OR
dim.account_owner != stg.account_owner OR
dim.company_name != stg.company_name
);
-- Step 3: Insert new versions for changed records + brand-new records
INSERT INTO dim_customer_t2 (
customer_id, company_name, segment, region, account_owner,
effective_from, effective_to, is_current, source_updated_at
)
SELECT
stg.customer_id,
stg.company_name,
stg.segment,
stg.region,
stg.account_owner,
CURRENT_DATE AS effective_from,
NULL AS effective_to,
TRUE AS is_current,
stg.updated_at AS source_updated_at
FROM staging_crm_customers stg
LEFT JOIN dim_customer_t2 dim
ON stg.customer_id = dim.customer_id
AND dim.is_current = TRUE
WHERE
-- New customer (no current record exists)
dim.customer_id IS NULL
OR
-- Changed customer (current record was just closed out above)
(
dim.customer_id IS NOT NULL AND (
dim.segment != stg.segment OR
dim.region != stg.region OR
dim.account_owner != stg.account_owner OR
dim.company_name != stg.company_name
)
);
Warning: These two steps must execute in the same transaction. If step 2 succeeds and step 3 fails, you've closed out active records without inserting replacements — your dimension is now broken. Wrap them in
BEGIN; ... COMMIT;or use your orchestration layer's transaction handling.
The Python version is more explicit about the logic, which makes it easier to test and debug:
import pandas as pd
import numpy as np
from datetime import date, timedelta
import sqlalchemy as sa
from sqlalchemy import text
def apply_scd_type2(
engine,
new_data: pd.DataFrame,
target_table: str,
natural_key: str,
tracked_columns: list,
effective_date: date = None
):
"""
Apply SCD Type 2 logic.
tracked_columns: list of column names that, when changed, trigger a new version
effective_date: the date the new version becomes effective (defaults to today)
"""
if effective_date is None:
effective_date = date.today()
close_date = effective_date - timedelta(days=1)
with engine.begin() as conn:
# Load only current records from the dimension
existing_current = pd.read_sql(
f"SELECT * FROM {target_table} WHERE is_current = TRUE",
conn
)
if existing_current.empty:
# First load — insert everything as-is
print("First load detected. Inserting all records as initial versions.")
new_data = new_data.copy()
new_data['effective_from'] = effective_date
new_data['effective_to'] = None
new_data['is_current'] = True
new_data['dw_created_at'] = pd.Timestamp.now()
new_data.to_sql(target_table, conn, if_exists='append', index=False)
return
# Merge new data with existing current records
merged = new_data.merge(
existing_current,
on=natural_key,
how='outer',
suffixes=('_new', '_old'),
indicator=True
)
# --- Identify the three cases ---
# Case 1: Brand new customers (in new data, not in dimension)
new_customers = merged[merged['_merge'] == 'left_only'].copy()
# Case 2: Existing customers with no changes (skip these entirely)
# Case 3: Existing customers with changes
existing_customers = merged[merged['_merge'] == 'both'].copy()
# Build a mask for changed records
changed_mask = pd.Series(False, index=existing_customers.index)
for col in tracked_columns:
new_col = f'{col}_new' if f'{col}_new' in existing_customers.columns else col
old_col = f'{col}_old' if f'{col}_old' in existing_customers.columns else col
# Handle NaN comparisons correctly
changed_mask = changed_mask | (
existing_customers[new_col].fillna('__NULL__') !=
existing_customers[old_col].fillna('__NULL__')
)
changed_customers = existing_customers[changed_mask].copy()
unchanged_count = len(existing_customers) - len(changed_customers)
print(f"New customers: {len(new_customers)}")
print(f"Changed customers: {len(changed_customers)}")
print(f"Unchanged customers: {unchanged_count} (skipped)")
if changed_customers.empty and new_customers.empty:
print("No changes detected. Nothing to do.")
return
# --- Step 1: Close out changed records ---
if not changed_customers.empty:
changed_sks = changed_customers['customer_sk'].dropna().tolist()
# Build parameterized query
sk_placeholders = ','.join([str(int(sk)) for sk in changed_sks])
conn.execute(text(f"""
UPDATE {target_table}
SET effective_to = :close_date,
is_current = FALSE
WHERE customer_sk IN ({sk_placeholders})
"""), {"close_date": close_date})
print(f"Closed out {len(changed_sks)} existing records.")
# --- Step 2: Insert new versions for changed records ---
if not changed_customers.empty:
# Extract new-value columns
new_cols = {col: f'{col}_new' for col in tracked_columns
if f'{col}_new' in changed_customers.columns}
new_versions = changed_customers[[natural_key]].copy()
for col, new_col in new_cols.items():
new_versions[col] = changed_customers[new_col].values
# Add non-tracked columns that we still need
other_cols = [c for c in new_data.columns
if c != natural_key and c not in tracked_columns]
for col in other_cols:
col_new = f'{col}_new' if f'{col}_new' in changed_customers.columns else col
if col_new in changed_customers.columns:
new_versions[col] = changed_customers[col_new].values
new_versions['effective_from'] = effective_date
new_versions['effective_to'] = None
new_versions['is_current'] = True
new_versions['dw_created_at'] = pd.Timestamp.now()
new_versions.to_sql(target_table, conn, if_exists='append', index=False)
# --- Step 3: Insert entirely new customers ---
if not new_customers.empty:
insert_cols = [c.replace('_new', '') for c in new_customers.columns
if c.endswith('_new')]
new_inserts = pd.DataFrame()
for col in new_data.columns:
col_key = f'{col}_new' if f'{col}_new' in new_customers.columns else col
if col_key in new_customers.columns:
new_inserts[col] = new_customers[col_key].values
new_inserts['effective_from'] = effective_date
new_inserts['effective_to'] = None
new_inserts['is_current'] = True
new_inserts['dw_created_at'] = pd.Timestamp.now()
new_inserts.to_sql(target_table, conn, if_exists='append', index=False)
This is where the investment pays off. Want to know what every customer's segment was when they placed orders in Q4 2023?
-- Join facts to the correct historical version of the customer dimension
SELECT
o.order_id,
o.order_date,
o.revenue,
c.customer_id,
c.company_name,
c.segment,
c.region
FROM fact_orders o
JOIN dim_customer_t2 c
ON o.customer_sk = c.customer_sk -- surrogate key join — pinned to exact version
-- Alternative: if your fact table only has the natural key (less ideal)
SELECT
o.order_id,
o.order_date,
o.revenue,
c.company_name,
c.segment
FROM fact_orders o
JOIN dim_customer_t2 c
ON o.customer_id = c.customer_id
AND o.order_date BETWEEN c.effective_from AND COALESCE(c.effective_to, '9999-12-31')
The surrogate key join is vastly more efficient than the date-range join. If your fact table only stores the natural key, every query has to evaluate the
BETWEENpredicate across all dimension versions. At scale, this kills performance. Store the surrogate key in your fact table and populate it at load time.
Here's a tricky production scenario: a CRM record that changed on March 10th arrives in your pipeline on March 20th (maybe the source system had an outage). How do you handle it?
With Type 2, you have two options:
effective_from to March 10th and adjust the previous version's effective_to accordingly. Accurate, but complex — you may need to re-close existing records and potentially re-open ones that were already closed.For most teams, option 1 is the pragmatic choice. Document the decision clearly. If you need option 2, you'll also want to trigger a downstream reprocessing of any fact records that fall in the affected date range.
Type 3 takes a different trade-off: instead of adding rows over time, it adds columns. You keep the current value and the previous value side by side in the same row.
When to use it: When you need to support "as-was" analysis but history beyond one change isn't needed — and when you want query simplicity over full auditing. A common use case is tracking reorganizations: "Show me revenue by the new territory structure, but also let me see the old territory breakdown for comparison."
When to avoid it: If attributes change more than once (and they will), you lose the intermediate history. Type 3 also gets messy fast if you have many tracked attributes — you end up with a proliferating column structure.
CREATE TABLE dim_customer_t3 (
customer_id INTEGER PRIMARY KEY,
-- Current values
company_name VARCHAR(200),
current_segment VARCHAR(50),
current_region VARCHAR(50),
current_account_owner VARCHAR(100),
-- Previous values (populated when a change occurs)
prior_segment VARCHAR(50),
prior_region VARCHAR(50),
prior_account_owner VARCHAR(100),
-- When the most recent change happened
segment_changed_at TIMESTAMP,
region_changed_at TIMESTAMP,
-- Standard audit fields
source_updated_at TIMESTAMP,
dw_updated_at TIMESTAMP DEFAULT NOW()
);
-- Update Type 3 dimension
-- The key: move current values to prior_ columns BEFORE overwriting them
UPDATE dim_customer_t3 AS dim
SET
-- Shift current values to prior columns
prior_segment = CASE WHEN stg.segment != dim.current_segment
THEN dim.current_segment
ELSE dim.prior_segment END,
prior_region = CASE WHEN stg.region != dim.current_region
THEN dim.current_region
ELSE dim.prior_region END,
prior_account_owner = CASE WHEN stg.account_owner != dim.current_account_owner
THEN dim.current_account_owner
ELSE dim.prior_account_owner END,
-- Set change timestamps
segment_changed_at = CASE WHEN stg.segment != dim.current_segment
THEN NOW()
ELSE dim.segment_changed_at END,
region_changed_at = CASE WHEN stg.region != dim.current_region
THEN NOW()
ELSE dim.region_changed_at END,
-- Apply new current values
current_segment = stg.segment,
current_region = stg.region,
current_account_owner = stg.account_owner,
company_name = stg.company_name,
source_updated_at = stg.updated_at,
dw_updated_at = NOW()
FROM staging_crm_customers stg
WHERE dim.customer_id = stg.customer_id
AND (
dim.current_segment != stg.segment OR
dim.current_region != stg.region OR
dim.current_account_owner != stg.account_owner OR
dim.company_name != stg.company_name
);
-- Insert new customers (no prior values yet)
INSERT INTO dim_customer_t3 (
customer_id, company_name,
current_segment, current_region, current_account_owner,
prior_segment, prior_region, prior_account_owner,
source_updated_at
)
SELECT
stg.customer_id, stg.company_name,
stg.segment, stg.region, stg.account_owner,
NULL, NULL, NULL,
stg.updated_at
FROM staging_crm_customers stg
WHERE NOT EXISTS (
SELECT 1 FROM dim_customer_t3 dim WHERE dim.customer_id = stg.customer_id
);
def apply_scd_type3(
engine,
new_data: pd.DataFrame,
target_table: str,
natural_key: str,
tracked_columns: list # list of column names to track with prior_ versions
):
"""
Apply SCD Type 3 logic.
Assumes dimension table has current_{col} and prior_{col} columns
for each column in tracked_columns.
"""
with engine.begin() as conn:
existing = pd.read_sql(f"SELECT * FROM {target_table}", conn)
if existing.empty:
print("First load. Inserting all records with null prior values.")
insert_df = new_data.copy()
for col in tracked_columns:
insert_df[f'current_{col}'] = insert_df[col]
insert_df[f'prior_{col}'] = None
insert_df[f'{col}_changed_at'] = None
insert_df = insert_df.drop(columns=[col])
insert_df['dw_updated_at'] = pd.Timestamp.now()
insert_df.to_sql(target_table, conn, if_exists='append', index=False)
return
merged = new_data.merge(existing, on=natural_key, how='outer',
suffixes=('_incoming', ''), indicator=True)
new_records = merged[merged['_merge'] == 'left_only'].copy()
existing_records = merged[merged['_merge'] == 'both'].copy()
# Process updates
for _, row in existing_records.iterrows():
updates = {natural_key: row[natural_key]}
changed = False
for col in tracked_columns:
incoming_val = row.get(f'{col}_incoming', row.get(col))
current_val = row.get(f'current_{col}')
if str(incoming_val) != str(current_val):
changed = True
updates[f'prior_{col}'] = current_val
updates[f'current_{col}'] = incoming_val
updates[f'{col}_changed_at'] = pd.Timestamp.now()
if changed:
updates['dw_updated_at'] = pd.Timestamp.now()
set_clause = ', '.join([f"{k} = :{k}" for k in updates if k != natural_key])
conn.execute(
text(f"UPDATE {target_table} SET {set_clause} WHERE {natural_key} = :{natural_key}"),
updates
)
# Insert new records
if not new_records.empty:
for _, row in new_records.iterrows():
insert_data = {natural_key: row[natural_key]}
for col in tracked_columns:
col_key = f'{col}_incoming' if f'{col}_incoming' in row.index else col
insert_data[f'current_{col}'] = row.get(col_key)
insert_data[f'prior_{col}'] = None
insert_data[f'{col}_changed_at'] = None
insert_data['dw_updated_at'] = pd.Timestamp.now()
cols = ', '.join(insert_data.keys())
placeholders = ', '.join([f':{k}' for k in insert_data.keys()])
conn.execute(
text(f"INSERT INTO {target_table} ({cols}) VALUES ({placeholders})"),
insert_data
)
print(f"Processed {len(existing_records)} existing + {len(new_records)} new records.")
Here's how to think about it systematically, not just by memorizing rules:
| Question | If Yes → Consider |
|---|---|
| Do users ever ask "what was X at the time of Y?" | Type 2 |
| Does the attribute correct an error rather than represent real change? | Type 1 |
| Is storage/cost a serious constraint? | Type 1 or Type 3 |
| Do you only need current vs. one-version-ago comparisons? | Type 3 |
| Do attributes change frequently (daily or weekly)? | Type 1 (if history not needed) or Type 2 with compression strategy |
| Are multiple attributes changing at different rates? | Type 2 with per-attribute tracking, or hybrid |
Hybrid approaches are common and legitimate. It's entirely reasonable to apply Type 1 to corrective attributes (phone number, email) and Type 2 to analytically meaningful attributes (segment, territory) on the same dimension table. The key is being explicit about which columns get which treatment — document it, or future-you will be very confused.
Build a complete SCD Type 2 pipeline for a dim_product dimension. Here's your scenario:
A retailer's product catalog changes regularly — prices get adjusted, categories get reorganized, and products move between suppliers. You need to preserve the full history so that margin analysis always reflects the cost and price at the time of sale.
Your tasks:
Create the dim_product_t2 table with appropriate surrogate key, natural key (product_sku), tracked attributes (price, cost, category, supplier_id), and versioning columns.
Write the initial load SQL that takes the full product catalog and inserts it as the first version of each product (all records with effective_from = '2024-01-01' and is_current = TRUE).
Using this simulated change batch, write the SQL to process the Type 2 update:
-- Simulated incoming changes
WITH product_changes AS (
SELECT 'SKU-1001' as product_sku, 'Wireless Headphones Pro' as product_name,
149.99 as price, 62.00 as cost, 'Audio' as category, 201 as supplier_id
UNION ALL
SELECT 'SKU-1002', 'USB-C Hub 7-Port', 79.99, 31.50, 'Accessories', 205
UNION ALL
SELECT 'SKU-1003', 'Mechanical Keyboard TKL', 129.99, 54.00, 'Peripherals', 203
UNION ALL
SELECT 'SKU-9001', 'Bluetooth Speaker Mini', 59.99, 22.00, 'Audio', 201 -- new product
)
Assume SKU-1001's price changed from 129.99 to 149.99, SKU-1002 is unchanged, and SKU-1003 moved to a new supplier (was 202, now 203). SKU-9001 is brand new.
Write a query that shows, for each product, all historical versions ordered by effective date — so analysts can see the full change history.
Bonus: Add a column version_number (1 for the first version, 2 for the second, etc.) to your historical query using a window function.
The close-then-insert pattern in Type 2 must be atomic. If your pipeline crashes between the UPDATE (closing records) and the INSERT (adding new versions), you've orphaned fact records — they point to surrogate keys that are now flagged as historical, and there's no current version to replace them.
Fix: Always wrap the two operations in a single database transaction. In your orchestration layer, ensure retries replay the entire unit — not just the failed step.
If your fact table stores customer_id (the natural business key) instead of customer_sk (the surrogate key), you've lost the precise version link. Every historical query now has to compute which dimension version was active for each fact row via a date range join — which is slow and error-prone.
Fix: At fact load time, look up the current customer_sk and store it. One lookup at load time is vastly cheaper than recomputing it on every query.
In SQL, NULL != NULL evaluates to NULL (not TRUE). If an attribute changes from a real value to NULL, a naive WHERE dim.segment != stg.segment will miss it.
Fix: Use IS DISTINCT FROM in PostgreSQL, or the equivalent COALESCE(dim.segment, '') != COALESCE(stg.segment, '') pattern in databases that don't support it.
-- Correct NULL-safe comparison
WHERE dim.segment IS DISTINCT FROM stg.segment
OR dim.region IS DISTINCT FROM stg.region
Loading your entire source table into a staging area and comparing every row on every pipeline run is wasteful and slow. At 10 million customers, this is a problem.
Fix: Use the source system's updated_at timestamp to filter only changed records since your last run. Store your pipeline's high-water mark:
def get_high_water_mark(engine, pipeline_name: str) -> datetime:
with engine.connect() as conn:
result = conn.execute(text("""
SELECT last_processed_at
FROM pipeline_control
WHERE pipeline_name = :name
"""), {"name": pipeline_name})
row = result.fetchone()
return row[0] if row else datetime(2000, 1, 1)
# Then filter source data:
# SELECT * FROM crm_customers WHERE updated_at > :high_water_mark
If a source record changes twice in a single day and your pipeline runs once a day, you only see the final state. In Type 2, you'll generate one new version when there should logically be two.
This is often acceptable — you document that your SCD Type 2 captures daily snapshots, not sub-daily changes. But if it's not acceptable, you need either an event-driven pipeline (CDC from the source system's transaction log) or you need the source to provide a full change history, not just a current snapshot.
If you ever update is_current without updating effective_to (or vice versa), your dimension becomes inconsistent. Some queries use is_current, others use effective_to IS NULL, and they'll return different results.
Fix: Always update both columns together. Better yet, create a database trigger or application-level constraint that enforces consistency, and add a data quality check to your pipeline that flags any customer_id where COUNT(*) WHERE is_current = TRUE > 1.
Here's what you've built an understanding of:
SCD Type 1 is appropriate for corrective changes and non-historical attributes. The implementation is a standard upsert with a change-detection predicate. Use it freely for attributes where history adds no analytical value.
SCD Type 2 is the cornerstone of proper historical dimension management. It requires surrogate keys in your fact tables, careful transaction handling, and thoughtful index design — but it makes your analytics trustworthy. The core pattern (close old version, insert new version) is straightforward; the complexity lives in edge cases like late-arriving data and NULL handling.
SCD Type 3 offers a middle ground with limited history. It's best for structured comparisons ("current vs. prior") and situations where full versioning is overkill. It degrades gracefully for attributes with low change frequency.
For most production data warehouses, you'll end up using all three — applied at the column level based on the analytical requirements for each attribute. That's not a cop-out; it's good data modeling.
Where to go from here:
Change Data Capture (CDC): Instead of batch-processing daily snapshots, CDC reads from your source database's transaction log to capture every individual change event. This enables sub-daily SCD processing and is the pattern you need for near-real-time warehousing. Tools like Debezium and Fivetran implement this.
dbt Snapshots: If your stack includes dbt, its snapshot feature implements SCD Type 2 with a declarative YAML configuration. Understanding the raw SQL implementation (as you do now) will help you configure dbt snapshots correctly and debug them when they produce unexpected results.
Fact Table Loading: SCD Type 2 only works if your fact table correctly captures the surrogate key at load time. The next layer of complexity is building a fact table loader that looks up the right dimension version for each incoming record — particularly when facts and dimension changes arrive in the same batch.
SCD Type 4 and Type 6: These less common variants extend the ideas you've learned here. Type 4 splits current data into a separate "mini-dimension" for query performance. Type 6 is a hybrid of Types 1, 2, and 3 that adds "current value" columns to the Type 2 table for easier querying.
Learning Path: Data Pipeline Fundamentals