feat(datafusion): implement repartition node for DataFusion, setting the best partition strategy for Iceberg for writing

- Implement hash partitioning for partitioned/bucketed tables
- Use round-robin partitioning for unpartitioned tables
- Support range distribution mode approximation via sort columns

Author: fvaleye

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 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

@@ -18,9 +18,11 @@
 pub(crate) mod commit;
 pub(crate) mod expr_to_predicate;
 pub(crate) mod metadata_scan;
+pub(crate) mod repartition;
 pub(crate) mod scan;
 pub(crate) mod write;
 
 pub(crate) const DATA_FILES_COL_NAME: &str = "data_files";
 
+pub use repartition::IcebergRepartitionExec;
 pub use scan::IcebergTableScan;