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

Typorithm · medium

Atomic facts and source checkpoints

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

Atomic facts and source checkpoints

0%

Concept and contract

Commit data and progress together, then reopen the database to verify rollback after failure.

A checkpoint records source progress, not merely that a loop visited some records. Advancing it before durable sink effects can lose data after a restart. Writing facts without recording matching progress can cause replay, which must also be safe. The lesson keeps both effects in one SQLite transaction.

The fixture uses an owned database file with an initial zero checkpoint. A controlled exception occurs after fact statements but before the checkpoint update. Correct rollback should leave both facts and progress unchanged, and a fresh connection must observe that state rather than trusting objects retained by the failing process.

Mechanism and worked example

The baseline configures sqlite3 with autocommit disabled and enters the connection context for the batch.

The baseline configures sqlite3 with autocommit disabled and enters the connection context for the batch. It writes facts, optionally raises the deliberate RuntimeError, then advances the checkpoint only on the successful path. Context exit commits success or rolls back the failed transaction.

After closing the original connection, the program opens another one to read rows and cursor. The failure case expects an empty fact table and checkpoint zero; the success case expects both effects. The empty-batch case remains explicit so progress behavior is not inferred from a conveniently populated happy path.

Complete worked example

python

import sqlite3

def inspect_example(values, fail):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.executescript('CREATE TABLE fact(pos INTEGER PRIMARY KEY, value INTEGER); CREATE TABLE checkpoint(pos INTEGER); INSERT INTO checkpoint VALUES(0);')
        connection.commit()
        try:
            with connection:
                for pos, value in enumerate(values, 1):
                    connection.execute('INSERT INTO fact VALUES(?,?)', [pos, value])
                if fail:
                    raise RuntimeError('controlled failure before checkpoint')
                connection.execute('UPDATE checkpoint SET pos=?', [len(values)])
        except RuntimeError:
            pass
    finally:
        connection.close()
    reopened = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        rows = [list(r) for r in reopened.execute('SELECT pos,value FROM fact ORDER BY pos')]
        return {'rows': rows, 'checkpoint': reopened.execute('SELECT pos FROM checkpoint').fetchone()[0]}
    finally:
        reopened.close()
Reasoning
The baseline configures sqlite3 with autocommit disabled and enters the connection context for the batch. It writes facts, optionally raises the deliberate RuntimeError, then advances the checkpoint only on the successful path. Context exit commits success or rolls back the failed transaction.
Download the ETL and Data Engineering computer lab ↗

Focused decision: use the complete example for context

python

with connection:
    write_facts(connection, batch)
    advance_checkpoint(connection, position)
Decision
Keep facts and checkpoint in the same transaction and report the total read through a fresh connection.
Download the ETL and Data Engineering computer lab ↗

Counterexamples and common mistakes

The mutation calls commit before raising the controlled fault.

The mutation calls commit before raising the controlled fault. The fact writes now survive while the checkpoint remains zero. Catching the exception later cannot undo an already committed transaction, and merely wrapping the remaining statements in another context would not restore the original atomic boundary.

The inverse ordering, checkpoint first and facts later, is more dangerous because restart can skip missing data. A source offset in another system cannot automatically join this local transaction. That situation needs a deliberate protocol and reconciliation; the example's guarantee is limited to tables participating in the same owned SQLite database.

Controlled bug: observe the rejected data contract

python

import sqlite3

def inspect_example(values, fail):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.executescript('CREATE TABLE fact(pos INTEGER PRIMARY KEY, value INTEGER); CREATE TABLE checkpoint(pos INTEGER); INSERT INTO checkpoint VALUES(0);')
        connection.commit()
        try:
            with connection:
                for pos, value in enumerate(values, 1):
                    connection.execute('INSERT INTO fact VALUES(?,?)', [pos, value])
                connection.commit()
                if fail:
                    raise RuntimeError('controlled failure before checkpoint')
                connection.execute('UPDATE checkpoint SET pos=?', [len(values)])
        except RuntimeError:
            pass
    finally:
        connection.close()
    reopened = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        rows = [list(r) for r in reopened.execute('SELECT pos,value FROM fact ORDER BY pos')]
        return {'rows': rows, 'checkpoint': reopened.execute('SELECT pos FROM checkpoint').fetchone()[0]}
    finally:
        reopened.close()
Reasoning
The mutation calls commit before raising the controlled fault. The fact writes now survive while the checkpoint remains zero. Catching the exception later cannot undo an already committed transaction, and merely wrapping the remaining statements in another context would not restore the original atomic boundary.
Download the ETL and Data Engineering computer lab ↗

Independent practice

Add total to the readback report using the newly opened connection.

Add total to the readback report using the newly opened connection. Keep the fault location, transaction boundary and fact/checkpoint observations unchanged. The failed transaction should still report zero, while the successful path reports the sum of committed canonical facts. Do not calculate the result from the original input alone.

