diff --git a/crates/polars-core/src/chunked_array/ops/aggregate/mod.rs b/crates/polars-core/src/chunked_array/ops/aggregate/mod.rs
index 36c65a14d1cbf..7fe43517b3651 100644
--- a/crates/polars-core/src/chunked_array/ops/aggregate/mod.rs
+++ b/crates/polars-core/src/chunked_array/ops/aggregate/mod.rs
@@ -20,25 +20,24 @@ use super::float_sorted_arg_max::{
 use crate::chunked_array::ChunkedArray;
 use crate::datatypes::{BooleanChunked, PolarsNumericType};
 use crate::prelude::*;
-use crate::series::implementations::SeriesWrap;
 use crate::series::IsSorted;
 
 /// Aggregations that return [`Series`] of unit length. Those can be used in broadcasting operations.
 pub trait ChunkAggSeries {
     /// Get the sum of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn sum_as_series(&self) -> Scalar {
+    fn sum_reduce(&self) -> Scalar {
         unimplemented!()
     }
     /// Get the max of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn max_as_series(&self) -> Series {
+    fn max_reduce(&self) -> Scalar {
         unimplemented!()
     }
     /// Get the min of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn min_as_series(&self) -> Series {
+    fn min_reduce(&self) -> Scalar {
         unimplemented!()
     }
     /// Get the product of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn prod_as_series(&self) -> Series {
+    fn prod_reduce(&self) -> Scalar {
         unimplemented!()
     }
 }
@@ -263,75 +262,70 @@ where
         Add<Output = <T::Native as Simd>::Simd> + compute::aggregate::Sum<T::Native>,
     ChunkedArray<T>: IntoSeries,
 {
-    fn sum_as_series(&self) -> Scalar {
+    fn sum_reduce(&self) -> Scalar {
         let v: Option<T::Native> = self.sum();
-        Scalar::new(T::get_dtype().clone(), v.into())
+        Scalar::new(T::get_dtype(), v.into())
     }
 
-    fn max_as_series(&self) -> Series {
+    fn max_reduce(&self) -> Scalar {
         let v = ChunkAgg::max(self);
-        let mut ca: ChunkedArray<T> = [v].iter().copied().collect();
-        ca.rename(self.name());
-        ca.into_series()
+        Scalar::new(T::get_dtype(), v.into())
     }
 
-    fn min_as_series(&self) -> Series {
+    fn min_reduce(&self) -> Scalar {
         let v = ChunkAgg::min(self);
-        let mut ca: ChunkedArray<T> = [v].iter().copied().collect();
-        ca.rename(self.name());
-        ca.into_series()
+        Scalar::new(T::get_dtype(), v.into())
     }
 
-    fn prod_as_series(&self) -> Series {
+    fn prod_reduce(&self) -> Scalar {
         let mut prod = T::Native::one();
-        for opt_v in self.into_iter().flatten() {
-            prod = prod * opt_v;
+
+        for arr in self.downcast_iter() {
+            for v in arr.into_iter().flatten() {
+                prod = prod * *v
+            }
         }
-        Self::from_slice_options(self.name(), &[Some(prod)]).into_series()
+        Scalar::new(T::get_dtype(), prod.into())
     }
 }
 
-fn as_series<T>(name: &str, v: Option<T::Native>) -> Series
-where
-    T: PolarsNumericType,
-    SeriesWrap<ChunkedArray<T>>: SeriesTrait,
-{
-    let mut ca: ChunkedArray<T> = [v].into_iter().collect();
-    ca.rename(name);
-    ca.into_series()
-}
-
 impl<T> VarAggSeries for ChunkedArray<T>
 where
     T: PolarsIntegerType,
     ChunkedArray<T>: ChunkVar,
 {
-    fn var_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float64Type>(self.name(), self.var(ddof))
+    fn var_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.var(ddof);
+        Scalar::new(DataType::Float64, v.into())
     }
 
-    fn std_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float64Type>(self.name(), self.std(ddof))
+    fn std_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.std(ddof);
+        Scalar::new(DataType::Float64, v.into())
     }
 }
 
 impl VarAggSeries for Float32Chunked {
-    fn var_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float32Type>(self.name(), self.var(ddof).map(|x| x as f32))
+    fn var_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.var(ddof).map(|v| v as f32);
+        Scalar::new(DataType::Float32, v.into())
     }
 
-    fn std_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float32Type>(self.name(), self.std(ddof).map(|x| x as f32))
+    fn std_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.std(ddof).map(|v| v as f32);
+        Scalar::new(DataType::Float32, v.into())
     }
 }
 
 impl VarAggSeries for Float64Chunked {
-    fn var_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float64Type>(self.name(), self.var(ddof))
+    fn var_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.var(ddof);
+        Scalar::new(DataType::Float64, v.into())
     }
 
-    fn std_as_series(&self, ddof: u8) -> Series {
-        as_series::<Float64Type>(self.name(), self.std(ddof))
+    fn std_reduce(&self, ddof: u8) -> Scalar {
+        let v = self.std(ddof);
+        Scalar::new(DataType::Float64, v.into())
     }
 }
 
@@ -342,68 +336,65 @@ where
     <T::Native as Simd>::Simd:
         Add<Output = <T::Native as Simd>::Simd> + compute::aggregate::Sum<T::Native>,
 {
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         quantile: f64,
         interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
-        Ok(as_series::<Float64Type>(
-            self.name(),
-            self.quantile(quantile, interpol)?,
-        ))
+    ) -> PolarsResult<Scalar> {
+        let v = self.quantile(quantile, interpol)?;
+        Ok(Scalar::new(DataType::Float64, v.into()))
     }
 
-    fn median_as_series(&self) -> Series {
-        as_series::<Float64Type>(self.name(), self.median())
+    fn median_reduce(&self) -> Scalar {
+        let v = self.median();
+        Scalar::new(DataType::Float64, v.into())
     }
 }
 
 impl QuantileAggSeries for Float32Chunked {
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         quantile: f64,
         interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
-        Ok(as_series::<Float32Type>(
-            self.name(),
-            self.quantile(quantile, interpol)?,
-        ))
+    ) -> PolarsResult<Scalar> {
+        let v = self.quantile(quantile, interpol)?;
+        Ok(Scalar::new(DataType::Float32, v.into()))
     }
 
-    fn median_as_series(&self) -> Series {
-        as_series::<Float32Type>(self.name(), self.median())
+    fn median_reduce(&self) -> Scalar {
+        let v = self.median();
+        Scalar::new(DataType::Float32, v.into())
     }
 }
 
 impl QuantileAggSeries for Float64Chunked {
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         quantile: f64,
         interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
-        Ok(as_series::<Float64Type>(
-            self.name(),
-            self.quantile(quantile, interpol)?,
-        ))
+    ) -> PolarsResult<Scalar> {
+        let v = self.quantile(quantile, interpol)?;
+        Ok(Scalar::new(DataType::Float64, v.into()))
     }
 
-    fn median_as_series(&self) -> Series {
-        as_series::<Float64Type>(self.name(), self.median())
+    fn median_reduce(&self) -> Scalar {
+        let v = self.median();
+        Scalar::new(DataType::Float64, v.into())
     }
 }
 
 impl ChunkAggSeries for BooleanChunked {
-    fn sum_as_series(&self) -> Scalar {
+    fn sum_reduce(&self) -> Scalar {
         let v = self.sum();
         Scalar::new(IDX_DTYPE, v.into())
     }
-    fn max_as_series(&self) -> Series {
+    fn max_reduce(&self) -> Scalar {
         let v = self.max();
-        Series::new(self.name(), [v])
+        Scalar::new(DataType::Boolean, v.into())
     }
-    fn min_as_series(&self) -> Series {
+    fn min_reduce(&self) -> Scalar {
         let v = self.min();
-        Series::new(self.name(), [v])
+        Scalar::new(DataType::Boolean, v.into())
     }
 }
 
@@ -457,14 +448,16 @@ impl StringChunked {
 }
 
 impl ChunkAggSeries for StringChunked {
-    fn sum_as_series(&self) -> Scalar {
+    fn sum_reduce(&self) -> Scalar {
         Scalar::new(DataType::String, AnyValue::Null)
     }
-    fn max_as_series(&self) -> Series {
-        Series::new(self.name(), &[self.max_str()])
+    fn max_reduce(&self) -> Scalar {
+        let av: AnyValue = self.max_str().into();
+        Scalar::new(DataType::String, av.into_static().unwrap())
     }
-    fn min_as_series(&self) -> Series {
-        Series::new(self.name(), &[self.min_str()])
+    fn min_reduce(&self) -> Scalar {
+        let av: AnyValue = self.min_str().into();
+        Scalar::new(DataType::String, av.into_static().unwrap())
     }
 }
 
@@ -529,11 +522,13 @@ impl CategoricalChunked {
 
 #[cfg(feature = "dtype-categorical")]
 impl ChunkAggSeries for CategoricalChunked {
-    fn min_as_series(&self) -> Series {
-        Series::new(self.name(), &[self.min_categorical()])
+    fn min_reduce(&self) -> Scalar {
+        let av: AnyValue = self.min_categorical().into();
+        Scalar::new(DataType::String, av.into_static().unwrap())
     }
-    fn max_as_series(&self) -> Series {
-        Series::new(self.name(), &[self.max_categorical()])
+    fn max_reduce(&self) -> Scalar {
+        let av: AnyValue = self.max_categorical().into();
+        Scalar::new(DataType::String, av.into_static().unwrap())
     }
 }
 
@@ -588,14 +583,16 @@ impl BinaryChunked {
 }
 
 impl ChunkAggSeries for BinaryChunked {
-    fn sum_as_series(&self) -> Scalar {
+    fn sum_reduce(&self) -> Scalar {
         unimplemented!()
     }
-    fn max_as_series(&self) -> Series {
-        Series::new(self.name(), [self.max_binary()])
+    fn max_reduce(&self) -> Scalar {
+        let av: AnyValue = self.max_binary().into();
+        Scalar::new(self.dtype().clone(), av.into_static().unwrap())
     }
-    fn min_as_series(&self) -> Series {
-        Series::new(self.name(), [self.min_binary()])
+    fn min_reduce(&self) -> Scalar {
+        let av: AnyValue = self.min_binary().into();
+        Scalar::new(self.dtype().clone(), av.into_static().unwrap())
     }
 }
 
@@ -686,10 +683,9 @@ mod test {
         assert_eq!(ca.mean().unwrap(), 1.5);
         assert_eq!(
             ca.into_series()
-                .mean_as_series()
-                .f32()
-                .unwrap()
-                .get(0)
+                .mean_reduce()
+                .value()
+                .extract::<f32>()
                 .unwrap(),
             1.5
         );
@@ -697,7 +693,7 @@ mod test {
         let ca = Float32Chunked::full_null("", 3);
         assert_eq!(ca.mean(), None);
         assert_eq!(
-            ca.into_series().mean_as_series().f32().unwrap().get(0),
+            ca.into_series().mean_reduce().value().extract::<f32>(),
             None
         );
     }
diff --git a/crates/polars-core/src/chunked_array/ops/aggregate/quantile.rs b/crates/polars-core/src/chunked_array/ops/aggregate/quantile.rs
index ce528337f0c12..d6218e81d4634 100644
--- a/crates/polars-core/src/chunked_array/ops/aggregate/quantile.rs
+++ b/crates/polars-core/src/chunked_array/ops/aggregate/quantile.rs
@@ -2,13 +2,13 @@ use super::*;
 
 pub trait QuantileAggSeries {
     /// Get the median of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn median_as_series(&self) -> Series;
+    fn median_reduce(&self) -> Scalar;
     /// Get the quantile of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         _quantile: f64,
         _interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series>;
+    ) -> PolarsResult<Scalar>;
 }
 
 /// helper
diff --git a/crates/polars-core/src/chunked_array/ops/aggregate/var.rs b/crates/polars-core/src/chunked_array/ops/aggregate/var.rs
index 95690cf5b3d7e..aff0aeb696417 100644
--- a/crates/polars-core/src/chunked_array/ops/aggregate/var.rs
+++ b/crates/polars-core/src/chunked_array/ops/aggregate/var.rs
@@ -2,9 +2,9 @@ use super::*;
 
 pub trait VarAggSeries {
     /// Get the variance of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn var_as_series(&self, ddof: u8) -> Series;
+    fn var_reduce(&self, ddof: u8) -> Scalar;
     /// Get the standard deviation of the [`ChunkedArray`] as a new [`Series`] of length 1.
-    fn std_as_series(&self, ddof: u8) -> Series;
+    fn std_reduce(&self, ddof: u8) -> Scalar;
 }
 
 impl<T> ChunkVar for ChunkedArray<T>
diff --git a/crates/polars-core/src/datatypes/any_value.rs b/crates/polars-core/src/datatypes/any_value.rs
index f9059c74f6c44..7773a02e85278 100644
--- a/crates/polars-core/src/datatypes/any_value.rs
+++ b/crates/polars-core/src/datatypes/any_value.rs
@@ -19,12 +19,18 @@ impl Scalar {
         Self { dtype, value }
     }
 
-    pub fn value(&self) -> &AnyValue {
+    pub fn value(&self) -> &AnyValue<'static> {
         &self.value
     }
 
+    pub fn as_any_value(&self) -> AnyValue {
+        self.value
+            .strict_cast(&self.dtype)
+            .unwrap_or_else(|| self.value.clone())
+    }
+
     pub fn into_series(self, name: &str) -> Series {
-        Series::from_any_values_and_dtype(name, &[self.value], &self.dtype, true).unwrap()
+        Series::from_any_values_and_dtype(name, &[self.as_any_value()], &self.dtype, true).unwrap()
     }
 }
 
@@ -1178,7 +1184,7 @@ impl GetAnyValue for ArrayRef {
     }
 }
 
-impl<K: NumericNative> From<K> for AnyValue<'_> {
+impl<K: NumericNative> From<K> for AnyValue<'static> {
     fn from(value: K) -> Self {
         unsafe {
             match K::PRIMITIVE {
@@ -1219,6 +1225,24 @@ impl<K: NumericNative> From<K> for AnyValue<'_> {
     }
 }
 
+impl<'a> From<&'a [u8]> for AnyValue<'a> {
+    fn from(value: &'a [u8]) -> Self {
+        AnyValue::Binary(value)
+    }
+}
+
+impl<'a> From<&'a str> for AnyValue<'a> {
+    fn from(value: &'a str) -> Self {
+        AnyValue::String(value)
+    }
+}
+
+impl From<bool> for AnyValue<'static> {
+    fn from(value: bool) -> Self {
+        AnyValue::Boolean(value)
+    }
+}
+
 #[cfg(test)]
 mod test {
     #[cfg(feature = "dtype-categorical")]
diff --git a/crates/polars-core/src/series/implementations/binary.rs b/crates/polars-core/src/series/implementations/binary.rs
index b39f95a7203bd..8964ed83262f3 100644
--- a/crates/polars-core/src/series/implementations/binary.rs
+++ b/crates/polars-core/src/series/implementations/binary.rs
@@ -228,11 +228,11 @@ impl SeriesTrait for SeriesWrap<BinaryChunked> {
         ChunkShift::shift(&self.0, periods).into_series()
     }
 
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::max_as_series(&self.0))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::max_reduce(&self.0))
     }
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::min_as_series(&self.0))
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::min_reduce(&self.0))
     }
     fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
         Arc::new(SeriesWrap(Clone::clone(&self.0)))
diff --git a/crates/polars-core/src/series/implementations/boolean.rs b/crates/polars-core/src/series/implementations/boolean.rs
index 63f61710fa852..938e74d56972c 100644
--- a/crates/polars-core/src/series/implementations/boolean.rs
+++ b/crates/polars-core/src/series/implementations/boolean.rs
@@ -253,52 +253,34 @@ impl SeriesTrait for SeriesWrap<BooleanChunked> {
         ChunkShift::shift(&self.0, periods).into_series()
     }
 
-    fn _sum_as_series(&self) -> PolarsResult<Scalar> {
-        Ok(ChunkAggSeries::sum_as_series(&self.0))
+    fn sum_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::sum_reduce(&self.0))
     }
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::max_as_series(&self.0))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::max_reduce(&self.0))
     }
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::min_as_series(&self.0))
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::min_reduce(&self.0))
     }
