quantile agg; update grouptuples (#3252)

polars/polars-lazy/src/physical_plan/expressions/aggregation.rs

@@ -306,15 +306,18 @@ impl PhysicalAggregation for AggQuantileExpr {
         groups: &GroupsProxy,
         state: &ExecutionState,
     ) -> Result<Option<Series>> {
-        let series = self.expr.evaluate(df, state)?;
-        let new_name = series.name().to_string();
-        let opt_agg = series.agg_quantile(groups, self.quantile, self.interpol);
+        let mut ac = self.expr.evaluate_on_groups(df, groups, state)?;
+        // don't change names by aggregations as is done in polars-core
+        let keep_name = ac.series().name().to_string();
 
+        let opt_agg =
+            ac.flat_naive()
+                .into_owned()
+                .agg_quantile(ac.groups(), self.quantile, self.interpol);
         let opt_agg = opt_agg.map(|mut agg| {
-            agg.rename(&new_name);
+            agg.rename(&keep_name);
             agg.into_series()
         });
-
         Ok(opt_agg)
     }
 }

py-polars/tests/test_lazy.py

@@ -1223,3 +1223,18 @@ def test_group_lengths() -> None:
             )
         )
     )
+
+
+def test_quantile_filtered_agg() -> None:
+    assert (
+        pl.DataFrame(
+            {
+                "group": [0, 0, 0, 0, 1, 1, 1, 1],
+                "value": [1, 2, 3, 4, 1, 2, 3, 4],
+            }
+        )
+        .groupby("group")
+        .agg(pl.col("value").filter(pl.col("value") < 2).quantile(0.5))["value"]
+        .to_list()
+        == [1.0, 1.0]
+    )