Part 6 — Transformations Part 2: SCD Types, Fact Loading, Spark, and Merge Optimization

Part 5 covered ELT strategy, dbt architecture, incremental Silver models, and MERGE patterns for Silver-layer conformance. This part goes deeper into the engine room of your pipeline: the advanced mechanics of dimension management, fact table loading, Spark-based transformations for lakehouses, and performance tuning for merge-heavy workloads.

Misconfiguring SCD logic, fact grain, or merge operations leads to downstream errors, incorrect historical analysis, late-arriving dimension gaps, and inconsistent KPI calculations. The patterns in this part are proven across Snowflake, Fabric, Databricks, BigQuery, Synapse, and Redshift. Master them here and your Gold layer will be reliable, testable, and maintainable at production scale.

1 Understanding Slowly Changing Dimensions Beginner

Dimensions change over time. A customer moves to a different city. A product changes category. An employee transfers to a new department. A sales territory gets reassigned. How your warehouse handles these changes determines whether you can answer historical questions accurately — “what was true at the time this transaction occurred?” — or only current-state questions.

The three types used in modern warehouses are defined by how they respond to changes in source attribute values.

2 SCD Type 1 — Overwrite Changes Beginner

Type 1 is the simplest dimension management pattern. When source data changes, the existing row is updated in place. No history is preserved. The dimension always reflects the current state of the source.

-- SCD Type 1: overwrite pattern -- works across all platforms
-- Use for corrections and attributes where history is irrelevant

-- Snowflake Type 1 MERGE
MERGE INTO gold.dim_customer AS tgt
USING silver.customer AS src
    ON tgt.customer_natural_key = src.customer_id
WHEN MATCHED AND (
        tgt.customer_name <> src.customer_name
     OR tgt.email         <> src.email
     OR tgt.city          <> src.city
     OR tgt.loyalty_tier  <> src.loyalty_tier
) THEN UPDATE SET
    tgt.customer_name  = src.customer_name,
    tgt.email          = src.email,
    tgt.city           = src.city,
    tgt.state_province = src.state_province,
    tgt.loyalty_tier   = src.loyalty_tier,
    tgt.updated_at     = src.updated_at
WHEN NOT MATCHED THEN INSERT (
    customer_natural_key, customer_name, email,
    city, state_province, country, loyalty_tier, updated_at
)
VALUES (
    src.customer_id, src.customer_name, src.email,
    src.city, src.state_province, src.country,
    src.loyalty_tier, src.updated_at
);

-- BigQuery Type 1 MERGE
MERGE gold.dim_customer AS tgt
USING silver.customer AS src
    ON tgt.customer_natural_key = src.customer_id
WHEN MATCHED AND (
        tgt.customer_name != src.customer_name
     OR tgt.email         != src.email
     OR tgt.loyalty_tier  != src.loyalty_tier
) THEN UPDATE SET
    customer_name  = src.customer_name,
    email          = src.email,
    city           = src.city,
    loyalty_tier   = src.loyalty_tier,
    updated_at     = src.updated_at
WHEN NOT MATCHED THEN INSERT ROW;

3 SCD Type 2 — Track Full History Intermediate

SCD Type 2 is the most important dimension management technique in analytical systems. Each change in a tracked attribute creates a new row with a new surrogate key, while the previous row is closed by setting its effective end date. Fact tables always join to the dimension version that was current at the time of the transaction — enabling accurate historical analysis across any time period.

The Type 2 Row Structure

4 SCD Type 2 — Complete Two-Step Pattern Intermediate

The complete SCD Type 2 implementation requires two operations: first close the changed rows (set effective_to and is_current = 0), then insert the new current versions. Most SQL MERGE implementations handle both in a single statement for inserts, but closing old rows and inserting new versions in one MERGE is platform-dependent. The two-step pattern shown below works reliably on all platforms.

Step 1 — Close Changed Rows

-- Step 1: Close rows where tracked attributes have changed
-- Sets effective_to = today and is_current = 0 for changed rows
-- Does NOT insert the new version -- that is Step 2

