polars_core/series/implementations/
decimal.rs1use polars_compute::decimal::{DEC128_MAX_PREC, dec128_add_scaled};
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_ord_inner<'a>(&'a self) -> Box<dyn TotalOrdInner + 'a> {
126 self.0.physical().into_total_ord_inner()
127 }
128
129 fn vec_hash(
130 &self,
131 random_state: PlSeedableRandomStateQuality,
132 buf: &mut Vec<u64>,
133 ) -> PolarsResult<()> {
134 self.0.physical().vec_hash(random_state, buf)?;
135 Ok(())
136 }
137
138 fn vec_hash_combine(
139 &self,
140 build_hasher: PlSeedableRandomStateQuality,
141 hashes: &mut [u64],
142 ) -> PolarsResult<()> {
143 self.0.physical().vec_hash_combine(build_hasher, hashes)?;
144 Ok(())
145 }
146
147 #[cfg(feature = "algorithm_group_by")]
148 unsafe fn agg_sum(&self, groups: &GroupsType) -> Series {
149 self.agg_helper(
150 |ca| ca.agg_sum(groups),
151 polars_compute::decimal::DEC128_MAX_PREC,
152 )
153 }
154
155 #[cfg(feature = "algorithm_group_by")]
156 unsafe fn agg_min(&self, groups: &GroupsType) -> Series {
157 self.agg_helper(|ca| ca.agg_min(groups), self.0.precision())
158 }
159
160 #[cfg(feature = "algorithm_group_by")]
161 unsafe fn agg_max(&self, groups: &GroupsType) -> Series {
162 self.agg_helper(|ca| ca.agg_max(groups), self.0.precision())
163 }
164
165 #[cfg(feature = "algorithm_group_by")]
166 unsafe fn agg_arg_min(&self, groups: &GroupsType) -> Series {
167 self.0.physical().agg_arg_min(groups)
168 }
169
170 #[cfg(feature = "algorithm_group_by")]
171 unsafe fn agg_arg_max(&self, groups: &GroupsType) -> Series {
172 self.0.physical().agg_arg_max(groups)
173 }
174
175 #[cfg(feature = "algorithm_group_by")]
176 unsafe fn agg_list(&self, groups: &GroupsType) -> Series {
177 self.agg_helper(|ca| ca.agg_list(groups), self.0.precision())
178 }
179
180 #[cfg(feature = "algorithm_group_by")]
181 unsafe fn agg_var(&self, groups: &GroupsType, ddof: u8) -> Series {
182 self.0
183 .cast(&DataType::Float64)
184 .unwrap()
185 .agg_var(groups, ddof)
186 }
187
188 #[cfg(feature = "algorithm_group_by")]
189 unsafe fn agg_std(&self, groups: &GroupsType, ddof: u8) -> Series {
190 self.0
191 .cast(&DataType::Float64)
192 .unwrap()
193 .agg_std(groups, ddof)
194 }
195
196 fn subtract(&self, rhs: &Series) -> PolarsResult<Series> {
197 let rhs = rhs.decimal()?;
198 ((&self.0) - rhs).map(|ca| ca.into_series())
199 }
200 fn add_to(&self, rhs: &Series) -> PolarsResult<Series> {
201 let rhs = rhs.decimal()?;
202 ((&self.0) + rhs).map(|ca| ca.into_series())
203 }
204 fn multiply(&self, rhs: &Series) -> PolarsResult<Series> {
205 let rhs = rhs.decimal()?;
206 ((&self.0) * rhs).map(|ca| ca.into_series())
207 }
208 fn divide(&self, rhs: &Series) -> PolarsResult<Series> {
209 let rhs = rhs.decimal()?;
210 ((&self.0) / rhs).map(|ca| ca.into_series())
211 }
212 fn remainder(&self, rhs: &Series) -> PolarsResult<Series> {
213 let rhs = rhs.decimal()?;
214 self.0.rem_with(rhs, true).map(|ca| ca.into_series())
215 }
216 #[cfg(feature = "algorithm_group_by")]
217 fn group_tuples(&self, multithreaded: bool, sorted: bool) -> PolarsResult<GroupsType> {
218 self.0.physical().group_tuples(multithreaded, sorted)
219 }
220 fn arg_sort_multiple(
221 &self,
222 by: &[Column],
223 options: &SortMultipleOptions,
224 ) -> PolarsResult<IdxCa> {
225 self.0.physical().arg_sort_multiple(by, options)
226 }
227}
228
229impl SeriesTrait for SeriesWrap<DecimalChunked> {
230 fn rename(&mut self, name: PlSmallStr) {
231 self.0.rename(name)
232 }
233
234 fn chunk_lengths(&self) -> ChunkLenIter<'_> {
235 self.0.physical().chunk_lengths()
236 }
237
238 fn name(&self) -> &PlSmallStr {
239 self.0.name()
240 }
241
242 fn chunks(&self) -> &Vec<ArrayRef> {
243 self.0.physical().chunks()
244 }
245 unsafe fn chunks_mut(&mut self) -> &mut Vec<ArrayRef> {
246 self.0.physical_mut().chunks_mut()
247 }
248
249 fn slice(&self, offset: i64, length: usize) -> Series {
250 self.apply_physical_to_s(|ca| ca.slice(offset, length))
251 }
252
253 fn split_at(&self, offset: i64) -> (Series, Series) {
254 let (a, b) = self.0.split_at(offset);
255 (a.into_series(), b.into_series())
256 }
257
258 fn append(&mut self, other: &Series) -> PolarsResult<()> {
259 polars_ensure!(self.0.dtype() == other.dtype(), append);
260 let mut other = other.to_physical_repr().into_owned();
261 self.0
262 .physical_mut()
263 .append_owned(std::mem::take(other._get_inner_mut().as_mut()))
264 }
265 fn append_owned(&mut self, mut other: Series) -> PolarsResult<()> {
266 polars_ensure!(self.0.dtype() == other.dtype(), append);
267 self.0.physical_mut().append_owned(std::mem::take(
268 &mut other
269 ._get_inner_mut()
270 .as_any_mut()
271 .downcast_mut::<DecimalChunked>()
272 .unwrap()
273 .phys,
274 ))
275 }
276
277 fn extend(&mut self, other: &Series) -> PolarsResult<()> {
278 polars_ensure!(self.0.dtype() == other.dtype(), extend);
279 let other = other.to_physical_repr();
284 self.0
285 .physical_mut()
286 .extend(other.as_ref().as_ref().as_ref())?;
287 Ok(())
288 }
289
290 fn filter(&self, filter: &BooleanChunked) -> PolarsResult<Series> {
291 Ok(self
292 .0
293 .physical()
294 .filter(filter)?
295 .into_decimal_unchecked(self.0.precision(), self.0.scale())
296 .into_series())
297 }
298
299 fn take(&self, indices: &IdxCa) -> PolarsResult<Series> {
300 Ok(self
301 .0
302 .physical()
303 .take(indices)?
304 .into_decimal_unchecked(self.0.precision(), self.0.scale())
305 .into_series())
306 }
307
308 unsafe fn take_unchecked(&self, indices: &IdxCa) -> Series {
309 self.0
310 .physical()
311 .take_unchecked(indices)
312 .into_decimal_unchecked(self.0.precision(), self.0.scale())
313 .into_series()
314 }
315
316 fn take_slice(&self, indices: &[IdxSize]) -> PolarsResult<Series> {
317 Ok(self
318 .0
319 .physical()
320 .take(indices)?
321 .into_decimal_unchecked(self.0.precision(), self.0.scale())
322 .into_series())
323 }
324
325 unsafe fn take_slice_unchecked(&self, indices: &[IdxSize]) -> Series {
326 self.0
327 .physical()
328 .take_unchecked(indices)
329 .into_decimal_unchecked(self.0.precision(), self.0.scale())
330 .into_series()
331 }
332
333 fn deposit(&self, validity: &Bitmap) -> Series {
334 self.0
335 .physical()
336 .deposit(validity)
337 .into_decimal_unchecked(self.0.precision(), self.0.scale())
338 .into_series()
339 }
340
341 fn len(&self) -> usize {
342 self.0.len()
343 }
344
345 fn rechunk(&self) -> Series {
346 let ca = self.0.physical().rechunk().into_owned();
347 ca.into_decimal_unchecked(self.0.precision(), self.0.scale())
348 .into_series()
349 }
350
351 fn with_validity(&self, validity: Option<Bitmap>) -> Series {
352 self.0
353 .physical()
354 .clone()
355 .with_validity(validity)
356 .into_decimal_unchecked(self.0.precision(), self.0.scale())
357 .into_series()
358 }
359
360 fn new_from_index(&self, index: usize, length: usize) -> Series {
361 self.0
362 .physical()
363 .new_from_index(index, length)
364 .into_decimal_unchecked(self.0.precision(), self.0.scale())
365 .into_series()
366 }
367
368 fn cast(&self, dtype: &DataType, cast_options: CastOptions) -> PolarsResult<Series> {
369 self.0.cast_with_options(dtype, cast_options)
370 }
371
372 #[inline]
373 unsafe fn get_unchecked(&self, index: usize) -> AnyValue<'_> {
374 self.0.get_any_value_unchecked(index)
375 }
376
377 fn sort_with(&self, options: SortOptions) -> PolarsResult<Series> {
378 Ok(self
379 .0
380 .physical()
381 .sort_with(options)
382 .into_decimal_unchecked(self.0.precision(), self.0.scale())
383 .into_series())
384 }
385
386 fn arg_sort(&self, options: SortOptions) -> IdxCa {
387 self.0.physical().arg_sort(options)
388 }
389
390 fn null_count(&self) -> usize {
391 self.0.null_count()
392 }
393
394 fn has_nulls(&self) -> bool {
395 self.0.has_nulls()
396 }
397
398 #[cfg(feature = "algorithm_group_by")]
399 fn unique(&self) -> PolarsResult<Series> {
400 Ok(self.apply_physical_to_s(|ca| ca.unique().unwrap()))
401 }
402
403 #[cfg(feature = "algorithm_group_by")]
404 fn n_unique(&self) -> PolarsResult<usize> {
405 self.0.physical().n_unique()
406 }
407
408 #[cfg(feature = "algorithm_group_by")]
409 fn arg_unique(&self) -> PolarsResult<IdxCa> {
410 self.0.physical().arg_unique()
411 }
412
413 #[cfg(feature = "algorithm_group_by")]
414 fn unique_id(&self) -> PolarsResult<(IdxSize, Vec<IdxSize>)> {
415 ChunkUnique::unique_id(self.0.physical())
416 }
417
418 fn is_null(&self) -> BooleanChunked {
419 self.0.is_null()
420 }
421
422 fn is_not_null(&self) -> BooleanChunked {
423 self.0.is_not_null()
424 }
425
426 fn reverse(&self) -> Series {
427 self.apply_physical_to_s(|ca| ca.reverse())
428 }
429
430 fn shift(&self, periods: i64) -> Series {
431 self.apply_physical_to_s(|ca| ca.shift(periods))
432 }
433
434 #[cfg(feature = "approx_unique")]
435 fn approx_n_unique(&self) -> PolarsResult<IdxSize> {
436 Ok(ChunkApproxNUnique::approx_n_unique(self.0.physical()))
437 }
438
439 fn clone_inner(&self) -> Arc<dyn SeriesTrait> {
440 Arc::new(SeriesWrap(Clone::clone(&self.0)))
441 }
442
443 fn sum_reduce(&self) -> PolarsResult<Scalar> {
444 let DataType::Decimal(_, scale) = self.dtype() else {
445 unreachable!()
446 };
447 let scale = *scale;
448 let prec = DEC128_MAX_PREC;
449 let sum = self
450 .0
451 .physical()
452 .iter()
453 .flatten()
454 .try_fold(0i128, |acc, v| {
455 dec128_add_scaled(acc, scale, v, scale, scale)
456 .ok_or_else(|| polars_err!(ComputeError: "overflow in decimal addition in sum"))
457 })?;
458 let av = AnyValue::Decimal(sum, prec, scale);
459 Ok(Scalar::new(DataType::Decimal(prec, scale), av))
460 }
461
462 fn min_reduce(&self) -> PolarsResult<Scalar> {
463 Ok(self.apply_physical(|ca| {
464 let min = ca.min();
465 let DataType::Decimal(prec, scale) = self.dtype() else {
466 unreachable!()
467 };
468 let av = if let Some(min) = min {
469 AnyValue::Decimal(min, *prec, *scale)
470 } else {
471 AnyValue::Null
472 };
473 Scalar::new(self.dtype().clone(), av)
474 }))
475 }
476
477 fn max_reduce(&self) -> PolarsResult<Scalar> {
478 Ok(self.apply_physical(|ca| {
479 let max = ca.max();
480 let DataType::Decimal(prec, scale) = self.dtype() else {
481 unreachable!()
482 };
483 let av = if let Some(m) = max {
484 AnyValue::Decimal(m, *prec, *scale)
485 } else {
486 AnyValue::Null
487 };
488 Scalar::new(self.dtype().clone(), av)
489 }))
490 }
491
492 fn _sum_as_f64(&self) -> f64 {
493 self.0.physical()._sum_as_f64() / self.scale_factor() as f64
494 }
495
496 fn mean(&self) -> Option<f64> {
497 self.0
498 .physical()
499 .mean()
500 .map(|v| v / self.scale_factor() as f64)
501 }
502 fn mean_reduce(&self) -> PolarsResult<Scalar> {
503 Ok(Scalar::new(DataType::Float64, self.mean().into()))
504 }
505
506 fn median(&self) -> Option<f64> {
507 self.0
508 .physical()
509 .median()
510 .map(|v| v / self.scale_factor() as f64)
511 }
512
513 fn median_reduce(&self) -> PolarsResult<Scalar> {
514 Ok(self.apply_scale(self.0.physical().median_reduce()))
515 }
516
517 fn std(&self, ddof: u8) -> Option<f64> {
518 self.0.cast(&DataType::Float64).ok()?.std(ddof)
519 }
520
521 fn std_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
522 self.0.cast(&DataType::Float64)?.std_reduce(ddof)
523 }
524
525 fn var(&self, ddof: u8) -> Option<f64> {
526 self.0.cast(&DataType::Float64).ok()?.var(ddof)
527 }
528
529 fn var_reduce(&self, ddof: u8) -> PolarsResult<Scalar> {
530 self.0.cast(&DataType::Float64)?.var_reduce(ddof)
531 }
532
533 fn quantile_reduce(&self, quantile: f64, method: QuantileMethod) -> PolarsResult<Scalar> {
534 self.0
535 .physical()
536 .quantile_reduce(quantile, method)
537 .map(|v| self.apply_scale(v))
538 }
539
540 fn quantiles_reduce(&self, quantiles: &[f64], method: QuantileMethod) -> PolarsResult<Scalar> {
541 let result = self.0.physical().quantiles_reduce(quantiles, method)?;
542 if let AnyValue::List(float_s) = result.value() {
543 let scale_factor = self.scale_factor() as f64;
544 let float_ca = float_s.f64().unwrap();
545 let scaled_s = float_ca
546 .iter()
547 .map(|v: Option<f64>| v.map(|f| f / scale_factor))
548 .collect::<Float64Chunked>()
549 .into_series();
550 Ok(Scalar::new(
551 DataType::List(Box::new(self.dtype().clone())),
552 AnyValue::List(scaled_s),
553 ))
554 } else {
555 polars_bail!(ComputeError: "expected list scalar from quantiles_reduce")
556 }
557 }
558
559 fn find_validity_mismatch(&self, other: &Series, idxs: &mut Vec<IdxSize>) {
560 self.0.physical().find_validity_mismatch(other, idxs)
561 }
562
563 fn as_any(&self) -> &dyn Any {
564 &self.0
565 }
566
567 fn as_any_mut(&mut self) -> &mut dyn Any {
568 &mut self.0
569 }
570
571 fn as_phys_any(&self) -> &dyn Any {
572 self.0.physical()
573 }
574
575 fn as_arc_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
576 self as _
577 }
578}