[EPIC] Distributed Execution for Reads/Writes #23

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.

• feat(iceberg): scan Iceberg tables in Ballista datafusion-ballista#2638 (ballista-iceberg, reads only) builds on
  this repo's main and needs nothing below to merge.
• feat: Add Iceberg Support datafusion-ballista#2217 is the full read/write integration. Commits
  are moving to the Ballista scheduler ([Iceberg] Run a job's final commit step on the scheduler, not an executor 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.

• feat: plan Iceberg file tasks before execution #67 (@alexanderbianchi): opt-in eager file planning
  (iceberg.eager_file_planning) that groups the planned file tasks into up
  to target_partitions partitions. Carries feat(datafusion): Add opt-in eager file scan planning with output partitioning iceberg-rust#2671
  forward. For Ballista it also needs:
  • 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. 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 feat: report the sort order of sorted Iceberg scans to DataFusion #55 (sort-order reporting), which
    also lists data files in TableProvider::scan.

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 ends in a projection. PR: fix: read partition source columns through PartitionExpr children #92
• 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. 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.
• Clustered (fanout disabled) writes fail on sorted input because batches are split in HashMap order #39 Fanout-disabled writes fail on sorted input (partitions emitted in
  HashMap order). Needed on top of INSERT into a partitioned table with fanout disabled always fails: the optimizer removes the partition sort #38. PR: fix: write each batch's partitions in first-row order for the clustered writer #81
• fix: accept NOT NULL input columns in partitioned INSERT #19 Accept NOT NULL input columns in partitioned INSERT (closes INSERT into a partitioned table rejects NOT NULL source columns for optional table columns #27).
  • 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.

Scans

• convert_filters_to_predicate negates a weakened AND, producing a predicate that drops matching rows #44 convert_filters_to_predicate negates a weakened AND and drops
  matching rows. PR: fix: preserve filter semantics for casts and negation #59
• 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. Fixed
  in iceberg-rust by fix(scan): don't project a stale schema after a schema update with no write 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, feat(datafusion): Expose DataFusion statistics on an IcebergTableScan 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.
• Expose Parquet read-ahead controls for Iceberg SQL scans #58 Expose Parquet read-ahead controls (needs feat(scan): expose Parquet read-ahead controls iceberg-rust#3395).

Catalogs and cached schemas

• CatalogProvider Limitations: Query Scoped Resolution #24 Query-scoped catalog resolution (@alexanderbianchi). Providers and
  their schemas are cached for the catalog provider's lifetime, the cause of
  INSERT fails after a column is added to the table, until the provider is rebuilt #56 and Wrong results and panics after a column is dropped and added back under the same name #57, and a problem for a long-lived scheduler.
• INSERT fails after a column is added to the table, until the provider is rebuilt #56 INSERT fails after a column is added, until the provider is rebuilt.
• Wrong results and panics after a column is dropped and added back under the same name #57 Wrong results and panics after a column is dropped and re-added under
  the same name.
• 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 runtime and
  panic outside one. PR: fix: run sync catalog calls on a separate runtime #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

• feat: make PartitionExpr reconstructible #14 PartitionExpr reconstructible (closes PartitionExpr cannot be reconstructed from its public API #13)
• feat: make the plan nodes inspectable and rebuildable #20 Plan nodes inspectable and rebuildable (closes Iceberg plan nodes can't be rebuilt by a PhysicalExtensionCodec #28)
• refactor: replace IcebergDataSource::new_with_predicate with a builder #63 IcebergDataSource builder, replacing new_with_predicate
• chore: rename crate back to iceberg-datafusion and start at version 0.12.0 #61 Crate renamed back to iceberg-datafusion (closes Crate name datafusion-iceberg is taken on crates.io; switch back to iceberg-datafusion #30)
• Use avro for datafiles collection from write-exec #54 Data files passed from write to commit as Avro instead of JSON
• Enable page-index pruning for Iceberg scans #65 Forward enable_page_index to the scan
• fix: refuse to look inside casts in expr_to_predicate #91 Filter pushdown keeps casts (closes Filter pushdown strips casts on columns, which drops matching rows or fails valid queries #36)
• fix: unescape LIKE prefixes before pushing them down as STARTS WITH #46 LIKE prefix pushdown handles escapes (closes LIKE prefix pushdown ignores backslash escapes and drops matching rows #40)
• fix: project columns correctly when scanning iceberg table metadata #47 Metadata tables honour the scan projection (closes Metadata tables ignore the scan projection, so projected and aggregate queries return wrong columns or fail #41)
• feat(catalog-loader): add with_runtime to BoxedCatalogBuilder iceberg-rust#3286 with_runtime on the catalog loader's
  BoxedCatalogBuilder
• fix(arrow): strip metadata from nested fields and normalize container field names iceberg-rust#3303 strip_metadata_from_schema handles list and map
  columns
• feat(table): add Table::with_runtime that keeps the manifest cache iceberg-rust#3406 Table::with_runtime, keeping the manifest cache
• fix(scan): don't project a stale schema after a schema update with no write iceberg-rust#2904 Scans project the current schema after a schema
  change

Dropped

• feat: add IcebergCatalogTableProvider::new_with_schema #48 / refactor: derive the catalog provider's schema from its table #80 IcebergCatalogTableProvider::new_with_schema and its replacement:
  feat(iceberg): scan Iceberg tables in Ballista 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 feat(transaction): add idempotency key to fast append iceberg-rust#3372 (idempotency key on
  fast_append): commits are moving to the Ballista scheduler, where each runs
  once.