From 820452a87ccaadf3abef04eb35ade38f5b762272 Mon Sep 17 00:00:00 2001
From: Ritchie Vink <ritchie46@gmail.com>
Date: Sat, 8 Oct 2022 16:13:15 +0200
Subject: [PATCH] refactor(rust): relax sync requirement on Executor trait
 impls (#5142)

---
 .../src/physical_plan/executors/executor.rs       |  2 +-
 .../executors/groupby_partitioned.rs              | 15 ++++++++++++---
 2 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/polars/polars-lazy/src/physical_plan/executors/executor.rs b/polars/polars-lazy/src/physical_plan/executors/executor.rs
index 75524fc49cdb..5a13cd03ed15 100644
--- a/polars/polars-lazy/src/physical_plan/executors/executor.rs
+++ b/polars/polars-lazy/src/physical_plan/executors/executor.rs
@@ -7,7 +7,7 @@ use super::*;
 ///
 /// Executors have other executors as input. By having a tree of executors we can execute the
 /// physical plan until the last executor is evaluated.
-pub trait Executor: Send + Sync {
+pub trait Executor: Send {
     fn execute(&mut self, cache: &mut ExecutionState) -> PolarsResult<DataFrame>;
 }
 
diff --git a/polars/polars-lazy/src/physical_plan/executors/groupby_partitioned.rs b/polars/polars-lazy/src/physical_plan/executors/groupby_partitioned.rs
index f2ddfb0e4d55..0822d13d3c2c 100644
--- a/polars/polars-lazy/src/physical_plan/executors/groupby_partitioned.rs
+++ b/polars/polars-lazy/src/physical_plan/executors/groupby_partitioned.rs
@@ -34,10 +34,18 @@ impl PartitionGroupByExec {
     }
 
     fn keys(&self, df: &DataFrame, state: &ExecutionState) -> PolarsResult<Vec<Series>> {
-        self.keys.iter().map(|s| s.evaluate(df, state)).collect()
+        compute_keys(&self.keys, df, state)
     }
 }
 
+fn compute_keys(
+    keys: &[Arc<dyn PhysicalExpr>],
+    df: &DataFrame,
+    state: &ExecutionState,
+) -> PolarsResult<Vec<Series>> {
+    keys.iter().map(|s| s.evaluate(df, state)).collect()
+}
+
 fn run_partitions(
     df: &mut DataFrame,
     exec: &PartitionGroupByExec,
@@ -50,11 +58,12 @@ fn run_partitions(
     // split on several threads. Than the final result we apply the same groupby again.
     let dfs = split_df(df, n_threads)?;
 
+    let phys_aggs = &exec.phys_aggs;
+    let keys = &exec.keys;
     POOL.install(|| {
         dfs.into_par_iter()
             .map(|df| {
-                let keys = exec.keys(&df, state)?;
-                let phys_aggs = &exec.phys_aggs;
+                let keys = compute_keys(keys, &df, state)?;
                 let gb = df.groupby_with_series(keys, false, maintain_order)?;
                 let groups = gb.get_groups();