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