-    fn median_as_series(&self) -> PolarsResult<Series> {
-        // first convert array to f32 as that's cheaper
-        // finally the single value to f64
-        Ok(self
-            .0
-            .cast(&DataType::Float32)
-            .unwrap()
-            .median_as_series()
-            .unwrap()
-            .cast(&DataType::Float64)
-            .unwrap())
+    fn median_reduce(&self) -> PolarsResult<Scalar> {
+        let ca = self.0.cast(&DataType::Int8).unwrap();
+        let sc = ca.median_reduce()?;
+        let v = sc.value().cast(&DataType::Float64);
+        Ok(Scalar::new(DataType::Float64, v))
     }
     /// Get the variance of the Series as a new Series of length 1.
-    fn var_as_series(&self, _ddof: u8) -> PolarsResult<Series> {
-        // first convert array to f32 as that's cheaper
-        // finally the single value to f64
-        Ok(self
-            .0
-            .cast(&DataType::Float32)
-            .unwrap()
-            .var_as_series(_ddof)
-            .unwrap()
-            .cast(&DataType::Float64)
-            .unwrap())
+    fn var_reduce(&self, _ddof: u8) -> PolarsResult<Scalar> {
+        let ca = self.0.cast(&DataType::Int8).unwrap();
+        let sc = ca.var_reduce(_ddof)?;
+        let v = sc.value().cast(&DataType::Float64);
+        Ok(Scalar::new(DataType::Float64, v))
     }
     /// Get the standard deviation of the Series as a new Series of length 1.
-    fn std_as_series(&self, _ddof: u8) -> PolarsResult<Series> {
-        // first convert array to f32 as that's cheaper
-        // finally the single value to f64
-        Ok(self
-            .0
-            .cast(&DataType::Float32)
-            .unwrap()
-            .std_as_series(_ddof)
-            .unwrap()
-            .cast(&DataType::Float64)
-            .unwrap())
+    fn std_reduce(&self, _ddof: u8) -> PolarsResult<Scalar> {
+        let ca = self.0.cast(&DataType::Int8).unwrap();
+        let sc = ca.std_reduce(_ddof)?;
+        let v = sc.value().cast(&DataType::Float64);
+        Ok(Scalar::new(DataType::Float64, v))
     }
     fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
         Arc::new(SeriesWrap(Clone::clone(&self.0)))
diff --git a/crates/polars-core/src/series/implementations/categorical.rs b/crates/polars-core/src/series/implementations/categorical.rs
index 056b6a435e3fc..cab3b9a8620c3 100644
--- a/crates/polars-core/src/series/implementations/categorical.rs
+++ b/crates/polars-core/src/series/implementations/categorical.rs
@@ -285,12 +285,12 @@ impl SeriesTrait for SeriesWrap<CategoricalChunked> {
         Arc::new(SeriesWrap(Clone::clone(&self.0)))
     }
 
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::min_as_series(&self.0))
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::min_reduce(&self.0))
     }
 
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::max_as_series(&self.0))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::max_reduce(&self.0))
     }
     fn as_any(&self) -> &dyn Any {
         &self.0
diff --git a/crates/polars-core/src/series/implementations/dates_time.rs b/crates/polars-core/src/series/implementations/dates_time.rs
index 9028871227d0b..9103542ad69dc 100644
--- a/crates/polars-core/src/series/implementations/dates_time.rs
+++ b/crates/polars-core/src/series/implementations/dates_time.rs
@@ -321,14 +321,22 @@ macro_rules! impl_dyn_series {
                 self.0.shift(periods).$into_logical().into_series()
             }
 
-            fn max_as_series(&self) -> PolarsResult<Series> {
-                Ok(self.0.max_as_series().$into_logical())
-            }
-            fn min_as_series(&self) -> PolarsResult<Series> {
-                Ok(self.0.min_as_series().$into_logical())
-            }
-            fn median_as_series(&self) -> PolarsResult<Series> {
-                Series::new(self.name(), &[self.median().map(|v| v as i64)]).cast(self.dtype())
+            fn max_reduce(&self) -> PolarsResult<Scalar> {
+                let sc = self.0.max_reduce();
+                let av = sc.value().cast(self.dtype()).into_static().unwrap();
+                Ok(Scalar::new(self.dtype().clone(), av))
+            }
+            fn min_reduce(&self) -> PolarsResult<Scalar> {
+                let sc = self.0.min_reduce();
+                let av = sc.value().cast(self.dtype()).into_static().unwrap();
+                Ok(Scalar::new(self.dtype().clone(), av))
+            }
+            fn median_reduce(&self) -> PolarsResult<Scalar> {
+                let av = AnyValue::from(self.median().map(|v| v as i64))
+                    .cast(self.dtype())
+                    .into_static()
+                    .unwrap();
+                Ok(Scalar::new(self.dtype().clone(), av))
             }
 
             fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
diff --git a/crates/polars-core/src/series/implementations/datetime.rs b/crates/polars-core/src/series/implementations/datetime.rs
index c237c63e1d9c1..15ada3f419104 100644
--- a/crates/polars-core/src/series/implementations/datetime.rs
+++ b/crates/polars-core/src/series/implementations/datetime.rs
@@ -334,32 +334,29 @@ impl SeriesTrait for SeriesWrap<DatetimeChunked> {
             .into_series()
     }
 
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(self
-            .0
-            .max_as_series()
-            .into_datetime(self.0.time_unit(), self.0.time_zone().clone()))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        let sc = self.0.max_reduce();
+
+        Ok(Scalar::new(self.dtype().clone(), sc.value().clone()))
     }
 
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(self
-            .0
-            .min_as_series()
-            .into_datetime(self.0.time_unit(), self.0.time_zone().clone()))
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        let sc = self.0.min_reduce();
+
+        Ok(Scalar::new(self.dtype().clone(), sc.value().clone()))
     }
 
