Incremental Loads and Change Data Capture

A full reload is correct and eventually too slow. Every incremental strategy trades that simplicity for speed, and each one loses something — usually deletes.

Four extraction strategies

Timestamp extraction, and its blind spot

import duckdb
con = duckdb.connect()

con.execute("""
    create or replace table source_customers as
    select * from (values
        (1, 'Ada Lovelace',  'GB', timestamp '2026-01-01 09:00:00'),
        (2, 'Grace Hopper',  'US', timestamp '2026-01-01 09:00:00'),
        (3, 'Alan Turing',   'GB', timestamp '2026-01-01 09:00:00')
    ) t(customer_id, full_name, country_code, updated_at)
""")
con.execute("create or replace table target_customers as select * from source_customers")
con.execute("create or replace table watermark as select timestamp '2026-01-01 09:00:00' as last_seen")

def incremental_load():
    mark = con.execute("select last_seen from watermark").fetchone()[0]
    con.execute("""
        merge into target_customers t
        using (select * from source_customers where updated_at > ?) s
        on t.customer_id = s.customer_id
        when matched then update set full_name = s.full_name, country_code = s.country_code,
                                     updated_at = s.updated_at
        when not matched then insert by name
    """, [mark])
    con.execute("update watermark set last_seen = (select max(updated_at) from source_customers)")
    print(con.execute("select count(*) as rows from target_customers").df().to_string(index=False))

Update one row and add another:

con.execute("update source_customers set country_code = 'NL', updated_at = timestamp '2026-01-02 10:00:00' where customer_id = 1")
con.execute("insert into source_customers values (4, 'Katherine Johnson', 'US', timestamp '2026-01-02 11:00:00')")
incremental_load()
print(con.execute("select customer_id, country_code from target_customers order by 1").df().to_string(index=False))

 rows
    4

 customer_id country_code
           1           NL
           2           US
           3           GB
           4           US

Correct. Now delete a row at the source:

con.execute("delete from source_customers where customer_id = 2")
incremental_load()
print(con.execute("""
    select 'source' as side, count(*) from source_customers
    union all select 'target', count(*) from target_customers
""").df().to_string(index=False))

   side  count_star()
 source             3
 target             4

The target keeps Grace forever. No error, no warning — the deleted row simply has no timestamp to select. Every aggregate over the target now counts a customer who no longer exists.

Three ways out: soft deletes at the source (deleted_at is not null), a periodic reconciliation of the key sets, or CDC.

def reconcile_deletes():
    removed = con.execute("""
        delete from target_customers
        where customer_id not in (select customer_id from source_customers)
        returning customer_id
    """).fetchall()
    print(f"removed {len(removed)}: {[r[0] for r in removed]}")

reconcile_deletes()

removed 1: [2]

Cheap when the key set fits in memory, which for most dimension tables it does. Schedule it weekly alongside the incremental load.

CDC

A CDC connector reads the database’s write-ahead log and emits one event per change:

con.execute("""
    create or replace table cdc_events as
    select * from (values
        (1, 'c', 1, 'Ada Lovelace',      'GB',   timestamp '2026-01-01 09:00:00'),
        (2, 'c', 2, 'Grace Hopper',      'US',   timestamp '2026-01-01 09:00:01'),
        (3, 'c', 3, 'Alan Turing',       'GB',   timestamp '2026-01-01 09:00:02'),
        (4, 'u', 1, 'Ada Lovelace',      'NL',   timestamp '2026-01-02 10:00:00'),
        (5, 'd', 2,  NULL,                NULL,  timestamp '2026-01-03 14:22:10'),
        (6, 'c', 4, 'Katherine Johnson', 'US',   timestamp '2026-01-03 15:00:00'),
        (7, 'u', 1, 'Ada L.',            'NL',   timestamp '2026-01-04 08:15:00')
    ) t(seq, op, customer_id, full_name, country_code, event_ts)
""")

op is the operation — c create, u update, d delete — and seq is the log position, which is monotonic. Debezium emits exactly this shape, with before and after images.

con.execute("create or replace table dim_customers as select * from target_customers where false")

def apply_cdc():
    # collapse to the latest event per key, then apply
    con.execute("""
        create or replace temp table latest as
        select * exclude (rn) from (
            select *, row_number() over (partition by customer_id order by seq desc) as rn
            from cdc_events
        ) where rn = 1
    """)
    con.execute("delete from dim_customers where customer_id in (select customer_id from latest where op = 'd')")
    con.execute("""
        merge into dim_customers t
        using (select * from latest where op in ('c','u')) s
        on t.customer_id = s.customer_id
        when matched then update set full_name = s.full_name, country_code = s.country_code,
                                     updated_at = s.event_ts
        when not matched then insert values (s.customer_id, s.full_name, s.country_code, s.event_ts)
    """)

