fix: Don't stackoverflow on all/any horizontal (#16287)

crates/polars-plan/src/dsl/function_expr/boolean.rs

@@ -1,9 +1,12 @@
+use std::ops::{BitAnd, BitOr};
+
+use polars_core::POOL;
+use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
+
 use super::*;
-use crate::map;
-#[cfg(feature = "is_between")]
-use crate::map_as_slice;
 #[cfg(feature = "is_in")]
 use crate::wrap;
+use crate::{map, map_as_slice};
 
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[derive(Clone, PartialEq, Debug, Eq, Hash)]
@@ -115,7 +118,8 @@ impl From<BooleanFunction> for SpecialEq<Arc<dyn SeriesUdf>> {
             #[cfg(feature = "is_in")]
             IsIn => wrap!(is_in),
             Not => map!(not),
-            AllHorizontal | AnyHorizontal => unreachable!(),
+            AllHorizontal => map_as_slice!(all_horizontal),
+            AnyHorizontal => map_as_slice!(any_horizontal),
         }
     }
 }
@@ -206,3 +210,41 @@ fn is_in(s: &mut [Series]) -> PolarsResult<Option<Series>> {
 fn not(s: &Series) -> PolarsResult<Series> {
     polars_ops::series::negate_bitwise(s)
 }
+
+// We shouldn't hit these often only on very wide dataframes where we don't reduce to & expressions.
+fn any_horizontal(s: &[Series]) -> PolarsResult<Series> {
+    let out = POOL
+        .install(|| {
+            s.par_iter()
+                .try_fold(
+                    || BooleanChunked::new("", &[false]),
+                    |acc, b| {
+                        let b = b.cast(&DataType::Boolean)?;
+                        let b = b.bool()?;
+                        PolarsResult::Ok((&acc).bitor(b))
+                    },
+                )
+                .try_reduce(|| BooleanChunked::new("", [false]), |a, b| Ok(a.bitor(b)))
+        })?
+        .with_name(s[0].name());
+    Ok(out.into_series())
+}
+
+// We shouldn't hit these often only on very wide dataframes where we don't reduce to & expressions.
+fn all_horizontal(s: &[Series]) -> PolarsResult<Series> {
+    let out = POOL
+        .install(|| {
+            s.par_iter()
+                .try_fold(
+                    || BooleanChunked::new("", &[true]),
+                    |acc, b| {
+                        let b = b.cast(&DataType::Boolean)?;
+                        let b = b.bool()?;
+                        PolarsResult::Ok((&acc).bitand(b))
+                    },
+                )
+                .try_reduce(|| BooleanChunked::new("", [true]), |a, b| Ok(a.bitand(b)))
+        })?
+        .with_name(s[0].name());
+    Ok(out.into_series())
+}

crates/polars-plan/src/logical_plan/conversion/expr_to_ir.rs

@@ -287,21 +287,26 @@ fn to_aexpr_impl(expr: Expr, arena: &mut Arena<AExpr>, state: &mut ConversionSta
                     return to_aexpr_impl(lit(true), arena, state);
                 },
                 // Convert to binary expression as the optimizer understands those.
+                // Don't exceed 128 expressions as we might stackoverflow.
                 FunctionExpr::Boolean(BooleanFunction::AllHorizontal) => {
-                    let expr = input
-                        .into_iter()
-                        .reduce(|l, r| l.logical_and(r))
-                        .unwrap()
-                        .cast(DataType::Boolean);
-                    return to_aexpr_impl(expr, arena, state);
+                    if input.len() < 128 {
+                        let expr = input
+                            .into_iter()
+                            .reduce(|l, r| l.logical_and(r))
+                            .unwrap()
+                            .cast(DataType::Boolean);
+                        return to_aexpr_impl(expr, arena, state);
+                    }
                 },
                 FunctionExpr::Boolean(BooleanFunction::AnyHorizontal) => {
-                    let expr = input
-                        .into_iter()
-                        .reduce(|l, r| l.logical_or(r))
-                        .unwrap()
-                        .cast(DataType::Boolean);
-                    return to_aexpr_impl(expr, arena, state);
+                    if input.len() < 128 {
+                        let expr = input
+                            .into_iter()
+                            .reduce(|l, r| l.logical_or(r))
+                            .unwrap()
+                            .cast(DataType::Boolean);
+                        return to_aexpr_impl(expr, arena, state);
+                    }
                 },
                 _ => {},
             }