-- Snowflake / Fabric / Synapse / Redshift
MERGE INTO gold.dim_customer AS tgt
USING silver.customer AS src
    ON  tgt.customer_natural_key = src.customer_id
    AND tgt.is_current           = 1           -- only match current rows
WHEN MATCHED AND (
        tgt.customer_name <> src.customer_name
     OR tgt.email         <> src.email
     OR tgt.city          <> src.city
     OR tgt.loyalty_tier  <> src.loyalty_tier
) THEN UPDATE SET
    tgt.effective_to = CURRENT_DATE(),
    tgt.is_current   = 0;
    -- Note: do NOT update the attribute columns here
    -- The old values must be preserved for historical analysis

Step 2 — Insert New Current Versions

-- Step 2: Insert new current rows for customers whose previous row was just closed
-- Also inserts rows for brand-new customers (no existing row in the dimension)

-- Snowflake
INSERT INTO gold.dim_customer (
    customer_natural_key,
    customer_name,
    email,
    city,
    state_province,
    country,
    loyalty_tier,
    effective_from,
    effective_to,
    is_current
)
SELECT
    src.customer_id,
    src.customer_name,
    src.email,
    src.city,
    src.state_province,
    src.country,
    src.loyalty_tier,
    CURRENT_DATE()           AS effective_from,
    '9999-12-31'::DATE       AS effective_to,
    1                        AS is_current
FROM silver.customer src
-- Include new customers (no row in dimension at all)
-- AND customers whose previous row was just closed (is_current = 0 after Step 1)
WHERE NOT EXISTS (
    SELECT 1
    FROM gold.dim_customer tgt
    WHERE tgt.customer_natural_key = src.customer_id
    AND   tgt.is_current           = 1
);
-- After Step 1, closed rows have is_current = 0
-- This INSERT finds customers with no current row (either new or just closed)
-- and creates their new current row

Verify SCD Type 2 Correctness

-- Verify: each customer should have exactly one current row (is_current = 1)
SELECT
    customer_natural_key,
    COUNT(*) AS current_row_count
FROM gold.dim_customer
WHERE is_current = 1
GROUP BY customer_natural_key
HAVING COUNT(*) > 1;
-- Should return 0 rows -- any result here means a SCD2 logic bug

-- Verify: effective_to for closed rows should never be 9999-12-31
SELECT COUNT(*) AS bad_rows
FROM gold.dim_customer
WHERE is_current = 0
AND effective_to = '9999-12-31';
-- Should return 0

-- Verify: history is preserved -- one customer with multiple versions
SELECT
    customer_natural_key,
    customer_name,
    loyalty_tier,
    effective_from,
    effective_to,
    is_current
FROM gold.dim_customer
WHERE customer_natural_key = 'CUST-12345'
ORDER BY effective_from;

5 SCD Type 2 — Spark and Databricks Delta MERGE Advanced

Databricks Delta MERGE can handle the close-and-insert pattern in a single statement using the whenMatchedUpdate + whenNotMatchedInsert combination. This is one of Delta Lake’s most powerful capabilities — ACID-compliant upserts on a lakehouse table without the two-step SQL pattern required on some other platforms.

# Databricks: SCD Type 2 with Delta MERGE
# Single statement handles both closing old rows and inserting new versions
# Requires Delta Lake 2.0+ (Databricks Runtime 11.x+)

from delta.tables import DeltaTable
from pyspark.sql.functions import current_date, lit

# Load the current Gold dimension table
dim_customer = DeltaTable.forName(spark, "gold.dim_customer")

# Load the Silver source with only changed records
# (only process rows where attributes have changed since last run)
updates = spark.sql("""
    SELECT
        s.customer_id           AS customer_natural_key,
        s.customer_name,
        s.email,
        s.city,
        s.state_province,
        s.country,
        s.loyalty_tier,
        CURRENT_DATE()          AS effective_from,
        DATE('9999-12-31')      AS effective_to,
        1                       AS is_current
    FROM silver.customer s
    LEFT JOIN gold.dim_customer d
        ON  d.customer_natural_key = s.customer_id
        AND d.is_current           = 1
    WHERE d.customer_natural_key IS NULL    -- new customers
       OR d.customer_name  <> s.customer_name  -- changed attributes
       OR d.email          <> s.email
       OR d.loyalty_tier   <> s.loyalty_tier
""")

