[EPIC] Distributed Execution for Reads/Writes

Start with the merge order and #20: inspect IcebergCommitExec, IcebergWriteExec, IcebergMetadataScan, IcebergTableScan, IcebergTableProvider::new_with_schema, and repartition.rs. Use apache/datafusion-ballista#2217 and the listed iceberg-rust issues to identify dependencies. Done means the selected distributed execution changes are implemented and compatible with Ballista's scheduler and executors.

Goal

Run Iceberg reads and writes in parallel on
Ballista 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
datafusion-iceberg and iceberg-rust changes that make that work well.

• apache/datafusion-ballista#2638 (ballista-iceberg, reads only) builds on
  this repo's main and needs nothing below to merge.
• apache/datafusion-ballista#2217 is the full read/write integration. Commits
  are moving to the Ballista scheduler (apache/datafusion-ballista#2622), where
  each commit runs once.

Most of this is likely relevant to datafusion-distributed too, though it's
written from Ballista's perspective. datafusion-distributed keeps its own
DataSourceExec integration (built on WorkUnitFeeds), so it needs the
planned file tasks to be reachable, but not necessarily the grouping below.
If I've missed anything, let me know and I'll update this epic.

Last updated 2026-11-03.

Open

Partitioned scans (highest priority)

IcebergDataSource 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 INSERT INTO t SELECT … FROM <iceberg table> into a table with no identity or bucket partition field writes
from a single task.

• #67 (@alexanderbianchi): opt-in eager file planning
  (iceberg.eager_file_planning) that groups the planned file tasks into up
  to target_partitions partitions. Carries apache/iceberg-rust#2671
  forward. For Ballista it also needs:
  • The node exposes its task groups and has a constructor that takes them,
    like #20's nodes. Files must be planned once, on the scheduler, and
    shipped to executors: re-planning on an executor could produce a
    different grouping, since planning order isn't deterministic.
  • No dependency on an iceberg-rust accessor that main doesn't have, such
    as #2671's TableScan::arrow_reader_builder().
  • The opt-in setting reaches the Ballista scheduler, or ballista-iceberg
    turns it on.
  • A scan with no files reports one partition, not zero.
  • One eager file listing shared with #55 (sort-order reporting), which
    also lists data files in TableProvider::scan.

Writes

• #35 Partitioned INSERT computes partition values from the wrong columns
  when the input ends in a projection. PR: #92
• #38 INSERT with fanout disabled always fails: IcebergWriteExec declares
  no required input ordering or distribution, so the optimizer removes the
  partition sort. A required hash distribution on _partition would also
  fix the next item.
• Hash-partition INSERT input for every partition transform. repartition.rs
  hashes on _partition only for identity or bucket fields and otherwise
  round-robins, which Ballista drops, so writers produce many small files.
• #39 Fanout-disabled writes fail on sorted input (partitions emitted in
  HashMap order). Needed on top of #38. PR: #81
• #19 Accept NOT NULL input columns in partitioned INSERT (closes #27).
  • #22 follow-up: unpartitioned tables write NULLs into required columns.

Scans

• #44 convert_filters_to_predicate negates a weakened AND and drops
  matching rows. PR: #59
• #42 SELECT * fails after a column is added, until the next write. Fixed
  in iceberg-rust by apache/iceberg-rust#2904; needs a revision bump here.
• Scan statistics: with eager planning, file record counts give row counts
  nearly for free (an earlier attempt, apache/iceberg-rust#880, went stale).
• Pass the session's batch_size to the reader, and bound per-partition
  file-read concurrency once a scan has several partitions per executor.
• #58 Expose Parquet read-ahead controls (needs apache/iceberg-rust#3395).

Catalogs and cached schemas

• #24 Query-scoped catalog resolution (@alexanderbianchi). Providers and
  their schemas are cached for the catalog provider's lifetime, the cause of
  #56 and #57, and a problem for a long-lived scheduler.
• #56 INSERT fails after a column is added, until the provider is rebuilt.
• #57 Wrong results and panics after a column is dropped and re-added under
  the same name.
• #37 CREATE TABLE / DROP TABLE hang on a current-thread runtime and
  panic outside one. PR: #45

iceberg-rust (not opened yet, most important first)

• 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.
• To investigate: whether FileIO can refresh catalog-vended storage
  credentials during a long job.

Done

• #14 PartitionExpr reconstructible (closes #13)
• #20 Plan nodes inspectable and rebuildable (closes #28)
• #63 IcebergDataSource builder, replacing new_with_predicate
• #61 Crate renamed back to iceberg-datafusion (closes #30)
• #54 Data files passed from write to commit as Avro instead of JSON
• #65 Forward enable_page_index to the scan
• #91 Filter pushdown keeps casts (closes #36)
• #46 LIKE prefix pushdown handles escapes (closes #40)
• #47 Metadata tables honour the scan projection (closes #41)
• apache/iceberg-rust#3286 with_runtime on the catalog loader's
  BoxedCatalogBuilder
• apache/iceberg-rust#3303 strip_metadata_from_schema handles list and map
  columns
• apache/iceberg-rust#3406 Table::with_runtime, keeping the manifest cache
• apache/iceberg-rust#2904 Scans project the current schema after a schema
  change

Dropped

• #48 / #80 IcebergCatalogTableProvider::new_with_schema and its replacement:
  apache/datafusion-ballista#2638 pins catalog-backed tables when the query is
  planned and sends their metadata, so no provider is rebuilt from a schema.
• Idempotent commits, and apache/iceberg-rust#3372 (idempotency key on
  fast_append): commits are moving to the Ballista scheduler, where each runs
  once.