[EPIC] Distributed Execution for Reads/Writes

[NoahKusaba]

Goal

Run Iceberg reads and writes in parallel on
Ballista executors: apache/datafusion-ballista#2217 adds an iceberg-ballista crate whose
logical and physical extension codecs send Iceberg plans to the scheduler and
executors.

A codec in another process has to name each node's type, read what it was built
from, and rebuild an equivalent node from those parts alone. This epic tracks the
changes that make that possible, split into what blocks
apache/datafusion-ballista#2217 and what improves the integration once it has
landed.

This likely is all relevant to datafusion-distributed as well, but written from the perspective of Ballista, so it may not be consistent to their needs.

I'm likely missing to-do's as well, so please let me know if someone notices anything and I can update this epic.

Required (blocking)

apache/datafusion-ballista#2217 can't merge until these are on
datafusion-iceberg main. Today it depends on a fork branch that carries all of
them except partitioned scans.

• feat: make PartitionExpr reconstructible #14 Make PartitionExpr reconstructible (closes PartitionExpr cannot be reconstructed from its public API #13)
• feat: make the plan nodes inspectable and rebuildable #20 Make the plan nodes inspectable and rebuildable: public
  IcebergCommitExec, IcebergWriteExec, IcebergMetadataScan, accessors,
  and constructors that take only what the accessors return