# Execute the SCD2 MERGE
(
    dim_customer.alias("tgt")
    .merge(
        updates.alias("src"),
        "tgt.customer_natural_key = src.customer_natural_key AND tgt.is_current = 1"
    )
    # Close the old row when attributes have changed
    .whenMatchedUpdate(
        condition = """
            tgt.customer_name <> src.customer_name
         OR tgt.email         <> src.email
         OR tgt.loyalty_tier  <> src.loyalty_tier
        """,
        set = {
            "effective_to": "current_date()",
            "is_current":   lit(0)
        }
    )
    # Insert new rows for new customers
    .whenNotMatchedInsert(
        values = {
            "customer_natural_key": "src.customer_natural_key",
            "customer_name":        "src.customer_name",
            "email":                "src.email",
            "city":                 "src.city",
            "state_province":       "src.state_province",
            "country":              "src.country",
            "loyalty_tier":         "src.loyalty_tier",
            "effective_from":       "current_date()",
            "effective_to":         "date('9999-12-31')",
            "is_current":           lit(1)
        }
    )
    .execute()
)

# Step 2 still needed for Databricks: insert new current versions for closed rows
# The MERGE above closes old rows but does NOT insert new versions for changed customers
# (Delta MERGE cannot insert AND close in the same matched clause)
# Run a second INSERT for changed customers:
spark.sql("""
    INSERT INTO gold.dim_customer
    SELECT
        s.customer_id           AS customer_natural_key,
        s.customer_name,
        s.email,
        s.city,
        s.state_province,
        s.country,
        s.loyalty_tier,
        CURRENT_DATE()          AS effective_from,
        DATE('9999-12-31')      AS effective_to,
        1                       AS is_current
    FROM silver.customer s
    WHERE NOT EXISTS (
        SELECT 1 FROM gold.dim_customer d
        WHERE d.customer_natural_key = s.customer_id
        AND   d.is_current = 1
    )
""")

6 SCD Type 2 — dbt Snapshots Intermediate

dbt snapshots are the recommended way to implement SCD Type 2 in a dbt project. They handle the open-close lifecycle automatically — you define the source, the unique key, and the strategy, and dbt manages row versioning on every run. Snapshots are stored in a dedicated snapshots/ folder in your dbt project.

-- snapshots/snap_crm_customer.sql
-- dbt snapshot: automatically manages SCD Type 2 row lifecycle

{% snapshot snap_crm_customer %}

{{
    config(
        target_schema = 'snapshots',
        unique_key     = 'customer_id',
        -- Strategy options:
        -- 'timestamp': uses an updated_at column to detect changes
        -- 'check':     compares specific columns for changes
        strategy       = 'timestamp',
        updated_at     = 'updated_at',
        -- invalidate_hard_deletes: marks rows as deleted when they
        -- disappear from source (requires dbt 1.x+)
        invalidate_hard_deletes = True
    )
}}

SELECT
    customer_id,
    customer_name,
    email,
    city,
    state_province,
    country,
    loyalty_tier,
    updated_at
FROM {{ ref('stg_crm_customer') }}

{% endsnapshot %}

-- dbt adds these columns automatically to snapshot tables:
-- dbt_scd_id:        unique hash of (natural_key + dbt_updated_at)
-- dbt_updated_at:    when this version was created in the snapshot
-- dbt_valid_from:    effective_from equivalent
-- dbt_valid_to:      effective_to equivalent (NULL = current row)
-- Note: dbt uses NULL for current rows instead of 9999-12-31
-- In your Gold mart models, translate to match your convention:

