Data Engineering Design Patterns: Initial Load versus Daily Load

This is part 3 of the data engineering design patterns series. Part 2 decided which rows are the next batch, and every run in it read a few rows from one hour. This part is about the load before the first scheduled run: the history that already exists when the pipeline goes live.

There are two ways to load it. One is a separate script, written once, that copies everything up to the go-live date. The other is the daily pipeline itself, handed the past as its interval.

My position is that history never gets its own script. It goes through the same extract, transform and write as every daily run, and the only thing that changes is the interval. This post shows four ways a one-off script drifts away from the pipeline, then the two ways of running history through the pipeline, and the version of that which still breaks.

#The failure

A daily pipeline for card transactions is scheduled from 1 October. The business wants September in the same table, so someone writes a script:

insert into fact_transactions (txn_id, account_id, txn_type, amount, updated_at, interval_start, loaded_at)
select txn_id, account_id, txn_type, amount, updated_at, '2026-09-01', now()
from src_transactions
where updated_at < '2026-10-01';

It runs in seconds and the row count matches the source: 74 rows read, 74 rows written. The daily pipeline starts the next morning, and every run succeeds.

The September total in the table is 302.00 higher than it should be. The source stores a refund as a positive amount with txn_type = 'REFUND', and the pipeline turns it negative before writing it. The script copied it as it was. In one table, a refund on 2 October is -25.00 and a refund on 30 September is +25.00.

Nothing crashed. The script did what it was written to do, and so did the pipeline. They were written to do different things.

#The companion code

Everything below comes from data-engineering-design-patterns, part 3. It uses the same Docker setup as parts 1 and 2: Postgres 16 in one container, and Python 3.11 with Spark 3.5 and Delta Lake 3.2 in another. From the part's folder:

docker compose build
docker compose up -d postgres
docker compose run --rm lab python postgres_loads.py
docker compose run --rm lab python delta_loads.py
docker compose run --rm lab python -m unittest -v test_initial_load

Each scenario starts with an empty source and target. After every step it writes the whole target to out/<engine>/<scenario>/ as a CSV, and its report compares the target with the expected table: the pipeline's transform applied to every source row committed so far. The report lists rows that are missing, duplicated, not supposed to be there, or carrying the wrong amount. The reports used in this post are in example-results.

#The data

The source is one Postgres table, src_transactions. It stores every amount as a positive number and says what the row is in txn_type. September is the history, and it stands in for the years a real source would hold, the way three hourly runs stood in for a pipeline in part 2.

That is 83 source rows, 74 of them in September. A txn_id carries its date, so T-0912-1 is the first row of 12 September.

The pipeline applies two rules between reading and writing:

The sign rule has a bug that nobody has noticed yet: a chargeback also returns money to the customer, and it stays positive. The fix arrives after 3 October.

The load itself is part 2's interval load. Each run reads the half-open interval it is handed, stamps its rows with interval_start, and replaces the rows carrying that stamp in one transaction.

#The script, and four ways it drifts

#The transform

The script in the opening story skipped both rules. After it runs, and again after the first three daily runs:

October is right. The pipeline dropped its test-account row and signed its refund. September has two rows that should not exist and six amounts with the wrong sign, and the difference stays at 302.00 for as long as the table lives.

The obvious answer is to write the script properly. The next three scenarios do that. Their script applies both rules, correctly:

select txn_id, account_id, txn_type,
       case when txn_type = 'REFUND' then -amount else amount end,
       updated_at, '2026-09-01', now()
from src_transactions
where updated_at < '2026-10-01'
  and account_id not like 'TEST-%';

#The cut-over

The script has to run at some moment, and the pipeline has to start at some moment. Here the script runs at 18:00 on 30 September, so the table is ready before go-live. Two more transactions are committed that evening.

After the script, the table matches the source exactly. That is the moment someone checks it. The two evening rows arrive later. The script has already finished, and the pipeline's first interval starts at midnight, so neither load owns the evening of 30 September.

The obvious repair is to start the pipeline one day early. The third row shows what that does. The run for 30 September loads the two evening rows, and it also loads the three daytime rows the script had already written. The gap has become an overlap.

#Ownership

The overlap is one case of a general problem. In this scenario the script runs after midnight, so there is no gap, and the first load is correct. Then 12 September is rerun through the pipeline, as any past day eventually is:

The two copies of T-0912-1 differ in interval_start:

T-0912-1,A-001,PURCHASE,112.00,2026-09-12 09:15:00+00:00,2026-09-01 00:00:00+00:00
T-0912-1,A-001,PURCHASE,112.00,2026-09-12 09:15:00+00:00,2026-09-12 00:00:00+00:00

The pipeline's rerun deletes the rows stamped 12 September and inserts that day again. The script stamped every history row with one value, so the delete finds nothing to remove. Part 1 called this column the record of which load owns a row. The script's rows belong to the script, and the pipeline cannot replace them.

