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

Typorithm · medium

Composite cursors and tied source timestamps

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

Composite cursors and tied source timestamps

0%

Concept and contract

Page a fixed ordered source with a strict tuple cursor so equal timestamps are not skipped.

Incremental extraction needs a total ordering that the source can reproduce. A timestamp alone is often insufficient because several records can share it. The fixture uses an integer logical tick plus a stable identity, ordered lexicographically as a pair. The pair, not the tick alone, is the cursor.

This is pagination over a fixed owned snapshot. It does not claim that an arbitrary mutable API behaves like a stable list, nor that textual timestamp ordering is always safe. A real source must define consistent ordering, cursor encoding and snapshot or upper-bound behavior before this extraction pattern can be adopted.

Mechanism and worked example

The program sorts records by tick and id, then selects the next size records whose tuple is strictly greater than the current cursor.

The program sorts records by tick and id, then selects the next size records whose tuple is strictly greater than the current cursor. After a nonempty page, the last record's tuple becomes the next cursor. An empty page terminates the extraction, preserving source order across pages.

The important case places more records at one tick than fit in the first page. The next page must continue through the remaining identities at that same tick before moving forward. Positive exact integer page size is required, and empty input should return no pages rather than inventing an initial record.

Complete worked example

python

def inspect_example(rows, size):
    if type(size) is not int or size <= 0:
        raise ValueError('positive page size required')
    ordered = sorted(rows, key=lambda r: (r['tick'], r['id']))
    cursor = (-1, '')
    pages = []
    while True:
        page = [r for r in ordered if (r['tick'], r['id']) > cursor][:size]
        if not page:
            break
        pages.append([r['id'] for r in page])
        cursor = (page[-1]['tick'], page[-1]['id'])
    return {'pages': pages, 'cursor': list(cursor)}
Reasoning
The program sorts records by tick and id, then selects the next size records whose tuple is strictly greater than the current cursor. After a nonempty page, the last record's tuple becomes the next cursor. An empty page terminates the extraction, preserving source order across pages.
Download the ETL and Data Engineering computer lab ↗

Focused decision: use the complete example for context

python

page = [r for r in ordered if (r['tick'], r['id']) > cursor][:size]
Decision
Count the completely traversed fixed snapshot without losing rows that share the last page timestamp.
Download the ETL and Data Engineering computer lab ↗

Counterexamples and common mistakes

The bug filters only for tick greater than the cursor tick.

The bug filters only for tick greater than the cursor tick. After the first page ends within a tied group, remaining records at that tick disappear. Increasing the page size can hide the defect for one dataset, but it does not repair the missing tie-breaker in the progress contract.

An unstable identity or a changing source order creates a related failure even with tuple comparison. If an API sorts equal timestamps differently on each request, a locally constructed cursor cannot guarantee coverage. Ask for a documented continuation token, snapshot identifier or reliable secondary key, and state how source changes during extraction are handled.

Controlled bug: observe the rejected data contract

python

def inspect_example(rows, size):
    if type(size) is not int or size <= 0:
        raise ValueError('positive page size required')
    ordered = sorted(rows, key=lambda r: (r['tick'], r['id']))
    cursor = (-1, '')
    pages = []
    while True:
        page = [r for r in ordered if r['tick'] > cursor[0]][:size]
        if not page:
            break
        pages.append([r['id'] for r in page])
        cursor = (page[-1]['tick'], page[-1]['id'])
    return {'pages': pages, 'cursor': list(cursor)}
Reasoning
The bug filters only for tick greater than the cursor tick. After the first page ends within a tied group, remaining records at that tick disappear. Increasing the page size can hide the defect for one dataset, but it does not repair the missing tie-breaker in the progress contract.
Download the ETL and Data Engineering computer lab ↗

Independent practice

Add count to the report, measuring how many records were emitted across all pages.

Add count to the report, measuring how many records were emitted across all pages. Keep page grouping, tuple comparison, positive-size validation and the final cursor behavior unchanged. The count should reconcile emitted source records rather than the configured size multiplied by the number of pages.

