Path: blob/main/crates/polars-ops/src/series/ops/cum_agg.rs
6939 views
use std::ops::{AddAssign, Mul};12use arity::unary_elementwise_values;3use arrow::array::{Array, BooleanArray};4use arrow::bitmap::{Bitmap, BitmapBuilder};5use num_traits::{Bounded, One, Zero};6use polars_core::prelude::*;7use polars_core::series::IsSorted;8use polars_core::utils::{CustomIterTools, NoNull};9use polars_core::with_match_physical_numeric_polars_type;10use polars_utils::float::IsFloat;11use polars_utils::min_max::MinMax;1213fn det_max<T>(state: &mut T, v: Option<T>) -> Option<Option<T>>14where15T: Copy + MinMax,16{17match v {18Some(v) => {19*state = MinMax::max_ignore_nan(*state, v);20Some(Some(*state))21},22None => Some(None),23}24}2526fn det_min<T>(state: &mut T, v: Option<T>) -> Option<Option<T>>27where28T: Copy + MinMax,29{30match v {31Some(v) => {32*state = MinMax::min_ignore_nan(*state, v);33Some(Some(*state))34},35None => Some(None),36}37}3839fn det_sum<T>(state: &mut T, v: Option<T>) -> Option<Option<T>>40where41T: Copy + AddAssign,42{43match v {44Some(v) => {45*state += v;46Some(Some(*state))47},48None => Some(None),49}50}5152fn det_prod<T>(state: &mut T, v: Option<T>) -> Option<Option<T>>53where54T: Copy + Mul<Output = T>,55{56match v {57Some(v) => {58*state = *state * v;59Some(Some(*state))60},61None => Some(None),62}63}6465fn cum_scan_numeric<T, F>(66ca: &ChunkedArray<T>,67reverse: bool,68init: T::Native,69update: F,70) -> ChunkedArray<T>71where72T: PolarsNumericType,73ChunkedArray<T>: FromIterator<Option<T::Native>>,74F: Fn(&mut T::Native, Option<T::Native>) -> Option<Option<T::Native>>,75{76let out: ChunkedArray<T> = match reverse {77false => ca.iter().scan(init, update).collect_trusted(),78true => ca.iter().rev().scan(init, update).collect_reversed(),79};80out.with_name(ca.name().clone())81}8283fn cum_max_numeric<T>(84ca: &ChunkedArray<T>,85reverse: bool,86init: Option<T::Native>,87) -> ChunkedArray<T>88where89T: PolarsNumericType,90T::Native: MinMax + Bounded,91ChunkedArray<T>: FromIterator<Option<T::Native>>,92{93let init = init.unwrap_or(if T::Native::is_float() {94T::Native::nan_value()95} else {96Bounded::min_value()97});98cum_scan_numeric(ca, reverse, init, det_max)99}100101fn cum_min_numeric<T>(102ca: &ChunkedArray<T>,103reverse: bool,104init: Option<T::Native>,105) -> ChunkedArray<T>106where107T: PolarsNumericType,108T::Native: MinMax + Bounded,109ChunkedArray<T>: FromIterator<Option<T::Native>>,110{111let init = init.unwrap_or(if T::Native::is_float() {112T::Native::nan_value()113} else {114Bounded::max_value()115});116cum_scan_numeric(ca, reverse, init, det_min)117}118119fn cum_max_bool(ca: &BooleanChunked, reverse: bool, init: Option<bool>) -> BooleanChunked {120if ca.len() == ca.null_count() {121return ca.clone();122}123124if init == Some(true) {125return unsafe {126BooleanChunked::from_chunks(127ca.name().clone(),128ca.downcast_iter()129.map(|arr| {130arr.with_values(Bitmap::new_with_value(true, arr.len()))131.to_boxed()132})133.collect(),134)135};136}137138let mut out;139if !reverse {140// TODO: efficient bitscan.141let Some(first_true_idx) = ca.iter().position(|x| x == Some(true)) else {142return ca.clone();143};144out = BitmapBuilder::with_capacity(ca.len());145out.extend_constant(first_true_idx, false);146out.extend_constant(ca.len() - first_true_idx, true);147} else {148// TODO: efficient bitscan.149let Some(last_true_idx) = ca.iter().rposition(|x| x == Some(true)) else {150return ca.clone();151};152out = BitmapBuilder::with_capacity(ca.len());153out.extend_constant(last_true_idx + 1, true);154out.extend_constant(ca.len() - 1 - last_true_idx, false);155}156157let arr: BooleanArray = out.freeze().into();158BooleanChunked::with_chunk_like(ca, arr.with_validity(ca.rechunk_validity()))159}160161fn cum_min_bool(ca: &BooleanChunked, reverse: bool, init: Option<bool>) -> BooleanChunked {162if ca.len() == ca.null_count() {163return ca.clone();164}165166if init == Some(false) {167return unsafe {168BooleanChunked::from_chunks(169ca.name().clone(),170ca.downcast_iter()171.map(|arr| {172arr.with_values(Bitmap::new_with_value(false, arr.len()))173.to_boxed()174})175.collect(),176)177};178}179180let mut out;181if !reverse {182// TODO: efficient bitscan.183let Some(first_false_idx) = ca.iter().position(|x| x == Some(false)) else {184return ca.clone();185};186out = BitmapBuilder::with_capacity(ca.len());187out.extend_constant(first_false_idx, true);188out.extend_constant(ca.len() - first_false_idx, false);189} else {190// TODO: efficient bitscan.191let Some(last_false_idx) = ca.iter().rposition(|x| x == Some(false)) else {192return ca.clone();193};194out = BitmapBuilder::with_capacity(ca.len());195out.extend_constant(last_false_idx + 1, false);196out.extend_constant(ca.len() - 1 - last_false_idx, true);197}198199let arr: BooleanArray = out.freeze().into();200BooleanChunked::with_chunk_like(ca, arr.with_validity(ca.rechunk_validity()))201}202203fn cum_sum_numeric<T>(204ca: &ChunkedArray<T>,205reverse: bool,206init: Option<T::Native>,207) -> ChunkedArray<T>208where209T: PolarsNumericType,210ChunkedArray<T>: FromIterator<Option<T::Native>>,211{212let init = init.unwrap_or(T::Native::zero());213cum_scan_numeric(ca, reverse, init, det_sum)214}215216fn cum_prod_numeric<T>(217ca: &ChunkedArray<T>,218reverse: bool,219init: Option<T::Native>,220) -> ChunkedArray<T>221where222T: PolarsNumericType,223ChunkedArray<T>: FromIterator<Option<T::Native>>,224{225let init = init.unwrap_or(T::Native::one());226cum_scan_numeric(ca, reverse, init, det_prod)227}228229pub fn cum_prod_with_init(230s: &Series,231reverse: bool,232init: &AnyValue<'static>,233) -> PolarsResult<Series> {234use DataType::*;235let out = match s.dtype() {236Boolean | Int8 | UInt8 | Int16 | UInt16 | Int32 | UInt32 => {237let s = s.cast(&Int64)?;238cum_prod_numeric(s.i64()?, reverse, init.extract()).into_series()239},240Int64 => cum_prod_numeric(s.i64()?, reverse, init.extract()).into_series(),241UInt64 => cum_prod_numeric(s.u64()?, reverse, init.extract()).into_series(),242#[cfg(feature = "dtype-i128")]243Int128 => cum_prod_numeric(s.i128()?, reverse, init.extract()).into_series(),244Float32 => cum_prod_numeric(s.f32()?, reverse, init.extract()).into_series(),245Float64 => cum_prod_numeric(s.f64()?, reverse, init.extract()).into_series(),246dt => polars_bail!(opq = cum_prod, dt),247};248Ok(out)249}250251/// Get an array with the cumulative product computed at every element.252///253/// If the [`DataType`] is one of `{Int8, UInt8, Int16, UInt16, Int32, UInt32}` the `Series` is254/// first cast to `Int64` to prevent overflow issues.255pub fn cum_prod(s: &Series, reverse: bool) -> PolarsResult<Series> {256cum_prod_with_init(s, reverse, &AnyValue::Null)257}258259pub fn cum_sum_with_init(260s: &Series,261reverse: bool,262init: &AnyValue<'static>,263) -> PolarsResult<Series> {264use DataType::*;265let out = match s.dtype() {266Boolean => {267let s = s.cast(&UInt32)?;268cum_sum_numeric(s.u32()?, reverse, init.extract()).into_series()269},270Int8 | UInt8 | Int16 | UInt16 => {271let s = s.cast(&Int64)?;272cum_sum_numeric(s.i64()?, reverse, init.extract()).into_series()273},274Int32 => cum_sum_numeric(s.i32()?, reverse, init.extract()).into_series(),275UInt32 => cum_sum_numeric(s.u32()?, reverse, init.extract()).into_series(),276Int64 => cum_sum_numeric(s.i64()?, reverse, init.extract()).into_series(),277UInt64 => cum_sum_numeric(s.u64()?, reverse, init.extract()).into_series(),278#[cfg(feature = "dtype-i128")]279Int128 => cum_sum_numeric(s.i128()?, reverse, init.extract()).into_series(),280Float32 => cum_sum_numeric(s.f32()?, reverse, init.extract()).into_series(),281Float64 => cum_sum_numeric(s.f64()?, reverse, init.extract()).into_series(),282#[cfg(feature = "dtype-decimal")]283Decimal(precision, scale) => {284let ca = s.decimal().unwrap().physical();285cum_sum_numeric(ca, reverse, init.clone().to_physical().extract())286.into_decimal_unchecked(*precision, scale.unwrap())287.into_series()288},289#[cfg(feature = "dtype-duration")]290Duration(tu) => {291let s = s.to_physical_repr();292let ca = s.i64()?;293cum_sum_numeric(ca, reverse, init.extract()).cast(&Duration(*tu))?294},295dt => polars_bail!(opq = cum_sum, dt),296};297Ok(out)298}299300/// Get an array with the cumulative sum computed at every element301///302/// If the [`DataType`] is one of `{Int8, UInt8, Int16, UInt16}` the `Series` is303/// first cast to `Int64` to prevent overflow issues.304pub fn cum_sum(s: &Series, reverse: bool) -> PolarsResult<Series> {305cum_sum_with_init(s, reverse, &AnyValue::Null)306}307308pub fn cum_min_with_init(309s: &Series,310reverse: bool,311init: &AnyValue<'static>,312) -> PolarsResult<Series> {313match s.dtype() {314DataType::Boolean => {315Ok(cum_min_bool(s.bool()?, reverse, init.extract_bool()).into_series())316},317#[cfg(feature = "dtype-decimal")]318DataType::Decimal(precision, scale) => {319let ca = s.decimal().unwrap().physical();320let out = cum_min_numeric(ca, reverse, init.clone().to_physical().extract())321.into_decimal_unchecked(*precision, scale.unwrap())322.into_series();323Ok(out)324},325dt if dt.to_physical().is_primitive_numeric() => {326let s = s.to_physical_repr();327with_match_physical_numeric_polars_type!(s.dtype(), |$T| {328let ca: &ChunkedArray<$T> = s.as_ref().as_ref().as_ref();329let out = cum_min_numeric(ca, reverse, init.extract()).into_series();330if dt.is_logical() {331out.cast(dt)332} else {333Ok(out)334}335})336},337dt => polars_bail!(opq = cum_min, dt),338}339}340341/// Get an array with the cumulative min computed at every element.342pub fn cum_min(s: &Series, reverse: bool) -> PolarsResult<Series> {343cum_min_with_init(s, reverse, &AnyValue::Null)344}345346pub fn cum_max_with_init(347s: &Series,348reverse: bool,349init: &AnyValue<'static>,350) -> PolarsResult<Series> {351match s.dtype() {352DataType::Boolean => {353Ok(cum_max_bool(s.bool()?, reverse, init.extract_bool()).into_series())354},355#[cfg(feature = "dtype-decimal")]356DataType::Decimal(precision, scale) => {357let ca = s.decimal().unwrap().physical();358let out = cum_max_numeric(ca, reverse, init.clone().to_physical().extract())359.into_decimal_unchecked(*precision, scale.unwrap())360.into_series();361Ok(out)362},363dt if dt.to_physical().is_primitive_numeric() => {364let s = s.to_physical_repr();365with_match_physical_numeric_polars_type!(s.dtype(), |$T| {366let ca: &ChunkedArray<$T> = s.as_ref().as_ref().as_ref();367let out = cum_max_numeric(ca, reverse, init.extract()).into_series();368if dt.is_logical() {369out.cast(dt)370} else {371Ok(out)372}373})374},375dt => polars_bail!(opq = cum_max, dt),376}377}378379/// Get an array with the cumulative max computed at every element.380pub fn cum_max(s: &Series, reverse: bool) -> PolarsResult<Series> {381cum_max_with_init(s, reverse, &AnyValue::Null)382}383384pub fn cum_count(s: &Series, reverse: bool) -> PolarsResult<Series> {385cum_count_with_init(s, reverse, 0)386}387388pub fn cum_count_with_init(s: &Series, reverse: bool, init: IdxSize) -> PolarsResult<Series> {389let mut out = if s.null_count() == 0 {390// Fast paths for no nulls391cum_count_no_nulls(s.name().clone(), s.len(), reverse, init)392} else {393let ca = s.is_not_null();394let out: IdxCa = if reverse {395let mut count = init + (s.len() - s.null_count()) as IdxSize;396let mut prev = false;397unary_elementwise_values(&ca, |v: bool| {398if prev {399count -= 1;400}401prev = v;402count403})404} else {405let mut count = init;406unary_elementwise_values(&ca, |v: bool| {407if v {408count += 1;409}410count411})412};413414out.into()415};416417out.set_sorted_flag([IsSorted::Ascending, IsSorted::Descending][reverse as usize]);418419Ok(out)420}421422fn cum_count_no_nulls(name: PlSmallStr, len: usize, reverse: bool, init: IdxSize) -> Series {423let start = 1 as IdxSize;424let end = len as IdxSize + 1;425let ca: NoNull<IdxCa> = if reverse {426(start..end).rev().map(|v| v + init).collect()427} else {428(start..end).map(|v| v + init).collect()429};430let mut ca = ca.into_inner();431ca.rename(name);432ca.into_series()433}434435436