-    fn median_as_series(&self) -> PolarsResult<Series> {
-        Series::new(self.name(), &[self.median().map(|v| v as i64)]).cast(self.dtype())
+    fn median_reduce(&self) -> PolarsResult<Scalar> {
+        let av: AnyValue = self.median().map(|v| v as i64).into();
+        Ok(Scalar::new(self.dtype().clone(), av))
     }
 
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         _quantile: f64,
         _interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
-        Ok(Int32Chunked::full_null(self.name(), 1)
-            .cast(self.dtype())
-            .unwrap())
+    ) -> PolarsResult<Scalar> {
+        Ok(Scalar::new(self.dtype().clone(), AnyValue::Null))
     }
 
     fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
diff --git a/crates/polars-core/src/series/implementations/decimal.rs b/crates/polars-core/src/series/implementations/decimal.rs
index f1d00c60db24b..678edabe45318 100644
--- a/crates/polars-core/src/series/implementations/decimal.rs
+++ b/crates/polars-core/src/series/implementations/decimal.rs
@@ -316,29 +316,42 @@ impl SeriesTrait for SeriesWrap<DecimalChunked> {
         Arc::new(SeriesWrap(Clone::clone(&self.0)))
     }
 
-    fn _sum_as_series(&self) -> PolarsResult<Scalar> {
+    fn sum_reduce(&self) -> PolarsResult<Scalar> {
         Ok(self.apply_physical(|ca| {
             let sum = ca.sum();
             let DataType::Decimal(_, Some(scale)) = self.dtype() else {
                 unreachable!()
             };
-            let av = match sum {
-                None => AnyValue::Null,
-                Some(v) => AnyValue::Decimal(v, *scale),
-            };
+            let av = AnyValue::Decimal(sum.unwrap(), *scale);
             Scalar::new(self.dtype().clone(), av)
         }))
     }
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(self.apply_physical_to_s(|ca| {
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(self.apply_physical(|ca| {
             let min = ca.min();
-            Int128Chunked::from_slice_options(self.name(), &[min])
+            let DataType::Decimal(_, Some(scale)) = self.dtype() else {
+                unreachable!()
+            };
+            let av = if let Some(min) = min {
+                AnyValue::Decimal(min, *scale)
+            } else {
+                AnyValue::Null
+            };
+            Scalar::new(self.dtype().clone(), av)
         }))
     }
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(self.apply_physical_to_s(|ca| {
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(self.apply_physical(|ca| {
             let max = ca.max();
-            Int128Chunked::from_slice_options(self.name(), &[max])
+            let DataType::Decimal(_, Some(scale)) = self.dtype() else {
+                unreachable!()
+            };
+            let av = if let Some(m) = max {
+                AnyValue::Decimal(m, *scale)
+            } else {
+                AnyValue::Null
+            };
+            Scalar::new(self.dtype().clone(), av)
         }))
     }
     fn as_any(&self) -> &dyn Any {
diff --git a/crates/polars-core/src/series/implementations/duration.rs b/crates/polars-core/src/series/implementations/duration.rs
index f3c6237a01b7e..7ba49d17ba95c 100644
--- a/crates/polars-core/src/series/implementations/duration.rs
+++ b/crates/polars-core/src/series/implementations/duration.rs
@@ -393,49 +393,70 @@ impl SeriesTrait for SeriesWrap<DurationChunked> {
             .into_series()
     }
 
-    fn _sum_as_series(&self) -> PolarsResult<Scalar> {
-        let sc = self.0.sum_as_series();
+    fn sum_reduce(&self) -> PolarsResult<Scalar> {
+        let sc = self.0.sum_reduce();
         let v = sc.value().as_duration(self.0.time_unit());
         Ok(Scalar::new(self.dtype().clone(), v))
     }
 
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(self.0.max_as_series().into_duration(self.0.time_unit()))
-    }
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(self.0.min_as_series().into_duration(self.0.time_unit()))
-    }
-    fn std_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-        Ok(self
-            .0
-            .std_as_series(ddof)
-            .cast(&self.dtype().to_physical())
-            .unwrap()
-            .into_duration(self.0.time_unit()))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        let sc = self.0.max_reduce();
+        let v = sc.value().as_duration(self.0.time_unit());
+        Ok(Scalar::new(self.dtype().clone(), v))
     }
-
-    fn var_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-        Ok(self
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        let sc = self.0.min_reduce();
+        let v = sc.value().as_duration(self.0.time_unit());
+        Ok(Scalar::new(
+            self.dtype().clone(),
+            v.as_duration(self.0.time_unit()),
+        ))
+    }
+    fn std_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+        let sc = self.0.std_reduce(ddof);
+        let to = self.dtype().to_physical();
+        let v = sc.value().cast(&to);
+        Ok(Scalar::new(
+            self.dtype().clone(),
+            v.as_duration(self.0.time_unit()),
+        ))
+    }
+
+    fn var_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+        // Why do we go via MilliSeconds here? Seems wrong to me.
+        // I think we should fix/inspect the tests that fail if we remain on the time-unit here.
+        let sc = self
             .0
             .cast_time_unit(TimeUnit::Milliseconds)
-            .var_as_series(ddof)
-            .cast(&self.dtype().to_physical())
-            .unwrap()
-            .into_duration(TimeUnit::Milliseconds))
-    }
-    fn median_as_series(&self) -> PolarsResult<Series> {
-        Series::new(self.name(), &[self.median().map(|v| v as i64)]).cast(self.dtype())
-    }
-    fn quantile_as_series(
+            .var_reduce(ddof);
+        let to = self.dtype().to_physical();
+        let v = sc.value().cast(&to);
+        Ok(Scalar::new(
+            DataType::Duration(TimeUnit::Milliseconds),
+            v.as_duration(TimeUnit::Milliseconds),
+        ))
+    }
+    fn median_reduce(&self) -> PolarsResult<Scalar> {
+        let v: AnyValue = self.median().map(|v| v as i64).into();
+        let to = self.dtype().to_physical();
+        let v = v.cast(&to);
+        Ok(Scalar::new(
+            self.dtype().clone(),
+            v.as_duration(self.0.time_unit()),
+        ))
+    }
+    fn quantile_reduce(
         &self,
         quantile: f64,
         interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
-        self.0
-            .quantile_as_series(quantile, interpol)?
-            .cast(&self.dtype().to_physical())
-            .unwrap()
-            .cast(self.dtype())
+    ) -> PolarsResult<Scalar> {
+        let v = self.0.quantile_reduce(quantile, interpol)?;
+        let to = self.dtype().to_physical();
+        let v = v.value().cast(&to);
+        Ok(Scalar::new(
+            self.dtype().clone(),
+            v.as_duration(self.0.time_unit()),
+        ))
     }
 
     fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
diff --git a/crates/polars-core/src/series/implementations/floats.rs b/crates/polars-core/src/series/implementations/floats.rs
index 72bc27a94101c..9817d15a04c92 100644
--- a/crates/polars-core/src/series/implementations/floats.rs
+++ b/crates/polars-core/src/series/implementations/floats.rs
@@ -284,30 +284,30 @@ macro_rules! impl_dyn_series {
                 ChunkShift::shift(&self.0, periods).into_series()
             }
 
-            fn _sum_as_series(&self) -> PolarsResult<Scalar> {
-                Ok(ChunkAggSeries::sum_as_series(&self.0))
+            fn sum_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::sum_reduce(&self.0))
             }