Find the reopened connection used for the verification query. Add a database aggregate there and handle an empty table as integer zero. The failure case proves the aggregate measures committed state rather than the amount temporarily visible inside a transaction that was later rolled back. Preserve unchanged caller arguments as well.

Step 1: Keep facts and checkpoint in the same transaction and report the total read through a fresh connection. Step 2: Edit the owned exercise file and run verify.py --practice etl-checkpoint. Preserve caller inputs and the independently declared failure cases.

Practice starter: solve before reading the solution

python

import sqlite3

def inspect_example(values, fail):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.executescript('CREATE TABLE fact(pos INTEGER PRIMARY KEY, value INTEGER); CREATE TABLE checkpoint(pos INTEGER); INSERT INTO checkpoint VALUES(0);')
        connection.commit()
        try:
            with connection:
                for pos, value in enumerate(values, 1):
                    connection.execute('INSERT INTO fact VALUES(?,?)', [pos, value])
                if fail:
                    raise RuntimeError('controlled failure before checkpoint')
                connection.execute('UPDATE checkpoint SET pos=?', [len(values)])
        except RuntimeError:
            pass
    finally:
        connection.close()
    reopened = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        rows = [list(r) for r in reopened.execute('SELECT pos,value FROM fact ORDER BY pos')]
        return {'rows': rows, 'checkpoint': reopened.execute('SELECT pos FROM checkpoint').fetchone()[0]}
    finally:
        reopened.close()

# Independent task: Keep facts and checkpoint in the same transaction and report the total read through a fresh connection.
Reasoning
Keep facts and checkpoint in the same transaction and report the total read through a fresh connection.
Download the ETL and Data Engineering computer lab ↗
  • Hint 1: Add total to the readback report using the newly opened connection. Keep the fault location, transaction boundary and fact/checkpoint observations unchanged. The failed transaction should still report zero, while the successful path reports the sum of committed canonical facts. Do not calculate the result from the original input alone.
  • Hint 2: Find the reopened connection used for the verification query. Add a database aggregate there and handle an empty table as integer zero. The failure case proves the aggregate measures committed state rather than the amount temporarily visible inside a transaction that was later rolled back. Preserve unchanged caller arguments as well.

Solution, tests and trade-offs

The solution extends post-recovery observation without weakening atomicity.

The solution extends post-recovery observation without weakening atomicity. Acceptance compares the fresh connection's rows, checkpoint and total, so a premature commit remains visible. This is a failure probe with an actual database file, not an in-memory simulation of what a rollback is expected to do.

A controlled Python exception still differs from sudden process death and from a machine losing power. The capstone adds a separate process-exit probe. Filesystem and hardware durability depend on configuration and environment, so describe the evidence precisely instead of claiming that one local exception test proves every possible failure mode.

Practice solution and acceptance checks

python

import sqlite3

def inspect_example(values, fail):
    connection = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        connection.executescript('CREATE TABLE fact(pos INTEGER PRIMARY KEY, value INTEGER); CREATE TABLE checkpoint(pos INTEGER); INSERT INTO checkpoint VALUES(0);')
        connection.commit()
        try:
            with connection:
                for pos, value in enumerate(values, 1):
                    connection.execute('INSERT INTO fact VALUES(?,?)', [pos, value])
                if fail:
                    raise RuntimeError('controlled failure before checkpoint')
                connection.execute('UPDATE checkpoint SET pos=?', [len(values)])
        except RuntimeError:
            pass
    finally:
        connection.close()
    reopened = sqlite3.connect('warehouse.db', autocommit=False)
    try:
        rows = [list(r) for r in reopened.execute('SELECT pos,value FROM fact ORDER BY pos')]
        return {'rows': rows, 'checkpoint': reopened.execute('SELECT pos FROM checkpoint').fetchone()[0], 'total': sum(r[1] for r in rows)}
    finally:
        reopened.close()
Reasoning
The solution extends post-recovery observation without weakening atomicity. Acceptance compares the fresh connection's rows, checkpoint and total, so a premature commit remains visible. This is a failure probe with an actual database file, not an in-memory simulation of what a rollback is expected to do.
Download the ETL and Data Engineering computer lab ↗

Interview reasoning and next step

Explain where the transaction begins and ends, which tables it covers and what a restart should do after each failure point.

Explain where the transaction begins and ends, which tables it covers and what a restart should do after each failure point. Distinguish consumed source position, attempted work and committed checkpoint. They may coincide on a successful run but have different meanings during recovery.

For a stretch, inject faults before writes, after writes and after the successful commit, recording the reopened state each time. Keep those failure positions explicit and expected. A robust review should reason from committed effects and a replay policy rather than relying on a broad catch-and-continue block.

Optional stretch — after the core task works: Inject faults at three named boundaries and compare fresh-connection state after each one.

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-checkpoint.py and run --practice etl-checkpoint. 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.

Commit data and progress together, then reopen the database to verify rollback after failure.