-- models/marts/core/dim_customer.sql
-- Build Gold dimension from dbt snapshot
SELECT
    {{ dbt_utils.generate_surrogate_key(['customer_id', 'dbt_valid_from']) }}
                                AS customer_key,
    customer_id                 AS customer_natural_key,
    customer_name,
    email,
    city,
    state_province,
    country,
    loyalty_tier,
    dbt_valid_from              AS effective_from,
    COALESCE(dbt_valid_to,
             '9999-12-31'::DATE) AS effective_to,
    CASE WHEN dbt_valid_to IS NULL THEN 1 ELSE 0 END AS is_current
FROM {{ ref('snap_crm_customer') }}

7 Late-Arriving Dimensions — The Unknown Key Pattern Intermediate

Late-arriving dimensions occur when fact records arrive before the corresponding dimension record has been loaded. This is a common situation in real pipelines — an order arrives before the new customer’s first CRM record has synced, or a game event arrives before the player profile has been created. If you attempt to join the fact to the dimension and no match exists, the fact row is either dropped or has a null foreign key — both break your analytics.

The Unknown Key Pattern

Pre-insert a special “unknown” row in every dimension table with a sentinel surrogate key value of -1 or 0. When a fact record joins to a dimension and no match exists, it gets the unknown key instead of being dropped or having a null foreign key. When the real dimension record arrives later, the fact rows can be updated to point to the correct key.

-- Step 1: Insert the unknown dimension row (run once at Gold layer setup)
-- All dimensions that participate in facts need an unknown row

INSERT INTO gold.dim_customer (
    customer_key, customer_natural_key, customer_name,
    email, city, state_province, country, loyalty_tier,
    effective_from, effective_to, is_current
)
VALUES (
    -1,         -- sentinel surrogate key
    'UNKNOWN',  -- sentinel natural key
    'Unknown Customer',
    'unknown@unknown.com',
    'Unknown', 'Unknown', 'Unknown', 'Unknown',
    '1900-01-01', '9999-12-31', 1
);

-- Same pattern for every dimension:
INSERT INTO gold.dim_product (product_key, product_natural_key, product_name, ...)
VALUES (-1, 'UNKNOWN', 'Unknown Product', ...);

INSERT INTO gold.dim_date (date_key, full_date, ...)
VALUES (-1, '1900-01-01', ...);  -- unknown date for missing date keys

-- Step 2: Fact load uses COALESCE to substitute unknown key when no match found
-- This prevents null foreign keys and lost fact rows

INSERT INTO gold.fct_order_line (
    order_line_key,
    order_date_key,
    customer_key,
    product_key,
    quantity,
    unit_price,
    extended_amount
)
SELECT
    ROW_NUMBER() OVER (ORDER BY o.order_id, o.line_number)  AS order_line_key,
    COALESCE(d.date_key,     -1)   AS order_date_key,     -- -1 if date not found
    COALESCE(c.customer_key, -1)   AS customer_key,       -- -1 if customer not found
    COALESCE(p.product_key,  -1)   AS product_key,        -- -1 if product not found
    o.quantity,
    o.unit_price,
    o.quantity * o.unit_price - COALESCE(o.discount, 0)    AS extended_amount
FROM silver.orders o
LEFT JOIN gold.dim_date     d ON d.full_date            = o.order_date
LEFT JOIN gold.dim_customer c ON c.customer_natural_key = o.customer_id
                              AND c.is_current          = 1
LEFT JOIN gold.dim_product  p ON p.product_natural_key  = o.product_id
                              AND p.is_current          = 1;
-- LEFT JOIN ensures fact rows are never dropped due to missing dimensions
-- COALESCE assigns the unknown key (-1) when no dimension match exists

-- Step 3: Backfill fact rows after late-arriving dimension arrives
-- When the real dimension record finally loads, update fact rows to correct key

UPDATE gold.fct_order_line f
SET    f.customer_key = c.customer_key
FROM   gold.dim_customer c
WHERE  f.customer_key = -1                          -- only rows with unknown key
AND    c.customer_natural_key = (
           -- look up the natural key from Silver using the fact's order context
           SELECT o.customer_id
           FROM   silver.orders o
           WHERE  o.order_id = f.order_natural_key
       )