-            fn max_as_series(&self) -> PolarsResult<Series> {
-                Ok(ChunkAggSeries::max_as_series(&self.0))
+            fn max_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::max_reduce(&self.0))
             }
-            fn min_as_series(&self) -> PolarsResult<Series> {
-                Ok(ChunkAggSeries::min_as_series(&self.0))
+            fn min_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::min_reduce(&self.0))
             }
-            fn median_as_series(&self) -> PolarsResult<Series> {
-                Ok(QuantileAggSeries::median_as_series(&self.0))
+            fn median_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(QuantileAggSeries::median_reduce(&self.0))
             }
-            fn var_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-                Ok(VarAggSeries::var_as_series(&self.0, ddof))
+            fn var_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+                Ok(VarAggSeries::var_reduce(&self.0, ddof))
             }
-            fn std_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-                Ok(VarAggSeries::std_as_series(&self.0, ddof))
+            fn std_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+                Ok(VarAggSeries::std_reduce(&self.0, ddof))
             }
-            fn quantile_as_series(
+            fn quantile_reduce(
                 &self,
                 quantile: f64,
                 interpol: QuantileInterpolOptions,
-            ) -> PolarsResult<Series> {
-                QuantileAggSeries::quantile_as_series(&self.0, quantile, interpol)
+            ) -> PolarsResult<Scalar> {
+                QuantileAggSeries::quantile_reduce(&self.0, quantile, interpol)
             }
 
             fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
diff --git a/crates/polars-core/src/series/implementations/mod.rs b/crates/polars-core/src/series/implementations/mod.rs
index ef36dd7636761..abbdcbe192feb 100644
--- a/crates/polars-core/src/series/implementations/mod.rs
+++ b/crates/polars-core/src/series/implementations/mod.rs
@@ -387,30 +387,30 @@ macro_rules! impl_dyn_series {
                 ChunkShift::shift(&self.0, periods).into_series()
             }
 
-            fn _sum_as_series(&self) -> PolarsResult<Scalar> {
-                Ok(ChunkAggSeries::sum_as_series(&self.0))
+            fn sum_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::sum_reduce(&self.0))
             }
