refactor(rust): Split Reduction into it plus ReductionState

crates/polars-core/src/scalar/mod.rs

@@ -12,10 +12,17 @@ pub struct Scalar {
 }
 
 impl Scalar {
+    #[inline(always)]
     pub fn new(dtype: DataType, value: AnyValue<'static>) -> Self {
         Self { dtype, value }
     }
 
+    #[inline(always)]
+    pub fn is_null(&self) -> bool {
+        self.value.is_null()
+    }
+
+    #[inline(always)]
     pub fn value(&self) -> &AnyValue<'static> {
         &self.value
     }
@@ -30,10 +37,12 @@ impl Scalar {
         Series::from_any_values_and_dtype(name, &[self.as_any_value()], &self.dtype, true).unwrap()
     }
 
+    #[inline(always)]
     pub fn dtype(&self) -> &DataType {
         &self.dtype
     }
 
+    #[inline(always)]
     pub fn update(&mut self, value: AnyValue<'static>) {
         self.value = value;
     }

crates/polars-expr/Cargo.toml

@@ -13,6 +13,7 @@ ahash = { workspace = true }
 arrow = { workspace = true }
 bitflags = { workspace = true }
 once_cell = { workspace = true }
+polars-compute = { workspace = true }
 polars-core = { workspace = true, features = ["lazy", "zip_with", "random"] }
 polars-io = { workspace = true, features = ["lazy"] }
 polars-json = { workspace = true, optional = true }

crates/polars-expr/src/reduce/convert.rs

@@ -2,11 +2,13 @@ use polars_core::error::feature_gated;
 use polars_plan::prelude::*;
 use polars_utils::arena::{Arena, Node};
 
-use super::extrema::*;
+use super::len::LenReduce;
+use super::mean::MeanReduce;
+use super::min_max::{MaxReduce, MinReduce};
+#[cfg(feature = "propagate_nans")]
+use super::nan_min_max::{NanMaxReduce, NanMinReduce};
 use super::sum::SumReduce;
 use super::*;
-use crate::reduce::len::LenReduce;
-use crate::reduce::mean::MeanReduce;
 
 /// Converts a node into a reduction + its associated selector expression.
 pub fn into_reduction(
@@ -29,8 +31,8 @@ pub fn into_reduction(
                 if *propagate_nans && field.dtype.is_float() {
                     feature_gated!("propagate_nans", {
                         let out: Box<dyn Reduction> = match field.dtype {
-                            DataType::Float32 => Box::new(MinNanReduce::<Float32Type>::new()),
-                            DataType::Float64 => Box::new(MinNanReduce::<Float64Type>::new()),
+                            DataType::Float32 => Box::new(NanMinReduce::<Float32Type>::new()),
+                            DataType::Float64 => Box::new(NanMinReduce::<Float64Type>::new()),
                             _ => unreachable!(),
                         };
                         (out, *input)
@@ -49,8 +51,8 @@ pub fn into_reduction(
                 if *propagate_nans && field.dtype.is_float() {
                     feature_gated!("propagate_nans", {
                         let out: Box<dyn Reduction> = match field.dtype {
-                            DataType::Float32 => Box::new(MaxNanReduce::<Float32Type>::new()),
-                            DataType::Float64 => Box::new(MaxNanReduce::<Float64Type>::new()),
+                            DataType::Float32 => Box::new(NanMaxReduce::<Float32Type>::new()),
+                            DataType::Float64 => Box::new(NanMaxReduce::<Float64Type>::new()),
                             _ => unreachable!(),
                         };
                         (out, *input)

crates/polars-expr/src/reduce/len.rs

@@ -3,37 +3,37 @@ use polars_core::error::constants::LENGTH_LIMIT_MSG;
 use super::*;
 
 #[derive(Clone)]
-pub struct LenReduce {
-    len: u64,
-}
+pub struct LenReduce {}
 
 impl LenReduce {
-    pub(crate) fn new() -> Self {
-        Self { len: 0 }
+    pub fn new() -> Self {
+        Self {}
     }
 }
 
 impl Reduction for LenReduce {
-    fn init_dyn(&self) -> Box<dyn Reduction> {
-        Box::new(Self::new())
+    fn new_reducer(&self) -> Box<dyn ReductionState> {
+        Box::new(LenReduceState { len: 0 })
     }
+}
 
-    fn reset(&mut self) {
-        self.len = 0;
-    }
+pub struct LenReduceState {
+    len: u64,
+}
 
+impl ReductionState for LenReduceState {
     fn update(&mut self, batch: &Series) -> PolarsResult<()> {
         self.len += batch.len() as u64;
         Ok(())
     }
 
-    fn combine(&mut self, other: &dyn Reduction) -> PolarsResult<()> {
+    fn combine(&mut self, other: &dyn ReductionState) -> PolarsResult<()> {
         let other = other.as_any().downcast_ref::<Self>().unwrap();
         self.len += other.len;
         Ok(())
     }
 
-    fn finalize(&mut self) -> PolarsResult<Scalar> {
+    fn finalize(&self) -> PolarsResult<Scalar> {
         #[allow(clippy::useless_conversion)]
         let as_idx: IdxSize = self.len.try_into().expect(LENGTH_LIMIT_MSG);
         Ok(Scalar::new(IDX_DTYPE, as_idx.into()))