Checkpoint V2 — Design Proposal

[rohitkulshreshtha]

Checkpoint V2 — Design Proposal (Revised)

*Revised 2026-04-20 against the implementation in PR #6725.

Changelog since the original proposal

Substantive changes to the design, in priority of user-visible impact:

• Pushdown guards on PushDownFilter / PushDownProjection. Filters and projections cannot push past a checkpointed source — trades scan-time cost for the "filtered-out keys still get checkpointed" contract.
• Sink-less pipelines supported via CheckpointTerminusNode. Plans ending in .collect() or pure-side-effect UDFs (LLMs, Kafka, HTTP) now drive store.checkpoint() at task end.
• Checkpointed-key files are Parquet, not Arrow IPC. Reuses Daft's Parquet scanner; potentially opens the door to bloom-filter pruning as future leverage.
• Sink categorization is two-bucket, not a compatibility matrix. Framework-atomic sinks (Iceberg/Delta/Parquet/Lance-native) vs externally-visible side-effect endpoints (write_sql/Lance-REST/UDFs-with-external-effects). User responsibility for idempotency in the second bucket.
• concat / union_all rejected at optimizer time. Closes the "multiple sources" open question as deferred — awaiting concrete use cases.
• Per-input CheckpointId via CheckpointIdMap. Correction to "one UUID per task" — under Flotilla's shared compiled pipeline, a per-input map is required.
• Limit / Offset / Sample added to the reject list. Expands "map-only" from "no shuffles" to "no shuffles and no row-selectors." Each has a worked example of how it breaks semantics.

Section-level additions: Pipeline Operators, Sink Compatibility, Commit Idempotency, Sink-less Pipelines. Section-level updates: Assumptions, Glossary, CheckpointStore (format), Crash Recovery (idempotency row), What This Handles, What This Doesn't Handle, Open Questions.

---

The Problem

Pipelines with expensive operations — UDFs, FFMPEG, embeddings, API calls — shouldn't redo completed work on failure or re-run. We need a way to track what's been processed and skip it next time.

Additionally, 2PC sinks (Iceberg, Delta Lake, Lance) have a gap between writing data files and committing them to the catalog. A crash in that gap means orphaned files and lost work. We need a way to recover from this without re-running the entire pipeline.

---

Summary

Track progress at the source, not the sink. The user associates a checkpoint store with a source and specifies a key column (e.g., file_id). On each run, rows with keys already in the checkpoint store are skipped. The pipeline between source and sink is a black box — we don't care what happens inside. When a task finishes (including writing to the sink), its source keys are checkpointed.

Track what was produced alongside what was processed. The checkpoint store records both the source keys (what went in) and the output file metadata (what came out). This means:

• On re-run, we know what to skip (keys) and what to commit (files)
• On crash recovery, we can roll forward — commit the files that were written but never made it to the catalog
• Progress tracking and transaction management collapse into one abstraction

The checkpoint store is the source of truth. It tracks both keys and files, so the system can always recover — either by skipping already-processed keys or by committing files that were written but never made it to the catalog. No data loss. No silent corruption — for sinks where Daft controls the commit boundary (see Sink Compatibility).

Explodes are handled, with a known trade-off. Progress is tracked at the source level (e.g., per input file), not per exploded row. If a crash happens mid-explode, the entire input file is re-processed on the next run — but fully completed files are skipped. A second-level checkpoint filter after the explode (tracking at per-row granularity) could be added in a similar fashion if customers need it.

Assumptions

1. Map-only pipelines. This design relies on Flotilla's task model where one task = one input partition = the full fused pipeline. Pipelines with shuffles (joins, aggregates, sorts, distinct) and row-selector operators (Limit, Offset, Sample) that use a checkpoint store fail at plan construction time. See Pipeline Operators.
2. Source data has a checkpoint key. The user specifies a column that uniquely identifies source inputs. This key should be lightweight relative to the data itself — it gets written to the checkpoint store on every run. Key types supported today: string. Int, binary, and composite keys are planned follow-ups.
3. Checkpoint store is durable, available, and strongly consistent. Read-after-write consistency is required — after checkpoint(), the head node must be able to see the checkpointed keys and files.
4. Filter / projection pushdown is suppressed past a checkpointed source. Trade-off for the "filtered-out keys still get checkpointed" guarantee — see Pushdown Guards.

---

Glossary