-            fn max_as_series(&self) -> PolarsResult<Series> {
-                Ok(ChunkAggSeries::max_as_series(&self.0))
+            fn max_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::max_reduce(&self.0))
             }
-            fn min_as_series(&self) -> PolarsResult<Series> {
-                Ok(ChunkAggSeries::min_as_series(&self.0))
+            fn min_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(ChunkAggSeries::min_reduce(&self.0))
             }
-            fn median_as_series(&self) -> PolarsResult<Series> {
-                Ok(QuantileAggSeries::median_as_series(&self.0))
+            fn median_reduce(&self) -> PolarsResult<Scalar> {
+                Ok(QuantileAggSeries::median_reduce(&self.0))
             }
-            fn var_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-                Ok(VarAggSeries::var_as_series(&self.0, ddof))
+            fn var_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+                Ok(VarAggSeries::var_reduce(&self.0, ddof))
             }
-            fn std_as_series(&self, ddof: u8) -> PolarsResult<Series> {
-                Ok(VarAggSeries::std_as_series(&self.0, ddof))
+            fn std_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
+                Ok(VarAggSeries::std_reduce(&self.0, ddof))
             }
-            fn quantile_as_series(
+            fn quantile_reduce(
                 &self,
                 quantile: f64,
                 interpol: QuantileInterpolOptions,
-            ) -> PolarsResult<Series> {
-                QuantileAggSeries::quantile_as_series(&self.0, quantile, interpol)
+            ) -> PolarsResult<Scalar> {
+                QuantileAggSeries::quantile_reduce(&self.0, quantile, interpol)
             }
 
             fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
diff --git a/crates/polars-core/src/series/implementations/string.rs b/crates/polars-core/src/series/implementations/string.rs
index 77ced4ef3c185..ce7a730267e9f 100644
--- a/crates/polars-core/src/series/implementations/string.rs
+++ b/crates/polars-core/src/series/implementations/string.rs
@@ -235,14 +235,14 @@ impl SeriesTrait for SeriesWrap<StringChunked> {
         ChunkShift::shift(&self.0, periods).into_series()
     }
 
-    fn _sum_as_series(&self) -> PolarsResult<Scalar> {
-        Ok(ChunkAggSeries::sum_as_series(&self.0))
+    fn sum_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::sum_reduce(&self.0))
     }
-    fn max_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::max_as_series(&self.0))
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::max_reduce(&self.0))
     }
-    fn min_as_series(&self) -> PolarsResult<Series> {
-        Ok(ChunkAggSeries::min_as_series(&self.0))
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
+        Ok(ChunkAggSeries::min_reduce(&self.0))
     }
     fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
         Arc::new(SeriesWrap(Clone::clone(&self.0)))
