Remove EliminateAggDistinct optimizer rule, implement distinct aggs in Python

dask_planner/src/expression.rs

@@ -711,6 +711,25 @@ impl PyExpr {
         }
     }
 
+    #[pyo3(name = "isDistinctAgg")]
+    pub fn is_distinct_aggregation(&self) -> PyResult<bool> {
+        // TODO refactor to avoid duplication
+        match &self.expr {
+            Expr::AggregateFunction { distinct, .. } => Ok(*distinct),
+            Expr::AggregateUDF { .. } => Ok(false),
+            Expr::Alias(expr, _) => match expr.as_ref() {
+                Expr::AggregateFunction { distinct, .. } => Ok(*distinct),
+                Expr::AggregateUDF { .. } => Ok(false),
+                _ => Err(py_type_err(
+                    "isDistinctAgg() - Non-aggregate expression encountered",
+                )),
+            },
+            _ => Err(py_type_err(
+                "getFilterExpr() - Non-aggregate expression encountered",
+            )),
+        }
+    }
+
     /// Returns if a sort expressions is an ascending sort
     #[pyo3(name = "isSortAscending")]
     pub fn is_sort_ascending(&self) -> PyResult<bool> {

dask_planner/src/sql/optimizer.rs

@@ -27,9 +27,6 @@ use datafusion_optimizer::{
 };
 use log::trace;
 
-mod eliminate_agg_distinct;
-use eliminate_agg_distinct::EliminateAggDistinct;
-
 /// Houses the optimization logic for Dask-SQL. This optimization controls the optimizations
 /// and their ordering in regards to their impact on the underlying `LogicalPlan` instance
 pub struct DaskSqlOptimizer {
@@ -65,7 +62,6 @@ impl DaskSqlOptimizer {
             Arc::new(PushDownFilter::new()),
             Arc::new(LimitPushDown::new()),
             // Dask-SQL specific optimizations
-            Arc::new(EliminateAggDistinct::new()),
             // The previous optimizations added expressions and projections,
             // that might benefit from the following rules
             Arc::new(SimplifyExpressions::new()),