allow literal expr in window functions (#1855)

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

@@ -224,6 +224,20 @@ impl<'a> AggregationContext<'a> {
         }
         match &self.series {
             AggState::NotAggregated(s) => {
+                // literal series
+                // the literal series needs to be expanded to the number of indices in the groups
+                let s = if s.len() == 1
+                    // or more then one group
+                    && (self.groups.len() > 1
+                    // or single groups with more than on index
+                    || !self.groups.as_ref().is_empty()
+                    && self.groups[0].1.len() > 1)
+                {
+                    Cow::Owned(s.expand_at_index(0, self.groups.iter().map(|g| g.1.len()).sum()))
+                } else {
+                    Cow::Borrowed(s)
+                };
+
                 let out = Cow::Owned(
                     s.agg_list(&self.groups)
                         .expect("should be able to aggregate this to list"),

polars/polars-lazy/src/physical_plan/planner.rs

@@ -472,9 +472,7 @@ impl DefaultPlanner {
 
                 if apply_columns.is_empty() {
                     if has_aexpr(function, expr_arena, |e| matches!(e, AExpr::Literal(_))) {
-                        return Err(PolarsError::ValueError(
-                            "Cannot apply a window function over literals".into(),
-                        ));
+                        apply_columns.push(Arc::from("literal"))
                     } else {
                         let e = node_to_exp(function, expr_arena);
                         return Err(PolarsError::ValueError(

polars/polars-lazy/src/test.rs

@@ -2220,3 +2220,26 @@ fn test_single_group_result() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_literal_window_fn() -> Result<()> {
+    let df = df![
+        "chars" => ["a", "a", "b"]
+    ]?;
+
+    let out = df
+        .lazy()
+        .select([lit(1).cumsum(false).over([col("chars")]).alias("foo")])
+        .collect()?;
+
+    let out = out.column("foo")?;
+    assert!(matches!(out.dtype(), DataType::List(_)));
+    let flat = out.explode()?;
+    let flat = flat.i32()?;
+    assert_eq!(
+        Vec::from(flat),
+        &[Some(1), Some(2), Some(1), Some(2), Some(1)]
+    );
+
+    Ok(())
+}