A script could stamp each row with its own day and avoid this. It would then be copying a third thing from the pipeline, after the two transform rules, and it would have to keep copying.

#The fix that arrives later

In the last script scenario, the script copies both rules exactly as they stood, and it runs after midnight. Everything matches after the first load. Then the chargeback bug is found and fixed in the pipeline, and the October days are rerun:

The rerun repairs October. September keeps the old rule, because the fix went into the pipeline and September was not loaded by the pipeline. Pushing the rerun back into September does not help: that is the ownership scenario, and it duplicates every day it touches.

This script had no bug on the day it ran. It was a faithful copy of the pipeline's logic, and the logic moved on without it. A one-off script is a second implementation of the pipeline, and it is frozen on the day it runs.

#The pattern: history is more intervals

The daily load already takes an interval. Nothing in it says the interval has to be yesterday:

def load_run_key(conn, run, negative_types):
    extracted = extract_interval(conn, run)
    transformed = transform(extracted, negative_types)
    stamped = stamp_run_key(transformed, run)
    with conn.transaction():
        conn.execute("delete from fact_transactions where interval_start = %s", (run.interval_start,))
        insert_rows(conn, stamped)

Loading history means calling this function for the intervals before the first scheduled one. There is no second implementation, so there is nothing to drift. There are two ways to choose the intervals.

#One run per day

The simplest is to run the daily pipeline once for every past day. This is what a scheduler's backfill does:

for run in daily_runs(day(9, 1), day(9, 30)):
    load_run_key(conn, run, NEGATIVE_TYPES_V1)

The history is loaded at 18:00 on 30 September, as the script was in the cut-over scenario:

All four drifts are gone:

• The transform is the pipeline's, because the pipeline ran it.
• The cut-over is no longer an event. At 18:00, the history load covers the 29 intervals that have ended. 30 September shows as missing in the first row because its interval is still open, and it runs after midnight like any other day, with the evening rows in it.
• Ownership is uniform. Every row carries its own day, so the rerun of 12 September replaces two rows with the same two rows.
• The fix reaches September the same way it reaches October: rerun the days.

The cost is the run count. September took 29 runs, and the reload after the fix took 33 more. Three years of history is 1,095 runs before the first scheduled one.

#One wide run

The other way is to hand the same function one interval that covers the whole history:

load_run_key(conn, wide_run(day(9, 1), day(9, 30)), NEGATIVE_TYPES_V1)

Up to the first three daily runs, it produces the same rows in five pipeline runs instead of 33:

The transform is the pipeline's, and there is still no cut-over. If the post stopped here, the wide run would look like the better choice.

#The variant that breaks: a wide run keyed on the run

Two more steps on the same table:

The rerun of 12 September duplicates that day, exactly as it did after the script. The load stamps its rows with the run's interval_start. For a daily run that is the row's own day. For the wide run it is 1 September for all 67 rows, so the daily rerun's delete finds nothing stamped 12 September.

The rerun of 1 September is worse. Its delete removes everything stamped 1 September, which is the whole history, and its insert puts back the two rows of 1 September. The table drops from 82 rows to 17. The duplicates are gone only because their originals went with the rest. The run reported success.

This is the pipeline's own code, unchanged, and it has reproduced the script's ownership problem and added a destructive one. Sharing the code is not enough. The rows have to come out the same whichever width of run wrote them.

#Keying each row on its own day

The fix is to take the stamp from the row, not from the run. Each row is stamped with the daily interval it falls in, and the delete covers every daily interval inside the run's interval:

def load_day_key(conn, run, negative_types):
    extracted = extract_interval(conn, run)
    transformed = transform(extracted, negative_types)
    stamped = stamp_day_key(transformed)
    with conn.transaction():
        conn.execute(
            "delete from fact_transactions where interval_start >= %s and interval_start < %s",
            (run.interval_start, run.interval_end),
        )
        insert_rows(conn, stamped)

For a daily run this is the same delete and the same stamp as before. For a wide run, every row gets the stamp a daily run would have given it.

Both reruns leave the table unchanged. The fix is applied to September and October together by one run with the interval 1 September to 3 October. The companion tests also compare this table with the one that 67 daily runs produced, on every column except loaded_at, and the two are identical. A wide run and a daily run are now interchangeable, which is what makes it safe to load history one way and rerun it the other.

Part 1 warned against keying the delete on a column the source controls, and updated_at is such a column. The difference is where the scope comes from. In part 1 the delete took its scope from business_date and the insert took whatever the file held, and the two disagreed by one row. Here the read, the stamp and the delete all take their scope from the run's interval. The run reads the rows inside it, stamps each one with a day inside it, and deletes the days inside it.

That holds on one condition: every interval starts and ends on a day boundary. A run from 06:00 to 06:00 would stamp rows with two different days and delete neither day completely. Nothing else in the load depends on the width, so a week or a month works the same way, and the width can be chosen for memory and transaction size.