apply_cdc()
print(con.execute("select * from dim_customers order by customer_id").df().to_string(index=False))

 customer_id     full_name country_code          updated_at
           1        Ada L.           NL 2026-01-04 08:15:00
           3   Alan Turing           GB 2026-01-01 09:00:02
           4 Katherine Johnson       US 2026-01-03 15:00:00

Grace is gone, Ada has her latest name, and the intermediate update (event 4) was collapsed away. Three rows, matching the source exactly.

Two details make this correct. Collapsing to the latest event per key before applying means the merge never sees two rows for one key. And ordering by seq, not by event_ts or arrival order, means a redelivered old event cannot win:

con.execute("insert into cdc_events values (4, 'u', 1, 'Ada Lovelace', 'GB', timestamp '2026-01-02 10:00:00')")
apply_cdc()
print(con.execute("select customer_id, full_name, country_code from dim_customers where customer_id = 1").df().to_string(index=False))

 customer_id full_name country_code
           1    Ada L.           NL

The redelivered event 4 lost to event 7, as it must. At-least-once delivery makes this scenario routine, not exotic.

Type 1 versus type 2

What we just built is SCD type 1 — the current value only. History needs type 2:

con.execute("""
    create or replace table dim_customers_scd2 (
        customer_id bigint, full_name varchar, country_code varchar,
        valid_from timestamp, valid_to timestamp, is_current boolean
    )
""")

def apply_scd2():
    changes = con.execute("""
        select * exclude (rn) from (
            select *, row_number() over (partition by customer_id order by seq) as rn
            from cdc_events where op in ('c','u')
        ) order by seq
    """).df()

    for _, e in changes.iterrows():
        current = con.execute("""
            select full_name, country_code from dim_customers_scd2
            where customer_id = ? and is_current
        """, [e.customer_id]).fetchone()

        if current and current == (e.full_name, e.country_code):
            continue                                    # nothing changed
        if current:
            con.execute("""
                update dim_customers_scd2 set valid_to = ?, is_current = false
                where customer_id = ? and is_current
            """, [e.event_ts, e.customer_id])
        con.execute("insert into dim_customers_scd2 values (?, ?, ?, ?, null, true)",
                    [e.customer_id, e.full_name, e.country_code, e.event_ts])

apply_scd2()
print(con.execute("""
    select customer_id, full_name, country_code, valid_from, valid_to, is_current
    from dim_customers_scd2 where customer_id = 1 order by valid_from
""").df().to_string(index=False))

 customer_id    full_name country_code          valid_from            valid_to  is_current
           1 Ada Lovelace           GB 2026-01-01 09:00:00 2026-01-02 10:00:00       False
           1 Ada Lovelace           NL 2026-01-02 10:00:00 2026-01-04 08:15:00       False
           1       Ada L.           NL 2026-01-04 08:15:00                None        True

Three versions, with intervals that abut exactly — the previous valid_to equals the next valid_from, no gaps and no overlaps. That property is what makes the point-in-time join correct:

print(con.execute("""
    select o.order_id, o.ordered_at, d.country_code as country_at_order_time
    from orders o
    join dim_customers_scd2 d
      on d.customer_id = o.customer_id
     and o.ordered_at >= d.valid_from
     and o.ordered_at <  coalesce(d.valid_to, timestamp '9999-12-31')
    order by o.order_id
""").df().to_string(index=False))

 order_id ordered_at country_at_order_time
     1001 2026-01-04                    NL
     1003 2026-01-04                    NL

A January order attributed to the country the customer was in then. Join to the type 1 dimension instead and last quarter’s signed-off revenue changes every time someone moves.

The coalesce on valid_to is essential: without it the current row’s null upper bound makes the comparison null and every recent order drops out of the join.

Querying a type 2 dimension

print(con.execute("select count(*) from dim_customers_scd2").fetchone())
print(con.execute("select count(*) from dim_customers_scd2 where is_current").fetchone())

(6,)
(3,)

Six rows for three customers. Every join to an SCD2 table needs either is_current or a point-in-time predicate — forgetting it multiplies rows silently, and is the first bug everyone hits with these tables.

Choosing

• Full reload while the table is small. It is correct and has no failure modes.
• Timestamp incremental with a lookback and a merge, plus periodic delete reconciliation, for large tables where CDC is unavailable.
• CDC when you need deletes, low source load, or intermediate states — and accept the operational cost of a connector and a topic.
• Type 2 only for dimensions where history is actually asked about. It doubles the complexity of every join, so do not apply it by default.

Practice

   side  count_star()
 source             3
 target             4

 customer_id full_name country_code
           1    Ada L.           NL

 order_id country_at_order_time
     1001                    NL

 total_rows  distinct_customers
          6                   3

Next: orchestration — dependencies, retries, and DAGs that recover on their own.