feat(datafusion): Add opt-in eager file scan planning with output partitioning

[toutane]

Which issue does this PR close?

• Closes Enable parallel file-level scanning for IcebergTableScan Datafusion Integration #2220
• Related to
  • EPIC: Support parallel scan in iceberg-datafusion #1604
  • Plan file scan task according scan file size. #128
  • feat!(datafusion): enable parallel file scanning with eager task bucketing #2298

What changes are included in this PR?

This PR adds an opt-in eager scan planning path for the DataFusion integration.

When iceberg.enable_eager_scan_planning is enabled, IcebergTableProvider::scan() plans FileScanTasks during physical planning, groups them across DataFusion target_partitions, and exposes the resulting output partition count as UnknownPartitioning(N). Each execute(partition) call then reads only the task group assigned to that output partition through ArrowReaderBuilder.

The default behavior is unchanged: eager planning is disabled by default, so scans keep the existing lazy single-partition planning path unless explicitly enabled.

This is a narrower split from #2298, focused on file-level eager planning and output partition count reporting. Some of the broader design discussions and review feedback happened in #2298; this PR keeps only the first scoped step and intentionally does not include hash partitioning, size-aware bin-packing, or row-group/sub-file planning.

Trade-offs:

• Eager planning does catalog/metadata work during TableProvider::scan(), so it is kept opt-in
• Task grouping is round-robin and count-based, not size-aware. This is simple and deterministic, but can be imbalanced when file sizes vary
• Parallelism is at the FileScanTask level only. A table with one large file will not benefit from this change
• The scan reports UnknownPartitioning(N), not hash partitioning. This exposes the number of output partitions without claiming stronger partitioning semantics

Follow-up work:

• Add size-aware task bin-packing using file_size_in_bytes (Plan file scan task according scan file size. #128)
• Add sub-file / row-group level parallelism as the longer-term direction (EPIC: Support parallel scan in iceberg-datafusion #1604)
• Cache planned tasks for repeated scans with the same snapshot, projection, filter, and target partition count
• Investigate stronger output partitioning declarations for safe identity-partitioned cases
• Add benchmarks for multi-file tables and skewed file-size distributions

Are these changes tested?

Yes. Integration tests cover:

• eager scan planning disabled by default
• enabling eager scan planning through iceberg.enable_eager_scan_planning
• exposing multiple scan output partitions when eager planning is enabled
• preserving query results between single-partition and multi-partition scans

[toutane]

👋 cc @mbutrovich @timsaucer

[mbutrovich]

First pass, thanks @toutane!

[mbutrovich]

execute's partition argument is read at line 146 (file_task_groups.get(_partition)) and interpolated into the error message at line 148 (partition {_partition} does not exist). The leading underscore signals "unused" to readers and lint, and it leaks into user-facing error text. Rename to partition. (_context at line 140 is still legitimately unused; leave it.)

[mbutrovich]

Both IcebergTableProvider::scan (:158, used at :178) and IcebergStaticTableProvider::scan (:343, used at :352) pass _state into create_scan_plan, which reads it via enable_eager_scan_planning(state). Rename both to state. Note: _state at :365 (insert_into is genuinely unused; leave it.)

[mbutrovich]

When target_partitions > tasks.len() (which the tests deliberately hit with data_file_count + 1), this allocates extra empty vecs and then walks the whole vec to remove them. The result is always min(target_partitions, tasks.len()) non-empty groups, so clamp up front and the retain disappears:

// tasks is non-empty here (early return above handles the empty case)
let target_partitions = target_partitions.max(1).min(tasks.len());

let mut groups: Vec<Vec<FileScanTask>> = vec![Vec::new(); target_partitions];
for (i, task) in tasks.into_iter().enumerate() {
    groups[i % target_partitions].push(task);
}

groups

[mbutrovich]

task_count is used once; inline it as (0..file_task_group.len()). Heads-up for anyone tempted to go further: file_task_group.iter().cloned().map(Ok) does not compile here, because the stream must be 'static and iter() would borrow the local Arc<[FileScanTask]>. The move-plus-index form is the allocation-free way to clone lazily out of the Arc, so keep that shape; just remove the extra binding.

[toutane]

Thanks @mbutrovich, addressed all four comments in d99603f