feat(datafusion): remove needs_repartitioning since DataFusion will use repartition one single time

Signed-off-by: Florian Valeye <[email protected]>

Author: fvaleye

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

@@ -96,45 +96,12 @@ pub(crate) fn repartition(
     let partitioning_strategy =
         determine_partitioning_strategy(&input, &table_metadata, target_partitions)?;
 
-    if !needs_repartitioning(&input, &partitioning_strategy) {
-        return Ok(input);
-    }
-
     Ok(Arc::new(RepartitionExec::try_new(
         input,
         partitioning_strategy,
     )?))
 }
 
-/// Returns whether repartitioning is actually needed by comparing input and desired partitioning
-fn needs_repartitioning(input: &Arc<dyn ExecutionPlan>, desired: &Partitioning) -> bool {
-    let input_partitioning = input.properties().output_partitioning();
-    match (input_partitioning, desired) {
-        (Partitioning::RoundRobinBatch(a), Partitioning::RoundRobinBatch(b)) => a != b,
-        (Partitioning::Hash(a_exprs, a_n), Partitioning::Hash(b_exprs, b_n)) => {
-            a_n != b_n || !same_columns(a_exprs, b_exprs)
-        }
-        _ => true,
-    }
-}
-
-/// Helper function to check if two sets of column expressions are the same
-fn same_columns(a_exprs: &[Arc<dyn PhysicalExpr>], b_exprs: &[Arc<dyn PhysicalExpr>]) -> bool {
-    if a_exprs.len() != b_exprs.len() {
-        return false;
-    }
-    a_exprs.iter().zip(b_exprs.iter()).all(|(a, b)| {
-        if let (Some(a_col), Some(b_col)) = (
-            a.as_any().downcast_ref::<Column>(),
-            b.as_any().downcast_ref::<Column>(),
-        ) {
-            a_col.name() == b_col.name() && a_col.index() == b_col.index()
-        } else {
-            std::ptr::eq(a.as_ref(), b.as_ref())
-        }
-    })
-}
-
 /// Determine the optimal partitioning strategy based on table metadata.
 ///
 /// Analyzes the table's partition specification to select the most appropriate