You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Ray Data's current checkpointing is row-identity based: the user must designate a globally unique ID column, the IDs of every successfully written row are persisted, and on restart the union of all such IDs is loaded and handed to every read task, which filters its input against the set.
This brings the following limitations:
The mandatory ID column limits use cases.id_column must be unique across the entire dataset and survive every operator. Many pipelines have no such column and cannot cheaply synthesize one.
Resume still re-reads all input. There is no source offset anywhere in the checkpoint, so a job that is 99% complete still opens and scans 100% of its input files, only to discard the rows afterwards.
Goal: design an ID-free checkpointing mechanism that expresses progress as offsets in the source rather than as a set of processed rows, assuming the source supports deterministic split enumeration.
Design principles
No in-band barriers: nothing is injected into the data stream and no operator input is ever aligned or blocked. The epoch boundary exists only as a stamp in block metadata plus a row-count ledger on the driver side.
Reuse existing channels: the epoch stamp rides on BlockMetadataWithSchema, which already flows from every worker back to the driver; row counts are collected in the existing task-completion callback. No new data path is introduced.
No user-visible contract: no ID column, no schema requirement, no change to user code.
Proposed approach: implicit barriers via row-count reconciliation
Introduce a Coordinator actor on the head node. It owns a ledger keyed by (operator, split, epoch) and is the only component that decides whether an epoch has been fully processed by the entire DAG. Once an epoch is completed, the source read offsets of that epoch are persisted; on restart each split resumes from its recorded offset.
The main steps can be described as followed:
1. Stamping. The source tags every block it produces with (split, epoch). Downstream operators inherit the stamp from their input blocks. The stamp is metadata only; it costs no extra data movement.
2. Per-task reporting. When a task finishes, the driver reports, for the (split, epoch) of its input, how many rows the task consumed (in_rows) and how many it produced (out_rows). Both numbers travel in the same message.
3. Sealing an epoch. When a checkpoint is triggered, each source tells the Coordinator how much data it read in that epoch: (split s, epoch e, offset f, rows n). This single message serves two purposes — it is both the offset to be persisted and the expected row count that seeds the reconciliation chain.
An operator is considered to have consumed everything of that epoch when in_rows == expected_in. At that instant out_rows is final, and it becomes the downstream operator's expected_in for the same (split, epoch). expected_in therefore propagates from the source down the DAG, one stage at a time.
5. Commit. When every operator has reconciled for every split of epoch e, the epoch is complete. The Coordinator atomically persists {(split, offset)} for that epoch to external storage and advances its watermark. If reconciliation does not complete, the epoch is discarded and its counts are merged into the next epoch.
6. Restore. Load the most recent committed epoch and resume each split from its recorded offset. Input before the offset is never opened, so restore cost is independent of how much work the previous run had completed.
Assumptions
The source must provide deterministic intra-split ordering, so that a recorded offset identifies the same resume point across runs. Split enumeration itself need not be deterministic — the split assignment is persisted as part of the checkpoint and restored verbatim.
Example: the Coordinator ledger
A snapshot of the ledger mid-execution. The DAG is src → map → sink with three source splits. The last committed epoch is 1, checkpoint 3 is in flight, and checkpoint 2 failed to reconcile in time and was merged into 3.
Two columns extend the value tuple of step 4: carry holds counts inherited from an aborted epoch, and final marks a source split that has been read to exhaustion.
Columns.expected is the row count the upstream is known to have produced for this unit; NIL means it is not known yet, which is different from 0 — the former must wait, the latter settles immediately. offset_end and final apply to source splits only: offset_end is the read position at the end of the epoch. out_rows is not tracked for the terminal operator. — means not applicable.
Source is the only stage judged on its output, because it has no input. Its expected is the only externally supplied number in the whole chain, which also makes it the single point where a self-reported count is cross-checked against a driver-measured one.
Reading epoch 3.("src", 0, 3) reconciles at 1000 == 600 + 400: 600 rows were read in epoch 3, plus 400 inherited from the aborted epoch 2. Its final out_rows of 1000 becomes expected for ("map", 0, 3), whose own out_rows of 700 in turn becomes expected for ("sink", 0, 3) — this is how the expected value propagates down the DAG one stage at a time.
Epoch 3 cannot commit: ("sink", 1, 3) has consumed 300 of the 420 rows the map stage produced. The remaining 120 rows are still in flight, so the epoch stays open until they land.
Reading epoch 4. No checkpoint has been triggered for epoch 4 yet, so most source units have no expected value. Splits 1 and 2 are exceptions: reaching the end of a split also fixes expected and records offset_end, independently of any checkpoint trigger. This is why ("src", 1, 4) can already settle while ("src", 0, 4) still shows NIL despite having produced 180 rows — counts routinely arrive before the number they will be compared against.
("src", 2, 4) shows that final and settled are orthogonal: the split has been read to exhaustion and its total is known to be 260, but only 210 rows have been accounted for so far. Knowing the denominator does not mean the numerator has caught up.
("map", 0, 4) is waiting on ("src", 0, 4): an operator's expected cannot be assigned until its upstream has settled.
Problem
Ray Data's current checkpointing is row-identity based: the user must designate a globally unique ID column, the IDs of every successfully written row are persisted, and on restart the union of all such IDs is loaded and handed to every read task, which filters its input against the set.
This brings the following limitations:
id_columnmust be unique across the entire dataset and survive every operator. Many pipelines have no such column and cannot cheaply synthesize one.Goal: design an ID-free checkpointing mechanism that expresses progress as offsets in the source rather than as a set of processed rows, assuming the source supports deterministic split enumeration.
Design principles
BlockMetadataWithSchema, which already flows from every worker back to the driver; row counts are collected in the existing task-completion callback. No new data path is introduced.Proposed approach: implicit barriers via row-count reconciliation
Introduce a Coordinator actor on the head node. It owns a ledger keyed by
(operator, split, epoch)and is the only component that decides whether an epoch has been fully processed by the entire DAG. Once an epoch is completed, the source read offsets of that epoch are persisted; on restart each split resumes from its recorded offset.The main steps can be described as followed:
1. Stamping. The source tags every block it produces with
(split, epoch). Downstream operators inherit the stamp from their input blocks. The stamp is metadata only; it costs no extra data movement.2. Per-task reporting. When a task finishes, the driver reports, for the
(split, epoch)of its input, how many rows the task consumed (in_rows) and how many it produced (out_rows). Both numbers travel in the same message.3. Sealing an epoch. When a checkpoint is triggered, each source tells the Coordinator how much data it read in that epoch:
(split s, epoch e, offset f, rows n). This single message serves two purposes — it is both the offset to be persisted and the expected row count that seeds the reconciliation chain.4. Reconciliation. The Coordinator maintains:
An operator is considered to have consumed everything of that epoch when
in_rows == expected_in. At that instantout_rowsis final, and it becomes the downstream operator'sexpected_infor the same(split, epoch).expected_intherefore propagates from the source down the DAG, one stage at a time.5. Commit. When every operator has reconciled for every split of epoch
e, the epoch is complete. The Coordinator atomically persists{(split, offset)}for that epoch to external storage and advances its watermark. If reconciliation does not complete, the epoch is discarded and its counts are merged into the next epoch.6. Restore. Load the most recent committed epoch and resume each split from its recorded offset. Input before the offset is never opened, so restore cost is independent of how much work the previous run had completed.
Assumptions
The source must provide deterministic intra-split ordering, so that a recorded offset identifies the same resume point across runs. Split enumeration itself need not be deterministic — the split assignment is persisted as part of the checkpoint and restored verbatim.
Example: the Coordinator ledger
A snapshot of the ledger mid-execution. The DAG is
src → map → sinkwith three source splits. The last committed epoch is 1, checkpoint 3 is in flight, and checkpoint 2 failed to reconcile in time and was merged into 3.Two columns extend the value tuple of step 4:
carryholds counts inherited from an aborted epoch, andfinalmarks a source split that has been read to exhaustion.Columns.
expectedis the row count the upstream is known to have produced for this unit;NILmeans it is not known yet, which is different from0— the former must wait, the latter settles immediately.offset_endandfinalapply to source splits only:offset_endis the read position at the end of the epoch.out_rowsis not tracked for the terminal operator.—means not applicable.Settling. A unit settles when
Source is the only stage judged on its output, because it has no input. Its
expectedis the only externally supplied number in the whole chain, which also makes it the single point where a self-reported count is cross-checked against a driver-measured one.Reading epoch 3.
("src", 0, 3)reconciles at1000 == 600 + 400: 600 rows were read in epoch 3, plus 400 inherited from the aborted epoch 2. Its finalout_rowsof 1000 becomesexpectedfor("map", 0, 3), whose ownout_rowsof 700 in turn becomesexpectedfor("sink", 0, 3)— this is how the expected value propagates down the DAG one stage at a time.Epoch 3 cannot commit:
("sink", 1, 3)has consumed 300 of the 420 rows the map stage produced. The remaining 120 rows are still in flight, so the epoch stays open until they land.Reading epoch 4. No checkpoint has been triggered for epoch 4 yet, so most source units have no
expectedvalue. Splits 1 and 2 are exceptions: reaching the end of a split also fixesexpectedand recordsoffset_end, independently of any checkpoint trigger. This is why("src", 1, 4)can already settle while("src", 0, 4)still showsNILdespite having produced 180 rows — counts routinely arrive before the number they will be compared against.("src", 2, 4)shows thatfinalandsettledare orthogonal: the split has been read to exhaustion and its total is known to be 260, but only 210 rows have been accounted for so far. Knowing the denominator does not mean the numerator has caught up.("map", 0, 4)is waiting on("src", 0, 4): an operator'sexpectedcannot be assigned until its upstream has settled.