Term	Definition
Flotilla	Daft's distributed execution engine
Head node	Flotilla's coordinator process that builds plans, dispatches tasks to workers, and performs catalog commits
Worker	A Ray actor that executes tasks — runs the full fused pipeline for one input partition
StageCheckpointKeysOperator	Operator that extracts the checkpoint key column from each morsel and stages it into the checkpoint store. Inserted by RewriteCheckpointSource above the anti-join and below user ops so filtered-out rows still stage
RewriteCheckpointSource	Optimizer rule that rewrites Source{checkpoint=Some(_)} into the anti-join + StageCheckpointKeysOperator subtree
CheckpointTerminusNode	Terminal wrapper injected at plan root when a checkpoint source is present but no native write sink is downstream. Drives store.checkpoint() at task completion
CheckpointIdMap	Lazy per-InputId → CheckpointId map, shared between StageCheckpointKeysOperator and the blocking sink via BuilderContext. Replaces the "one UUID per task" single-cell design
BlobStoreCheckpointedKeysScanOperator	Scan operator that reads checkpointed-key Parquet files. Feeds the right side of the anti-join
KFJ / KeyFilteringJoin	Distributed anti-join feature (see PR #5931) that the rewrite rule reuses
2PC sink	A sink where data files are written first, then committed to a catalog in a separate step (e.g., Iceberg, Delta Lake, Lance)
Framework-atomic sink	A sink where Daft controls the commit boundary — framework can offer exactly-once output. See Sink Compatibility
Side-effect endpoint	A write target where data leaves Daft's control mid-pipeline (DB inserts, HTTP POSTs, Kafka publishes). Exactly-once output is the user's or the external system's responsibility

---

The Proposal: Source-Oriented Checkpointing

A checkpoint store is associated with a source. It tracks which source inputs have been fully processed. On re-run, the source skips them.

checkpoint = daft.CheckpointStore("s3://checkpoints/my-job")

df = daft.read_parquet("s3://input/", checkpoint=checkpoint, on="file_id")
df = df.explode("pages")
df = df.with_column("embedding", embed_udf)   # expensive
df = df.filter(quality_check)
df.write_iceberg(catalog, table="output")

The on parameter specifies the checkpoint key — the column that uniquely identifies a source input. This must be a column present in the source data (e.g., a document ID or file hash).

On re-run, the RewriteCheckpointSource optimizer rule rewrites the source node into:

Sink
  └─ [user ops: Filter / Project / UDF / Explode / Unpivot ...]
     └─ StageCheckpointKeys             (stages keys BEFORE user ops see them)
        └─ Join(Anti, KeyFiltering) {
             left:  Source { checkpoint cleared }
             right: Source(BlobStoreCheckpointedKeysScanOperator)
           }

StageCheckpointKeysOperator sits below user ops and above the anti-join. Every row that emerges from the anti-join (i.e., every not-yet-checkpointed source row) has its key staged before user operators run. This is what makes "filtered-out keys still get checkpointed" a hard guarantee — a user filter downstream cannot drop keys that haven't already been staged.

Pushdown guards

Daft's optimizer normally pushes filters and projections into the Source node. For checkpointed sources, that would let rows drop at the Parquet scan before StageCheckpointKeysOperator sees them, breaking the guarantee above. The RewriteCheckpointSource pass runs early; pushdown rules now check for checkpoint.is_some():

• PushDownFilter unconditionally skips checkpointed sources.
• PushDownProjection is narrower — it only skips when pushing the projection would prune the on= key column. Non-key columns still get pruned normally.

Trade-off (user-facing). Opting into checkpointing means the source materializes every row and retains the key column up to StageCheckpointKeysOperator, even if a downstream filter will drop most of them. Checkpointing ⇒ reduced pushdown, paid at scan time. This is the expected cost of the "filtered keys still get checkpointed" contract.

Why this trade is OK (current assumption). Checkpointing is aimed at workloads where the per-row processing cost (the expensive UDF, the API call, the embedding generation) dominates the per-row reading cost. Wasting a scan on rows that a downstream filter will drop is cheap relative to re-running the expensive operation on those rows on every rerun. If/when we encounter workloads where this assumption breaks — cheap processing, costly scans — the trade-off needs revisiting.

Per-input CheckpointId

Under Flotilla, the pipeline translator compiles the operator set once per query, and every task that runs the pipeline reuses those same operator instances. A single pre-allocated CheckpointId would cause the second task to fail because the checkpoint was already finalized by the first. CheckpointIdMap is a lazy per-InputId map held on BuilderContext; StageCheckpointKeysOperator and the blocking sink both read it via the shared context so each task lands under its own CheckpointId.

---

Why Source-Oriented?

The alternative is sink-oriented: query the sink to find what's already been written, anti-join against it. This breaks in several important cases.

The filter problem. An expensive filter (e.g., a UDF that classifies rows and drops the ones that don't qualify) runs on every input row. Rows that get filtered out never reach the sink. On resume, the sink has no record of them — so they get re-processed every single run. The expensive classification runs again, reaches the same conclusion (drop), and the work is wasted. With source-oriented tracking, the task completed for that input, so it's marked done regardless of whether any rows survived to the sink.

The explode problem. A PDF gets exploded into 1000 pages. 200 are written to the sink before a crash. On resume, the sink sees rows with file_id=A in the output and concludes "file A has been processed" — but only 200 of 1000 pages were written. The remaining 800 are silently lost. You can't determine from the sink alone whether an input file was fully processed, only that some output exists for it. Source-oriented tracking doesn't have this problem — the file is only marked done when the task completes, meaning all pages made it through.

Non-queryable sinks. Some sinks have no public API for reading uncommitted data. Plain file formats (Parquet, CSV) have no key semantics — you'd have to scan everything. Source-oriented tracking is independent of the sink — it works with any sink.

Eventually consistent sinks. Many sinks — particularly vector databases — are eventually consistent by design. A write may return success before the data is indexed or queryable. Sink-oriented tracking can't safely read its own writes in these cases. Source-oriented tracking avoids this entirely.

---

Pipeline Operators

The checkpoint rule validates the subtree between source and sink at optimizer time. Two categories of operators are rejected:

Rejected: shuffles (original design)

Sort, TopN, Repartition, Shuffle, Distinct, Aggregate, Pivot, Window, Join, AsofJoin, Concat, Union, Intersect. These reorder or redistribute rows in ways that break the one-task-per-input-partition fusion the design relies on.

Rejected: row-selectors (added during implementation)

Limit, Offset, Sample. These aren't shuffles but still break checkpoint semantics in non-obvious ways:

• Limit picks rows from the anti-join output, not the original source. On a source that grows between runs, source.limit(2) returns [a, b] on run 1 and [e, f] on run 2 — the meaning of "first 2 rows" silently drifts.
• Offset has the dual problem. source.offset(2) intends to drop the first 2 rows. After run 1 checkpoints rows 3+, run 2's anti-join returns just rows 1-2; offset drops them. The rows the user said "never process" are now checkpointed as if processed — the user's intent is violated.
• Sample is non-deterministic across runs. Checkpointed keys shrink the sample universe each run, so sample(0.3) draws from a dwindling set — silently regressing coverage until it returns empty.

Allowed

Filter, Project / with_column, UDF (map), Explode, Unpivot (melt), IntoBatches. These are map-only at the physical level: row-in → 0..N rows-out, no cross-partition dependencies.

---

CheckpointStore

The CheckpointStore is the core abstraction — a durable store that tracks both what was processed (source keys) and what was produced (output files), enabling both progress tracking and crash recovery from a single interface.

erDiagram
    CheckpointStore ||--o{ Checkpoint : contains
    Checkpoint ||--o{ SourceKey : "staged keys"
    Checkpoint ||--o{ FileMetadata : "staged files"
    Checkpoint {
        CheckpointId id
        QueryId query_id
        Status status "checkpointed | committed"
    }
    SourceKey {
        string value
    }
    FileMetadata {
        string file_path
    }

Loading

Why track both keys and files?

A simple progress store could just track source keys — mark them done at task completion. This works for non-2PC sinks (plain Parquet, CSV) where the sink write is final.

But for 2PC sinks (Iceberg, Delta Lake, Lance), "task completes" ≠ "data committed." A task writes data files to storage, but the catalog commit happens later on the head node. If we only track keys and the head node crashes before commit, the keys say "done" but the files are orphaned — data loss.

By tracking both keys and files together:

1. The checkpoint filter skips existing keys — rows with already-checkpointed keys are dropped during pipeline execution, even if the files haven't been committed to the catalog yet
2. The head node knows exactly what to commit — it reads the files associated with those keys from the checkpoint store

On recovery: read the checkpoint store → skip the keys → commit the files → done.

This merges the progress store and the transaction manager into one abstraction. One store, both concerns resolved. A useful side-effect: the same mechanism gives plain Parquet/CSV a 2PC-like capability (orphan cleanup by reconciling on-disk files against the checkpoint store's authoritative file list — follow-up work).

Checkpointed-key file format: Parquet

Checkpointed source keys are persisted as Parquet files under <store>/keys/*.parquet. The right side of the anti-join is literally a read_parquet scan over this path — reusing Daft's first-class Parquet scanner, same code path, same perf, same operational surface. An Arrow IPC variant would have required maintaining a parallel reader.

Interface

#[async_trait]
pub trait CheckpointStore: Send + Sync {
    // Worker: stage source keys into a checkpoint. Keys are a columnar
    // `Series` — any Arrow-compatible type (Utf8, Int64, Struct for composite
    // keys). Append semantics; not idempotent.
    async fn stage_keys(&self, id: &CheckpointId, keys: Series) -> CheckpointResult<()>;

    // Worker: stage output file metadata. Append semantics; not idempotent.
    async fn stage_files(&self, id: &CheckpointId, files: Vec<FileMetadata>) -> CheckpointResult<()>;

    // Worker: finalize the checkpoint — keys and files become coupled and
    // visible. Idempotent (no-op if already finalized).
    async fn checkpoint(&self, id: &CheckpointId) -> CheckpointResult<()>;

    // Head node: stream all checkpointed keys as columnar chunks.
    async fn get_checkpointed_keys(&self)
        -> CheckpointResult<BoxStream<'_, CheckpointResult<Series>>>;

    // Head node: stream checkpointed-but-not-yet-committed file metadata.
    async fn get_checkpointed_files(&self)
        -> CheckpointResult<BoxStream<'_, CheckpointResult<FileMetadata>>>;

    // Head node: single-checkpoint lookup and full listing.
    async fn get_checkpoint(&self, id: &CheckpointId) -> CheckpointResult<Checkpoint>;
    async fn list_checkpoints(&self)
        -> CheckpointResult<BoxStream<'_, CheckpointResult<Checkpoint>>>;

    // Head node: mark checkpoints as committed. Keys stay visible for
    // skip-on-rerun; files drop out of `get_checkpointed_files`. Idempotent.
    async fn mark_committed(&self, ids: &[CheckpointId]) -> CheckpointResult<()>;
}

Lifecycle

checkpointed → committed

• Staged (invisible): Workers write keys and files independently during task execution. Not visible to readers. Keys are staged early (StageCheckpointKeysOperator, before user ops). Files are staged late (after writer finalize). These are separate calls because the key column may not even survive the pipeline — projections, filters, or transforms may strip it before the sink.
• Checkpointed: checkpoint() called at task completion. Keys and files are coupled and visible. The head node reads from this state.
• Committed: mark_committed() called after catalog commit. Indicates that downstream data is durable in the external catalog (Iceberg snapshot, Delta log, etc.).

One CheckpointId per task, resolved per-InputId at runtime via CheckpointIdMap.

---

Sink Compatibility

Sinks fall into two buckets. Category 1 sinks get exactly-once output from the framework. Category 2 sinks require the user or the external system to handle idempotency.

Category 1 — Framework-atomic

Daft controls the commit boundary. The framework can offer exactly-once output on rerun-after-crash.

• Iceberg / Delta / Paimon — native 2PC (write files, then commit to catalog). Exactly-once requires idempotent commits — see Commit Idempotency.
• Parquet / CSV — via the checkpoint-store-as-2PC mechanism. The store knows which files belong to committed checkpoints, so orphaned files from failed tasks can be reconciled against the authoritative list. Orphan cleanup is follow-up work; in the current implementation, failed-task Parquet files remain on disk until manually reconciled.
• Lance (native 2PC) — atomic-per-task writes. Untested against the checkpoint store but structurally compatible.

Category 2 — Externally-visible side effects

The write leaves Daft's control mid-pipeline. Daft cannot unwind the effect, so duplicate writes on rerun will produce duplicate external state unless the user or the downstream system handles it.

• write_sql (append mode, no PK) — each micropartition commits independently; no task-level rollback. Mid-task crash → partial rows committed → rerun → duplicates.
• write_sql (upsert target, keyed writes) — safe. The user's write pattern handles idempotency.
• Lance REST — same as write_sql (per-micropartition commits, no staging).
• UDFs with external side effects — API calls, Kafka publishes, HTTP POSTs. Rerun of a failed task re-invokes the UDF on rows that already left the boundary. CheckpointTerminusNode exists specifically to drive store.checkpoint() for these pipelines, but it cannot undo the external effect — it just bounds rerun to whole-task granularity.

What checkpointing offers in Category 2

Even when exactly-once output isn't achievable, checkpointing still provides "don't redo expensive work at task granularity." Checkpointed keys from past completed tasks are always skipped on rerun. A crash within a task does cause that task's work to be redone, but completed-and-checkpointed tasks stay checkpointed. For most workloads this is the valuable guarantee — the exactly-once output contract is a bonus available when the sink (or the user) cooperates.

How to get exactly-once output in Category 2

Pick any of:

• Use a framework-atomic sink from Category 1.
• Use an idempotent write pattern (upsert on PK, INSERT ... ON CONFLICT UPDATE, MERGE INTO, keyed overwrite).
• Let the downstream system dedup (Kafka keyed topics, vector DB with PK, logs reconciled by row ID).
• Accept whole-task-level duplicates as a workload-appropriate cost (rare, but valid for some pipelines).

---

Commit Idempotency

For Category 1 sinks with native catalogs (Iceberg, Delta, Paimon), "data committed" ≠ "mark_committed() returned." A head-node crash between the catalog commit and mark_committed() means the next run sees the same files in get_checkpointed_files() and re-commits them. Without idempotency at the catalog layer, this creates duplicate snapshots (Iceberg/Paimon) or duplicate rows (Delta without txn).

Daft's responsibility here:

• Iceberg / Paimon — tag each snapshot with daft.checkpoint-store (store path) + daft.epoch-id (monotonic counter per mark_committed). On recovery, walk snapshot history; if the tag already exists, skip the re-commit.
• Delta Lake — use a txn action with {appId: store_path, version: epoch_id}. Delta's transaction protocol dedups at the log level.
• Parquet / CSV — orphan cleanup using the checkpoint store's file list as the source of truth. File-system operations are idempotent.

Implementation follow-up.

---

Sink-less Pipelines

Not every pipeline ends in a Daft-native write sink. Pipelines that end in .collect() after a UDF with external side effects (LLMs, embeddings APIs, Kafka publishes) have no place to attach the checkpoint hook on the write arm of the translator. In the original design these would silently no-op — rows anti-join correctly, but nothing ever gets checkpointed, so every re-run redoes everything. That's the worst possible failure mode.

CheckpointTerminusNode

A terminal wrapper injected at the plan root when a checkpoint source is present with no native write sink downstream. It drives store.checkpoint() on task completion at the same per-task checkpoint granularity as the sink case.

Trade-off. Checkpoint granularity is per-task. A task that processes 5 rows, writes to an external system for each, and crashes on row 6 will have all 6 source keys not checkpointed on rerun — and the external system may have already received rows 1-5. This is consistent with Category 2 semantics: the framework can't unwind external effects. Users who want row-level recovery against external systems need to either make the external writes idempotent or structure the pipeline differently (e.g., write to Parquet first, then drive external writes in a second idempotent pass).

---

Crash Recovery

Crash point	State	Recovery
Before worker stages anything	Nothing persisted	Redo task
After staging, before checkpoint()	Keys + files exist but invisible	Redo task — staged data is orphaned, cleaned up
After checkpoint(), before worker reports to head node	Checkpoint visible, head node unaware	Head node re-dispatches task; anti-join skips already-checkpointed keys — no duplicate work
After all tasks reported, before sink commit	Checkpoints visible with file list	Read checkpoint store → commit files to sink
After sink commit, before mark_committed()	Data committed, checkpoint not marked	Depends on sink: Category 1 with idempotent commits (Iceberg tag, Delta txn, Parquet orphan cleanup) → safe. Category 1 without those mechanisms → catalog duplication on rerun. Category 2 → depends on user's write pattern (see Sink Compatibility)
After mark_committed()	Done	Nothing to do

At every crash point for Category 1 sinks with commit idempotency configured, the system either redoes work (correct) or rolls forward (correct). Category 2 sinks require user-side or system-side idempotency for the "after commit" window.

---

What This Handles

• Expensive UDFs — the canonical use case. Don't re-embed, re-classify, or re-call APIs across runs that span a full task.
• Expensive explodes — skip fully processed input files. Re-explode only the one file that was in-flight at crash time.
• Expensive filters — filtered rows are "processed" because the task completed. Never re-evaluated, because StageCheckpointKeysOperator stages keys before user filters see the rows.
• Non-queryable sinks — progress is tracked at the source, not the sink.
• 2PC sinks — checkpoint store carries file metadata, enabling roll-forward recovery with commit idempotency.
• Plain Parquet / CSV as pseudo-2PC sinks — the same checkpoint store that tracks source keys also tracks output file metadata. The authoritative file list enables orphan cleanup on rerun, giving Parquet/CSV a 2PC-like capability without a catalog.
• Sink-less pipelines — UDFs with external side effects (LLM calls, Kafka publishes, HTTP POSTs) drive store.checkpoint() at task end via CheckpointTerminusNode.
• Any map-only pipeline — source-level tracking at the task boundary covers all operators that don't require shuffles and don't depend on row position (Limit, Offset, Sample).

What This Doesn't Handle (Yet)

• Shuffle-based operations — joins, aggregates, sorts, distinct. Progress tracking across shuffles is fundamentally harder. Out of scope for v1.
• Row-selectors (Limit, Offset, Sample) — rejected at optimizer time. Supporting them would require a separate design for "partially processed partitions."
• Multi-source pipelines (concat, union_all, intersect) — rejected at optimizer time. The design space (one store per source vs shared store; key-space collision handling) depends on concrete use cases we haven't seen yet.
• write_sql / Lance REST in non-idempotent write patterns — these commit per-micropartition with no rollback. Exactly-once output requires the user to write idempotently (upsert, ON CONFLICT UPDATE, MERGE) or accept duplicates.
• Partial explode recovery — if a file was mid-explode at crash time, we re-explode the whole file. Materialize explode output as a separate pipeline step if unacceptable.
• Orphan file cleanup for Parquet/CSV — designed but not implemented. Category 1 Parquet/CSV guarantee relies on this.

---

Open Questions / Follow-ups

• Key types beyond string. Int, binary, and composite (multi-column) keys are untested or unsupported. Planned: integer first, composite later.
• Partition / file-level filter. Row-level on= is the general fallback but forces the source to materialize and retain the key column. For file-addressable connectors (Parquet, Iceberg, Delta, Hudi, Paimon), a file-level filter — intercepting each connector's to_scan_tasks() and skipping files whose paths are already checkpointed — would be more efficient. Complementary to row-level, not a replacement. Open design question: do we pursue this as an optimization, or is row-level sufficient?
• Non-file-addressable sources — SQL/JDBC partition-range checkpointing was sketched but not designed. Probably a separate proposal.
• KFJ anti-join sizing — num_workers, cpus_per_worker, concurrency are hardcoded. Needs to either scale with the cluster or be configurable via CheckpointConfig.
• Canonical checkpointed-keys column name — checkpointed-key files currently use the source's on= column name. A canonical name ("key") would decouple checkpointed-key file stability from schema renames.
• In-flight write tracking in the store — stage_keys / stage_files block the pipeline on I/O. If the store tracked outstanding writes and checkpoint() became the sync point, the pipeline would be non-blocking. This also fixes the empty-source edge case where a task processes 0 rows and checkpoint() errors with "checkpoint not found."
• Post-explode checkpoint filter — should users be able to place a second checkpoint filter after an explode, tracking at a finer granularity (e.g., per-row instead of per-file)? Handles partial failure within a single input file but adds complexity.
• Store lifecycle — when is a checkpoint store created and cleaned up? Orphaned staged data needs eventual cleanup; the mechanism is TBD.

---

Related Work / References

• Implementation PR — #6725
• Original design discussion — #6446
• Checkpoint V1 / skip_existing design — Discussion #5868
• Checkpoint V1 / skip_existing implementation — PR #5931
• S3 CheckpointStore backend — PR #6599
• daft-checkpoint crate + CheckpointStore trait — PR #6511

---

Feature Matrix

Last updated: 2026-04-20. Tracks where each dimension of checkpoint support stands as of PR #6725 (in review).

Legend

• ✅ Done + validated
• 🟨 Works with caveats / partial
• ⏳ In progress or planned
• 🔬 Untested / unvalidated (not blocked, just not exercised)
• ❌ Not supported / explicitly blocked

---

Source connectors (where checkpoint= can be attached)

Connector	Status	Notes
Parquet	✅	Primary focus; MinIO integration tests cover full cycle
CSV / JSON	🔬	Should work via the same ScanOperator path — untested
Iceberg (read)	✅	Read side works; write side needs idempotent-commit follow-up
Delta Lake (read)	✅	Same
Hudi	🔬	Daft's own Hudi support is limited
Paimon	🔬	Untested end-to-end
Lance	🔬	Untested
SQL / JDBC	❌	Row-level on= doesn't map cleanly; partition-range checkpointing is a future design
Kafka / streaming	❌	Out of scope — offset-based semantics are orthogonal

Sink compatibility

Two guarantees are separable: (G1) don't redo expensive work, and (G2) don't write duplicate output rows. G1 holds for any sink where Daft defers externally-visible commits to task-end. G2 requires either framework-level 2PC, sink-native atomicity, or user-level idempotency (upserts, keyed writes).

Category 1 — Framework-atomic (Daft controls the commit boundary)

G1 holds; G2 is a framework/sink property.

Sink	Status	Notes
Iceberg	⏳	G2 via daft.checkpoint-store + daft.epoch-id snapshot tags (follow-up)
Delta Lake	⏳	G2 via {appId: store_path, version: epoch_id} txn action (follow-up)
Paimon	🔬	Snapshot tags like Iceberg; untested end-to-end
Lance (native 2PC)	🔬	Atomic-per-task; untested but structurally compatible
Parquet / CSV	🟨	G2 via checkpoint store as 2PC layer + orphan cleanup (follow-up)
Hudi	🔬	Untested

Category 2 — Side-effect endpoints (data leaves Daft mid-pipeline)

G1 depends on whether the sink defers commits to task-end; G2 is the user's or the external system's responsibility.

Endpoint	G1: don't redo work	G2: exactly-once output	Notes
write_sql (append mode, no PK)	❌	❌	Per-micropartition commits, no rollback — mid-task crash leaves partial rows, rerun duplicates them
write_sql (upsert target / ON CONFLICT UPDATE / MERGE)	🟨	✅	Rows may be re-written on crash, but upsert semantics collapse to correct output
Lance REST	❌	❌	Same pattern as write_sql append
UDF external effects (LLM / Kafka / HTTP)	✅ (whole-task granularity)	user's responsibility	CheckpointTerminusNode drives store.checkpoint() at task completion; framework can't unwind external effects, so mid-task crash re-sends rows that already left the boundary

Object store backends

Backend	Status	Notes
S3 / S3-compatible	✅	MinIO integration tests
Local FS (file://)	✅	Used for local dev + testing
GCS	🔬	Should work via daft-io, unvalidated
Azure Blob	⏳	validation pending

Pipeline operators allowed between source and sink

Operator	Allowed?	Reason
Filter	✅	Core use case; filtered-out keys still get checkpointed
Project / with_column	✅
UDF (map)	✅	The canonical expensive-work case
Explode	✅	Per-task granularity (documented trade-off)
Unpivot (melt)	✅	Map-only (wide→long row transformation, no shuffle)
IntoBatches	✅
Limit	❌	Reads from anti-join output — "first N rows" drifts across runs
Offset	❌	Checkpoints rows the user intended to drop
Sample	❌	Non-deterministic, shrinks coverage each run
Sort / TopN	❌	Shuffle
GroupBy / Aggregate	❌	Shuffle
Distinct	❌	Shuffle
Join / AsofJoin	❌	Shuffle
Window	❌	Shuffle
Pivot	❌	Shuffle — has group_by + aggregation
Repartition	❌	Shuffle
Concat / Union / Intersect	❌	multi-source design deferred

Key types

Type	Status	Notes
String (utf8)	✅	Primary
Integer	🔬	Key type follow-up
Binary	🔬	Key type follow-up
Composite (multi-column)	❌	design work needed

Capabilities

Capability	Status	Notes
Row-level checkpointing (on=)	✅
Task-level (partition) checkpointing	✅	Natural boundary — task completion = checkpoint
Sink-less pipelines	✅	CheckpointTerminusNode
Crash recovery — mid-task	✅	Staged work invisible until checkpoint()
Crash recovery — after checkpoint, before head commit	✅	Store is source of truth; get_checkpointed_files returns what needs committing
Crash recovery — after commit, before mark_committed()	⏳	Safe only with idempotent commits (follow-up)
Multi-source (concat/union) pipelines	❌	deferred
File-level filter (connector-aware — intercept to_scan_tasks())	❌	Future optimization — complementary to row-level
Partition-level task-skip (entire ScanTask skipped if all files checkpointed)	❌	Future optimization
Orphan file cleanup (staged-but-never-checkpointed artifacts)	❌	Future
KFJ anti-join parameters configurable	⏳	currently hardcoded
Canonical checkpointed-keys column name (vs source-matching)	⏳	follow-up
In-flight write tracking (fixes empty-source edge)	⏳	follow-up

Deployment

Environment	Status	Notes
Flotilla (Ray multi-worker)	✅	Primary target
Native runner	❌	Gated with ValueError + remediation guidance

Validation state

Area	Status	Notes
Rust unit tests	✅	14 in daft-checkpoint, 252 in daft-logical-plan
MinIO (S3-compatible) integration tests	✅	tests/integration/checkpoint/test_checkpoint_s3.py
Azure validation	⏳	pending
Large-dataset functional tests	⏳	Week of 2026-04-20 goal
Performance benchmarks	❌	Not started
Iceberg / Delta end-to-end with real catalog	❌	Coverage lands with the idempotent-commit follow-up

[chenghuichen]

Coming from the conversation on #6520 — the V2 design is a cleaner answer to the problem I was working around with ignore_corrupt_files.

Would it be okay to take a stab at an S3-backed CheckpointStore? The implementation would go through daft-io's IOClient, so GCS and Azure would come along for free.

[rohitkulshreshtha]

Absolutely! #6511 is the PR that merged in the interface.

#6554 - These are raw notes from an early brainstorm for the S3 implementation — I haven't thought through the details deeply yet, so I'd really appreciate your help shaping this into something solid.

[rohitkulshreshtha]

#6511

[chenghuichen]

The on= question might be worth revisiting from a slightly different angle.

Given the design's core assumption — one task per input partition, task completion implies a successful sink write — the checkpoint is fundamentally tracking completed partitions, not processed rows. If that holds, row-level keys feel like the wrong abstraction: since a task is the atomic unit, there's no meaningful partial-partition state to record.

The natural key would then be partition identity — something each connector derives from the partition itself, rather than something the user specifies from their data schema:

• File-based connectors (Parquet, CSV, Iceberg, Delta, Hudi, Paimon): (file_path, ChunkSpec) — stable across re-runs as long as files are immutable
• SQL/JDBC: the partition range condition (WHERE col >= lower AND col < upper) — usable as a checkpoint key when both data and planning config are stable across re-runs
• Streaming (Kafka, etc.): likely out of scope — streaming checkpointing has well-established offset-based semantics that are somewhat orthogonal to this design

One practical implication: rather than a user-facing on= parameter, each connector would implement a small interface that exposes its partition identity. Connectors that don't implement it simply don't support checkpointing yet — which also naturally handles the "no natural unique key" case without burdening users.

Row-level checkpointing is genuinely valuable, but it likely requires deeper infrastructure work that seems better suited to a follow-on design rather than the current scope. Happy to be corrected if there are use cases driving the on= design that this framing misses.

[rohitkulshreshtha]

@everySympathy - Thanks for sharing — I know you've been thinking about checkpointing for a while, and it's great to see that thinking here. Will dig into the partition identity framing properly and respond.

[everySympathy]

The on= question might be worth revisiting from a slightly different angle.

Given the design's core assumption — one task per input partition, task completion implies a successful sink write — the checkpoint is fundamentally tracking completed partitions, not processed rows. If that holds, row-level keys feel like the wrong abstraction: since a task is the atomic unit, there's no meaningful partial-partition state to record.

The natural key would then be partition identity — something each connector derives from the partition itself, rather than something the user specifies from their data schema:

• File-based connectors (Parquet, CSV, Iceberg, Delta, Hudi, Paimon): (file_path, ChunkSpec) — stable across re-runs as long as files are immutable
• SQL/JDBC: the partition range condition (WHERE col >= lower AND col < upper) — usable as a checkpoint key when both data and planning config are stable across re-runs
• Streaming (Kafka, etc.): likely out of scope — streaming checkpointing has well-established offset-based semantics that are somewhat orthogonal to this design

One practical implication: rather than a user-facing on= parameter, each connector would implement a small interface that exposes its partition identity. Connectors that don't implement it simply don't support checkpointing yet — which also naturally handles the "no natural unique key" case without burdening users.

Row-level checkpointing is genuinely valuable, but it likely requires deeper infrastructure work that seems better suited to a follow-on design rather than the current scope. Happy to be corrected if there are use cases driving the on= design that this framing misses.

Row-level and partition-level checkpointing differ in fault-tolerance granularity.

When using partition-level checkpointing, we also need to control partition size, and carefully balance the trade-off between fault tolerance(partition not too big) and scheduling overhead(partition not too small).

[rohitkulshreshtha]

Great discussion — I want to share the thinking behind the on= / row-level design choice, since we explored this space in depth before landing here.

We think of checkpointing + resumption as a sharp tool. Its usage isn't limited to availability failures (hosts going away) — it also covers application-level failures: misconfigurations, UDFs that need tweaking, bad input data (which is actually chenghuichen's original use case with ignore_corrupt_files).

There are roughly three families of solutions in terms of data burden:

1. O(data) — row-level key tracking. What this design proposes.
2. O(metadata) — partition-level tracking. Track partition identity, not row keys.
3. O(tasks + workers) — fully data-agnostic durable execution. Pure infrastructure state.

Options 2 and 3 are lighter — but they place a determinism burden on the rest of the system. Partitioning must be repeatable, which implies source immutability and a deterministic partitioning scheme. Plans must be deterministic, which limits the kinds of changes you can make between runs — UDFs, filters, projections. @everySympathy's analysis of glob ordering, merge instability, and scan task splits illustrates exactly this cost.

This has two consequences. First, it constrains engine design — we'd be committing to deterministic partitioning across versions. Second, and more importantly, it burdens the user. They'd need to reason about Daft internals — partition schemes, scan task merging, glob ordering — to understand what they can and can't change between runs.

Users may not care about partitioning or planning. They definitely do understand something simpler: my input data has these rows, my output has those rows. Row-level on= maps directly to that mental model. The checkpoint contract is: "have I seen this key before?" — not "did the engine partition your data the same way this time?"

That said, partition-level tracking is a natural optimization for cases where determinism can be guaranteed — and could serve as a default when users don't specify on=. Worth exploring as a follow-on.

[chenghuichen]

Thanks for the responses — I think there's a framing mismatch worth clarifying, because the determinism concern may not apply to what I had in mind.

When I said "partition identity," I didn't mean a fingerprint that must match across runs. I meant the file paths that a partition covers, used as a flat filter applied directly to the data source — not matched against new partition structures.

Concrete example: source has file1, file2, file3. Run 1 merges them into two tasks: Task A [file1, file2] completes, Task B [file3] fails. Checkpoint records {file1, file2}. On re-run, glob ordering shifts and tasks are now Task A [file1], Task B [file2, file3]. The re-run loads {file1, file2} as a flat set and filters the source directly: file1 → skip, file2 → skip, file3 → process. Partition structure is regenerated fresh and is irrelevant — there's no partition-to-partition matching at all. The determinism burden disappears.

That said, thinking through the implementation, there are actually three levels of filtering with different trade-offs:

• Task-level skip: before execution, check if all files in a task are checkpointed → skip the whole task. Connector-agnostic, simplest to implement.
• Row-level filter: build checkpointed keys into a NOT IN predicate, push it down via standard scan machinery. Also connector-agnostic — @everySympathy has already implemented this.
• File-level filter: intercept the file listing inside each connector's to_scan_tasks(), filtering out already-processed files before tasks are generated. Connector-aware — Delta filters add_actions, Iceberg filters plan_files() output, etc.

Row-level on= makes sense where source data isn't file-addressable, or where cross-partition deduplication is genuinely needed — agreed that's a real use case. The two approaches seem complementary rather than competing.

[everySympathy]

Thanks for the responses — I think there's a framing mismatch worth clarifying, because the determinism concern may not apply to what I had in mind.

When I said "partition identity," I didn't mean a fingerprint that must match across runs. I meant the file paths that a partition covers, used as a flat filter applied directly to the data source — not matched against new partition structures.

Concrete example: source has file1, file2, file3. Run 1 merges them into two tasks: Task A [file1, file2] completes, Task B [file3] fails. Checkpoint records {file1, file2}. On re-run, glob ordering shifts and tasks are now Task A [file1], Task B [file2, file3]. The re-run loads {file1, file2} as a flat set and filters the source directly: file1 → skip, file2 → skip, file3 → process. Partition structure is regenerated fresh and is irrelevant — there's no partition-to-partition matching at all. The determinism burden disappears.

That said, thinking through the implementation, there are actually three levels of filtering with different trade-offs:

• Task-level skip: before execution, check if all files in a task are checkpointed → skip the whole task. Connector-agnostic, simplest to implement.
• Row-level filter: build checkpointed keys into a NOT IN predicate, push it down via standard scan machinery. Also connector-agnostic — @everySympathy has already implemented this.
• File-level filter: intercept the file listing inside each connector's to_scan_tasks(), filtering out already-processed files before tasks are generated. Connector-aware — Delta filters add_actions, Iceberg filters plan_files() output, etc.

Row-level on= makes sense where source data isn't file-addressable, or where cross-partition deduplication is genuinely needed — agreed that's a real use case. The two approaches seem complementary rather than competing.

Thanks for your additional explanation! It makes sense.

[chenghuichen]

The "What This Doesn't Handle" section covers source-side limitations, but there's a sink-side constraint worth adding explicitly.

Checkpoint rerun creates a natural expectation of idempotency. This holds for most sinks — Iceberg, Delta, Paimon, and native Lance all write-then-atomic-commit, so failed tasks leave no visible trace. Two current sinks may not satisfy this, though:

• write_sql: Commits each micropartition in a separate database transaction with no task-level rollback. In overwrite mode, the table is cleared upfront — a mid-task failure leaves partial rows permanently in the database. Rerun duplicates them.
• Lance REST: Same issue — each micropartition is independently committed with no staging or rollback.

The concern is silent data corruption: checkpoint would mark the task incomplete and retry it, but the sink has already partially written. No error is surfaced to the user.

Explicitly flagging these sinks as not yet compatible with checkpoint — until they adopt transaction or upsert semantics — might help users feel more confident in the correctness guarantee.

[rohitkulshreshtha]

I agree with your viewpoint about sink-side guarantees being made explicit. In addition to your points about 2PC style protocols that Iceberg, Delta, etc. support - there's an additional point about idempotence of the commit() operation itself, which is worth mentioning. I will update the document to add this. It's 100% must be stated that ideal behavior requires some protocol-level co-operation from the sinks.

[rohitkulshreshtha]

Thanks for your thoughtful comments, chenghuichen. Much appreciated.

[chenghuichen]

One concern about the crash recovery table: the scenario "After sink commit, before mark_committed()" carries a hidden assumption that catalog commits are idempotent.

In practice, they're not. If a crash occurs between a successful catalog commit and mark_committed(), the next run will still see those files returned by get_checkpointed_files() — because from the checkpoint store's perspective, they're still in Checkpointed state. The recovery logic would then attempt to commit the same files again.

For Iceberg and Paimon, re-committing the same DataFile objects in a new transaction doesn't deduplicate — it creates a new snapshot referencing those files a second time, resulting in duplicate rows. Delta Lake has similar behavior unless an explicit transaction ID is used for idempotency, which Daft's current sink doesn't leverage.

Safe recovery in this window would require either checking catalog state before re-committing, or using format-native idempotency mechanisms (e.g., Iceberg's txnId). Worth calling out explicitly in the design, as it's one of the trickier failure modes.

[rohitkulshreshtha]

Agree. This is what I was referring to as "idempotence of the commit operation" #6446 (reply in thread).

[rohitkulshreshtha]

One nice side-effect of this design: since the checkpoint store tracks both source keys and output file metadata, we effectively have a 2PC layer that works for any sink — including plain Parquet. The checkpoint store knows which files were produced by each task. After all tasks complete, the head node can treat the checkpointed file list as the authoritative output, and clean up any orphaned files from failed/partial tasks. Parquet writes become atomic without needing a catalog.

[rohitkulshreshtha]

@chenghuichen @everySympathy
I wanted to highlight some design decisions that are worth capturing:

Pushdown guards on PushDownFilter / PushDownProjection

The original design treated the checkpoint filter as "injected after the source." Daft's optimizer normally pushes filters and projections into the Source node — which would drop rows before StageCheckpointKeysOperator sees them, breaking the core promise that filtered-out keys still get checkpointed. (An expensive UDF drops a row → key never staged → re-run re-evaluates that row → drops it again → repeat forever.)

Decision. PushDownFilter unconditionally skips sources with checkpoint.is_some(). PushDownProjection is narrower — it only skips when the on= key column would be pruned; otherwise it still prunes unused columns normally.

Trade-off (user-facing). Opting into checkpointing means the source must materialize every row + retain the key column up to StageCheckpointKeysOperator, even if a downstream filter would drop most of them. Checkpointing ⇒ reduced pushdown, paid at scan time. Worth being explicit in the docs.

---

Sink-less pipelines via CheckpointTerminusNode

The original design assumed every pipeline ends in a Daft-native write sink, where task completion is the natural checkpoint boundary. That excludes a category Daft actively pitches: UDFs that call external systems — LLMs, embeddings APIs, Kafka publishes, web scrapers.

Decision. Introduce CheckpointTerminusNode: a terminal wrapper injected at the plan root when a checkpoint source is present with no native write sink downstream. It drives store.checkpoint() on task completion, at the same per-task seal granularity as the sink case.

Trade-off. Seal granularity is still per-task, so a crash mid-batch loses work on rows that already hit the external system. We can potentially improve this to be row-level in the future, but I have not made it a goal at the moment.

---

Key file format: Parquet (not Arrow IPC)

The store persists sealed source keys so the anti-join on re-run knows what to skip. Original design left the format open. A straightforward choice would be Arrow IPC — same shape as the in-memory data.

Decision. Parquet. The right side of the anti-join is literally a read_parquet scan of <store>/keys/*.parquet — reusing Daft's first-class Parquet scanner, same code path, same perf, same operational surface. Arrow IPC would mean maintaining a parallel reader.

Future leverage. Parquet statistics (min/max per row group) and optional bloom filters give a straight path to pruning sealed-key scans without building new infra. If a checkpoint store ever accumulates millions of keys and the anti-join becomes a bottleneck, enabling bloom filters per column is a single writer-side flag and the scanner picks it up for free. Arrow IPC would have required building pruning from scratch.

---

concat / union_all rejected at optimizer time

Original design listed multi-source pipelines as an open question. The answer shapes the API: does CheckpointConfig attach per-source? Does one source's StageCheckpointKeysOperator see another's keys? Share one store across sources or one each?

Decision for now. Reject at optimizer time (validate_map_only_downstream rejects Concat/Union/Intersect downstream of a checkpointed source).

Why defer, not design. The right answer depends on concrete customer intent we haven't seen yet. Deferring until a real use case lands avoids baking wrong assumptions into the API.

---

Per-input CheckpointId (instead of per-task)

Original said "one CheckpointId per task (UUID)." That implicitly assumed one task = one operator instance. Under Flotilla it isn't — the pipeline translator compiles the operator set once, and every task that runs the pipeline reuses those same instances (StageCheckpointKeysOperator, blocking sink, etc.). A single pre-allocated UUID means every task's checkpoint(id) call hits the same id — second task errors with "already sealed." We've changed the mapping to be per input-id to avoid this.

---

Limit / Offset / Sample added to the reject list

Original said "fail at plan construction time for shuffles" — joins, aggregates, sorts, distinct. In review we found three non-shuffle operators that still break checkpoint semantics in non-obvious ways:

• Limit picks rows from the anti-join output, not the original source. On a source that grows between runs, source.limit(2) returns [a, b] on run 1 and [e, f] on run 2 — meaning of "first 2 rows" silently drifts.
• Offset has the dual problem. source.offset(2) intends to drop the first 2 rows. After run 1 seals rows 3+, run 2's anti-join returns just rows 1-2, offset drops them — so rows the user said "never process" are now sealed as if processed, the intent is violated.
• Sample is non-deterministic across runs. Sealed keys shrink the sample universe each run, so sample(0.3) draws from a dwindling set — silently regressing coverage until it returns empty.

Decision. Add all three to the denylist with worked examples in the code comments. Fail at optimizer time with an error pointing to the offending op.

[rohitkulshreshtha]

#6725

[rohitkulshreshtha]

Checkpointing v2 — Feature Matrix

Last updated: 2026-04-20. Tracks where each dimension of checkpoint support stands as of PR #6725 (in review).

Legend

• ✅ Done + validated
• 🟨 Works with caveats / partial
• ⏳ In progress or planned
• 🔬 Untested / unvalidated (not blocked, just not exercised)
• ❌ Not supported / explicitly blocked

---

Source connectors (where checkpoint= can be attached)

Connector	Status	Notes
Parquet	✅	Primary focus; MinIO integration tests cover full cycle
CSV / JSON	🔬	Should work via the same ScanOperator path — untested
Iceberg (read)	✅	Read side works; write side needs idempotent-commit follow-up
Delta Lake (read)	✅	Same
Hudi	🔬	Daft's own Hudi support is limited
Paimon	🔬	Untested end-to-end
Lance	🔬	Untested
SQL / JDBC	❌	Row-level on= doesn't map cleanly; partition-range checkpointing is a future design
Kafka / streaming	❌	Out of scope — offset-based semantics are orthogonal

Sink compatibility

Two guarantees are separable: (G1) don't redo expensive work, and (G2) don't write duplicate output rows. G1 holds for any sink where Daft defers externally-visible commits to task-end. G2 requires either framework-level 2PC, sink-native atomicity, or user-level idempotency (upserts, keyed writes).

Category 1 — Framework-atomic (Daft controls the commit boundary)

G1 holds; G2 is a framework/sink property.

Sink	Status	Notes
Iceberg	⏳	G2 via daft.checkpoint-store + daft.epoch-id snapshot tags (follow-up)
Delta Lake	⏳	G2 via {appId: store_path, version: epoch_id} txn action (follow-up)
Paimon	🔬	Snapshot tags like Iceberg; untested end-to-end
Lance (native 2PC)	🔬	Atomic-per-task; untested but structurally compatible
Parquet / CSV	🟨	G2 via checkpoint store as 2PC layer + orphan cleanup (follow-up)
Hudi	🔬	Untested

Category 2 — Side-effect endpoints (data leaves Daft mid-pipeline)

G1 depends on whether the sink defers commits to task-end; G2 is the user's or the external system's responsibility.

Endpoint	G1: don't redo work	G2: exactly-once output	Notes
write_sql (append mode, no PK)	❌	❌	Per-micropartition commits, no rollback — mid-task crash leaves partial rows, rerun duplicates them
write_sql (upsert target / ON CONFLICT UPDATE / MERGE)	🟨	✅	Rows may be re-written on crash, but upsert semantics collapse to correct output
Lance REST	❌	❌	Same pattern as write_sql append
UDF external effects (LLM / Kafka / HTTP)	✅ (whole-task granularity)	user's responsibility	CheckpointTerminusNode drives store.checkpoint() at task completion; framework can't unwind external effects, so mid-task crash re-sends rows that already left the boundary

Object store backends

Backend	Status	Notes
S3 / S3-compatible	✅	MinIO integration tests
Local FS (file://)	✅	Used for local dev + testing
GCS	🔬	Should work via daft-io, unvalidated
Azure Blob	⏳	validation pending

Pipeline operators allowed between source and sink

Operator	Allowed?	Reason
Filter	✅	Core use case; filtered-out keys still get checkpointed
Project / with_column	✅
UDF (map)	✅	The canonical expensive-work case
Explode	✅	Per-task granularity (documented trade-off)
Unpivot (melt)	✅	Map-only (wide→long row transformation, no shuffle)
IntoBatches	✅
Limit	❌	Reads from anti-join output — "first N rows" drifts across runs
Offset	❌	Checkpoints rows the user intended to drop
Sample	❌	Non-deterministic, shrinks coverage each run
Sort / TopN	❌	Shuffle
GroupBy / Aggregate	❌	Shuffle
Distinct	❌	Shuffle
Join / AsofJoin	❌	Shuffle
Window	❌	Shuffle
Pivot	❌	Shuffle — has group_by + aggregation
Repartition	❌	Shuffle
Concat / Union / Intersect	❌	multi-source design deferred

Key types

Type	Status	Notes
String (utf8)	✅	Primary
Integer	🔬	Key type follow-up
Binary	🔬	Key type follow-up
Composite (multi-column)	❌	design work needed

Capabilities

Capability	Status	Notes
Row-level checkpointing (on=)	✅
Task-level (partition) checkpointing	✅	Natural boundary — task completion = checkpoint
Sink-less pipelines	✅	CheckpointTerminusNode
Crash recovery — mid-task	✅	Staged work invisible until checkpoint()
Crash recovery — after checkpoint, before head commit	✅	Store is source of truth; get_checkpointed_files returns what needs committing
Crash recovery — after commit, before mark_committed()	⏳	Safe only with idempotent commits (follow-up)
Multi-source (concat/union) pipelines	❌	deferred
File-level filter (connector-aware — intercept to_scan_tasks())	❌	Future optimization — complementary to row-level
Partition-level task-skip (entire ScanTask skipped if all files checkpointed)	❌	Future optimization
Orphan file cleanup (staged-but-never-checkpointed artifacts)	❌	Future
KFJ anti-join parameters configurable	⏳	currently hardcoded
Canonical checkpointed-keys column name (vs source-matching)	⏳	follow-up
In-flight write tracking (fixes empty-source edge)	⏳	follow-up

Deployment

Environment	Status	Notes
Flotilla (Ray multi-worker)	✅	Primary target
Native runner	❌	Gated with ValueError + remediation guidance

Validation state

Area	Status	Notes
Rust unit tests	✅	14 in daft-checkpoint, 252 in daft-logical-plan
MinIO (S3-compatible) integration tests	✅	tests/integration/checkpoint/test_checkpoint_s3.py
Azure validation	⏳	pending
Large-dataset functional tests	⏳	Week of 2026-04-20 goal
Performance benchmarks	❌	Not started
Iceberg / Delta end-to-end with real catalog	❌	Coverage lands with the idempotent-commit follow-up

[rohitkulshreshtha]

@chenghuichen, copying this from #6725.

Thanks for the close read — genuinely appreciate the depth of engagement here. File-path identity is a real win for file-based scans — pushdown stays enabled, O(files) storage, simpler key. I hear you that if you focus on blob-storage-oriented systems, which is not an insignificant slice of the world, the general-purpose row-level approach feels overbuilt. The current PR stays row-level on= because DAFT also needs to cover Kafka, explodes, SQL (non-file sources) — really a matter of prioritization on our end, not architectural disagreement. If your team is willing to contribute the file-path fast-path, there's no reason for us to serialize the effort on the Daft side. Happy to have someone from your team drive a follow-up PR with me reviewing and integrating.

Before code, a short design doc would help us align — a few open questions off the top of my head:

1. At what level does the filtering occur? I'd guess the scheduler, filtering out files already covered before task dispatch.
2. How do we register that a file was covered? DAFT uses input IDs, which (IIRC) aren't a pure function of the file path — so the mapping from path → "covered" needs a deterministic identity mechanism.

There may be more — a design doc would surface them. Happy to move the deeper discussion to #6446 so this PR thread stays scoped.

[chenghuichen]

@rohitkulshreshtha Thanks for the openness to a follow-up — and for the two concrete questions (where does filtering occur, how do we register coverage). I've spent some time working through the details, and this is what I landed on.

---

File-Path Identity Checkpoint — Design Proposal

Positioning

File-path identity is a fast-path that activates when: (1) the source is file-based, (2) the user does not specify on=. When on= is provided, or the source is non-file (SQL, Kafka, Python factories), the existing row-level path handles it unchanged.

df.checkpoint(store)              # file-path mode
df.checkpoint(store, on="req_id") # row-level mode

pub enum CheckpointKeyMode {
    RowLevel { key_column: String },
    FilePath,
}

pub struct CheckpointConfig {
    pub store: CheckpointStoreConfig,
    pub key_mode: CheckpointKeyMode,
}

Source Atom Keys

The checkpoint key granularity is the source atom — the smallest indivisible unit of source data. For unsplit files, the atom is the file itself. For split files, each chunk defined by ChunkSpec produces its own atom key. The ChunkSpec enum is exhaustive-matched, so new file formats that introduce splitting automatically require an atom key encoding — the compiler enforces it.

fn source_atom_keys(source: &ScanSource) -> Vec<String> {
    let path = source.get_path();
    match source.chunk_spec() {
        None => vec![path.clone()],
        Some(spec) => spec.atom_keys(path),
    }
}

impl ChunkSpec {
    fn atom_keys(&self, path: &str) -> Vec<String> {
        match self {
            Self::Parquet(rgs) =>
                rgs.iter().map(|rg| format!("{path}#rg={rg}")).collect(),
            Self::Bytes { start, end } =>
                vec![format!("{path}#bytes={start}-{end}")],
        }
    }
}

Why per-atom, not per-file or per-chunk:

• Per-file is unsafe. split_and_merge_pass can distribute chunks of the same file across different InputIds. If one succeeds and seals "file_A.parquet" while another fails, re-run skips the entire file — silent data loss.
• Per-chunk is fragile. A key like "file.pq#rg=0,1" encodes the splitter's grouping config. Changing parquet_split_row_groups between runs invalidates all existing keys, causing full re-processing.
• Per-atom is both safe and stable. Row group indices (and analogous chunk identifiers in other formats) are intrinsic to the file — they don't change regardless of how the splitter groups them. Filtering checks each atom individually: all atoms present → skip the chunk; any atom missing → process it.

Optimizer Rule: Dual-Mode Dispatch

RewriteCheckpointSource dispatches on key_mode. Both modes share the same validation (map-only downstream) and reuse the StageCheckpointKeys plan node — they diverge in what that node wraps.

fn rewrite(&self, source: &Source, cfg: &CheckpointConfig) -> ... {
    validate_map_only_downstream(&plan, &node)?;

    match &cfg.key_mode {
        CheckpointKeyMode::RowLevel { key_column } =>
            self.rewrite_row_level(source, key_column, cfg),
        CheckpointKeyMode::FilePath =>
            self.rewrite_file_path(source, cfg),
    }
}

fn rewrite_file_path(&self, source: &Source, cfg: &CheckpointConfig) -> ... {
    // 1. Validate source is file-based
    // 2. Strip checkpoint from Source (prevent re-match)
    // 3. Wrap in StageCheckpointKeys(key_mode=FilePath)
    // No anti-join. No SCKO. No BlobStoreKeysScan. No pushdown guards.
}

Plan tree comparison:

Row-level:                        File-path:

StageCheckpointKeys(RowLevel)     StageCheckpointKeys(FilePath)
  │                                 │
Anti-Join(KFJ, key_col)           Source (unchanged)
  ├── Source (ckpt=None)
  └── BlobStoreKeysScan

+ pushdown guards active          + pushdown fully enabled

Pipeline Builder: Mode Branch

FilePath mode creates no intermediate operator. It creates a FilePathRegistry (InputId → atom keys, propagates source identity to
sink) and sets CheckpointPipelineState::FilePath on the builder context, then returns the child pipeline as-is.

The checkpointed-key HashSet is loaded lazily downstream — in ScanTaskSource (local) or at the driver in produce_tasks (distributed).

Filtering and Staging

In distributed mode, filtering operates at two layers:

Layer 1 — Driver-side coarse pruning (StageCheckpointKeysNode::produce_tasks): The driver loads all checkpointed keys into a HashSet once, then filters the task stream before dispatching. If every atom key in a task builder is already checkpointed, the task is dropped entirely — no worker scheduling, no network, no IO.

Layer 2 — Worker-side fine-grained filtering (ScanTaskSource): Same HashSet-based per-atom filtering as the native path. Handles partially checkpointed tasks that passed Layer 1 (e.g., a task with multiple scan sources where only some are done).

Layer 1 is a pure performance optimization; removing it does not affect correctness. Layer 2 is the correctness mechanism.

End-to-End Flow

┌─── Optimizer ──────────────────────────────────────────────────┐
│  RowLevel: StageCheckpointKeys → AntiJoin → [Source, KeysScan]│
│  FilePath: StageCheckpointKeys → Source                       │
└────────────────────────┬───────────────────────────────────────┘
                         ▼
┌─── Driver ────────────────────────────────────────────────────┐
│  load keys → HashSet                                          │
│    ├─ coarse task-level filter                                │
│    └─ dispatch surviving tasks                                │
└────────────────────────┬───────────────────────────────────────┘
                         ▼
┌─── Pipeline Builder (StageCheckpointKeys arm) ────────────────┐
│  RowLevel: SCKO operator + ctx(store, id_map, key_expr)       │
│  FilePath: checkpointed_set + registry + ctx(store, id_map)   │
└────────────────────────┬───────────────────────────────────────┘
                         ▼
┌─── Shared Execution ──────────────────────────────────────────┐
│                                                               │
│  ScanTaskSource ─── [FilePath: per-atom filter + register]    │
│       │                                                       │
│    Pipeline (map-only ops, UDFs)                              │
│       │                                                       │
│  Sink / Terminus                                              │
│    ├─ [RowLevel] keys staged by SCKO upstream                 │
│    ├─ [FilePath] registry.take() → store.stage_keys()         │
│    ├─ store.stage_files()          [sink only, shared]        │
│    └─ store.checkpoint(id)         [SEAL, shared]             │
│                                                               │
└───────────────────────────────────────────────────────────────┘

Shared: CheckpointStore + S3 impl, three-phase lifecycle, CheckpointIdMap, sink sealing, file metadata staging, mark_committed, Python API, map-only validation, StageCheckpointKeys plan node.

Independent only in: key source (row data vs. scan task metadata), filtering mechanism (anti-join vs. HashSet lookup), pushdown behavior (guarded vs. unrestricted).

Benefits

• Broadens checkpoint applicability. Row-level on= requires a globally unique key column — a constraint that many common scenarios (e.g., append-only tables, denormalized logs) simply cannot satisfy. File-path mode requires no user-specified key. Any file-based source is checkpointable out of the box.
• Filter and projection pushdown fully enabled. Keys come from scan task metadata, not the row stream. Pushdown guards are unnecessary — eliminates the scan-time tax of materializing every row up to SCKO.
• O(row_groups) checkpoint keys instead of O(rows). Two orders of magnitude fewer entries. Smaller store, less memory, cheaper filtering. This alone eliminates the anti-join at execution time, replacing it with a simple HashSet membership check — fewer keys, zero join-operator overhead, same checkpoint store I/O.

Open Items

1. HashSet memory ceiling. A typical batch job touches 10k–100k files; at ~100 bytes per atom key, the checkpointed-key HashSet fits in ~1–10 MB. For the vast majority of single-job workloads this is negligible. For extreme scale (millions of accumulated keys), file-path mode can fall back to the streaming anti-join used by row-level mode. Deferred until a concrete need arises.
2. JSONL byte-range stability. Parquet RG indices are file-intrinsic and config-immune. JSONL byte boundaries depend on the split algorithm — config changes cause safe re-processing but no data loss. May want to fall back to whole-file keys for JSONL.

---

Happy to iterate on any of this. If the direction looks right I can start on a PR against your branch.

[rohitkulshreshtha]

@chenghuichen — thank you for writing this up. I'm on-call this week and want to give this the careful read it deserves rather than a rushed one. I'll come back with a proper response by early next week, if that's ok with you.

Really appreciate the time you put into this.

[chenghuichen]

#6869

[rohitkulshreshtha]

Hey @chenghuichen — sorry for the long delay on this. After reading through the file-path identity proposal carefully, I'd like to surface three concerns to the wider team before we move forward:

1. Source immutability. Atom keys identify chunks by file location, so they silently skip files whose contents have changed between runs — a sharper constraint than today's row-level mode, which derives keys from row content.

2. Byte-range brittleness. For byte-range-split formats (JSONL today), the atom key includes splitter-chosen byte offsets, so config changes between runs produce overlapping-but-not-equal keys and duplicate emission.

3. ChunkSpec encoding is convention, not contract. Nothing in the type system prevents a future contributor from byte-range-splitting Parquet, which would silently re-attach Concern 2's brittleness.

I want to discuss these with the team first before responding on PR #6869. Will report back after our sync. Thanks for the very thorough writeup.

[rohitkulshreshtha]

Hey @chenghuichen — quick update on where my thinking is. I haven't looked at the PR code yet; still chewing on the design and wanted to share where I've landed so you can push back. Most of this is loud thinking, not conclusions.

What I'm convinced on:

• Row group as the atom for Parquet feels right. O(metadata) keys instead of O(rows) is a real win, and row group indices are file-intrinsic so they survive splitter config changes.
• For CSV/text and other byte-range-split formats, the splitter chooses byte boundaries at read time, so atom keys would be brittle there. Probably right to exclude those formats from this mode and require on= instead. WDYT?

What I'm still chewing on:

1. Source immutability is a real assumption we'd bake in. Atom keys check file location, not content, so a file replaced between runs gets silently skipped. Some customers genuinely have immutable sources and are fine with this; we've also heard pushback on assuming it as a default. Open question on how/whether to surface the constraint to users — doc warning, opt-in flag at checkpoint creation, plan-time error when the source is mutable-by-construction. Curious how you've been thinking about this.

2. Sub-row-group splitting as a future concern. Today Daft splits at row-group boundaries, so atom = row group is sufficient. But row-range splitting within a row group is plausible (e.g., parallelizing a single 10M-row row group across workers). If we ever add that, path#rg=N becomes too coarse and the natural extension (path#rg=N#rows=A-B) re-introduces splitter-config sensitivity. I'd rather not have checkpointing be the thing that institutionalizes "row group is the minimum forever" if there's a natural design that keeps it composable. Have you thought about what the atom encoding would look like if Daft moved to sub-row-group splitting later?

Will get to the PR code separately — just wanted to surface the live design questions first.

[chenghuichen]

@rohitkulshreshtha Thanks for sharing where your thinking is — really helpful to see what's settled vs. still open.

On byte-range-split formats: Fully agreed. Clean separation.

On source immutability: I've been thinking about this too — in fact some of my earlier commits on the PR experimented with addressing it. My current thinking is that Daft can actually detect file changes without baking in an immutability
assumption, using metadata that's already available at listing time.

The approach: encode file_size + etag (object stores) or file_size + mtime (local FS) into the atom key (e.g. path#rg=N#sz=12345#etag=abc123). Both come from the listing/stat call itself — S3's list_objects_v2 already returns e_tag() per
object, local FS's metadata() syscall already returns modified(). This needs FileMetadata extended to carry those fields through to ScanSource, but the information is free at the source — no additional IO.

The property is the same: if a file is replaced between runs, its atom key naturally differs from what's in the checkpoint store → no match → reprocessed. Users don't need to reason about immutability; the system handles it. In the rare case
where neither etag nor mtime is available, we raise an error directing the user to use on= instead.

On sub-row-group splitting: I hadn't thought about this before your message, so thanks for raising it. My thinking here is still rough given how abstract this scenario is, but my intuition is that the checkpoint granularity can stay at the row-group level even if execution goes finer. If Daft ever splits a single row group across workers, the checkpoint contract would be: all sub-splits of that row group must complete before the row-group-level key is sealed, same pattern your design already uses per task. If any sub-split fails, the entire row group reruns. This adds coordination complexity, but as far as I can tell, the design would not structurally prevent sub-row-group splitting from being layered on later.

Does that framing sit right with you?