APPLYDJINN
← ETL and Data Engineering: reliable pipelinesProgress & review ↗

Typorithm · medium

Versioned SQLite UPSERT and idempotent replay

pythonetldata-engineeringindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Versioned SQLite UPSERT and idempotent replay

0%

Concept and contract

Persist source ordering across batches so replay cannot duplicate or roll back current state.

In-memory deduplication cannot protect a sink across separate program runs. A persistent current-state table needs a unique business identity and a rule for handling an already stored row. Here the primary key identifies the entity and the source version decides whether its payload may advance.

Idempotent replay means applying the same delivery again leaves the same current-state result. It does not imply that every external side effect occurred exactly once. The lesson scopes its guarantee to one owned SQLite database, parameterized statements and the explicit version contract used by the supplied source rows.

Mechanism and worked example

The program opens an on-disk database, creates its fact table and applies the same delivery twice.

The program opens an on-disk database, creates its fact table and applies the same delivery twice. INSERT handles a new identity; ON CONFLICT performs an update only when excluded.version exceeds the stored version. A preliminary check rejects equal-version payload disagreements instead of choosing an arbitrary winner.

The final query sorts identities and returns the measured row count. Trace a newer value followed by an older value: the WHERE guard prevents regression. Repeating the whole batch cannot add another row for the same primary key, and the independent empty-input case remains a valid empty sink.

Complete worked example

python

import sqlite3

def inspect_example(rows):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.execute('CREATE TABLE fact(id TEXT PRIMARY KEY, version INTEGER NOT NULL, amount_cents INTEGER NOT NULL)')
        connection.commit()
        for batch in [rows, rows]:
            with connection:
                for row in batch:
                    previous = connection.execute('SELECT version, amount_cents FROM fact WHERE id=?', [row['id']]).fetchone()
                    if previous and previous[0] == row['version'] and previous[1] != row['amount_cents']:
                        raise ValueError('conflicting source version')
                    connection.execute('INSERT INTO fact VALUES (:id,:version,:amount_cents) ON CONFLICT(id) DO UPDATE SET version=excluded.version, amount_cents=excluded.amount_cents WHERE excluded.version > fact.version', row)
        stored = [list(r) for r in connection.execute('SELECT id,version,amount_cents FROM fact ORDER BY id')]
        return {'rows': stored, 'count': len(stored)}
    finally:
        connection.close()
Reasoning
The program opens an on-disk database, creates its fact table and applies the same delivery twice. INSERT handles a new identity; ON CONFLICT performs an update only when excluded.version exceeds the stored version. A preliminary check rejects equal-version payload disagreements instead of choosing an arbitrary winner.
Download the ETL and Data Engineering computer lab ↗

Focused decision: use the complete example for context

python

connection.execute(
    "INSERT INTO fact VALUES (:id,:version,:amount_cents) ON CONFLICT(id) "
    "DO UPDATE SET version=excluded.version, amount_cents=excluded.amount_cents "
    "WHERE excluded.version > fact.version", row)
Decision
Add an exact total over stored current facts while keeping replay idempotent and rejecting stale overwrites.
Download the ETL and Data Engineering computer lab ↗

Counterexamples and common mistakes

The counterexample reverses the source-version comparison.

The counterexample reverses the source-version comparison. A stale delivery can now replace a newer stored payload, so a primary key alone has not delivered correct replay behavior. It prevents duplicate identities but does not establish which version should be authoritative.

Another shortcut would use string interpolation to build SQL or commit after every individual row. Parameter binding protects the value boundary, while the transaction controls the batch boundary; neither replaces the other. A production source also needs a policy for version resets, identity reuse and source conflicts before numerical comparison is trustworthy.

Controlled bug: observe the rejected data contract

python

import sqlite3

def inspect_example(rows):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.execute('CREATE TABLE fact(id TEXT PRIMARY KEY, version INTEGER NOT NULL, amount_cents INTEGER NOT NULL)')
        connection.commit()
        for batch in [rows, rows]:
            with connection:
                for row in batch:
                    previous = connection.execute('SELECT version, amount_cents FROM fact WHERE id=?', [row['id']]).fetchone()
                    if previous and previous[0] == row['version'] and previous[1] != row['amount_cents']:
                        raise ValueError('conflicting source version')
                    connection.execute('INSERT INTO fact VALUES (:id,:version,:amount_cents) ON CONFLICT(id) DO UPDATE SET version=excluded.version, amount_cents=excluded.amount_cents WHERE excluded.version < fact.version', row)
        stored = [list(r) for r in connection.execute('SELECT id,version,amount_cents FROM fact ORDER BY id')]
        return {'rows': stored, 'count': len(stored)}
    finally:
        connection.close()
Reasoning
The counterexample reverses the source-version comparison. A stale delivery can now replace a newer stored payload, so a primary key alone has not delivered correct replay behavior. It prevents duplicate identities but does not establish which version should be authoritative.
Download the ETL and Data Engineering computer lab ↗

Independent practice

Add total_cents, calculated from the stored fact table after replay.

Add total_cents, calculated from the stored fact table after replay. Preserve the ordered rows, count, version guard and equal-version conflict rejection. The new total should describe the canonical sink, not the sum of every event delivered before deduplication or the sum of both replay passes.

