Preparing your workspaceloading the brief, studio & editor…

Loadingone moment…

Commit batches to a sink exactly once · Python scenario · D8LooP

Reading is open to everyone. Running code and saving drafts need an account so your work is yours and comes back on your next visit.

CODE WORKSPACE

Commit batches to a sink exactly once

The sink writes rows and then records the offset it reached. When the process dies between those two steps, the batch is replayed — and because the writer starts from the beginning of the batch, some rows are written twice. The fix is to make the write decision from the committed offsets rather than from the batch. Write commit_batches(batches, committed_offsets). Offsets are inclusive.

Function to write

commit_batches(batches: list[dict], committed_offsets: dict) -> dict

A dict with writes, the updated offsets per partition, and the batches blocked by a gap.

How to approach it

Sort each partition's batches by offset, then carry a frontier forward through them.

Sample cases

+ 2 held back until you submit

a retry after a partial write

One batch is half committed, one continues cleanly, one starts past a gap, and one partition is unknown.

Input

Argument 1

[
  {
    'partition': 0,
    'start_offset': 8,
    'end_offset': 12,
    'rows': [
      'r8',
      'r9',
      'r10',
      'r11',
      'r12'
    ]
  },
  {
    'partition': 0,
    'start_offset': 13,
    'end_offset': 14,
    'rows': [
      'r13',
      'r14'
    ]
  },
  {
    'partition': 0,
    'start_offset': 20,
    'end_offset': 21,
    'rows': [
      'r20',
      'r21'
    ]
  },
  {
    'partition': 1,
    'start_offset': 0,
    'end_offset': 1,
    'rows': [
      's0',
      's1'
    ]
  },
  {
    'partition': 2,
    'start_offset': 5,
    'end_offset': 6,
    'rows': [
      't5',
      't6'
    ]
  }
]

Argument 2

{
  '0': 10,
  '1': -1
}

Returns

{
  'writes': [
    {
      'partition': 1,
      'from_offset': 0,
      'rows': [
        's0',
        's1'
      ]
    }
  ],
  'offsets': {
    '0': -1,
    '1': 1,
    '2': -1
  },
  'blocked': [
    {
      'partition': 0,
      'start_offset': 8,
      'reason': 'gap'
    },
    {
      'partition': 2,
      'start_offset': 5,
      'reason': 'gap'
    }
  ]
}

nothing to commit

No batches means no writes, and the offsets come back exactly as they went in.

Input

Argument 1

[] (empty list)

Argument 2

{
  '0': 10
}

Returns

{
  'writes': [],
  'offsets': {
    '0': 10
  },
  'blocked': []
}

the same batches replayed

Everything is already committed, so a full replay writes nothing at all.

Input

Argument 1

[
  {
    'partition': 0,
    'start_offset': 8,
    'end_offset': 12,
    'rows': [
      'r8',
      'r9',
      'r10',
      'r11',
      'r12'
    ]
  },
  {
    'partition': 0,
    'start_offset': 13,
    'end_offset': 14,
    'rows': [
      'r13',
      'r14'
    ]
  },
  {
    'partition': 0,
    'start_offset': 20,
    'end_offset': 21,
    'rows': [
      'r20',
      'r21'
    ]
  },
  {
    'partition': 1,
    'start_offset': 0,
    'end_offset': 1,
    'rows': [
      's0',
      's1'
    ]
  },
  {
    'partition': 2,
    'start_offset': 5,
    'end_offset': 6,
    'rows': [
      't5',
      't6'
    ]
  }
]

Argument 2

{
  '0': 14,
  '1': 1,
  '2': 6
}

Returns

{
  'writes': [
    {
      'partition': 1,
      'from_offset': 0,
      'rows': [
        's0',
        's1'
      ]
    }
  ],
  'offsets': {
    '0': -1,
    '1': 1,
    '2': -1
  },
  'blocked': [
    {
      'partition': 0,
      'start_offset': 8,
      'reason': 'gap'
    },
    {
      'partition': 2,
      'start_offset': 5,
      'reason': 'gap'
    }
  ]
}

Constraints

1A partition's frontier is its committed offset, or -1 when it has none. Every offset at or below it has already been written.
2Process each partition's batches in ascending offset order, whatever order they arrive in.
3A batch entirely at or below the frontier is skipped, and never moves the frontier backwards.
4A batch that straddles the frontier writes only the rows above it. Row n of a batch is the row at offset start_offset + n.
5A batch starting more than one past the frontier would leave a gap. Do not write it: report it as blocked and stop processing that partition, because everything after it depends on the missing rows.
6A partition with no batches keeps its committed offset unchanged.
7Return writes — each with the partition, the first offset written and the rows — plus the new offsets and the blocked batches.
8A batch with no usable partition or offsets is skipped.

Worked example

Partition 0 is committed to offset 10, and the first batch covers 8 to 12. Rows r8, r9 and r10 are already written; only r11 and r12 are new. Writing the whole batch again duplicates three rows, and an aggregate over the sink is then wrong by exactly the size of whatever was in flight when the process died. The batch covering 20 to 21 is the harder decision. Offsets 15 to 19 are missing from this run, so writing 20 leaves a hole that the offset record will then claim is filled — the next run starts from 21 and those rows are lost forever. Blocking is the right answer even though the rows are sitting right there: an incomplete sink that knows it is incomplete can be repaired, and one that has recorded a lie about its progress cannot.

What this tests

Exactly-once as a property of the write decision rather than of the transport. The two halves — trimming the overlap and refusing the gap — are the same insight seen from both sides: the offset record must never claim more than was written.

PYTHONcommit_batches(batches: list[dict], committed_offsets: dict) -> dict

AI evaluation

The reviewer reads your query.

Submit for review to find out what your query gets right, what it gets wrong, and how it compares with the best working query for this exercise.

Desktop only

This scenario runs a full workspace — editor, canvas and results side by side. It needs a laptop or desktop to be usable. Open this page on a bigger screen to start building.

Read the library instead Back to modules