[Iceberg] Add basic iceberg integration

[gabotechs]

Closes #600

Note

The PR is large because it contains a lot of boilerplate for a basic Iceberg integration, but it was crafted by hand, and it contains just whatever is minimal to get some tests e2e running

Adds a very basic Iceberg integration as a new crate that depends on datafusion, datafusion-distributed and iceberg-rust.

This is mostly a rewrite of https://github.com/apache/iceberg-rust/tree/main/crates/integrations/datafusion, with a few exceptions for helpers like expression transpilation.

It heavily relies on WorkUnitFeeds for streaming FileScanTasks at runtime, following a pattern that looks like this today in single-node:

┌────────────────────────────────────────────┐
│             IcebergDataSource              │
│                                            │
│┌──────────────────────────────────────────┐│
││           IcebergWorkUnitFeed            ││
││┌────────────┐┌────────────┐┌────────────┐││
│││   Feed 0   ││   Feed 1   ││   Feed 2   │││
││└──────┬─────┘└──────┬─────┘└──────┬─────┘││
│└───────┼─────────────┼─────────────┼──────┘│
│  .─────▼─────. .─────▼─────. .─────▼─────. │
│ (FileScanTask (FileScanTask (FileScanTask )│
│  .───────────. `─────┬─────' .───────────. │
│ (FileScanTask )      │      (FileScanTask )│
│  `─────┬─────'       │       .───────────. │
│        │             │      (FileScanTask )│
│        │             │       `─────┬─────' │
│        │             │             │       │
│ ┌──────▼─────┐┌──────▼─────┐┌──────▼─────┐ │
│ │Partition 0 ││Partition 1 ││Partition 2 │ │
│ │ArrowReader ││ArrowReader ││ArrowReader │ │
│ └──────┬─────┘└──────┬─────┘└──────┬─────┘ │
│        │             │             │       │
│  .─────▼─────.       │       .─────▼─────. │
│ ( RecordBatch ).─────▼─────.( RecordBatch )│
│  `─────┬─────'( RecordBatch ).───────────. │
│        │       `─────┬─────'( RecordBatch )│
│        │             │       `───────────' │
└────────┼─────────────┼─────────────┼───────┘
         ▼             ▼             ▼

And that will look like this once distribution is supported:

 ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
                                     Coordinating Context                                   │
 │
  ┌────────────────────────────────────────────────────────────────────────────────────────┐│
 ││                                  IcebergWorkUnitFeed                                   │
  │┌─────────────┐┌─────────────┐┌────────────┐┌────────────┐┌─────────────┐┌─────────────┐││
 │││   Feed 0    ││   Feed 1    ││   Feed 2   ││   Feed 3   ││   Feed 4    ││   Feed 5    ││
  │└──────┬──────┘└─────┬───────┘└────┬───────┘└───────┬────┘└───────┬─────┘└──────┬──────┘││
 └└───────┼─────────────┼─────────────┼────────────────┼─────────────┼─────────────┼───────┴
    .─────▼─────. .─────▼─────. .─────▼─────.    .─────▼─────. .─────▼─────. .─────▼─────.
   (FileScanTask (FileScanTask (FileScanTask )  (FileScanTask (FileScanTask (FileScanTask )
    .───────────. `─────┬─────' .───────────.    `─────┬─────' .───────────. `─────┬─────'
   (FileScanTask )      │      (FileScanTask )         │      (FileScanTask )      │
    `─────┬─────'       │       .───────────.          │       `───────────'       │
          │             │      (FileScanTask )         │             │             │
 Worker 0 │             │       `─────┬─────'          │             │             │ Worker 1
┌ ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─ ─ ┼ ─ ─ ─ ┐┌ ─ ─ ─ ┼ ─ ─ ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─ ─ ┼ ─ ─ ─ ─ ┐
  ┌───────┼─────────────┼─────────────┼───────┐┌───────┼─────────────┼─────────────┼───────┐
│ │       │     IcebergD│taSource     │       ││       │     IcebergD│taSource     │       │ │
  │       │             │             │       ││       │             │             │       │
│ │┌──────▼─────┐┌──────▼─────┐┌──────▼─────┐ ││┌──────▼─────┐┌──────▼─────┐┌──────▼─────┐ │ │
  ││Partition 0 ││Partition 1 ││Partition 2 │ │││Partition 0 ││Partition 1 ││Partition 2 │ │
│ ││ArrowReader ││ArrowReader ││ArrowReader │ │││ArrowReader ││ArrowReader ││ArrowReader │ │ │
  │└──────┬─────┘└──────┬─────┘└──────┬─────┘ ││└──────┬─────┘└──────┬─────┘└──────┬─────┘ │
│ │       │             │             │       ││       │             │             │       │ │
  │ .─────▼─────.       │       .─────▼─────. ││       │             ▼             ▼       │
│ │( RecordBatch ).─────▼─────.( RecordBatch )││ .─────▼─────. .───────────. .───────────. │ │
  │ `─────┬─────'( RecordBatch ).───────────. ││( RecordBatch ( RecordBatch ) RecordBatch )│
│ │       │       `─────┬─────'( RecordBatch )││ `─────┬─────' `───────────' `─────┬─────' │ │
  │       │             │       `───────────' ││       │      ( RecordBatch )      │       │
│ │       │             │             │       ││       │       `─────┬─────'       │       │ │
  └───────┼─────────────┼─────────────┼───────┘└───────┼─────────────┼─────────────┼───────┘
│         ▼             ▼             ▼       ││       ▼             ▼             ▼         │
 ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─  ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─

However, this PR just adds a very basic integration that includes:

• Basic plumbing for TableProvider and TableProviderFactory implementations, very similar to what https://github.com/apache/iceberg-rust/tree/main/crates/integrations/datafusion has today on this front
• An IcebergWorkUnitFeed implementation that streams FileScanTasks messages at execution time as new files are discovered
• An IcebergDataSource that reads the FileScanTask streams, executing the tasks and yielding Arrow RecordBatches.
• Integration tests, for existing functionality and future one.

It's still laking:

• Any kind of distribution
• Any kind of [de]serialization for FileScanTasks and IcebergDataSources
• No filter pushdowns
• No sort pushdowns
• No benchmarks
• No sense of data partitioning

---

_{Stack created with GitHub Stacks CLI • Give Feedback 💬}

[shinzoxD]

Two blockers at this head:

1. The new Iceberg crate is not exercised by required CI. It is a workspace member but not a default member, while the unit job runs only cargo test --features integration; the current job log contains none of the Iceberg test targets. Please add cargo test -p datafusion-distributed-iceberg (or an appropriate workspace command) to required CI so this integration's tests actually run.

2. set_iceberg_integration always registers IcebergCodec, but both PhysicalExtensionCodec methods and every FileScanTaskMessage prost::Message method are unimplemented!(). The composed codec retries only normal Err results, so an Iceberg serialization attempt panics; it can also prevent a subsequently registered user codec from being reached. Since #600 explicitly scopes this first version to single-node execution, please omit codec registration and the distributed-execution claims until serialization exists, or make unsupported paths return structured errors and cover them with tests.

[sandugood]

Lgtm, looks nice as an initial integration.
Thanks @gabotechs