fix(pyspark): handle predicates in Aggregation

ibis-project · Jun 13, 2022 · 842c307 · 842c307
1 parent 1509b1d
commit 842c307
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/ibis/backends/pyspark/compiler.py b/ibis/backends/pyspark/compiler.py
@@ -395,6 +395,15 @@ def compile_aggregation(t, expr, scope, timecontext, **kwargs):
 
     src_table = t.translate(op.table, scope, timecontext)
 
+    if op.predicates:
+        src_table = src_table.filter(
+            t.translate(
+                functools.reduce(operator.and_, op.predicates),
+                scope,
+                timecontext,
+            )
+        )
+
     if op.by:
         context = AggregationContext.GROUP
         aggs = [