Use an aggregate over the database after the writes have completed. Handle the empty table explicitly so the report returns integer zero rather than SQL NULL. The independent cases distinguish a real sink total from a source total when several versions refer to the same identity or the entire delivery is repeated.

Step 1: Add an exact total over stored current facts while keeping replay idempotent and rejecting stale overwrites. Step 2: Edit the owned exercise file and run verify.py --practice etl-upsert. Preserve caller inputs and the independently declared failure cases.

Practice starter: solve before reading the solution

python

import sqlite3

def inspect_example(rows):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.execute('CREATE TABLE fact(id TEXT PRIMARY KEY, version INTEGER NOT NULL, amount_cents INTEGER NOT NULL)')
        connection.commit()
        for batch in [rows, rows]:
            with connection:
                for row in batch:
                    previous = connection.execute('SELECT version, amount_cents FROM fact WHERE id=?', [row['id']]).fetchone()
                    if previous and previous[0] == row['version'] and previous[1] != row['amount_cents']:
                        raise ValueError('conflicting source version')
                    connection.execute('INSERT INTO fact VALUES (:id,:version,:amount_cents) ON CONFLICT(id) DO UPDATE SET version=excluded.version, amount_cents=excluded.amount_cents WHERE excluded.version > fact.version', row)
        stored = [list(r) for r in connection.execute('SELECT id,version,amount_cents FROM fact ORDER BY id')]
        return {'rows': stored, 'count': len(stored)}
    finally:
        connection.close()

# Independent task: Add an exact total over stored current facts while keeping replay idempotent and rejecting stale overwrites.
Reasoning
Add an exact total over stored current facts while keeping replay idempotent and rejecting stale overwrites.
Download the ETL and Data Engineering computer lab ↗
  • Hint 1: Add total_cents, calculated from the stored fact table after replay. Preserve the ordered rows, count, version guard and equal-version conflict rejection. The new total should describe the canonical sink, not the sum of every event delivered before deduplication or the sum of both replay passes.
  • Hint 2: Use an aggregate over the database after the writes have completed. Handle the empty table explicitly so the report returns integer zero rather than SQL NULL. The independent cases distinguish a real sink total from a source total when several versions refer to the same identity or the entire delivery is repeated.

Solution, tests and trade-offs

The solution observes the canonical persisted state and returns its integer total alongside the original fields.

The solution observes the canonical persisted state and returns its integer total alongside the original fields. The same source replay still has no additional effect on current rows. The checker closes each case inside a separate directory, keeping this lesson's database independent from every other acceptance fixture.

If the sink also sends email or writes object storage, SQLite cannot atomically commit those external effects. An outbox, idempotency key or reconciliation process may be needed, depending on the boundary. State exactly which effects your guarantee covers instead of extending local transaction correctness into a claim about an entire distributed pipeline.

Practice solution and acceptance checks

python

import sqlite3

def inspect_example(rows):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.execute('CREATE TABLE fact(id TEXT PRIMARY KEY, version INTEGER NOT NULL, amount_cents INTEGER NOT NULL)')
        connection.commit()
        for batch in [rows, rows]:
            with connection:
                for row in batch:
                    previous = connection.execute('SELECT version, amount_cents FROM fact WHERE id=?', [row['id']]).fetchone()
                    if previous and previous[0] == row['version'] and previous[1] != row['amount_cents']:
                        raise ValueError('conflicting source version')
                    connection.execute('INSERT INTO fact VALUES (:id,:version,:amount_cents) ON CONFLICT(id) DO UPDATE SET version=excluded.version, amount_cents=excluded.amount_cents WHERE excluded.version > fact.version', row)
        stored = [list(r) for r in connection.execute('SELECT id,version,amount_cents FROM fact ORDER BY id')]
        return {'rows': stored, 'count': len(stored), 'total_cents': sum(r[2] for r in stored)}
    finally:
        connection.close()
Reasoning
The solution observes the canonical persisted state and returns its integer total alongside the original fields. The same source replay still has no additional effect on current rows. The checker closes each case inside a separate directory, keeping this lesson's database independent from every other acceptance fixture.
Download the ETL and Data Engineering computer lab ↗

Interview reasoning and next step

Explain the roles of the primary key, version guard and conflict check separately.

Explain the roles of the primary key, version guard and conflict check separately. A strong answer should account for a new identity, a newer update, an older replay and an equal-version contradiction. Then describe what additional state is needed to resume source extraction after the database commit.

For a stretch, split one delivery across two owned calls using the same file and reopen the connection between them. Verify that old versions still cannot replace newer ones. The next lesson adds a source checkpoint to the transaction so persisted business state and progress can recover together.

Optional stretch — after the core task works: Reopen the owned database between batches and verify persistent protection from stale updates.

Computer practice: https://applydjinn.com/typorithm/etl-data-engineering-lab-20261107.zip. Use an isolated Python 3.13 environment, install requirements.txt and run verify.py. Edit exercises/etl-upsert.py and run --practice etl-upsert. Each case owns its files and database; Parquet uses the pinned DuckDB engine. The recovery probe terminates a separate interpreter and reopens the SQLite file. The app retains static examples, prepared results and questions without an editor or code execution. Use the course Download button once to save every lesson and illustration; opening an unsaved course does not automatically download it.

Persist source ordering across batches so replay cannot duplicate or roll back current state.