AND    c.is_current = 1;                            -- use the current dimension row

8 Fact Table Loading — Grain Enforcement Intermediate

Fact tables must always respect the grain defined in Part 1. Every INSERT into a fact table is an opportunity to introduce grain violations — duplicate rows at the same grain, incorrect surrogate key lookups, or measures that are aggregated at the wrong level before loading. These errors produce incorrect analytics that are very difficult to detect after the fact.

The Fact Load Pattern

-- Gold fact load: complete pattern with grain enforcement
-- Snowflake / Fabric / BigQuery (adjust syntax per platform as needed)

-- Step 1: Load into a staging area first -- do not load directly to Gold fact
CREATE OR REPLACE TEMPORARY TABLE stg_fct_order_line AS
SELECT
    o.order_id          AS order_natural_key,
    o.line_number,
    -- Surrogate key lookups with COALESCE for late-arriving dimensions
    COALESCE(d.date_key,     -1) AS order_date_key,
    COALESCE(c.customer_key, -1) AS customer_key,
    COALESCE(p.product_key,  -1) AS product_key,
    COALESCE(s.store_key,    -1) AS store_key,
    -- Measures: atomic grain -- NO aggregation before Gold
    o.quantity,
    o.unit_price,
    COALESCE(o.discount_amount, 0)                  AS discount_amount,
    o.quantity * o.unit_price
        - COALESCE(o.discount_amount, 0)            AS extended_amount,
    CURRENT_TIMESTAMP()                             AS loaded_at
FROM silver.orders o
LEFT JOIN gold.dim_date     d ON d.full_date            = o.order_date
LEFT JOIN gold.dim_customer c ON c.customer_natural_key = o.customer_id
                              AND c.is_current          = 1
LEFT JOIN gold.dim_product  p ON p.product_natural_key  = o.product_id
                              AND p.is_current          = 1
LEFT JOIN gold.dim_store    s ON s.store_natural_key    = o.store_id
                              AND s.is_current          = 1;

-- Step 2: Validate grain before loading to Gold
-- This query should return 0 rows -- any result is a grain violation
SELECT
    order_natural_key,
    line_number,
    COUNT(*) AS duplicate_count
FROM stg_fct_order_line
GROUP BY order_natural_key, line_number
HAVING COUNT(*) > 1;

-- Step 3: Only load to Gold after grain validation passes
INSERT INTO gold.fct_order_line
SELECT
    ROW_NUMBER() OVER (ORDER BY order_natural_key, line_number) AS order_line_key,
    order_natural_key,
    order_date_key,
    customer_key,
    product_key,
    store_key,
    quantity,
    unit_price,
    discount_amount,
    extended_amount,
    loaded_at
FROM stg_fct_order_line;

9 Fact Table Loading — Spark Example Advanced

# Databricks / Fabric: fact table load in PySpark
from pyspark.sql.functions import (
    col, coalesce, lit, current_timestamp, monotonically_increasing_id
)

# Load Silver source
orders = spark.table("silver.orders")

# Load dimension lookup tables (cache small dimensions for join performance)
dim_date     = spark.table("gold.dim_date").cache()
dim_customer = spark.table("gold.dim_customer").filter(col("is_current") == 1).cache()
dim_product  = spark.table("gold.dim_product").filter(col("is_current") == 1).cache()
dim_store    = spark.table("gold.dim_store").filter(col("is_current") == 1).cache()