diff --git a/crates/polars-core/src/series/mod.rs b/crates/polars-core/src/series/mod.rs
index 43af08e02110e..a1be063e65dd8 100644
--- a/crates/polars-core/src/series/mod.rs
+++ b/crates/polars-core/src/series/mod.rs
@@ -391,9 +391,9 @@ impl Series {
     where
         T: NumCast,
     {
-        let sum = self.sum_as_series()?;
-        let sum = sum.value().cast(&DataType::Float64);
-        Ok(T::from(sum.extract::<f64>().unwrap()).unwrap())
+        let sum = self.sum_reduce()?;
+        let sum = sum.value().extract().unwrap();
+        Ok(sum)
     }
 
     /// Returns the minimum value in the array, according to the natural order.
@@ -402,8 +402,9 @@ impl Series {
     where
         T: NumCast,
     {
-        let min = self.min_as_series()?.cast(&DataType::Float64)?;
-        Ok(min.f64().unwrap().get(0).and_then(T::from))
+        let min = self.min_reduce()?;
+        let min = min.value().extract::<T>();
+        Ok(min)
     }
 
     /// Returns the maximum value in the array, according to the natural order.
@@ -412,8 +413,9 @@ impl Series {
     where
         T: NumCast,
     {
-        let max = self.max_as_series()?.cast(&DataType::Float64)?;
-        Ok(max.f64().unwrap().get(0).and_then(T::from))
+        let max = self.max_reduce()?;
+        let max = max.value().extract::<T>();
+        Ok(max)
     }
 
     /// Explode a list Series. This expands every item to a new row..
@@ -629,11 +631,11 @@ impl Series {
     ///
     /// If the [`DataType`] is one of `{Int8, UInt8, Int16, UInt16}` the `Series` is
     /// first cast to `Int64` to prevent overflow issues.
-    pub fn sum_as_series(&self) -> PolarsResult<Scalar> {
+    pub fn sum_reduce(&self) -> PolarsResult<Scalar> {
         use DataType::*;
         match self.dtype() {
-            Int8 | UInt8 | Int16 | UInt16 => self.cast(&Int64).unwrap().sum_as_series(),
-            _ => self._sum_as_series(),
+            Int8 | UInt8 | Int16 | UInt16 => self.cast(&Int64).unwrap().sum_reduce(),
+            _ => self.0.sum_reduce(),
         }
     }
 
@@ -641,7 +643,7 @@ impl Series {
     ///
     /// If the [`DataType`] is one of `{Int8, UInt8, Int16, UInt16}` the `Series` is
     /// first cast to `Int64` to prevent overflow issues.
-    pub fn product(&self) -> PolarsResult<Series> {
+    pub fn product(&self) -> PolarsResult<Scalar> {
         #[cfg(feature = "product")]
         {
             use DataType::*;
@@ -651,10 +653,10 @@ impl Series {
                     let s = self.cast(&Int64).unwrap();
                     s.product()
                 },
-                Int64 => Ok(self.i64().unwrap().prod_as_series()),
-                UInt64 => Ok(self.u64().unwrap().prod_as_series()),
-                Float32 => Ok(self.f32().unwrap().prod_as_series()),
-                Float64 => Ok(self.f64().unwrap().prod_as_series()),
+                Int64 => Ok(self.i64().unwrap().prod_reduce()),
+                UInt64 => Ok(self.u64().unwrap().prod_reduce()),
+                Float32 => Ok(self.f32().unwrap().prod_reduce()),
+                Float64 => Ok(self.f64().unwrap().prod_reduce()),
                 dt => {
                     polars_bail!(InvalidOperation: "`product` operation not supported for dtype `{dt}`")
                 },
@@ -798,33 +800,22 @@ impl Series {
         self.slice(-(len as i64), len)
     }
 
-    pub fn mean_as_series(&self) -> Series {
+    pub fn mean_reduce(&self) -> Scalar {
         match self.dtype() {
             DataType::Float32 => {
-                let val = &[self.mean().map(|m| m as f32)];
-                Series::new(self.name(), val)
+                let val = self.mean().map(|m| m as f32);
+                Scalar::new(self.dtype().clone(), val.into())
             },
             dt if dt.is_numeric() || matches!(dt, DataType::Boolean) => {
-                let val = &[self.mean()];
-                Series::new(self.name(), val)
+                let val = self.mean();
+                Scalar::new(DataType::Float64, val.into())
             },
-            #[cfg(feature = "dtype-datetime")]
-            dt @ DataType::Datetime(_, _) => {
-                Series::new(self.name(), &[self.mean().map(|v| v as i64)])
-                    .cast(dt)
-                    .unwrap()
+            dt if dt.is_temporal() => {
+                let val = self.mean().map(|v| v as i64);
+                let av: AnyValue = val.into();
+                Scalar::new(dt.clone(), av)
             },
-            #[cfg(feature = "dtype-duration")]
-            dt @ DataType::Duration(_) => {
-                Series::new(self.name(), &[self.mean().map(|v| v as i64)])
-                    .cast(dt)
-                    .unwrap()
-            },
-            #[cfg(feature = "dtype-time")]
-            dt @ DataType::Time => Series::new(self.name(), &[self.mean().map(|v| v as i64)])
-                .cast(dt)
-                .unwrap(),
-            _ => return Series::full_null(self.name(), 1, self.dtype()),
+            dt => Scalar::new(dt.clone(), AnyValue::Null),
         }
     }
 
diff --git a/crates/polars-core/src/series/series_trait.rs b/crates/polars-core/src/series/series_trait.rs
index 3ea809265deb6..e9dc8513073ba 100644
--- a/crates/polars-core/src/series/series_trait.rs
+++ b/crates/polars-core/src/series/series_trait.rs
@@ -413,39 +413,39 @@ pub trait SeriesTrait:
     /// ```
     fn shift(&self, _periods: i64) -> Series;
 
-    /// Get the sum of the Series as a new Series of length 1.
+    /// Get the sum of the Series as a new Scalar.
     ///
     /// If the [`DataType`] is one of `{Int8, UInt8, Int16, UInt16}` the `Series` is
     /// first cast to `Int64` to prevent overflow issues.
-    fn _sum_as_series(&self) -> PolarsResult<Scalar> {
+    fn sum_reduce(&self) -> PolarsResult<Scalar> {
         polars_bail!(opq = sum, self._dtype());
     }
     /// Get the max of the Series as a new Series of length 1.
-    fn max_as_series(&self) -> PolarsResult<Series> {
+    fn max_reduce(&self) -> PolarsResult<Scalar> {
         polars_bail!(opq = max, self._dtype());
     }
     /// Get the min of the Series as a new Series of length 1.
-    fn min_as_series(&self) -> PolarsResult<Series> {
+    fn min_reduce(&self) -> PolarsResult<Scalar> {
         polars_bail!(opq = min, self._dtype());
     }
     /// Get the median of the Series as a new Series of length 1.
-    fn median_as_series(&self) -> PolarsResult<Series> {
+    fn median_reduce(&self) -> PolarsResult<Scalar> {
         polars_bail!(opq = median, self._dtype());
     }
     /// Get the variance of the Series as a new Series of length 1.
-    fn var_as_series(&self, _ddof: u8) -> PolarsResult<Series> {
+    fn var_reduce(&self, _ddof: u8) -> PolarsResult<Scalar> {
         polars_bail!(opq = var, self._dtype());
     }
     /// Get the standard deviation of the Series as a new Series of length 1.
-    fn std_as_series(&self, _ddof: u8) -> PolarsResult<Series> {
+    fn std_reduce(&self, _ddof: u8) -> PolarsResult<Scalar> {
         polars_bail!(opq = std, self._dtype());
     }
     /// Get the quantile of the ChunkedArray as a new Series of length 1.
-    fn quantile_as_series(
+    fn quantile_reduce(
         &self,
         _quantile: f64,
         _interpol: QuantileInterpolOptions,
-    ) -> PolarsResult<Series> {
+    ) -> PolarsResult<Scalar> {
         polars_bail!(opq = quantile, self._dtype());
     }
 
diff --git a/crates/polars-lazy/src/physical_plan/expressions/aggregation.rs b/crates/polars-lazy/src/physical_plan/expressions/aggregation.rs
index dd2937dc3e57d..5799f31779340 100644
--- a/crates/polars-lazy/src/physical_plan/expressions/aggregation.rs
+++ b/crates/polars-lazy/src/physical_plan/expressions/aggregation.rs
@@ -573,7 +573,9 @@ impl PhysicalExpr for AggQuantileExpr {
     fn evaluate(&self, df: &DataFrame, state: &ExecutionState) -> PolarsResult<Series> {
         let input = self.input.evaluate(df, state)?;
         let quantile = self.get_quantile(df, state)?;
-        input.quantile_as_series(quantile, self.interpol)
+        input
+            .quantile_reduce(quantile, self.interpol)
+            .map(|sc| sc.into_series(input.name()))
     }
     #[allow(clippy::ptr_arg)]
     fn evaluate_on_groups<'a>(
diff --git a/crates/polars-lazy/src/physical_plan/planner/expr.rs b/crates/polars-lazy/src/physical_plan/planner/expr.rs
index 1815f0b253553..28a3a63882ceb 100644
--- a/crates/polars-lazy/src/physical_plan/planner/expr.rs
+++ b/crates/polars-lazy/src/physical_plan/planner/expr.rs
@@ -380,11 +380,14 @@ fn create_physical_expr_inner(
 
                                 match s.is_sorted_flag() {
                                     IsSorted::Ascending | IsSorted::Descending => {
-                                        s.min_as_series().map(Some)
-                                    },
-                                    IsSorted::Not => {
-                                        parallel_op_series(|s| s.min_as_series(), s, None, state)
+                                        s.min_reduce().map(|sc| Some(sc.into_series(s.name())))
                                     },
+                                    IsSorted::Not => parallel_op_series(
+                                        |s| s.min_reduce().map(|sc| sc.into_series(s.name())),
+                                        s,
+                                        None,
+                                        state,
+                                    ),
                                 }
                             }) as Arc<dyn SeriesUdf>)
                         },
@@ -414,17 +417,20 @@ fn create_physical_expr_inner(
 
                                 match s.is_sorted_flag() {
                                     IsSorted::Ascending | IsSorted::Descending => {
-                                        s.max_as_series().map(Some)
-                                    },
-                                    IsSorted::Not => {
-                                        parallel_op_series(|s| s.max_as_series(), s, None, state)
+                                        s.max_reduce().map(|sc| Some(sc.into_series(s.name())))
                                     },
+                                    IsSorted::Not => parallel_op_series(
+                                        |s| s.max_reduce().map(|sc| sc.into_series(s.name())),
+                                        s,
+                                        None,
+                                        state,
+                                    ),
                                 }
                             }) as Arc<dyn SeriesUdf>)
                         },
                         AAggExpr::Median(_) => SpecialEq::new(Arc::new(move |s: &mut [Series]| {
                             let s = std::mem::take(&mut s[0]);
-                            s.median_as_series().map(Some)
+                            s.median_reduce().map(|sc| Some(sc.into_series(s.name())))
                         })
                             as Arc<dyn SeriesUdf>),
                         AAggExpr::NUnique(_) => {
@@ -460,7 +466,7 @@ fn create_physical_expr_inner(
                             as Arc<dyn SeriesUdf>),
                         AAggExpr::Mean(_) => SpecialEq::new(Arc::new(move |s: &mut [Series]| {
                             let s = std::mem::take(&mut s[0]);
-                            Ok(Some(s.mean_as_series()))
+                            Ok(Some(s.mean_reduce().into_series(s.name())))
                         })
                             as Arc<dyn SeriesUdf>),
                         AAggExpr::Implode(_) => {
@@ -477,7 +483,7 @@ fn create_physical_expr_inner(
                             SpecialEq::new(Arc::new(move |s: &mut [Series]| {
                                 let s = std::mem::take(&mut s[0]);
                                 parallel_op_series(
-                                    |s| s.sum_as_series().map(|sc| sc.into_series(s.name())),
+                                    |s| s.sum_reduce().map(|sc| sc.into_series(s.name())),
                                     s,
                                     None,
                                     state,
@@ -498,14 +504,14 @@ fn create_physical_expr_inner(
                             let ddof = *ddof;
                             SpecialEq::new(Arc::new(move |s: &mut [Series]| {
                                 let s = std::mem::take(&mut s[0]);
-                                s.std_as_series(ddof).map(Some)
+                                s.std_reduce(ddof).map(|sc| Some(sc.into_series(s.name())))
                             }) as Arc<dyn SeriesUdf>)
                         },
                         AAggExpr::Var(_, ddof) => {
                             let ddof = *ddof;
                             SpecialEq::new(Arc::new(move |s: &mut [Series]| {
                                 let s = std::mem::take(&mut s[0]);
-                                s.var_as_series(ddof).map(Some)
+                                s.var_reduce(ddof).map(|sc| Some(sc.into_series(s.name())))
                             }) as Arc<dyn SeriesUdf>)
                         },
                         AAggExpr::AggGroups(_) => {
diff --git a/crates/polars-ops/src/chunked_array/list/min_max.rs b/crates/polars-ops/src/chunked_array/list/min_max.rs
index dd043110be2ee..5acc95a506e7d 100644
--- a/crates/polars-ops/src/chunked_array/list/min_max.rs
+++ b/crates/polars-ops/src/chunked_array/list/min_max.rs
@@ -95,7 +95,11 @@ pub(super) fn list_min_function(ca: &ListChunked) -> PolarsResult<Series> {
                 })
             },
             _ => Ok(ca
-                .try_apply_amortized(|s| s.as_ref().min_as_series())?
+                .try_apply_amortized(|s| {
+                    let s = s.as_ref();
+                    let sc = s.min_reduce()?;
+                    Ok(sc.into_series(s.name()))
+                })?
                 .explode()
                 .unwrap()
                 .into_series()),
@@ -201,7 +205,11 @@ pub(super) fn list_max_function(ca: &ListChunked) -> PolarsResult<Series> {
                 })
             },
             _ => Ok(ca
-                .try_apply_amortized(|s| s.as_ref().max_as_series())?
+                .try_apply_amortized(|s| {
+                    let s = s.as_ref();
+                    let sc = s.max_reduce()?;
+                    Ok(sc.into_series(s.name()))
+                })?
                 .explode()
                 .unwrap()
                 .into_series()),
diff --git a/crates/polars-ops/src/chunked_array/list/sum_mean.rs b/crates/polars-ops/src/chunked_array/list/sum_mean.rs
index 9387f3e6702f4..94255979ac19b 100644
--- a/crates/polars-ops/src/chunked_array/list/sum_mean.rs
+++ b/crates/polars-ops/src/chunked_array/list/sum_mean.rs
@@ -105,7 +105,7 @@ pub(super) fn sum_with_nulls(ca: &ListChunked, inner_dtype: &DataType) -> Polars
         },
         // slowest sum_as_series path
         _ => ca
-            .try_apply_amortized(|s| s.as_ref().sum_as_series().map(|sc| sc.into_series("")))?
+            .try_apply_amortized(|s| s.as_ref().sum_reduce().map(|sc| sc.into_series("")))?
             .explode()
             .unwrap()
             .into_series(),
diff --git a/crates/polars-ops/src/series/ops/log.rs b/crates/polars-ops/src/series/ops/log.rs
index a63017eec0091..559e9d0b5bace 100644
--- a/crates/polars-ops/src/series/ops/log.rs
+++ b/crates/polars-ops/src/series/ops/log.rs
@@ -92,7 +92,7 @@ pub trait LogSeries: SeriesSealed {
                 let pk = s.as_ref();
 
                 let pk = if normalize {
-                    let sum = pk.sum_as_series().unwrap().into_series("");
+                    let sum = pk.sum_reduce().unwrap().into_series("");
 
                     if sum.get(0).unwrap().extract::<f64>().unwrap() != 1.0 {
                         pk / &sum
diff --git a/crates/polars-plan/src/dsl/function_expr/shrink_type.rs b/crates/polars-plan/src/dsl/function_expr/shrink_type.rs
index fab3b88e22bca..cbd932ac1d785 100644
--- a/crates/polars-plan/src/dsl/function_expr/shrink_type.rs
+++ b/crates/polars-plan/src/dsl/function_expr/shrink_type.rs
@@ -5,12 +5,7 @@ pub(super) fn shrink(s: Series) -> PolarsResult<Series> {
         if s.dtype().is_float() {
             s.cast(&DataType::Float32)
         } else if s.dtype().is_unsigned_integer() {
-            let max = s
-                .max_as_series()?
-                .get(0)
-                .unwrap()
-                .extract::<u64>()
-                .unwrap_or(0_u64);
+            let max = s.max_reduce()?.value().extract::<u64>().unwrap_or(0_u64);
             if max <= u8::MAX as u64 {
                 s.cast(&DataType::UInt8)
             } else if max <= u16::MAX as u64 {
@@ -21,18 +16,8 @@ pub(super) fn shrink(s: Series) -> PolarsResult<Series> {
                 Ok(s)
             }
         } else {
-            let min = s
-                .min_as_series()?
-                .get(0)
-                .unwrap()
-                .extract::<i64>()
-                .unwrap_or(0_i64);
-            let max = s
-                .max_as_series()?
-                .get(0)
-                .unwrap()
-                .extract::<i64>()
-                .unwrap_or(0_i64);
+            let min = s.min_reduce()?.value().extract::<i64>().unwrap_or(0_i64);
+            let max = s.max_reduce()?.value().extract::<i64>().unwrap_or(0_i64);
 
             if min >= i8::MIN as i64 && max <= i8::MAX as i64 {
                 s.cast(&DataType::Int8)
diff --git a/crates/polars-plan/src/dsl/mod.rs b/crates/polars-plan/src/dsl/mod.rs
index 3c4b961305832..3357875e246af 100644
--- a/crates/polars-plan/src/dsl/mod.rs
+++ b/crates/polars-plan/src/dsl/mod.rs
@@ -799,7 +799,7 @@ impl Expr {
         };
 
         self.function_with_options(
-            move |s: Series| Some(s.product()).transpose(),
+            move |s: Series| Some(s.product().map(|sc| sc.into_series(s.name()))).transpose(),
             GetOutput::map_dtype(|dt| {
                 use DataType::*;
                 match dt {
diff --git a/crates/polars/tests/it/core/rolling_window.rs b/crates/polars/tests/it/core/rolling_window.rs
index 7d1b069ccbbc8..b823bf7d87366 100644
--- a/crates/polars/tests/it/core/rolling_window.rs
+++ b/crates/polars/tests/it/core/rolling_window.rs
@@ -177,7 +177,7 @@ fn test_rolling_map() {
 
     let out = ca
         .rolling_map(
-            &|s| s.sum_as_series().unwrap().into_series(s.name()),
+            &|s| s.sum_reduce().unwrap().into_series(s.name()),
             RollingOptionsFixedWindow {
                 window_size: 3,
                 min_periods: 3,
diff --git a/py-polars/src/series/aggregation.rs b/py-polars/src/series/aggregation.rs
index 626d3eb7cef52..5874ec08ac386 100644
--- a/py-polars/src/series/aggregation.rs
+++ b/py-polars/src/series/aggregation.rs
@@ -35,10 +35,9 @@ impl PySeries {
     fn max(&self, py: Python) -> PyResult<PyObject> {
         Ok(Wrap(
             self.series
-                .max_as_series()
+                .max_reduce()
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
@@ -49,18 +48,13 @@ impl PySeries {
                 self.series
                     .cast(&DataType::UInt8)
                     .unwrap()
-                    .mean_as_series()
-                    .get(0)
-                    .map_err(PyPolarsErr::from)?,
-            )
-            .into_py(py)),
-            DataType::Datetime(_, _) | DataType::Duration(_) | DataType::Time => Ok(Wrap(
-                self.series
-                    .mean_as_series()
-                    .get(0)
-                    .map_err(PyPolarsErr::from)?,
+                    .mean_reduce()
+                    .as_any_value(),
             )
             .into_py(py)),
+            DataType::Datetime(_, _) | DataType::Duration(_) | DataType::Time => {
+                Ok(Wrap(self.series.mean_reduce().as_any_value()).into_py(py))
+            },
             _ => Ok(self.series.mean().into_py(py)),
         }
     }
@@ -71,18 +65,16 @@ impl PySeries {
                 self.series
                     .cast(&DataType::UInt8)
                     .unwrap()
-                    .median_as_series()
+                    .median_reduce()
                     .map_err(PyPolarsErr::from)?
-                    .get(0)
-                    .map_err(PyPolarsErr::from)?,
+                    .as_any_value(),
             )
             .into_py(py)),
             DataType::Datetime(_, _) | DataType::Duration(_) | DataType::Time => Ok(Wrap(
                 self.series
-                    .median_as_series()
+                    .median_reduce()
                     .map_err(PyPolarsErr::from)?
-                    .get(0)
-                    .map_err(PyPolarsErr::from)?,
+                    .as_any_value(),
             )
             .into_py(py)),
             _ => Ok(self.series.median().into_py(py)),
@@ -92,10 +84,9 @@ impl PySeries {
     fn min(&self, py: Python) -> PyResult<PyObject> {
         Ok(Wrap(
             self.series
-                .min_as_series()
+                .min_reduce()
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
@@ -105,8 +96,7 @@ impl PySeries {
             self.series
                 .product()
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
@@ -116,22 +106,18 @@ impl PySeries {
         quantile: f64,
         interpolation: Wrap<QuantileInterpolOptions>,
     ) -> PyResult<PyObject> {
-        let tmp = self
-            .series
-            .quantile_as_series(quantile, interpolation.0)
-            .map_err(PyPolarsErr::from)?;
-        let out = tmp.get(0).unwrap_or(AnyValue::Null);
+        let bind = self.series.quantile_reduce(quantile, interpolation.0);
+        let sc = bind.map_err(PyPolarsErr::from)?;
 
-        Ok(Python::with_gil(|py| Wrap(out).into_py(py)))
+        Ok(Python::with_gil(|py| Wrap(sc.as_any_value()).into_py(py)))
     }
 
     fn std(&self, py: Python, ddof: u8) -> PyResult<PyObject> {
         Ok(Wrap(
             self.series
-                .std_as_series(ddof)
+                .std_reduce(ddof)
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
@@ -139,10 +125,9 @@ impl PySeries {
     fn var(&self, py: Python, ddof: u8) -> PyResult<PyObject> {
         Ok(Wrap(
             self.series
-                .var_as_series(ddof)
+                .var_reduce(ddof)
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
@@ -150,11 +135,9 @@ impl PySeries {
     fn sum(&self, py: Python) -> PyResult<PyObject> {
         Ok(Wrap(
             self.series
-                .sum_as_series()
-                .map(|sc| sc.into_series(""))
+                .sum_reduce()
                 .map_err(PyPolarsErr::from)?
-                .get(0)
-                .map_err(PyPolarsErr::from)?,
+                .as_any_value(),
         )
         .into_py(py))
     }
diff --git a/py-polars/tests/unit/dataframe/test_describe.py b/py-polars/tests/unit/dataframe/test_describe.py
index 2fdf0db614b2f..4c424bc3b61ce 100644
--- a/py-polars/tests/unit/dataframe/test_describe.py
+++ b/py-polars/tests/unit/dataframe/test_describe.py
@@ -30,6 +30,7 @@ def test_df_describe(lazy: bool) -> None:
 
     frame: pl.DataFrame | pl.LazyFrame = df.lazy() if lazy else df
     result = frame.describe()
+    print(result)
 
     expected = pl.DataFrame(
         {
diff --git a/py-polars/tests/unit/operations/aggregation/test_aggregations.py b/py-polars/tests/unit/operations/aggregation/test_aggregations.py
index b71a3394a2b1d..4aa6593783343 100644
--- a/py-polars/tests/unit/operations/aggregation/test_aggregations.py
+++ b/py-polars/tests/unit/operations/aggregation/test_aggregations.py
@@ -34,8 +34,8 @@ def test_boolean_aggs() -> None:
     ]
     assert df.select(aggs).to_dict(as_series=False) == {
         "mean": [0.6666666666666666],
-        "std": [0.5773502588272095],
-        "var": [0.3333333432674408],
+        "std": [0.5773502691896258],
+        "var": [0.33333333333333337],
     }
 
     assert df.group_by(pl.lit(1)).agg(aggs).to_dict(as_series=False) == {