polars_core/series/implementations/
decimal.rs1use polars_compute::arithmetic::DivRounding;
2use polars_compute::decimal::{DEC128_MAX_PREC, dec128_fits};
3use polars_compute::rolling::QuantileMethod;
4
5use super::*;
6use crate::prelude::*;
7
8unsafe impl IntoSeries for DecimalChunked {
9 fn into_series(self) -> Series {
10 Series(Arc::new(SeriesWrap(self)))
11 }
12}
13
14impl private::PrivateSeriesNumeric for SeriesWrap<DecimalChunked> {
15 fn bit_repr(&self) -> Option<BitRepr> {
16 Some(self.0.physical().to_bit_repr())
17 }
18}
19
20impl SeriesWrap<DecimalChunked> {
21 fn apply_physical_to_s<F: Fn(&Int128Chunked) -> Int128Chunked>(&self, f: F) -> Series {
22 f(self.0.physical())
23 .into_decimal_unchecked(self.0.precision(), self.0.scale())
24 .into_series()
25 }
26
27 fn apply_physical<T, F: Fn(&Int128Chunked) -> T>(&self, f: F) -> T {
28 f(self.0.physical())
29 }
30
31 fn scale_factor(&self) -> u128 {
32 10u128.pow(self.0.scale() as u32)
33 }
34
35 fn apply_scale(&self, mut scalar: Scalar) -> Scalar {
36 if scalar.is_null() {
37 return scalar;
38 }
39
40 debug_assert_eq!(scalar.dtype(), &DataType::Float64);
41 let v = scalar
42 .value()
43 .try_extract::<f64>()
44 .expect("should be f64 scalar");
45 scalar.update((v / self.scale_factor() as f64).into());
46 scalar
47 }
48
49 fn agg_helper<F: Fn(&Int128Chunked) -> Series>(&self, f: F, precision: usize) -> Series {
50 let agg_s = f(self.0.physical());
51 let scale = self.0.scale();
52 match agg_s.dtype() {
53 DataType::Int128 => {
54 let ca = agg_s.i128().unwrap();
55 let ca = ca.as_ref().clone();
56 ca.into_decimal_unchecked(precision, scale).into_series()
57 },
58 DataType::List(dtype) if matches!(dtype.as_ref(), DataType::Int128) => {
59 let dtype = self.0.dtype();
60 let ca = agg_s.list().unwrap();
61 let arr = ca.downcast_iter().next().unwrap();
62 let s = unsafe {
64 Series::from_chunks_and_dtype_unchecked(
65 PlSmallStr::EMPTY,
66 vec![arr.values().clone()],
67 dtype,
68 )
69 }
70 .into_decimal(precision, scale)
71 .unwrap();
72 let new_values = s.array_ref(0).clone();
73 let dtype = DataType::Int128;
74 let arrow_dtype =
75 ListArray::<i64>::default_datatype(dtype.to_arrow(CompatLevel::newest()));
76 let new_arr = ListArray::<i64>::new(
77 arrow_dtype,
78 arr.offsets().clone(),
79 new_values,
80 arr.validity().cloned(),
81 );
82 unsafe {
83 ListChunked::from_chunks_and_dtype_unchecked(
84 agg_s.name().clone(),
85 vec![Box::new(new_arr)],
86 DataType::List(Box::new(DataType::Decimal(precision, scale))),
87 )
88 .into_series()
89 }
90 },
91 _ => unreachable!(),
92 }
93 }
94}
95
96impl private::PrivateSeries for SeriesWrap<DecimalChunked> {
97 fn compute_len(&mut self) {
98 self.0.physical_mut().compute_len()
99 }
100
101 fn _field(&self) -> Cow<'_, Field> {
102 Cow::Owned(self.0.field())
103 }
104
105 fn _dtype(&self) -> &DataType {
106 self.0.dtype()
107 }
108 fn _get_flags(&self) -> StatisticsFlags {
109 self.0.physical().get_flags()
110 }
111 fn _set_flags(&mut self, flags: StatisticsFlags) {
112 self.0.physical_mut().set_flags(flags)
113 }
114
115 #[cfg(feature = "zip_with")]
116 fn zip_with_same_type(&self, mask: &BooleanChunked, other: &Series) -> PolarsResult<Series> {
117 let other = other.decimal()?;
118
119 Ok(self
120 .0
121 .physical()
122 .zip_with(mask, other.physical())?
123 .into_decimal_unchecked(self.0.precision(), self.0.scale())
124 .into_series())
125 }
126 fn into_total_ord_inner<'a>(&'a self) -> Box<dyn TotalOrdInner + 'a> {
127 self.0.physical().into_total_ord_inner()
128 }
129
130 fn vec_hash(
131 &self,
132 random_state: PlSeedableRandomStateQuality,
133 buf: &mut Vec<u64>,
134 ) -> PolarsResult<()> {
135 self.0.physical().vec_hash(random_state, buf)?;
136 Ok(())
137 }
138
139 fn vec_hash_combine(
140 &self,
141 build_hasher: PlSeedableRandomStateQuality,
142 hashes: &mut [u64],
143 ) -> PolarsResult<()> {
144 self.0.physical().vec_hash_combine(build_hasher, hashes)?;
145 Ok(())
146 }
147
148 #[cfg(feature = "algorithm_group_by")]
149 unsafe fn agg_sum(&self, groups: &GroupsType) -> Series {
150 self.agg_helper(
151 |ca| ca.agg_sum(groups),
152 polars_compute::decimal::DEC128_MAX_PREC,
153 )
154 }
155
156 #[cfg(feature = "algorithm_group_by")]
157 unsafe fn agg_min(&self, groups: &GroupsType) -> Series {
158 self.agg_helper(|ca| ca.agg_min(groups), self.0.precision())
159 }
160
161 #[cfg(feature = "algorithm_group_by")]
162 unsafe fn agg_max(&self, groups: &GroupsType) -> Series {
163 self.agg_helper(|ca| ca.agg_max(groups), self.0.precision())
164 }
165
166 #[cfg(feature = "algorithm_group_by")]
167 unsafe fn agg_arg_min(&self, groups: &GroupsType) -> Series {
168 self.0.physical().agg_arg_min(groups)
169 }
170
171 #[cfg(feature = "algorithm_group_by")]
172 unsafe fn agg_arg_max(&self, groups: &GroupsType) -> Series {
173 self.0.physical().agg_arg_max(groups)
174 }
175
176 #[cfg(feature = "algorithm_group_by")]
177 unsafe fn agg_list(&self, groups: &GroupsType) -> Series {
178 self.agg_helper(|ca| ca.agg_list(groups), self.0.precision())
179 }
180
181 #[cfg(feature = "algorithm_group_by")]
182 unsafe fn agg_var(&self, groups: &GroupsType, ddof: u8) -> Series {
183 self.0
184 .cast(&DataType::Float64)
185 .unwrap()
186 .agg_var(groups, ddof)
187 }
188
189 #[cfg(feature = "algorithm_group_by")]
190 unsafe fn agg_std(&self, groups: &GroupsType, ddof: u8) -> Series {
191 self.0
192 .cast(&DataType::Float64)
193 .unwrap()
194 .agg_std(groups, ddof)
195 }
196
197 fn subtract(&self, rhs: &Series) -> PolarsResult<Series> {
198 let rhs = rhs.decimal()?;
199 ((&self.0) - rhs).map(|ca| ca.into_series())
200 }
201 fn add_to(&self, rhs: &Series) -> PolarsResult<Series> {
202 let rhs = rhs.decimal()?;
203 ((&self.0) + rhs).map(|ca| ca.into_series())
204 }
205 fn multiply(&self, rhs: &Series) -> PolarsResult<Series> {
206 let rhs = rhs.decimal()?;
207 ((&self.0) * rhs).map(|ca| ca.into_series())
208 }
209 fn divide(&self, rhs: &Series) -> PolarsResult<Series> {
210 let rhs = rhs.decimal()?;
211 ((&self.0) / rhs).map(|ca| ca.into_series())
212 }
213 fn remainder(&self, rhs: &Series) -> PolarsResult<Series> {
214 let rhs = rhs.decimal()?;
215 self.0
216 .rem_with(rhs, DivRounding::Floor)
217 .map(|ca| ca.into_series())
218 }
219 #[cfg(feature = "algorithm_group_by")]
220 fn group_tuples(&self, multithreaded: bool, sorted: bool) -> PolarsResult<GroupsType> {
221 self.0.physical().group_tuples(multithreaded, sorted)
222 }
223 fn arg_sort_multiple(
224 &self,
225 by: &[Column],
226 options: &SortMultipleOptions,
227 ) -> PolarsResult<IdxCa> {
228 self.0.physical().arg_sort_multiple(by, options)
229 }
230}
231
232impl SeriesTrait for SeriesWrap<DecimalChunked> {
233 fn rename(&mut self, name: PlSmallStr) {
234 self.0.rename(name)
235 }
236
237 fn chunk_lengths(&self) -> ChunkLenIter<'_> {
238 self.0.physical().chunk_lengths()
239 }
240
241 fn name(&self) -> &PlSmallStr {
242 self.0.name()
243 }
244
245 fn chunks(&self) -> &Vec<ArrayRef> {
246 self.0.physical().chunks()
247 }
248 unsafe fn chunks_mut(&mut self) -> &mut Vec<ArrayRef> {
249 self.0.physical_mut().chunks_mut()
250 }
251
252 fn slice(&self, offset: i64, length: usize) -> Series {
253 self.apply_physical_to_s(|ca| ca.slice(offset, length))
254 }
255
256 fn split_at(&self, offset: i64) -> (Series, Series) {
257 let (a, b) = self.0.split_at(offset);
258 (a.into_series(), b.into_series())
259 }
260
261 fn append(&mut self, other: &Series) -> PolarsResult<()> {
262 polars_ensure!(self.0.dtype() == other.dtype(), append);
263 let mut other = other.to_physical_repr().into_owned();
264 self.0
265 .physical_mut()
266 .append_owned(std::mem::take(other._get_inner_mut().as_mut()))
267 }
268 fn append_owned(&mut self, mut other: Series) -> PolarsResult<()> {
269 polars_ensure!(self.0.dtype() == other.dtype(), append);
270 self.0.physical_mut().append_owned(std::mem::take(
271 &mut other
272 ._get_inner_mut()
273 .as_any_mut()
274 .downcast_mut::<DecimalChunked>()
275 .unwrap()
276 .phys,
277 ))
278 }
279
280 fn extend(&mut self, other: &Series) -> PolarsResult<()> {
281 polars_ensure!(self.0.dtype() == other.dtype(), extend);
282 let other = other.to_physical_repr();
287 self.0
288 .physical_mut()
289 .extend(other.as_ref().as_ref().as_ref())?;
290 Ok(())
291 }
292
293 fn filter(&self, filter: &BooleanChunked) -> PolarsResult<Series> {
294 Ok(self
295 .0
296 .physical()
297 .filter(filter)?
298 .into_decimal_unchecked(self.0.precision(), self.0.scale())
299 .into_series())
300 }
301
302 fn take(&self, indices: &IdxCa) -> PolarsResult<Series> {
303 Ok(self
304 .0
305 .physical()
306 .take(indices)?
307 .into_decimal_unchecked(self.0.precision(), self.0.scale())
308 .into_series())
309 }
310
311 unsafe fn take_unchecked(&self, indices: &IdxCa) -> Series {
312 self.0
313 .physical()
314 .take_unchecked(indices)
315 .into_decimal_unchecked(self.0.precision(), self.0.scale())
316 .into_series()
317 }
318
319 fn take_slice(&self, indices: &[IdxSize]) -> PolarsResult<Series> {
320 Ok(self
321 .0
322 .physical()
323 .take(indices)?
324 .into_decimal_unchecked(self.0.precision(), self.0.scale())
325 .into_series())
326 }
327
328 unsafe fn take_slice_unchecked(&self, indices: &[IdxSize]) -> Series {
329 self.0
330 .physical()
331 .take_unchecked(indices)
332 .into_decimal_unchecked(self.0.precision(), self.0.scale())
333 .into_series()
334 }
335
336 fn deposit(&self, validity: &Bitmap) -> Series {
337 self.0
338 .physical()
339 .deposit(validity)
340 .into_decimal_unchecked(self.0.precision(), self.0.scale())
341 .into_series()
342 }
343
344 fn len(&self) -> usize {
345 self.0.len()
346 }
347
348 fn rechunk(&self) -> Series {
349 let ca = self.0.physical().rechunk().into_owned();
350 ca.into_decimal_unchecked(self.0.precision(), self.0.scale())
351 .into_series()
352 }
353
354 fn with_validity(&self, validity: Option<Bitmap>) -> Series {
355 self.0
356 .physical()
357 .clone()
358 .with_validity(validity)
359 .into_decimal_unchecked(self.0.precision(), self.0.scale())
360 .into_series()
361 }
362
363 fn new_from_index(&self, index: usize, length: usize) -> Series {
364 self.0
365 .physical()
366 .new_from_index(index, length)
367 .into_decimal_unchecked(self.0.precision(), self.0.scale())
368 .into_series()
369 }
370
371 fn cast(&self, dtype: &DataType, cast_options: CastOptions) -> PolarsResult<Series> {
372 self.0.cast_with_options(dtype, cast_options)
373 }
374
375 #[inline]
376 unsafe fn get_unchecked(&self, index: usize) -> AnyValue<'_> {
377 self.0.get_any_value_unchecked(index)
378 }
379
380 fn sort_with(&self, options: SortOptions) -> PolarsResult<Series> {
381 Ok(self
382 .0
383 .physical()
384 .sort_with(options)
385 .into_decimal_unchecked(self.0.precision(), self.0.scale())
386 .into_series())
387 }
388
389 fn arg_sort(&self, options: SortOptions) -> IdxCa {
390 self.0.physical().arg_sort(options)
391 }
392
393 fn null_count(&self) -> usize {
394 self.0.null_count()
395 }
396
397 fn has_nulls(&self) -> bool {
398 self.0.has_nulls()
399 }
400
401 #[cfg(feature = "algorithm_group_by")]
402 fn unique(&self) -> PolarsResult<Series> {
403 Ok(self.apply_physical_to_s(|ca| ca.unique().unwrap()))
404 }
405
406 #[cfg(feature = "algorithm_group_by")]
407 fn n_unique(&self) -> PolarsResult<usize> {
408 self.0.physical().n_unique()
409 }
410
411 #[cfg(feature = "algorithm_group_by")]
412 fn arg_unique(&self) -> PolarsResult<IdxCa> {
413 self.0.physical().arg_unique()
414 }
415
416 #[cfg(feature = "algorithm_group_by")]
417 fn unique_id(&self) -> PolarsResult<(IdxSize, Vec<IdxSize>)> {
418 ChunkUnique::unique_id(self.0.physical())
419 }
420
421 fn is_null(&self) -> BooleanChunked {
422 self.0.is_null()
423 }
424
425 fn is_not_null(&self) -> BooleanChunked {
426 self.0.is_not_null()
427 }
428
429 fn reverse(&self) -> Series {
430 self.apply_physical_to_s(|ca| ca.reverse())
431 }
432
433 fn shift(&self, periods: i64) -> Series {
434 self.apply_physical_to_s(|ca| ca.shift(periods))
435 }
436
437 #[cfg(feature = "approx_unique")]
438 fn approx_n_unique(&self) -> PolarsResult<IdxSize> {
439 Ok(ChunkApproxNUnique::approx_n_unique(self.0.physical()))
440 }
441
442 fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
443 Arc::new(SeriesWrap(Clone::clone(&self.0)))
444 }
445
446 fn sum_reduce(&self) -> PolarsResult<Scalar> {
447 let DataType::Decimal(_, scale) = self.dtype() else {
448 unreachable!()
449 };
450 let scale = *scale;
451 let prec = DEC128_MAX_PREC;
452 let sum = self
453 .0
454 .physical()
455 .iter()
456 .flatten()
457 .try_fold(0i128, i128::checked_add)
458 .filter(|sum| dec128_fits(*sum, prec))
459 .ok_or_else(|| polars_err!(ComputeError: "overflow in decimal addition in sum"))?;
460 let av = AnyValue::Decimal(sum, prec, scale);
461 Ok(Scalar::new(DataType::Decimal(prec, scale), av))
462 }
463
464 fn min_reduce(&self) -> PolarsResult<Scalar> {
465 Ok(self.apply_physical(|ca| {
466 let min = ca.min();
467 let DataType::Decimal(prec, scale) = self.dtype() else {
468 unreachable!()
469 };
470 let av = if let Some(min) = min {
471 AnyValue::Decimal(min, *prec, *scale)
472 } else {
473 AnyValue::Null
474 };
475 Scalar::new(self.dtype().clone(), av)
476 }))
477 }
478
479 fn max_reduce(&self) -> PolarsResult<Scalar> {
480 Ok(self.apply_physical(|ca| {
481 let max = ca.max();
482 let DataType::Decimal(prec, scale) = self.dtype() else {
483 unreachable!()
484 };
485 let av = if let Some(m) = max {
486 AnyValue::Decimal(m, *prec, *scale)
487 } else {
488 AnyValue::Null
489 };
490 Scalar::new(self.dtype().clone(), av)
491 }))
492 }
493
494 fn _sum_as_f64(&self) -> f64 {
495 self.0.physical()._sum_as_f64() / self.scale_factor() as f64
496 }
497
498 fn mean(&self) -> Option<f64> {
499 self.0
500 .physical()
501 .mean()
502 .map(|v| v / self.scale_factor() as f64)
503 }
504 fn mean_reduce(&self) -> PolarsResult<Scalar> {
505 Ok(Scalar::new(DataType::Float64, self.mean().into()))
506 }
507
508 fn median(&self) -> Option<f64> {
509 self.0
510 .physical()
511 .median()
512 .map(|v| v / self.scale_factor() as f64)
513 }
514
515 fn median_reduce(&self) -> PolarsResult<Scalar> {
516 Ok(self.apply_scale(self.0.physical().median_reduce()))
517 }
518
519 fn std(&self, ddof: u8) -> Option<f64> {
520 self.0.cast(&DataType::Float64).ok()?.std(ddof)
521 }
522
523 fn std_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
524 self.0.cast(&DataType::Float64)?.std_reduce(ddof)
525 }
526
527 fn var(&self, ddof: u8) -> Option<f64> {
528 self.0.cast(&DataType::Float64).ok()?.var(ddof)
529 }
530
531 fn var_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
532 self.0.cast(&DataType::Float64)?.var_reduce(ddof)
533 }
534
535 fn quantile_reduce(&self, quantile: f64, method: QuantileMethod) -> PolarsResult<Scalar> {
536 self.0
537 .physical()
538 .quantile_reduce(quantile, method)
539 .map(|v| self.apply_scale(v))
540 }
541
542 fn quantiles_reduce(&self, quantiles: &[f64], method: QuantileMethod) -> PolarsResult<Scalar> {
543 let result = self.0.physical().quantiles_reduce(quantiles, method)?;
544 if let AnyValue::List(float_s) = result.value() {
545 let scale_factor = self.scale_factor() as f64;
546 let float_ca = float_s.f64().unwrap();
547 let scaled_s = float_ca
548 .iter()
549 .map(|v: Option<f64>| v.map(|f| f / scale_factor))
550 .collect::<Float64Chunked>()
551 .into_series();
552 Ok(Scalar::new(
553 DataType::List(Box::new(DataType::Float64)),
554 AnyValue::List(scaled_s),
555 ))
556 } else {
557 polars_bail!(ComputeError: "expected list scalar from quantiles_reduce")
558 }
559 }
560
561 fn find_validity_mismatch(&self, other: &Series, idxs: &mut Vec<IdxSize>) {
562 self.0.physical().find_validity_mismatch(other, idxs)
563 }
564
565 fn as_any(&self) -> &dyn Any {
566 &self.0
567 }
568
569 fn as_any_mut(&mut self) -> &mut dyn Any {
570 &mut self.0
571 }
572
573 fn as_phys_any(&self) -> &dyn Any {
574 self.0.physical()
575 }
576
577 fn as_arc_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
578 self as _
579 }
580}