Use the actual page lengths because the final page can be partial. Compare the tied-tick case, empty input and the one-record case. The independent solution contract rejects a baseline that returns only pages, while the controlled bug remains detectable because its emitted count cannot account for the skipped tied record.

Step 1: Count the completely traversed fixed snapshot without losing rows that share the last page timestamp. Step 2: Edit the owned exercise file and run verify.py --practice etl-watermark. Preserve caller inputs and the independently declared failure cases.

Practice starter: solve before reading the solution

python

def inspect_example(rows, size):
    if type(size) is not int or size <= 0:
        raise ValueError('positive page size required')
    ordered = sorted(rows, key=lambda r: (r['tick'], r['id']))
    cursor = (-1, '')
    pages = []
    while True:
        page = [r for r in ordered if (r['tick'], r['id']) > cursor][:size]
        if not page:
            break
        pages.append([r['id'] for r in page])
        cursor = (page[-1]['tick'], page[-1]['id'])
    return {'pages': pages, 'cursor': list(cursor)}

# Independent task: Count the completely traversed fixed snapshot without losing rows that share the last page timestamp.
Reasoning
Count the completely traversed fixed snapshot without losing rows that share the last page timestamp.
Download the ETL and Data Engineering computer lab ↗
  • Hint 1: Add count to the report, measuring how many records were emitted across all pages. Keep page grouping, tuple comparison, positive-size validation and the final cursor behavior unchanged. The count should reconcile emitted source records rather than the configured size multiplied by the number of pages.
  • Hint 2: Use the actual page lengths because the final page can be partial. Compare the tied-tick case, empty input and the one-record case. The independent solution contract rejects a baseline that returns only pages, while the controlled bug remains detectable because its emitted count cannot account for the skipped tied record.

Solution, tests and trade-offs

The solution adds reconciliation metadata to the same extraction rule.

The solution adds reconciliation metadata to the same extraction rule. A strict tuple comparison provides forward progress for the stated fixed ordering without duplicating the boundary record. Exact page observations still matter: a correct total alone could conceal a duplicated row balanced by a different missing row.

Source progress and business event time may be separate fields. A late business event can arrive with a recent ingestion sequence or an older timestamp. Choose the extraction cursor from the source's delivery contract, then use event time for analytical windows under an explicit lateness policy rather than confusing the two clocks.

Practice solution and acceptance checks

python

def inspect_example(rows, size):
    if type(size) is not int or size <= 0:
        raise ValueError('positive page size required')
    ordered = sorted(rows, key=lambda r: (r['tick'], r['id']))
    cursor = (-1, '')
    pages = []
    while True:
        page = [r for r in ordered if (r['tick'], r['id']) > cursor][:size]
        if not page:
            break
        pages.append([r['id'] for r in page])
        cursor = (page[-1]['tick'], page[-1]['id'])
    return {'pages': pages, 'cursor': list(cursor), 'count': sum(len(p) for p in pages)}
Reasoning
The solution adds reconciliation metadata to the same extraction rule. A strict tuple comparison provides forward progress for the stated fixed ordering without duplicating the boundary record. Exact page observations still matter: a correct total alone could conceal a duplicated row balanced by a different missing row.
Download the ETL and Data Engineering computer lab ↗

Interview reasoning and next step

Explain why the cursor includes an identity and why the comparison is strict.

Explain why the cursor includes an identity and why the comparison is strict. Then identify the assumption that makes this local example reliable: the same immutable source snapshot and a stable total ordering are used for every page. State what changes if rows are inserted behind an already saved cursor.

For a stretch, permute the initial input and retain identical sorted pages, then add a fixture with several full pages at one tick. Keep integer ticks as declared. The following lesson introduces bounded lookback and an upper bound to reason about a specific late-arrival recovery policy.

Optional stretch — after the core task works: Add several pages at the same tick and verify deterministic coverage under shuffled input.

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

Page a fixed ordered source with a strict tuple cursor so equal timestamps are not skipped.