[EPIC] Distributed Execution for Reads/Writes

未关闭
#23 5 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

还没有人认领这个 Issue。

评估

难度
5/5
预计耗时
一周以上
新手友好度
35/100
Issue 类型
功能
描述清晰度
基本清楚
活跃度
活跃
技术栈
rust

调研方向

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.

由索引模型根据 Issue 内容生成。

描述

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.
主要语言
Rust
星标
33
派生
34
平均合并
5 天 18 小时
30 天内合并 PR
5

环境准备

这个项目没有提供开发容器、Dockerfile 或贡献指南,环境需要你自己搭建:先看它的 README,通用步骤见我们的新手贡献指南。

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

apache/datafusion-iceberg 的其他 Issue

查看 apache/datafusion-iceberg 的全部 Issue

相似的 Issue

更多 Rust Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。