[EPIC] Iceberg integration

[gabotechs]

This is a collection of issues and central discussion for building a native integration to Iceberg inside this project.

The reason why having a native integration here is because:

• Serve as a testing use-case for this project. During CI, tests against real Iceberg tables using distributed execution will run, increasing coverage of the core of datafusion-distributed itself.
• Guide design decisions regarding public APIs. Right now, the only real use-case present in this project is raw parquet tables stored in a file system, but that use-case is relatively simple and misses on features like dynamically resolved files during execution. Having a more complex use-case like Iceberg will help shape the project's APIs.
• Lack of basic functionality in upstream integration https://github.com/apache/iceberg-rust/tree/main/crates/integrations/datafusion, not enough review power there (e.g., feat!(datafusion): enable parallel file scanning with eager task bucketing apache/iceberg-rust#2298, feat(datafusion): report IcebergTableScan metrics apache/iceberg-rust#2521, Report DataFusion operator metrics in IcebergTableScan apache/iceberg-rust#2364), and maintainers explicitly stating their willingness to not maintain a DataFusion integration there long term (e.g., feat(datafusion-ballista) Add Initial Integration for Ballisa Datafusion apache/iceberg-rust#2613 (comment))

The extension points in this project seem to fit very well for providing a good distributed integration with DataFusion and Iceberg, but the task is big and will need to be broken down in several sub issues that people can contribute to independently:

• [Iceberg] Establish a minimal e2e-tested read integration #600
• [Iceberg] Add local benchmarks #601
• [Iceberg] Add remote benchmarks #602
• [Iceberg] Serialize IcebergDataSource plans for distributed execution #603
• [Iceberg] Serialize FileScanTask work units #604
• [Iceberg] Estimate desired task counts from table metadata #605
• [Iceberg] Support partitioning in work-unit feeds #606
• [Iceberg] Populate planning statistics from table metadata #608
• [Iceberg] Support physical filter pushdown #609
• [Iceberg] Support sort-order pushdown #610
• [Iceberg] Add catalog and schema providers #611
• [Iceberg] Support CREATE TABLE and DROP TABLE through Iceberg catalogs #612
• [Iceberg] Support Iceberg metadata tables #613
• [Iceberg] Plan append inserts into catalog-backed tables #614
• [Iceberg] Write data files for append inserts #615
• [Iceberg] Commit append transactions #616

[gabotechs]

Here is some discussion around moving the Iceberg integration out of iceberg-rust into a separate crate:

• Moving the DataFusion integration into a separate repository apache/iceberg-rust#2992

Regardless of the outcome on that this discussion, we still need the following items to be completed in this project one way or another:

• [Iceberg] Establish a minimal e2e-tested read integration #600 <- yes, this one is still needed because of the integration with WorkUnitFeeds
• [Iceberg] Add local benchmarks #601
• [Iceberg] Add remote benchmarks #602
• [Iceberg] Serialize IcebergDataSource plans for distributed execution #603
• [Iceberg] Serialize FileScanTask work units #604
• [Iceberg] Estimate desired task counts from table metadata #605
• [Iceberg] Support partitioning in work-unit feeds #606
• [Iceberg] Populate planning statistics from table metadata #608

So we can start shipping those independently

[sandugood]

By the way, @gabotechs
Will we continue working on iceberg integration in this repo in a separate branch, or adopt the new repo, according to apache/iceberg-rust#3029

[gabotechs]

We'll start adopting the new repo for several things unrelated to execution, like planning for example.

However, we still need to maintain an integration here in this project, as execution is going to be different (this project will rely on WorkUnitFeeds).

What this implies to this project though, is that at least this set of issues are likely not going to be delivered here, and they will be delegated to upstream code:

• [Iceberg] Support physical filter pushdown #609
• [Iceberg] Support sort-order pushdown #610
• [Iceberg] Add catalog and schema providers #611
• [Iceberg] Support CREATE TABLE and DROP TABLE through Iceberg catalogs #612
• [Iceberg] Support Iceberg metadata tables #613
• [Iceberg] Plan append inserts into catalog-backed tables #614
• [Iceberg] Write data files for append inserts #615
• [Iceberg] Commit append transactions #616