feat(datafusion): parallel file scanning with eager task bucketing (#43)

Port of apache#2298 onto the spiceai-0.9.0 fork.

IcebergTableProvider::scan() now plans files eagerly and distributes
FileScanTasks into min(target_partitions, n_files) buckets, one bucket
per DataFusion partition, so file reads are scheduled concurrently
instead of streaming through a single UnknownPartitioning(1) partition.
When the table is identity-partitioned (single spec, supported column
types, partition columns projected) the scan declares Partitioning::Hash
so downstream joins/aggregates can skip a RepartitionExec.

- TableScan::to_arrow_from_tasks: replay pre-collected FileScanTasks
  through the Arrow reader; preserves the spice fork's ArrowReaderBuilder
  (file_io, runtime) signature and row-selection config.
- IcebergTableScan gains new_with_tasks (eager) alongside new (lazy,
  used by IcebergStaticTableProvider); execute(i) streams buckets[i].
  Constructors made pub; with_new_children now errors on children.
- New table/bucketing.rs: identity-hash bucketing via
  REPARTITION_RANDOM_STATE + create_hashes, fallback to data_file_path.
- Spice limit pushdown preserved: with_limit threaded into the planning
  builder and build_table_scan.
- Drop the unused convert_filters_to_predicate re-export.

Author: phillipleblanc

crates/iceberg/src/scan/mod.rs

@@ -472,6 +472,21 @@ impl TableScan {
 
     /// Returns an [`ArrowRecordBatchStream`].
     pub async fn to_arrow(&self) -> Result<ArrowRecordBatchStream> {
+        self.to_arrow_from_tasks(self.plan_files().await?)
+    }
+
+    /// Like [`TableScan::to_arrow`], but accepts a caller-supplied
+    /// [`FileScanTask`] stream instead of running [`TableScan::plan_files`]
+    /// internally.
+    ///
+    /// # Correctness
+    ///
+    /// Tasks must come from a [`TableScan`] with the same projection and
+    /// filter as `self`: predicates are baked into each task at planning
+    /// time and are not re-applied here. Reader-side configuration
+    /// (concurrency, batch size, row-group filtering, row selection) is
+    /// taken from `self` and may differ from the planning scan.
+    pub fn to_arrow_from_tasks(&self, tasks: FileScanTaskStream) -> Result<ArrowRecordBatchStream> {
         let mut arrow_reader_builder =
             ArrowReaderBuilder::new(self.file_io.clone(), self.runtime.clone())
                 .with_data_file_concurrency_limit(self.concurrency_limit_data_files)
@@ -484,7 +499,7 @@ impl TableScan {
 
         arrow_reader_builder
             .build()
-            .read(self.plan_files().await?)
+            .read(tasks)
             .map(|result| result.stream())
     }
 

crates/integrations/datafusion/src/physical_plan/mod.rs

@@ -26,6 +26,5 @@ pub(crate) mod write;
 
 pub(crate) const DATA_FILES_COL_NAME: &str = "data_files";
 
-pub use expr_to_predicate::convert_filters_to_predicate;
 pub use project::project_with_partition;
 pub use scan::IcebergTableScan;