Series Context
SCD2 Is Not Enough (Part 2) — Timeline Compression and Deterministic Integration
Merging Multiple SCD2 Timelines Without Explosion: ASOF Sampling, Gaps-and-Islands, and PIT Tables
Series Context
In Part 1, we standardized heterogeneous sources into controlled bi-temporal SCD2.
Now comes the hard part.
Once everything is SCD2…
Joining them becomes dangerous.
1. Why Multi-SCD2 Joins Explode
Each SCD2 table has its own change points.
The Explosion Problem
Consider a simple 360-degree customer view requiring three SCD2 sources:
Combined timeline:
9 change points from just 3 sources with 3 changes each.
Each change fragments the timeline. Add more sources → exponential fragmentation.
Even if business values don’t change.
The Naive JOIN Problem
A naive temporal JOIN looks like this:
-- DON'T DO THIS: Cartesian explosion
SELECT
c.customer_id,
c.name,
a.street,
a.city,
con.email,
con.phone
FROM customer_scd2 c
JOIN address_scd2 a ON c.customer_id = a.customer_id
AND c.valid_from <= a.valid_to
AND c.valid_to >= a.valid_from
JOIN contact_scd2 con ON c.customer_id = con.customer_id
AND c.valid_from <= con.valid_to
AND c.valid_to >= con.valid_from
This creates a Cartesian product of overlapping periods — timeline explosion.
2. The MDDE Pattern: Timeline Union + Resampling
MDDE solves this with the Temporal Building Blocks :
Step 1 — Collect All Change Points
from mdde.generator.building_blocks.temporal_blocks import (
TemporalBuildingBlocks,
TemporalConfig,
TemporalDialect,
)
config = TemporalConfig(
dialect=TemporalDialect.DATABRICKS,
valid_from_col="valid_from",
valid_to_col="valid_to",
)
blocks = TemporalBuildingBlocks(config)
# Collect all change dates from multiple sources
change_points_cte = blocks.change_points_union(
sources=["customer_scd2", "address_scd2", "contact_scd2"],
business_key="customer_id",
include_valid_to=True, # Include closure points
cte_name="change_points",
)
Generated SQL:
-- Collect all change dates from 3 source(s)
change_points AS (
SELECT DISTINCT valid_from AS change_date, customer_id
FROM customer_scd2
UNION
SELECT DISTINCT valid_to AS change_date, customer_id
FROM customer_scd2
WHERE valid_to IS NOT NULL
UNION
SELECT DISTINCT valid_from AS change_date, customer_id
FROM address_scd2
UNION
SELECT DISTINCT valid_to AS change_date, customer_id
FROM address_scd2
WHERE valid_to IS NOT NULL
UNION
SELECT DISTINCT valid_from AS change_date, customer_id
FROM contact_scd2
UNION
SELECT DISTINCT valid_to AS change_date, customer_id
FROM contact_scd2
WHERE valid_to IS NOT NULL
)
Step 2 — ASOF Sample Each Source
For each change point, sample the valid state from each source:
# Sample customer at each change point
customer_sample = blocks.asof_sample(
source="customer_scd2",
change_points_cte="change_points",
business_key="customer_id",
columns=["name", "status", "segment"],
cte_name="customer_sampled",
)
Generated SQL (Databricks/DuckDB with native ASOF JOIN):
-- Sample customer_scd2 at each change point
customer_sampled AS (
SELECT
cp.change_date,
cp.customer_id,
s.name, s.status, s.segment
FROM change_points cp
ASOF JOIN customer_scd2 s
ON cp.customer_id = s.customer_id
AND cp.change_date >= s.valid_from
)
Generated SQL (BigQuery/PostgreSQL/SQL Server fallback):
-- Sample customer_scd2 at each change point
customer_sampled AS (
SELECT change_date, customer_id, name, status, segment
FROM (
SELECT
cp.change_date,
cp.customer_id,
s.name, s.status, s.segment,
ROW_NUMBER() OVER (
PARTITION BY cp.customer_id, cp.change_date
ORDER BY s.valid_from DESC
) AS _rn
FROM change_points cp
JOIN customer_scd2 s
ON cp.customer_id = s.customer_id
AND cp.change_date >= s.valid_from
AND (s.valid_to IS NULL OR cp.change_date < s.valid_to)
)
WHERE _rn = 1
)
Key insight: MDDE detects dialect capabilities and generates optimal SQL:
Step 3 — Merge All Samples
from mdde.generator.timeline_merger_sql import (
TimelineMerger,
TimelineMergeConfig,
TimelineSource,
)
config = TimelineMergeConfig(
output_entity="customer_360_merged",
business_key="customer_id",
sources=[
TimelineSource(
entity_name="customer_scd2",
attributes=["name", "status", "segment"],
),
TimelineSource(
entity_name="address_scd2",
attributes=["street", "city", "postal_code"],
prefix="addr_", # Prefix output columns
),
TimelineSource(
entity_name="contact_scd2",
attributes=["email", "phone"],
prefix="contact_",
),
],
dialect="databricks",
compress_output=True, # Critical for performance
)
merger = TimelineMerger()
result = merger.generate(config)
print(result.merge_sql)
This produces a unified timeline with all attributes sampled at each change point.
But it’s still fragmented. Many adjacent intervals contain identical values.
3. Timeline Compression (Normalization)
After merging, the timeline contains technical splits where no business values changed:
Rows 1–2 are identical (no business change at Jan 10). Rows 3–4 are identical (no business change at Feb 1).
These must compress to:
The Gaps-and-Islands Pattern
MDDE uses hash-based compression with the classic gaps-and-islands SQL pattern:
from mdde.generator.timeline_compress_sql import (
TimelineCompressSqlGenerator,
TimelineCompressConfig,
TimelineCompressSpec,
TimelineDialect,
CompressionMethod,
)
config = TimelineCompressConfig(
dialect=TimelineDialect.DATABRICKS,
method=CompressionMethod.ROW_HASH, # Use existing _row_hash
include_comments=True,
)
spec = TimelineCompressSpec(
entity_id="customer_360",
entity_name="customer_360_merged",
source_sql="merged_timeline",
business_key_columns=["customer_id"],
attribute_columns=["name", "segment", "addr_city", "addr_street", "contact_email"],
config=config,
)
generator = TimelineCompressSqlGenerator(config=config)
result = generator.generate_compression_sql(spec)
print(result.sql)
Generated SQL:
-- Timeline Compression: customer_360_merged
-- Classification: timeline_compress
-- Method: gaps-and-islands (NORMALIZE equivalent)
-- Business keys: customer_id
WITH boundaries AS (
SELECT
customer_id,
name,
segment,
addr_city,
addr_street,
contact_email,
_valid_from,
_valid_to,
_row_hash AS _content_hash,
CASE
-- New island if content hash differs from previous
WHEN _content_hash != LAG(_content_hash) OVER (
PARTITION BY customer_id
ORDER BY _valid_from
) THEN 1
-- New island if there's a gap in the timeline
WHEN _valid_from != LAG(_valid_to) OVER (
PARTITION BY customer_id
ORDER BY _valid_from
) THEN 1
-- First record is always a boundary
WHEN LAG(_content_hash) OVER (
PARTITION BY customer_id
ORDER BY _valid_from
) IS NULL THEN 1
ELSE 0
END AS _is_boundary
FROM merged_timeline
),
islands AS (
SELECT
*,
SUM(_is_boundary) OVER (
PARTITION BY customer_id
ORDER BY _valid_from
ROWS UNBOUNDED PRECEDING
) AS _island_id
FROM boundaries
)
SELECT
customer_id,
name,
segment,
addr_city,
addr_street,
contact_email,
_content_hash,
MIN(_valid_from) OVER (
PARTITION BY customer_id, _island_id
) AS _valid_from,
CASE
WHEN MAX(CASE WHEN _valid_to IS NULL THEN 1 ELSE 0 END) OVER (
PARTITION BY customer_id, _island_id
) = 1 THEN NULL
ELSE MAX(_valid_to) OVER (
PARTITION BY customer_id, _island_id
)
END AS _valid_to
FROM islands
QUALIFY ROW_NUMBER() OVER (
PARTITION BY customer_id, _island_id
ORDER BY _valid_from
) = 1
Using the Temporal Building Block
For simpler use cases, use the building block directly:
# Generate island boundary detection
boundary_expr = blocks.island_boundary(
hash_column="_content_hash",
partition_key="customer_id",
order_col="valid_from",
boundary_alias="_is_boundary",
)
# Generate timeline compression CTE
compress_cte = blocks.timeline_compress(
source="merged_with_hash",
business_key="customer_id",
hash_column="_content_hash",
columns=["name", "segment", "addr_city"],
cte_name="compressed",
)
Hash-Based Compression Benefits
Content Hash Generation
MDDE generates NULL-safe hash expressions:
hash_expr = blocks.content_hash(
columns=["name", "segment", "addr_city", "addr_street"],
alias="_content_hash",
)
Generated SQL:
MD5(CONCAT_WS('|',
COALESCE(CAST(name AS STRING), '^^NULL^^'),
COALESCE(CAST(segment AS STRING), '^^NULL^^'),
COALESCE(CAST(addr_city AS STRING), '^^NULL^^'),
COALESCE(CAST(addr_street AS STRING), '^^NULL^^')
)) AS _content_hash
Key insight: NULL values are replaced with a marker (^^NULL^^) to ensure:
NULLvs'NULL'vs''are distinguishable- Hash is deterministic regardless of NULL positions
4. Point-in-Time (PIT) Tables
For high-performance reporting integration, MDDE generates PIT (Point-in-Time) helper tables:
Why PIT Tables?
PIT Table Generation
from mdde.generator.pit_generator import (
PITGenerator,
PITGeneratorConfig,
PITDialect,
SnapshotStrategy,
Materialization,
)
config = PITGeneratorConfig(
dialect=PITDialect.DATABRICKS,
snapshot_strategy=SnapshotStrategy.CHANGE_POINTS, # Or DAILY, WEEKLY, CALENDAR
snapshot_date_column="snapshot_date",
include_surrogate_keys=True,
include_business_keys=True,
include_timeline_columns=True,
materialization=Materialization.TABLE,
pit_name_template="pit_{entity_name}",
)
generator = PITGenerator(conn, config)
spec = generator.generate_pit_table("ENT_customer_scd2")
print(spec.create_ddl)
print(spec.populate_sql)
Generated DDL:
CREATE TABLE IF NOT EXISTS pit_customer_scd2 (
snapshot_date DATE,
customer_id STRING,
customer_sk BIGINT,
valid_from DATE,
valid_to DATE,
PRIMARY KEY (snapshot_date, customer_id)
)
USING DELTA;
Generated Populate SQL:
-- Populate PIT table: pit_customer_scd2
-- Source: customer_scd2
-- Strategy: change_points
WITH snapshot_dates AS (
-- All change points from source entity
SELECT DISTINCT CAST(valid_from AS DATE) AS snapshot_date
FROM customer_scd2
WHERE valid_from IS NOT NULL
UNION
SELECT DISTINCT CAST(valid_to AS DATE) AS snapshot_date
FROM customer_scd2
WHERE valid_to IS NOT NULL AND valid_to < '9999-12-31'
)
INSERT INTO pit_customer_scd2
SELECT
d.snapshot_date,
s.customer_id,
s.customer_sk,
s.valid_from,
s.valid_to
FROM snapshot_dates d
CROSS JOIN (
SELECT DISTINCT customer_id
FROM customer_scd2
) bk
LEFT JOIN customer_scd2 s
ON bk.customer_id = s.customer_id
AND d.snapshot_date >= s.valid_from
AND d.snapshot_date < COALESCE(s.valid_to, '9999-12-31')
AND s.is_latest = TRUE;
Snapshot Strategies
Using PIT Tables in Queries
Once PIT tables exist, temporal queries become simple:
-- Without PIT (complex temporal JOIN)
SELECT f.*, c.name, c.segment
FROM fact_orders f
JOIN customer_scd2 c
ON f.customer_id = c.customer_id
AND f.order_date >= c.valid_from
AND f.order_date < COALESCE(c.valid_to, '9999-12-31')
WHERE c.is_latest = TRUE;
-- With PIT (simple equi-join)
SELECT f.*, c.name, c.segment
FROM fact_orders f
JOIN pit_customer_scd2 p
ON f.customer_id = p.customer_id
AND f.order_date = p.snapshot_date
JOIN customer_scd2 c
ON p.customer_sk = c.customer_sk;
Performance gains: 10–100x depending on data volume and query complexity.
5. The Complete Timeline Merger
MDDE’s TimelineMerger composes all building blocks into a single operation:
from mdde.generator.timeline_merger_sql import (
generate_timeline_merge,
)
sql = generate_timeline_merge(
output_entity="customer_360",
business_key="customer_id",
sources=[
{"entity_name": "customer_scd2", "attributes": ["name", "status"]},
{"entity_name": "address_scd2", "attributes": ["street", "city"], "prefix": "addr_"},
{"entity_name": "contact_scd2", "attributes": ["email", "phone"], "prefix": "contact_"},
],
dialect="databricks",
compress=True, # Enable compression
)
This single call generates:
- Change points UNION from all sources
- ASOF sample for each source
- Merged timeline with all attributes
- Hash-based compression
- Final SELECT with metadata columns
6. Temporal Building Blocks Reference
MDDE’s TemporalBuildingBlocks provides 11 composable patterns:
Composing Building Blocks
blocks = TemporalBuildingBlocks(config)
# Build CTEs
ctes = [
blocks.change_points_union(sources, business_key),
blocks.asof_sample("customer_scd2", "change_points", business_key, columns),
blocks.asof_sample("address_scd2", "change_points", business_key, columns),
blocks.timeline_compress("merged", business_key, "_hash", columns),
]
# Generate final query
final_select = "SELECT * FROM compressed"
full_sql = blocks.compose_ctes(ctes, final_select)
7. What We Achieved in Part 2
Now — and only now — can we integrate safely.
Coming Next
In Part 3, we address the most complex case:
Valuation History Engineering
- How to calculate initial, previous, and current values correctly
- How to rebuild valuation timelines after late-arriving data
- How to refresh only impacted segments (impact-driven refresh)
- How to maintain determinism with incremental recalculation
We’ll use MDDE’s Impact-Driven Refresh patterns (ADR-271) to solve this.
Read Part 3 here →
SCD2 Is Not Enough (Part 3): Valuation Engineering and Impact-Driven Refresh
Key Takeaways
- Multi-SCD2 JOINs explode — Each source’s change points fragment the timeline
- Change point UNION — Collect all dates, then sample each source
- ASOF sampling — MDDE uses native ASOF JOIN where available, ROW_NUMBER fallback otherwise
- Hash-based compression — Eliminate technical splits, keep only business changes
- Gaps-and-islands pattern — Detect boundaries where hash changes
- PIT tables — Pre-compute surrogate keys for 10–100x query speedup
- Building blocks compose — 11 reusable patterns for any temporal need