feat(datafusion): implement the partitioning node for DataFusion to define the partitioning

crates/iceberg/src/table.rs

@@ -24,7 +24,7 @@ use crate::inspect::MetadataTable;
 use crate::io::FileIO;
 use crate::io::object_cache::ObjectCache;
 use crate::scan::TableScanBuilder;
-use crate::spec::{TableMetadata, TableMetadataRef};
+use crate::spec::{SchemaRef, TableMetadata, TableMetadataRef};
 use crate::{Error, ErrorKind, Result, TableIdent};
 
 /// Builder to create table scan.
@@ -235,6 +235,11 @@ impl Table {
         self.readonly
     }
 
+    /// Returns the current schema as a shared reference.
+    pub fn current_schema_ref(&self) -> SchemaRef {
+        self.metadata.current_schema().clone()
+    }
+
     /// Create a reader for the table.
     pub fn reader_builder(&self) -> ArrowReaderBuilder {
         ArrowReaderBuilder::new(self.file_io.clone())

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

@@ -19,6 +19,7 @@ pub(crate) mod commit;
 pub(crate) mod expr_to_predicate;
 pub(crate) mod metadata_scan;
 pub(crate) mod project;
+pub(crate) mod repartition;
 pub(crate) mod scan;
 pub(crate) mod write;
 

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

@@ -32,6 +32,9 @@ use iceberg::table::Table;
 
 use crate::to_datafusion_error;
 
+/// Column name for the combined partition values struct
+pub(crate) const PARTITION_VALUES_COLUMN: &str = "_partition";
+
 /// Extends an ExecutionPlan with partition value calculations for Iceberg tables.
 ///
 /// This function takes an input ExecutionPlan and extends it with an additional column