fix window expression case (#3937)

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

@@ -332,9 +332,18 @@ impl PhysicalExpr for BinaryExpr {
             // Both are or a flat series
             // so we can flatten the Series and apply the operators
             _ => {
+                // Check if the group state of `ac_a` differs from the original `GroupTuples`.
+                // If this is the case we might need to align the groups. But only if `ac_b` is not a
+                // `Literal` as literals don't have any groups, the changed group order does not matter
+                // in that case
+                let different_group_state =
+                    |ac_a: &AggregationContext, ac_b: &AggregationContext| {
+                        (ac_a.update_groups != UpdateGroups::No)
+                            && !matches!(ac_b.agg_state(), AggState::Literal(_))
+                    };
+
                 // the groups state differs, so we aggregate both and flatten again to make them align
-                if ac_l.update_groups != UpdateGroups::No || ac_r.update_groups != UpdateGroups::No
-                {
+                if different_group_state(&ac_l, &ac_r) || different_group_state(&ac_r, &ac_l) {
                     // use the aggregated state to determine the new groups
                     let lhs = ac_l.aggregated();
                     ac_l.with_update_groups(UpdateGroups::WithSeriesLenOwned(lhs.clone()));

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

@@ -258,6 +258,10 @@ impl PhysicalExpr for WindowExpr {
             // and every partition run the cache should be empty so we expect a max of 1.
             debug_assert!(gt_map.len() <= 1);
             if let Some(gt) = gt_map.get_mut(&cache_key) {
+                if df.height() > 0 {
+                    assert!(!gt.is_empty());
+                };
+
                 // We take now, but it is important that we set this before we return!
                 // a next windows function may get this cached key and get an empty if this
                 // does not happen

polars/tests/it/lazy/expressions/arity.rs

@@ -242,3 +242,32 @@ fn test_null_commutativity() {
     assert_eq!(out.column("a").unwrap().get(0), AnyValue::Boolean(true));
     assert_eq!(out.column("b").unwrap().get(0), AnyValue::Boolean(true));
 }
+
+#[test]
+fn test_binary_over_3930() -> Result<()> {
+    let df = df![
+        "class" => ["a", "a", "a", "b", "b", "b"],
+        "score" => [0.2, 0.5, 0.1, 0.3, 0.4, 0.2]
+    ]?;
+
+    let ss = col("score").pow(2);
+    let mdiff = (ss.clone().shift(-1) - ss.shift(1)) / lit(2);
+    let out = df.lazy().select([mdiff.over([col("class")])]).collect()?;
+
+    let out = out.column("score")?;
+    let out = out.f64()?;
+
+    assert_eq!(
+        Vec::from(out),
+        &[
+            None,
+            Some(-0.015000000000000003),
+            None,
+            None,
+            Some(-0.024999999999999994),
+            None
+        ]
+    );
+
+    Ok(())
+}