# Build the fact DataFrame with LEFT JOINs
fact = (
    orders
    .join(dim_date,
          orders.order_date == dim_date.full_date,
          "left")
    .join(dim_customer,
          orders.customer_id == dim_customer.customer_natural_key,
          "left")
    .join(dim_product,
          orders.product_id == dim_product.product_natural_key,
          "left")
    .join(dim_store,
          orders.store_id == dim_store.store_natural_key,
          "left")
    .select(
        # Surrogate keys with unknown key fallback
        coalesce(col("date_key"),     lit(-1)).alias("order_date_key"),
        coalesce(col("customer_key"), lit(-1)).alias("customer_key"),
        coalesce(col("product_key"),  lit(-1)).alias("product_key"),
        coalesce(col("store_key"),    lit(-1)).alias("store_key"),
        # Natural key for audit
        col("order_id").alias("order_natural_key"),
        col("line_number"),
        # Measures
        col("quantity"),
        col("unit_price"),
        coalesce(col("discount_amount"), lit(0)).alias("discount_amount"),
        (col("quantity") * col("unit_price")
         - coalesce(col("discount_amount"), lit(0))).alias("extended_amount"),
        current_timestamp().alias("loaded_at")
    )
)

# Grain validation before writing
from pyspark.sql.functions import count as spark_count

grain_check = (
    fact
    .groupBy("order_natural_key", "line_number")
    .agg(spark_count("*").alias("cnt"))
    .filter(col("cnt") > 1)
)

if grain_check.count() > 0:
    grain_check.show()
    raise Exception("Grain violation detected -- aborting fact load")

# Write to Gold as Delta (append mode for new orders)
(
    fact
    .write
    .format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .saveAsTable("gold.fct_order_line")
)

# Unpersist cached dimensions
dim_date.unpersist()
dim_customer.unpersist()
dim_product.unpersist()
dim_store.unpersist()

10 Merge Optimization for Performance Advanced

MERGE operations on large tables are the most expensive operations in a Gold pipeline. They read the entire target table to find matching rows, compare attributes, and write updates. Without optimization, a MERGE on a 500M-row dimension table can take hours. These platform-specific techniques dramatically reduce MERGE execution time.

Snowflake: Stream-Based Change Detection

-- Snowflake Streams capture changes to a table incrementally
-- Only rows that have changed since the last stream consumption are processed
-- This eliminates the need to scan the entire Silver table on every run

-- Create a stream on the Silver customer table
CREATE OR REPLACE STREAM silver_customer_stream
    ON TABLE silver.customer
    APPEND_ONLY = FALSE;  -- capture inserts, updates, AND deletes

-- Use the stream as the MERGE source -- only processes changed rows
MERGE INTO gold.dim_customer AS tgt
USING silver_customer_stream AS src
    ON  tgt.customer_natural_key = src.customer_id
    AND tgt.is_current           = 1
WHEN MATCHED AND src.METADATA$ACTION = 'DELETE' THEN
    UPDATE SET tgt.effective_to = CURRENT_DATE(), tgt.is_current = 0
WHEN MATCHED AND src.METADATA$ACTION = 'INSERT' AND (
        tgt.customer_name <> src.customer_name
     OR tgt.loyalty_tier  <> src.loyalty_tier
) THEN
    UPDATE SET tgt.effective_to = CURRENT_DATE(), tgt.is_current = 0
WHEN NOT MATCHED AND src.METADATA$ACTION = 'INSERT' THEN
    INSERT (...) VALUES (...);
-- Snowflake Streams: https://docs.snowflake.com/en/user-guide/streams

Databricks: Z-Order and File Compaction Before MERGE

-- Pre-optimize Delta table before running MERGE
-- Z-ordering on the join column co-locates related data
-- Reduces files scanned during MERGE match lookups

-- Run OPTIMIZE before bulk MERGE operations (weekly or after large loads)
OPTIMIZE gold.dim_customer
ZORDER BY (customer_natural_key);
-- After Z-ordering, MERGE reads far fewer files to find matching natural keys

-- Set Delta auto-optimize for ongoing small writes
ALTER TABLE gold.dim_customer
SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact'   = 'true'
);

BigQuery: Filter Source Before MERGE

-- BigQuery MERGE cost reduction: filter source to changed rows only
-- BigQuery charges by bytes scanned -- smaller source = lower cost