#The variant that breaks: an interval that has not ended

A scheduler that works in data intervals starts a run after its interval has ended. A history run is started by hand, and nothing stops it being handed an interval that is still open. Here the wide run at 18:00 on 30 September is given the whole month:

This is the script's cut-over gap again. The run loaded what 30 September held at 18:00, the scheduler considers that day done, and the evening rows are never read.

The difference from the script is the last row. 30 September is an interval the pipeline owns, so rerunning it loads the evening rows and replaces the daytime ones. The script's version of this repair duplicated three rows. The rule that avoids the repair is short: a history run ends at the last interval that has closed, and the scheduler takes the next one.

#Delta Lake: what each way costs

On Delta Lake the source stays in Postgres and the target is a Delta table. The load is the same function with the write swapped for a replaceWhere, the one-commit delete-then-insert from part 1:

predicate = (
    f"interval_start >= {timestamp_literal(run.interval_start)} "
    f"AND interval_start < {timestamp_literal(run.interval_end)}"
)
(
    batch.write.format("delta")
    .mode("overwrite")
    .option("replaceWhere", predicate)
    .save(path)
)

The results match Postgres. The wide run keyed on the run duplicates 12 September and then loses 63 rows when 1 September is rerun, and the wide run keyed on the row's day survives both reruns and the fix. What Delta adds is a count of what each way leaves behind:

Every run is one commit and at least one file. One run per day leaves 33 files holding 80 rows between them, and three years of history would leave over a thousand before the first scheduled run. The wide run writes the 29 days as one file. Part 17 covers compaction, which is how the one-run-per-day table gets repaired afterwards.

The wide file has its own cost, and it shows in the commit history of the day-keyed table:

Rerunning 12 September replaces two rows and writes 67. A Parquet file cannot be edited in place, so removing two rows from the 67-row history file means writing the other 65 again. In the one-run-per-day table the same rerun removed one two-row file and added another. At this size the difference is nothing. On a real history it is the price of a one-day rerun into a file that holds a month.

#Side by side

Two rows leave every column clean. One run per day needs no change to part 2's load and pays for it with a run, a commit and a file per day of history. The wide run keyed on the row's day needs the key change and pays with larger rewrites when a single day is rerun. The script is not one of the two.

#Checks worth keeping

The companion code's tests state the claims above as assertions:

• A script that copies the source leaves six September refunds with the wrong sign and two test-account rows in the table.
• A script run at 18:00 misses the two evening rows, and running 30 September through the pipeline duplicates the three daytime rows.
• Rerunning a script-loaded day through the pipeline duplicates it.
• A fix rerun over October leaves the four September chargebacks wrong.
• One run per day matches the expected table after the load, after a rerun and after 33 reruns with the fix.
• A wide run keyed on the run duplicates 12 September on rerun, and a rerun of 1 September leaves 17 rows with 63 missing, on both engines.
• A wide run keyed on the row's day survives both reruns and the fix, and its final table equals the one-run-per-day table.
• A wide run handed an open day misses the evening rows until that day is rerun.
• On Delta, 33 daily runs make 33 commits and 33 files, and the wide load makes five of each.

The check I would add to a real pipeline is the one from the cut-over scenario, run later than anyone thinks necessary. The script's table matched the source at 18:00 and was wrong by midnight. A count for each closed day, compared between source and target after the first week of daily runs, finds the missing evening rows and every duplicated day in this post. It has to apply the pipeline's rules to the source side, or it will pass the script that kept the test accounts. That is part 14.

#What is next

Every source row in this post was written once and never changed. A row that is updated gets a new updated_at, falls into a later interval, and leaves its older copy behind in the earlier one. Part 4 is upserts with MERGE: replacing a row by its key wherever it was first loaded, and the batch with duplicate source keys that makes MERGE fail or pick a row at random.

#Summary to save

• Load history through the daily pipeline's own code. A one-off script is a second implementation, and it is frozen on the day it runs.
• A script drifts in four ways: it skips or re-implements the transform, it leaves a gap or an overlap at the cut-over, its rows cannot be replaced by a pipeline rerun, and later fixes never reach them.
• A script that is correct on the day it runs still fails the last three. Checking it once, after it finishes, finds none of them.
• History is the intervals before the first scheduled one. Run the pipeline once per past day, or once with a wide interval.
• One run per day needs no change to the daily load. It costs one run, one commit and one data file per day of history.
• A wide run is only safe when each row is stamped with the daily interval it falls in, and the delete covers every daily interval inside the run's interval. Stamped with the run's own start, a daily rerun duplicates a day or deletes the history.
• Keep every interval on day boundaries, so that a wide run and a daily run write identical rows and either can replace the other.
• End a history run at the last interval that has closed. An open day loaded early loses whatever arrives after the run.
• On Delta Lake, a one-day rerun into a wide file rewrites the whole file. A wide history load trades many small files for larger rewrites later.