• IcebergTableProvider::new_with_schema (PR to be opened; draft, stacked on
  feat: make the plan nodes inspectable and rebuildable #20). Lets the scheduler rebuild a catalog-backed provider with the schema the
  client planned against, without loading the table.
• Opt-in idempotent commits (PR to be opened; draft, stacked on feat: make the plan nodes inspectable and rebuildable #20).
  IcebergCommitExec records a commit id in the snapshot summary and skips
  committing when an earlier execution already did, since Ballista reruns tasks
  after losing an executor. Two attempts running at the same time can still
  both commit; see the idempotency key under iceberg-rust below.
• Partitioned scans. IcebergTableScan reports UnknownPartitioning(1), so
  every scan runs as one task on one executor, which lists and reads every
  file. It also limits writes: Ballista drops round-robin repartitions, so an
  INSERT into an unpartitioned table, or one with no identity or bucket
  partition field, runs one writer task per input partition. Into such a
  table, INSERT INTO t SELECT … FROM <iceberg table> writes from a single
  task.
  feat(datafusion): Add opt-in eager file scan planning with output partitioning iceberg-rust#2671 (@toutane, closes Enable parallel file-level scanning for IcebergTableScan Datafusion Integration iceberg-rust#2220)
  implements this: scan() plans the file tasks eagerly, groups them across
  target_partitions, and execute(i) reads group i. It went through several
  review rounds there but was stranded by the move to this repo, and
  [Epic] Sort-order reporting for iceberg-datafusion scans iceberg-rust#3126 (sort-order reporting) builds on it. Porting it
  here also needs, for Ballista:
  • The node exposes its task groups and has a constructor that takes them,
    like feat: make the plan nodes inspectable and rebuildable #20's nodes. The files must be planned once, on the scheduler, and
    shipped to executors: an executor that planned again would read a
    different grouping, since file planning order isn't deterministic.
    FileScanTask is public and serializable.
  • No dependency on TableScan::arrow_reader_builder(), which #2671 adds to
    iceberg-rust but neither iceberg-rust main nor the revision chore(deps): align with current iceberg-rust revision #21 moves
    to has.
    Build the ArrowReaderBuilder here instead (its setters are public, and
    Runtime::try_current() works at execute time), or upstream the
    accessor first (see iceberg-rust below).
  • If eager planning stays opt-in (#2671 uses
    iceberg.enable_eager_scan_planning), the setting reaches the Ballista
    scheduler, or iceberg-ballista turns it on.
  • A scan with no files reports one partition, not zero.

Nice to have

Fixes and improvements that don't block apache/datafusion-ballista#2217. Most
of the iceberg-rust items remove a workaround in iceberg-ballista or close a
known gap.

datafusion-iceberg

• fix: accept NOT NULL input columns in partitioned INSERT #19 Accept NOT NULL input columns in partitioned INSERT. Without it, a
  partitioned INSERT into optional columns from a NOT NULL source (such as a
  VALUES list with no NULLs, whose columns DataFusion infers as
  non-nullable) fails at plan time.
  • INSERT nullability: unpartitioned path skips the runtime NULL check; partitioned path rejects nullable sources at plan time #22 follow-up: unpartitioned tables write NULLs into required columns;
    partitioned tables reject nullable sources at plan time
• Hash-partition INSERT input for every partition transform. repartition.rs
  hashes on _partition only when the spec has an identity or bucket field,
  and otherwise round-robins, which Ballista drops. Each writer task can then
  write to every table partition, and the INSERT produces many small files.
  _partition already holds the transformed value, so hashing on it works
  for any transform.
• Scan statistics. With eager planning, file record counts give row counts
  nearly for free, which helps join planning. An earlier attempt,
  feat(datafusion): Expose DataFusion statistics on an IcebergTableScan iceberg-rust#880, went stale.
• Pass the session's batch_size to the scan's reader, and bound per-partition
  file-read concurrency once a scan has several partitions per executor.

Correctness bugs

Opened by @andygrove. None of these blocks apache/datafusion-ballista#2217, but
each returns wrong results or fails in a way a distributed run hits too.

Writes:

• Partitioned INSERT computes partition values from the wrong columns when the input plan ends in a projection #35 Partitioned INSERT computes partition values from the wrong columns when
  the input plan ends in a projection. PartitionExpr::children() is empty, so
  ProjectionPushdown moves it onto a different column layout and data files
  get the wrong partition values.
• INSERT into a partitioned table with fanout disabled always fails: the optimizer removes the partition sort #38 INSERT with fanout disabled always fails: IcebergWriteExec declares no
  required input ordering or distribution, so the optimizer removes the
  partition sort. Also covers unpartitioned tables (INSERT into an unpartitioned table fails when fanout is disabled #43, closed as a
  duplicate). A required hash distribution on _partition would also help
  the hash-partitioned INSERT item above, since Ballista keeps hash
  repartitions but drops round-robin ones.
• Clustered (fanout disabled) writes fail on sorted input because batches are split in HashMap order #39 Fanout-disabled writes fail on sorted input, because iceberg-rust's
  RecordBatchPartitionSplitter emits a batch's partitions in HashMap order.
  Needed on top of INSERT into a partitioned table with fanout disabled always fails: the optimizer removes the partition sort #38 for fanout-disabled writes to work.

Scans and filter pushdown:

• Filter pushdown strips casts on columns, which drops matching rows or fails valid queries #36 Filter pushdown strips casts on columns, which drops matching rows or
  fails valid queries (for example a REAL column compared with a DOUBLE
  literal)
• LIKE prefix pushdown ignores backslash escapes and drops matching rows #40 LIKE prefix pushdown ignores backslash escapes and drops matching rows
• convert_filters_to_predicate negates a weakened AND, producing a predicate that drops matching rows #44 convert_filters_to_predicate negates a weakened AND, producing a
  predicate that drops matching rows
• Metadata tables ignore the scan projection, so projected and aggregate queries return wrong columns or fail #41 Metadata tables ignore the scan projection, so projected and aggregate
  queries return wrong columns or fail. The fix adds a projection to
  IcebergMetadataScan, which feat: make the plan nodes inspectable and rebuildable #20 makes public, so its accessors and
  constructor need to carry it.
• SELECT * fails after a column is added until the next write to the table #42 SELECT * fails after a column is added until the next write to the
  table. The datafusion-iceberg side of fix(scan): don't project a stale schema after a schema update with no write iceberg-rust#2904 below.

Catalog:

• CREATE TABLE / DROP TABLE hang on a current-thread tokio runtime and panic outside a runtime #37 CREATE TABLE / DROP TABLE hang on a current-thread tokio runtime and
  panic outside a runtime

iceberg-rust

In review:

• fix(scan): don't project a stale schema after a schema update with no write iceberg-rust#2904 A scan projects a stale schema after a schema change
  with no write since (fixes Scan projects a stale schema after a schema update with no write iceberg-rust#2905; surfaces here as SELECT * fails after a column is added until the next write to the table #42)
• feat(catalog-loader): add with_runtime to BoxedCatalogBuilder iceberg-rust#3286 with_runtime on the catalog loader's
  BoxedCatalogBuilder, so a loaded catalog can be bound to a chosen runtime
• fix(arrow): strip metadata from list elements and map entries iceberg-rust#3303 strip_metadata_from_schema fails on list and map
  columns. Blocks the list and map nullability tests staged in fix: accept NOT NULL input columns in partitioned INSERT #19.

Not opened yet, most important first:

• Idempotency key on fast_append, checked inside the commit retry loop.
  Closes the concurrent-commit gap above: today a conflicting commit is retried
  and re-applied on the new head without re-checking the commit id.
• A scan on a runtime that has shut down returns no rows instead of failing
• The REST catalog never refreshes its OAuth token, so a long-lived scheduler
  fails to plan once its token expires
• Debug output of FileIO/StorageConfig prints storage credentials
• Table::with_runtime that keeps the table's manifest cache
• TableScan::arrow_reader_builder(), from feat(datafusion): Add opt-in eager file scan planning with output partitioning iceberg-rust#2671, so a
  partitioned scan reads with exactly the reader settings its TableScan
  was built with

To investigate: whether FileIO can refresh catalog-vended storage credentials
during a long job.

Merge order

1. feat: make the plan nodes inspectable and rebuildable #20
2. The rest of the required items, which build on feat: make the plan nodes inspectable and rebuildable #20, in any order:
   • new_with_schema uses feat: make the plan nodes inspectable and rebuildable #20's public try_new and table_ident()
   • idempotent commits add to IcebergCommitExec, which feat: make the plan nodes inspectable and rebuildable #20 makes public
   • partitioned scans rework the IcebergTableScan code feat: make the plan nodes inspectable and rebuildable #20 changes

The nice-to-haves can land in any order.

[NoahKusaba]

cc @mbutrovich

At your suggestion I put together an EPIC to track unfinished work for distributed datafusion execution, which is relevant for both ballista and datafusion-distributed.

[NoahKusaba]

cc @andygrove and @milenkovicm

[NoahKusaba]

cc @alexanderbianchi --> I know you represent the interested party from datafusion-distributed .

[NoahKusaba]

As mbutrovich has stated there is more review bandwidth for iceberg-rust PR's, I'm going to first focus on the "idemopotent key change on fast_appends", as it closes the only gap in correct writes and the apache/iceberg-rust#3297 we reviewed, as it's mostly done.

On the datafusion-iceberg side, I'm going to wait on PR #20 to merge while I'm doing the iceberg-rust stuff, and then start opening the remaining PR's that stack on top of it.

I've also messaged @toutane on slack about re-opening his partitioned reads PR on this repo, and I'll look to re-open it myself by end-of-october as it's a major blocker.

[alexanderbianchi]

Thanks for the write-up! A lot of these primitives overlap with what DFD needs, which is great. In DFD, we’ll maintain our own DataSourceExec integration because it’s built on the WorkUnitFeeds abstraction. Ideally, we can reuse upstream for everything else.

As these APIs become public, I’d like to make sure they also accommodate DFD’s execution model—particularly running writers on workers and collecting their results for a coordinator-side commit.

Breaking this into sub-issues would help; I know a few engineers looking to pick up tickets, including myself.

I can ask @toutane whether he wants to move apache/iceberg-rust#2671 here or whether I should carry it forward—we work together.