MERGE gold.dim_customer AS tgt
USING (
    -- Only include rows changed in the last 2 days
    -- Reduces bytes scanned dramatically on large Silver tables
    SELECT *
    FROM silver.customer
    WHERE updated_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)
      AND customer_id IS NOT NULL
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY customer_id
        ORDER BY updated_at DESC
    ) = 1
) AS src
ON tgt.customer_natural_key = src.customer_id
WHEN MATCHED AND (
    tgt.customer_name != src.customer_name
    OR tgt.loyalty_tier != src.loyalty_tier
) THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ROW;

General MERGE Performance Rules

• Filter source before MERGE. Process only rows changed since the last run. A MERGE with 10,000 source rows is orders of magnitude cheaper than one with 10,000,000.
• Match on indexed columns. Join keys in the MERGE ON clause should be the same columns used for clustering, Z-ordering, or distribution keys.
• Minimize the WHEN MATCHED conditions. Check only the attributes that actually drive SCD decisions — not every column in the dimension.
• Use change-detection streams where available. Snowflake Streams and Databricks Change Data Feed eliminate full-table scans on the source.

11 Audit Tables — Tracking Pipeline Execution Intermediate

Audit tables record metadata about every pipeline run — row counts, timestamps, error states, and data quality metrics. They are essential for debugging pipeline failures, monitoring data freshness, and providing evidence for data governance and compliance requirements.

-- Create an audit table to track every Gold pipeline execution
CREATE TABLE etl_audit (
    audit_id         BIGINT IDENTITY(1,1) PRIMARY KEY,
    pipeline_name    VARCHAR(200)    NOT NULL,
    target_table     VARCHAR(200)    NOT NULL,
    run_started_at   TIMESTAMP       NOT NULL,
    run_completed_at TIMESTAMP,
    rows_inserted    BIGINT          DEFAULT 0,
    rows_updated     BIGINT          DEFAULT 0,
    rows_deleted     BIGINT          DEFAULT 0,
    rows_rejected    BIGINT          DEFAULT 0,
    status           VARCHAR(20)     NOT NULL,  -- 'RUNNING', 'SUCCESS', 'FAILED'
    error_message    VARCHAR(MAX),
    watermark_from   TIMESTAMP,
    watermark_to     TIMESTAMP
);

-- Record pipeline start
INSERT INTO etl_audit (
    pipeline_name, target_table, run_started_at, status
)
VALUES (
    'gold_dim_customer_scd2',
    'gold.dim_customer',
    CURRENT_TIMESTAMP(),
    'RUNNING'
);

-- After MERGE completes, update audit record with results
UPDATE etl_audit
SET
    run_completed_at = CURRENT_TIMESTAMP(),
    rows_inserted    = @inserted_count,
    rows_updated     = @updated_count,
    status           = 'SUCCESS',
    watermark_to     = CURRENT_TIMESTAMP()
WHERE pipeline_name = 'gold_dim_customer_scd2'
AND   status        = 'RUNNING'
AND   run_started_at = (
    SELECT MAX(run_started_at)
    FROM etl_audit
    WHERE pipeline_name = 'gold_dim_customer_scd2'
);

-- Query audit history for monitoring
SELECT
    pipeline_name,
    target_table,
    run_started_at,
    DATEDIFF('second', run_started_at, run_completed_at) AS duration_sec,
    rows_inserted,
    rows_updated,
    status
FROM etl_audit
WHERE run_started_at >= DATEADD('day', -7, CURRENT_TIMESTAMP())
ORDER BY run_started_at DESC;

12 Choosing SCD Type for Each Dimension Beginner

Not every dimension needs SCD Type 2. Over-applying Type 2 adds complexity and storage without analytical benefit. This reference table helps you make the right choice for common dimension attributes.

13 Workshops

References

• Microsoft — SCD Concepts and Types
• Kimball Group — Slowly Changing Dimensions
• BigQuery — MERGE Statement
• Snowflake — MERGE Statement
• Snowflake — Streams for Change Tracking
• Databricks — Delta Lake MERGE
• Databricks — Z-Order Optimization
• dbt — Snapshots (SCD Type 2)
• Delta Lake Documentation
• Microsoft Fabric — MERGE Statement