1#![allow(unsafe_op_in_unsafe_fn)]
2use std::borrow::Cow;
4
5use arrow::datatypes::ArrowSchemaRef;
6use polars_row::ArrayRef;
7use polars_utils::UnitVec;
8use polars_utils::itertools::Itertools;
9use rayon::prelude::*;
10
11use crate::chunked_array::flags::StatisticsFlags;
12#[cfg(feature = "algorithm_group_by")]
13use crate::chunked_array::ops::unique::is_unique_helper;
14use crate::prelude::gather::check_bounds_ca;
15use crate::prelude::*;
16#[cfg(feature = "row_hash")]
17use crate::utils::split_df;
18use crate::utils::{Container, NoNull, slice_offsets, try_get_supertype};
19use crate::{HEAD_DEFAULT_LENGTH, TAIL_DEFAULT_LENGTH};
20
21#[cfg(feature = "dataframe_arithmetic")]
22mod arithmetic;
23pub mod builder;
24mod chunks;
25pub use chunks::chunk_df_for_writing;
26pub mod column;
27mod dataframe;
28mod filter;
29mod projection;
30pub use dataframe::DataFrame;
31use filter::filter_zero_width;
32use projection::{AmortizedColumnSelector, LINEAR_SEARCH_LIMIT};
33
34pub mod explode;
35mod from;
36#[cfg(feature = "algorithm_group_by")]
37pub mod group_by;
38pub(crate) mod horizontal;
39#[cfg(any(feature = "rows", feature = "object"))]
40pub mod row;
41mod top_k;
42mod upstream_traits;
43mod validation;
44
45use arrow::record_batch::{RecordBatch, RecordBatchT};
46use polars_utils::pl_str::PlSmallStr;
47#[cfg(feature = "serde")]
48use serde::{Deserialize, Serialize};
49use strum_macros::IntoStaticStr;
50
51#[cfg(feature = "row_hash")]
52use crate::hashing::_df_rows_to_hashes_threaded_vertical;
53use crate::prelude::sort::arg_sort;
54use crate::runtime::RAYON;
55use crate::series::IsSorted;
56
57#[derive(Copy, Clone, Debug, PartialEq, Eq, Default, Hash, IntoStaticStr)]
58#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
59#[cfg_attr(feature = "dsl-schema", derive(schemars::JsonSchema))]
60#[strum(serialize_all = "snake_case")]
61pub enum UniqueKeepStrategy {
62 First,
64 Last,
66 None,
68 #[default]
71 Any,
72}
73
74#[derive(Copy, Clone, Debug, PartialEq, Eq, Default, Hash, IntoStaticStr)]
75#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
76#[cfg_attr(feature = "dsl-schema", derive(schemars::JsonSchema))]
77#[strum(serialize_all = "snake_case")]
78pub enum PivotColumnNaming {
80 Combine,
82 #[default]
85 Auto,
86}
87
88impl DataFrame {
89 pub fn materialized_column_iter(&self) -> impl ExactSizeIterator<Item = &Series> {
90 self.columns().iter().map(Column::as_materialized_series)
91 }
92
93 pub fn estimated_size(&self) -> usize {
106 self.columns().iter().map(Column::estimated_size).sum()
107 }
108
109 pub fn try_apply_columns(
110 &self,
111 func: impl Fn(&Column) -> PolarsResult<Column> + Send + Sync,
112 ) -> PolarsResult<Vec<Column>> {
113 return inner(self, &func);
114
115 fn inner(
116 slf: &DataFrame,
117 func: &(dyn Fn(&Column) -> PolarsResult<Column> + Send + Sync),
118 ) -> PolarsResult<Vec<Column>> {
119 slf.columns().iter().map(func).collect()
120 }
121 }
122
123 pub fn apply_columns(&self, func: impl Fn(&Column) -> Column + Send + Sync) -> Vec<Column> {
124 return inner(self, &func);
125
126 fn inner(slf: &DataFrame, func: &(dyn Fn(&Column) -> Column + Send + Sync)) -> Vec<Column> {
127 slf.columns().iter().map(func).collect()
128 }
129 }
130
131 pub fn try_apply_columns_par(
132 &self,
133 func: impl Fn(&Column) -> PolarsResult<Column> + Send + Sync,
134 ) -> PolarsResult<Vec<Column>> {
135 return inner(self, &func);
136
137 fn inner(
138 slf: &DataFrame,
139 func: &(dyn Fn(&Column) -> PolarsResult<Column> + Send + Sync),
140 ) -> PolarsResult<Vec<Column>> {
141 RAYON.install(|| slf.columns().par_iter().map(func).collect())
142 }
143 }
144
145 pub fn apply_columns_par(&self, func: impl Fn(&Column) -> Column + Send + Sync) -> Vec<Column> {
146 return inner(self, &func);
147
148 fn inner(slf: &DataFrame, func: &(dyn Fn(&Column) -> Column + Send + Sync)) -> Vec<Column> {
149 RAYON.install(|| slf.columns().par_iter().map(func).collect())
150 }
151 }
152
153 pub(crate) fn reserve_chunks(&mut self, additional: usize) {
155 for s in unsafe { self.columns_mut_retain_schema() } {
156 if let Column::Series(s) = s {
157 unsafe { s.chunks_mut().reserve(additional) }
160 }
161 }
162 }
163 pub fn new_from_index(&self, index: usize, height: usize) -> Self {
164 let new_cols = self.apply_columns(|c| c.new_from_index(index, height));
165
166 unsafe { Self::_new_unchecked_impl(height, new_cols).with_schema_from(self) }
167 }
168
169 pub fn full_null(schema: &Schema, height: usize) -> Self {
171 let columns = schema
172 .iter_fields()
173 .map(|f| Column::full_null(f.name().clone(), height, f.dtype()))
174 .collect();
175
176 unsafe { DataFrame::_new_unchecked_impl(height, columns) }
177 }
178
179 pub fn ensure_matches_schema(&mut self, schema: &Schema) -> PolarsResult<()> {
182 let mut did_cast = false;
183 let cached_schema = self.cached_schema().cloned();
184
185 for (col, (name, dt)) in unsafe { self.columns_mut() }.iter_mut().zip(schema.iter()) {
186 polars_ensure!(
187 col.name() == name,
188 SchemaMismatch: "column name mismatch: expected {:?}, found {:?}",
189 name,
190 col.name()
191 );
192
193 let needs_cast = col.dtype().matches_schema_type(dt)?;
194
195 if needs_cast {
196 *col = col.cast(dt)?;
197 did_cast = true;
198 }
199 }
200
201 if !did_cast {
202 unsafe { self.set_opt_schema(cached_schema) };
203 }
204
205 Ok(())
206 }
207
208 pub fn with_row_index(&self, name: PlSmallStr, offset: Option<IdxSize>) -> PolarsResult<Self> {
243 let mut new_columns = Vec::with_capacity(self.width() + 1);
244 let offset = offset.unwrap_or(0);
245
246 if self.get_column_index(&name).is_some() {
247 polars_bail!(duplicate = name)
248 }
249
250 let col = Column::new_row_index(name, offset, self.height())?;
251 new_columns.push(col);
252 new_columns.extend_from_slice(self.columns());
253
254 Ok(unsafe { DataFrame::new_unchecked(self.height(), new_columns) })
255 }
256
257 pub unsafe fn with_row_index_mut(
265 &mut self,
266 name: PlSmallStr,
267 offset: Option<IdxSize>,
268 ) -> &mut Self {
269 debug_assert!(
270 self.get_column_index(&name).is_none(),
271 "with_row_index_mut(): column with name {} already exists",
272 name
273 );
274
275 let offset = offset.unwrap_or(0);
276 let col = Column::new_row_index(name, offset, self.height()).unwrap();
277
278 unsafe { self.columns_mut() }.insert(0, col);
279 self
280 }
281
282 pub fn shrink_to_fit(&mut self) {
284 for s in unsafe { self.columns_mut_retain_schema() } {
286 s.shrink_to_fit();
287 }
288 }
289
290 pub fn rechunk_mut_par(&mut self) -> &mut Self {
293 if self.columns().iter().any(|c| c.n_chunks() > 1) {
294 RAYON.install(|| {
295 unsafe { self.columns_mut_retain_schema() }
296 .par_iter_mut()
297 .for_each(|c| *c = c.rechunk());
298 })
299 }
300
301 self
302 }
303
304 pub fn rechunk_mut(&mut self) -> &mut Self {
306 let columns = unsafe { self.columns_mut() };
308
309 for col in columns.iter_mut().filter(|c| c.n_chunks() > 1) {
310 *col = col.rechunk();
311 }
312
313 self
314 }
315
316 pub fn should_rechunk(&self) -> bool {
318 if !self
321 .columns()
322 .iter()
323 .filter_map(|c| c.as_series().map(|s| s.n_chunks()))
324 .all_equal()
325 {
326 return true;
327 }
328
329 let mut chunk_lengths = self.materialized_column_iter().map(|s| s.chunk_lengths());
331 match chunk_lengths.next() {
332 None => false,
333 Some(first_column_chunk_lengths) => {
334 if first_column_chunk_lengths.size_hint().0 == 1 {
336 return chunk_lengths.any(|cl| cl.size_hint().0 != 1);
337 }
338 let height = self.height();
341 let n_chunks = first_column_chunk_lengths.size_hint().0;
342 if n_chunks > height && !(height == 0 && n_chunks == 1) {
343 return true;
344 }
345 let v: Vec<_> = first_column_chunk_lengths.collect();
347 for cl in chunk_lengths {
348 if cl.enumerate().any(|(idx, el)| Some(&el) != v.get(idx)) {
349 return true;
350 }
351 }
352 false
353 },
354 }
355 }
356
357 pub fn align_chunks_par(&mut self) -> &mut Self {
359 if self.should_rechunk() {
360 self.rechunk_mut_par()
361 } else {
362 self
363 }
364 }
365
366 pub fn align_chunks(&mut self) -> &mut Self {
368 if self.should_rechunk() {
369 self.rechunk_mut()
370 } else {
371 self
372 }
373 }
374
375 pub fn get_column_names(&self) -> Vec<&PlSmallStr> {
386 self.columns().iter().map(|s| s.name()).collect()
387 }
388
389 pub fn get_column_names_owned(&self) -> Vec<PlSmallStr> {
391 self.columns().iter().map(|s| s.name().clone()).collect()
392 }
393
394 pub fn set_column_names<T>(&mut self, new_names: &[T]) -> PolarsResult<()>
406 where
407 T: AsRef<str>,
408 {
409 polars_ensure!(
410 new_names.len() == self.width(),
411 ShapeMismatch: "{} column names provided for a DataFrame of width {}",
412 new_names.len(), self.width()
413 );
414
415 validation::ensure_names_unique(new_names)?;
416
417 *unsafe { self.columns_mut() } = std::mem::take(unsafe { self.columns_mut() })
418 .into_iter()
419 .zip(new_names)
420 .map(|(c, name)| c.with_name(PlSmallStr::from_str(name.as_ref())))
421 .collect();
422
423 Ok(())
424 }
425
426 pub fn dtypes(&self) -> Vec<DataType> {
439 self.columns().iter().map(|s| s.dtype().clone()).collect()
440 }
441
442 pub fn first_col_n_chunks(&self) -> usize {
444 match self.columns().iter().find_map(|col| col.as_series()) {
445 None if self.width() == 0 => 0,
446 None => 1,
447 Some(s) => s.n_chunks(),
448 }
449 }
450
451 pub fn max_n_chunks(&self) -> usize {
453 self.columns()
454 .iter()
455 .map(|s| s.as_series().map(|s| s.n_chunks()).unwrap_or(1))
456 .max()
457 .unwrap_or(0)
458 }
459
460 pub fn fields(&self) -> Vec<Field> {
476 self.columns()
477 .iter()
478 .map(|s| s.field().into_owned())
479 .collect()
480 }
481
482 pub fn hstack(&self, columns: &[Column]) -> PolarsResult<Self> {
516 let mut new_cols = Vec::with_capacity(self.width() + columns.len());
517
518 new_cols.extend(self.columns().iter().cloned());
519 new_cols.extend_from_slice(columns);
520
521 DataFrame::new(self.height(), new_cols)
522 }
523 pub fn vstack(&self, other: &DataFrame) -> PolarsResult<Self> {
564 let mut df = self.clone();
565 df.vstack_mut(other)?;
566 Ok(df)
567 }
568
569 pub fn vstack_mut(&mut self, other: &DataFrame) -> PolarsResult<&mut Self> {
610 if self.width() != other.width() {
611 polars_ensure!(
612 self.shape() == (0, 0),
613 ShapeMismatch:
614 "unable to append to a DataFrame of shape {:?} with a DataFrame of width {}",
615 self.shape(), other.width(),
616 );
617
618 self.clone_from(other);
619
620 return Ok(self);
621 }
622
623 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
624
625 unsafe { self.columns_mut_retain_schema() }
626 .iter_mut()
627 .zip(other.columns())
628 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
629 ensure_can_extend(&*left, right)?;
630 left.append(right)
631 .with_context(|| format!("failed to vstack column '{}'", right.name()))?;
632 Ok(())
633 })?;
634
635 unsafe { self.set_height(new_height) };
636
637 Ok(self)
638 }
639
640 pub fn vstack_mut_owned(&mut self, other: DataFrame) -> PolarsResult<&mut Self> {
641 if self.width() != other.width() {
642 polars_ensure!(
643 self.shape() == (0, 0),
644 ShapeMismatch:
645 "unable to append to a DataFrame of width {} with a DataFrame of width {}",
646 self.width(), other.width(),
647 );
648
649 *self = other;
650
651 return Ok(self);
652 }
653
654 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
655
656 unsafe { self.columns_mut_retain_schema() }
657 .iter_mut()
658 .zip(other.into_columns())
659 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
660 ensure_can_extend(&*left, &right)?;
661 let right_name = right.name().clone();
662 left.append_owned(right)
663 .with_context(|| format!("failed to vstack column '{right_name}'"))?;
664 Ok(())
665 })?;
666
667 unsafe { self.set_height(new_height) };
668
669 Ok(self)
670 }
671
672 pub fn vstack_mut_unchecked(&mut self, other: &DataFrame) -> &mut Self {
679 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
680
681 unsafe { self.columns_mut_retain_schema() }
682 .iter_mut()
683 .zip(other.columns())
684 .for_each(|(left, right)| {
685 left.append(right)
686 .with_context(|| format!("failed to vstack column '{}'", right.name()))
687 .expect("should not fail");
688 });
689
690 unsafe { self.set_height(new_height) };
691
692 self
693 }
694
695 pub fn vstack_mut_owned_unchecked(&mut self, other: DataFrame) -> &mut Self {
702 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
703
704 unsafe { self.columns_mut_retain_schema() }
705 .iter_mut()
706 .zip(other.into_columns())
707 .for_each(|(left, right)| {
708 left.append_owned(right).expect("should not fail");
709 });
710
711 unsafe { self.set_height(new_height) };
712
713 self
714 }
715
716 pub fn extend(&mut self, other: &DataFrame) -> PolarsResult<()> {
731 polars_ensure!(
732 self.width() == other.width(),
733 ShapeMismatch:
734 "unable to extend a DataFrame of width {} with a DataFrame of width {}",
735 self.width(), other.width(),
736 );
737
738 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
739
740 unsafe { self.columns_mut_retain_schema() }
741 .iter_mut()
742 .zip(other.columns())
743 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
744 ensure_can_extend(&*left, right)?;
745 left.extend(right)
746 .with_context(|| format!("failed to extend column '{}'", right.name()))?;
747 Ok(())
748 })?;
749
750 unsafe { self.set_height(new_height) };
751
752 Ok(())
753 }
754
755 pub fn drop_in_place(&mut self, name: &str) -> PolarsResult<Column> {
772 let idx = self.try_get_column_index(name)?;
773 Ok(unsafe { self.columns_mut() }.remove(idx))
774 }
775
776 pub fn drop_nulls<S>(&self, subset: Option<&[S]>) -> PolarsResult<Self>
805 where
806 for<'a> &'a S: AsRef<str>,
807 {
808 if let Some(v) = subset {
809 let v = self.select_to_vec(v)?;
810 self._drop_nulls_impl(v.as_slice())
811 } else {
812 self._drop_nulls_impl(self.columns())
813 }
814 }
815
816 fn _drop_nulls_impl(&self, subset: &[Column]) -> PolarsResult<Self> {
817 if subset.iter().all(|s| !s.has_nulls()) {
819 return Ok(self.clone());
820 }
821
822 let mut iter = subset.iter();
823
824 let mask = iter
825 .next()
826 .ok_or_else(|| polars_err!(NoData: "no data to drop nulls from"))?;
827 let mut mask = mask.is_not_null();
828
829 for c in iter {
830 mask = mask & c.is_not_null();
831 }
832 self.filter(&mask)
833 }
834
835 pub fn drop(&self, name: &str) -> PolarsResult<Self> {
850 let idx = self.try_get_column_index(name)?;
851 let mut new_cols = Vec::with_capacity(self.width() - 1);
852
853 self.columns().iter().enumerate().for_each(|(i, s)| {
854 if i != idx {
855 new_cols.push(s.clone())
856 }
857 });
858
859 Ok(unsafe { DataFrame::_new_unchecked_impl(self.height(), new_cols) })
860 }
861
862 pub fn drop_many<I, S>(&self, names: I) -> Self
864 where
865 I: IntoIterator<Item = S>,
866 S: Into<PlSmallStr>,
867 {
868 let names: PlHashSet<PlSmallStr> = names.into_iter().map(|s| s.into()).collect();
869 self.drop_many_amortized(&names)
870 }
871
872 pub fn drop_many_amortized(&self, names: &PlHashSet<PlSmallStr>) -> DataFrame {
874 if names.is_empty() {
875 return self.clone();
876 }
877 let mut new_cols = Vec::with_capacity(self.width().saturating_sub(names.len()));
878 self.columns().iter().for_each(|s| {
879 if !names.contains(s.name()) {
880 new_cols.push(s.clone())
881 }
882 });
883
884 unsafe { DataFrame::new_unchecked(self.height(), new_cols) }
885 }
886
887 fn insert_column_no_namecheck(
890 &mut self,
891 index: usize,
892 column: Column,
893 ) -> PolarsResult<&mut Self> {
894 if self.shape() == (0, 0) {
895 unsafe { self.set_height(column.len()) };
896 }
897
898 polars_ensure!(
899 column.len() == self.height(),
900 ShapeMismatch:
901 "unable to add a column of length {} to a DataFrame of height {}",
902 column.len(), self.height(),
903 );
904
905 unsafe { self.columns_mut() }.insert(index, column);
906 Ok(self)
907 }
908
909 pub fn insert_column(&mut self, index: usize, column: Column) -> PolarsResult<&mut Self> {
911 let name = column.name();
912
913 polars_ensure!(
914 self.get_column_index(name).is_none(),
915 Duplicate:
916 "column with name {:?} is already present in the DataFrame", name
917 );
918
919 self.insert_column_no_namecheck(index, column)
920 }
921
922 pub fn with_column(&mut self, mut column: Column) -> PolarsResult<&mut Self> {
925 if self.shape() == (0, 0) {
926 unsafe { self.set_height(column.len()) };
927 }
928
929 column.broadcast_in_place_to(self.height())?;
930
931 if let Some(i) = self.get_column_index(column.name()) {
932 *unsafe { self.columns_mut() }.get_mut(i).unwrap() = column
933 } else {
934 unsafe { self.columns_mut() }.push(column)
935 };
936
937 Ok(self)
938 }
939
940 pub unsafe fn push_column_unchecked(&mut self, column: Column) -> &mut Self {
946 unsafe { self.columns_mut() }.push(column);
947 self
948 }
949
950 pub fn with_columns_mut(
953 &mut self,
954 columns: impl IntoIterator<Item = Column>,
955 output_schema: &Schema,
956 ) -> PolarsResult<()> {
957 let columns = columns.into_iter();
958
959 unsafe {
960 self.columns_mut_retain_schema()
961 .reserve(columns.size_hint().0)
962 }
963
964 for c in columns {
965 self.with_column_and_schema_mut(c, output_schema)?;
966 }
967
968 Ok(())
969 }
970
971 fn with_column_and_schema_mut(
972 &mut self,
973 mut column: Column,
974 output_schema: &Schema,
975 ) -> PolarsResult<&mut Self> {
976 if self.shape() == (0, 0) {
977 unsafe { self.set_height(column.len()) };
978 }
979
980 column.broadcast_in_place_to(self.height())?;
981
982 let i = output_schema
983 .index_of(column.name())
984 .or_else(|| self.get_column_index(column.name()))
985 .unwrap_or(self.width());
986
987 if i < self.width() {
988 *unsafe { self.columns_mut() }.get_mut(i).unwrap() = column
989 } else if i == self.width() {
990 unsafe { self.columns_mut() }.push(column)
991 } else {
992 panic!("{:?}, {}", output_schema, column.name());
994 }
995
996 Ok(self)
997 }
998
999 pub fn get(&self, idx: usize) -> Option<Vec<AnyValue<'_>>> {
1010 (idx < self.height()).then(|| self.columns().iter().map(|c| c.get(idx).unwrap()).collect())
1011 }
1012
1013 pub fn select_at_idx(&self, idx: usize) -> Option<&Column> {
1029 self.columns().get(idx)
1030 }
1031
1032 pub fn get_column_index(&self, name: &str) -> Option<usize> {
1050 if let Some(schema) = self.cached_schema() {
1051 schema.index_of(name)
1052 } else if self.width() <= LINEAR_SEARCH_LIMIT {
1053 self.columns().iter().position(|s| s.name() == name)
1054 } else {
1055 self.schema().index_of(name)
1056 }
1057 }
1058
1059 pub fn try_get_column_index(&self, name: &str) -> PolarsResult<usize> {
1061 self.get_column_index(name)
1062 .ok_or_else(|| polars_err!(col_not_found = name))
1063 }
1064
1065 pub fn column(&self, name: &str) -> PolarsResult<&Column> {
1079 let idx = self.try_get_column_index(name)?;
1080 Ok(self.select_at_idx(idx).unwrap())
1081 }
1082
1083 pub fn select<I, S>(&self, names: I) -> PolarsResult<Self>
1094 where
1095 I: IntoIterator<Item = S>,
1096 S: AsRef<str>,
1097 {
1098 DataFrame::new(self.height(), self.select_to_vec(names)?)
1099 }
1100
1101 pub unsafe fn select_unchecked<I, S>(&self, names: I) -> PolarsResult<Self>
1106 where
1107 I: IntoIterator<Item = S>,
1108 S: AsRef<str>,
1109 {
1110 Ok(unsafe { DataFrame::new_unchecked(self.height(), self.select_to_vec(names)?) })
1111 }
1112
1113 pub fn select_to_vec(
1131 &self,
1132 selection: impl IntoIterator<Item = impl AsRef<str>>,
1133 ) -> PolarsResult<Vec<Column>> {
1134 AmortizedColumnSelector::new(self).select_multiple(selection)
1135 }
1136
1137 pub fn filter(&self, mask: &BooleanChunked) -> PolarsResult<Self> {
1149 if self.width() == 0 {
1150 filter_zero_width(self.height(), mask)
1151 } else if mask.len() == 1 && self.len() >= 1 {
1152 if mask.all() && mask.null_count() == 0 {
1153 Ok(self.clone())
1154 } else {
1155 Ok(self.clear())
1156 }
1157 } else {
1158 let all_chunks_aligned = !self.should_rechunk()
1161 && self
1162 .materialized_column_iter()
1163 .next()
1164 .is_some_and(|s| s.chunk_lengths().eq(mask.chunk_lengths()));
1165
1166 let mask = if all_chunks_aligned {
1167 Cow::Borrowed(mask)
1168 } else {
1169 mask.rechunk()
1170 };
1171
1172 let new_columns: Vec<Column> =
1173 self.try_apply_columns_par(|s| s.filter(mask.as_ref()))?;
1174 let out = unsafe {
1175 DataFrame::new_unchecked(new_columns[0].len(), new_columns).with_schema_from(self)
1176 };
1177
1178 Ok(out)
1179 }
1180 }
1181
1182 pub fn filter_seq(&self, mask: &BooleanChunked) -> PolarsResult<Self> {
1184 if self.width() == 0 {
1185 filter_zero_width(self.height(), mask)
1186 } else if mask.len() == 1 && mask.null_count() == 0 && self.len() >= 1 {
1187 if mask.all() && mask.null_count() == 0 {
1188 Ok(self.clone())
1189 } else {
1190 Ok(self.clear())
1191 }
1192 } else {
1193 let all_chunks_aligned = !self.should_rechunk()
1194 && self
1195 .materialized_column_iter()
1196 .next()
1197 .is_some_and(|s| s.chunk_lengths().eq(mask.chunk_lengths()));
1198
1199 let mask = if all_chunks_aligned {
1200 Cow::Borrowed(mask)
1201 } else {
1202 mask.rechunk()
1203 };
1204
1205 let new_columns: Vec<Column> = self.try_apply_columns(|s| s.filter(mask.as_ref()))?;
1206 let out = unsafe {
1207 DataFrame::new_unchecked(new_columns[0].len(), new_columns).with_schema_from(self)
1208 };
1209
1210 Ok(out)
1211 }
1212 }
1213
1214 pub fn take(&self, indices: &IdxCa) -> PolarsResult<Self> {
1226 check_bounds_ca(indices, self.height().try_into().unwrap_or(IdxSize::MAX))?;
1227
1228 let new_cols = self.apply_columns_par(|c| {
1229 assert_eq!(c.len(), self.height());
1230 unsafe { c.take_unchecked(indices) }
1231 });
1232
1233 Ok(unsafe { DataFrame::new_unchecked(indices.len(), new_cols).with_schema_from(self) })
1234 }
1235
1236 pub unsafe fn take_unchecked(&self, idx: &IdxCa) -> Self {
1239 self.take_unchecked_impl(idx, true)
1240 }
1241
1242 #[cfg(feature = "algorithm_group_by")]
1245 pub unsafe fn gather_group_unchecked(&self, group: &GroupsIndicator) -> Self {
1246 match group {
1247 GroupsIndicator::Idx((_, indices)) => unsafe {
1248 self.take_slice_unchecked_impl(indices.as_slice(), false)
1249 },
1250 GroupsIndicator::Slice([offset, len]) => self.slice(*offset as i64, *len as usize),
1251 }
1252 }
1253
1254 pub unsafe fn take_unchecked_impl(&self, idx: &IdxCa, allow_threads: bool) -> Self {
1257 let cols = if allow_threads && RAYON.current_num_threads() > 1 {
1258 RAYON.install(|| {
1259 if RAYON.current_num_threads() > self.width() {
1260 let stride = usize::max(idx.len().div_ceil(RAYON.current_num_threads()), 256);
1261 if self.height() / stride >= 2 {
1262 self.apply_columns_par(|c| {
1263 let c = if c.dtype().is_nested() {
1266 &c.rechunk()
1267 } else {
1268 c
1269 };
1270
1271 (0..idx.len().div_ceil(stride))
1272 .into_par_iter()
1273 .map(|i| c.take_unchecked(&idx.slice((i * stride) as i64, stride)))
1274 .reduce(
1275 || Column::new_empty(c.name().clone(), c.dtype()),
1276 |mut a, b| {
1277 a.append_owned(b).unwrap();
1278 a
1279 },
1280 )
1281 })
1282 } else {
1283 self.apply_columns_par(|c| c.take_unchecked(idx))
1284 }
1285 } else {
1286 self.apply_columns_par(|c| c.take_unchecked(idx))
1287 }
1288 })
1289 } else {
1290 self.apply_columns(|s| s.take_unchecked(idx))
1291 };
1292
1293 unsafe { DataFrame::new_unchecked(idx.len(), cols).with_schema_from(self) }
1294 }
1295
1296 pub unsafe fn take_slice_unchecked(&self, idx: &[IdxSize]) -> Self {
1299 self.take_slice_unchecked_impl(idx, true)
1300 }
1301
1302 pub unsafe fn take_slice_unchecked_impl(&self, idx: &[IdxSize], allow_threads: bool) -> Self {
1305 let cols = if allow_threads && RAYON.current_num_threads() > 1 {
1306 RAYON.install(|| {
1307 if RAYON.current_num_threads() > self.width() {
1308 let stride = usize::max(idx.len().div_ceil(RAYON.current_num_threads()), 256);
1309 if self.height() / stride >= 2 {
1310 self.apply_columns_par(|c| {
1311 let c = if c.dtype().is_nested() {
1314 &c.rechunk()
1315 } else {
1316 c
1317 };
1318
1319 (0..idx.len().div_ceil(stride))
1320 .into_par_iter()
1321 .map(|i| {
1322 let idx = &idx[i * stride..];
1323 let idx = &idx[..idx.len().min(stride)];
1324 c.take_slice_unchecked(idx)
1325 })
1326 .reduce(
1327 || Column::new_empty(c.name().clone(), c.dtype()),
1328 |mut a, b| {
1329 a.append_owned(b).unwrap();
1330 a
1331 },
1332 )
1333 })
1334 } else {
1335 self.apply_columns_par(|s| s.take_slice_unchecked(idx))
1336 }
1337 } else {
1338 self.apply_columns_par(|s| s.take_slice_unchecked(idx))
1339 }
1340 })
1341 } else {
1342 self.apply_columns(|s| s.take_slice_unchecked(idx))
1343 };
1344 unsafe { DataFrame::new_unchecked(idx.len(), cols).with_schema_from(self) }
1345 }
1346
1347 pub fn rename(&mut self, column: &str, name: PlSmallStr) -> PolarsResult<&mut Self> {
1362 if column == name.as_str() {
1363 return Ok(self);
1364 }
1365 polars_ensure!(
1366 !self.schema().contains(&name),
1367 Duplicate: "column rename attempted with already existing name \"{name}\""
1368 );
1369
1370 self.get_column_index(column)
1371 .and_then(|idx| unsafe { self.columns_mut() }.get_mut(idx))
1372 .ok_or_else(|| polars_err!(col_not_found = column))
1373 .map(|c| c.rename(name))?;
1374
1375 Ok(self)
1376 }
1377
1378 pub fn rename_many<'a>(
1379 mut self,
1380 renames: impl Iterator<Item = (&'a str, PlSmallStr)>,
1381 ) -> PolarsResult<Self> {
1382 let schema = self.schema().clone();
1383
1384 for (from, to) in renames {
1385 if from == to.as_str() {
1386 continue;
1387 }
1388
1389 let idx = schema
1390 .index_of(from)
1391 .ok_or_else(|| polars_err!(col_not_found = from))?;
1392
1393 unsafe { self.columns_mut() }
1394 .get_mut(idx)
1395 .unwrap()
1396 .rename(to);
1397 }
1398
1399 let schema = Schema::from_iter_check_duplicates(
1401 self.columns()
1402 .iter()
1403 .map(|c| c.name().clone())
1404 .zip_eq(schema.iter_values().cloned()),
1405 )?;
1406
1407 unsafe { self.set_schema(Arc::new(schema)) };
1408
1409 Ok(self)
1410 }
1411
1412 pub fn sort_in_place(
1416 &mut self,
1417 by: impl IntoIterator<Item = impl AsRef<str>>,
1418 sort_options: SortMultipleOptions,
1419 ) -> PolarsResult<&mut Self> {
1420 let by_column = self.select_to_vec(by)?;
1421
1422 let mut out = self.sort_impl(by_column, sort_options, None)?;
1423 unsafe { out.set_schema_from(self) };
1424
1425 *self = out;
1426
1427 Ok(self)
1428 }
1429
1430 #[doc(hidden)]
1431 pub fn sort_impl(
1433 &self,
1434 by_column: Vec<Column>,
1435 sort_options: SortMultipleOptions,
1436 slice: Option<(i64, usize)>,
1437 ) -> PolarsResult<Self> {
1438 if by_column.is_empty() {
1439 return if let Some((offset, len)) = slice {
1441 Ok(self.slice(offset, len))
1442 } else {
1443 Ok(self.clone())
1444 };
1445 }
1446
1447 for column in &by_column {
1448 if column.dtype().is_object() {
1449 polars_bail!(
1450 InvalidOperation: "column '{}' has a dtype of '{}', which does not support sorting", column.name(), column.dtype()
1451 )
1452 }
1453 }
1454
1455 let first_descending = sort_options.descending[0];
1460 let first_by_column = by_column[0].name().to_string();
1461
1462 let set_sorted = |df: &mut DataFrame| {
1463 let _ = df.apply(&first_by_column, |s| {
1466 let mut s = s.clone();
1467 if first_descending {
1468 s.set_sorted_flag(IsSorted::Descending)
1469 } else {
1470 s.set_sorted_flag(IsSorted::Ascending)
1471 }
1472 s
1473 });
1474 };
1475
1476 if self.shape_has_zero() {
1477 let mut out = self.clone();
1478 set_sorted(&mut out);
1479 return Ok(out);
1480 }
1481
1482 if let Some((0, k)) = slice {
1483 if k < self.height() {
1484 return self.bottom_k_impl(k, by_column, sort_options);
1485 }
1486 }
1487 #[cfg(feature = "dtype-categorical")]
1491 let is_not_categorical_enum =
1492 !(matches!(by_column[0].dtype(), DataType::Categorical(_, _))
1493 || matches!(by_column[0].dtype(), DataType::Enum(_, _)));
1494
1495 #[cfg(not(feature = "dtype-categorical"))]
1496 #[allow(non_upper_case_globals)]
1497 const is_not_categorical_enum: bool = true;
1498
1499 if by_column.len() == 1 && is_not_categorical_enum {
1500 let required_sorting = if sort_options.descending[0] {
1501 IsSorted::Descending
1502 } else {
1503 IsSorted::Ascending
1504 };
1505 let no_sorting_required = (by_column[0].is_sorted_flag() == required_sorting)
1508 && ((by_column[0].null_count() == 0)
1509 || by_column[0].get(by_column[0].len() - 1).unwrap().is_null()
1510 == sort_options.nulls_last[0]);
1511
1512 if no_sorting_required {
1513 return if let Some((offset, len)) = slice {
1514 Ok(self.slice(offset, len))
1515 } else {
1516 Ok(self.clone())
1517 };
1518 }
1519 }
1520
1521 let has_nested = by_column.iter().any(|s| s.dtype().is_nested());
1522 let allow_threads = sort_options.multithreaded;
1523
1524 let mut df = self.clone();
1526 let df = df.rechunk_mut_par();
1527 let mut take = match (by_column.len(), has_nested) {
1528 (1, false) => {
1529 let s = &by_column[0];
1530 let options = SortOptions {
1531 descending: sort_options.descending[0],
1532 nulls_last: sort_options.nulls_last[0],
1533 multithreaded: sort_options.multithreaded,
1534 maintain_order: sort_options.maintain_order,
1535 limit: sort_options.limit,
1536 };
1537 if df.width() == 1 && df.try_get_column_index(s.name().as_str()).is_ok() {
1541 let mut out = s.sort_with(options)?;
1542 if let Some((offset, len)) = slice {
1543 out = out.slice(offset, len);
1544 }
1545 return Ok(out.into_frame());
1546 }
1547 s.arg_sort(options)
1548 },
1549 _ => arg_sort(&by_column, sort_options)?,
1550 };
1551
1552 if let Some((offset, len)) = slice {
1553 take = take.slice(offset, len);
1554 }
1555
1556 let mut df = unsafe { df.take_unchecked_impl(&take, allow_threads) };
1559 set_sorted(&mut df);
1560 Ok(df)
1561 }
1562
1563 pub fn _to_metadata(&self) -> DataFrame {
1568 let num_columns = self.width();
1569
1570 let mut column_names =
1571 StringChunkedBuilder::new(PlSmallStr::from_static("column_name"), num_columns);
1572 let mut repr_ca = StringChunkedBuilder::new(PlSmallStr::from_static("repr"), num_columns);
1573 let mut sorted_asc_ca =
1574 BooleanChunkedBuilder::new(PlSmallStr::from_static("sorted_asc"), num_columns);
1575 let mut sorted_dsc_ca =
1576 BooleanChunkedBuilder::new(PlSmallStr::from_static("sorted_dsc"), num_columns);
1577 let mut fast_explode_list_ca =
1578 BooleanChunkedBuilder::new(PlSmallStr::from_static("fast_explode_list"), num_columns);
1579 let mut materialized_at_ca =
1580 StringChunkedBuilder::new(PlSmallStr::from_static("materialized_at"), num_columns);
1581
1582 for col in self.columns() {
1583 let flags = col.get_flags();
1584
1585 let (repr, materialized_at) = match col {
1586 Column::Series(s) => ("series", s.materialized_at()),
1587 Column::Scalar(_) => ("scalar", None),
1588 };
1589 let sorted_asc = flags.contains(StatisticsFlags::IS_SORTED_ASC);
1590 let sorted_dsc = flags.contains(StatisticsFlags::IS_SORTED_DSC);
1591 let fast_explode_list = flags.contains(StatisticsFlags::CAN_FAST_EXPLODE_LIST);
1592
1593 column_names.append_value(col.name().clone());
1594 repr_ca.append_value(repr);
1595 sorted_asc_ca.append_value(sorted_asc);
1596 sorted_dsc_ca.append_value(sorted_dsc);
1597 fast_explode_list_ca.append_value(fast_explode_list);
1598 materialized_at_ca.append_option(materialized_at.map(|v| format!("{v:#?}")));
1599 }
1600
1601 unsafe {
1602 DataFrame::new_unchecked(
1603 self.width(),
1604 vec![
1605 column_names.finish().into_column(),
1606 repr_ca.finish().into_column(),
1607 sorted_asc_ca.finish().into_column(),
1608 sorted_dsc_ca.finish().into_column(),
1609 fast_explode_list_ca.finish().into_column(),
1610 materialized_at_ca.finish().into_column(),
1611 ],
1612 )
1613 }
1614 }
1615 pub fn sort(
1653 &self,
1654 by: impl IntoIterator<Item = impl AsRef<str>>,
1655 sort_options: SortMultipleOptions,
1656 ) -> PolarsResult<Self> {
1657 let mut df = self.clone();
1658 df.sort_in_place(by, sort_options)?;
1659 Ok(df)
1660 }
1661
1662 pub fn replace(&mut self, column: &str, new_col: Column) -> PolarsResult<&mut Self> {
1677 self.apply(column, |_| new_col)
1678 }
1679
1680 pub fn replace_column(&mut self, index: usize, new_column: Column) -> PolarsResult<&mut Self> {
1695 polars_ensure!(
1696 index < self.width(),
1697 ShapeMismatch:
1698 "unable to replace at index {}, the DataFrame has only {} columns",
1699 index, self.width(),
1700 );
1701
1702 polars_ensure!(
1703 new_column.len() == self.height(),
1704 ShapeMismatch:
1705 "unable to replace a column, series length {} doesn't match the DataFrame height {}",
1706 new_column.len(), self.height(),
1707 );
1708
1709 unsafe { *self.columns_mut().get_mut(index).unwrap() = new_column };
1710
1711 Ok(self)
1712 }
1713
1714 pub fn apply<F, C>(&mut self, name: &str, f: F) -> PolarsResult<&mut Self>
1755 where
1756 F: FnOnce(&Column) -> C,
1757 C: IntoColumn,
1758 {
1759 let idx = self.try_get_column_index(name)?;
1760 self.apply_at_idx(idx, f)?;
1761 Ok(self)
1762 }
1763
1764 pub fn apply_at_idx<F, C>(&mut self, idx: usize, f: F) -> PolarsResult<&mut Self>
1795 where
1796 F: FnOnce(&Column) -> C,
1797 C: IntoColumn,
1798 {
1799 let df_height = self.height();
1800 let width = self.width();
1801
1802 let cached_schema = self.cached_schema().cloned();
1803
1804 let col = unsafe { self.columns_mut() }.get_mut(idx).ok_or_else(|| {
1805 polars_err!(
1806 ComputeError: "invalid column index: {} for a DataFrame with {} columns",
1807 idx, width
1808 )
1809 })?;
1810
1811 let new_col = f(col)
1812 .into_column()
1813 .with_name(col.name().clone())
1814 .broadcast_owned_to(df_height)?;
1815 let col_before = std::mem::replace(col, new_col);
1816
1817 if col.dtype() == col_before.dtype() {
1818 unsafe { self.set_opt_schema(cached_schema) };
1819 }
1820
1821 Ok(self)
1822 }
1823
1824 pub fn try_apply_at_idx<F, C>(&mut self, idx: usize, f: F) -> PolarsResult<&mut Self>
1865 where
1866 F: FnOnce(&Column) -> PolarsResult<C>,
1867 C: IntoColumn,
1868 {
1869 let df_height = self.height();
1870 let width = self.width();
1871
1872 let cached_schema = self.cached_schema().cloned();
1873
1874 let col = unsafe { self.columns_mut() }.get_mut(idx).ok_or_else(|| {
1875 polars_err!(
1876 ComputeError: "invalid column index: {} for a DataFrame with {} columns",
1877 idx, width
1878 )
1879 })?;
1880
1881 let mut new_col = f(col).map(|c| c.into_column())?;
1882
1883 polars_ensure!(
1884 new_col.len() == df_height,
1885 ShapeMismatch:
1886 "try_apply_at_idx: resulting Series has length {} while the DataFrame has height {}",
1887 new_col.len(), df_height
1888 );
1889
1890 new_col = new_col.with_name(col.name().clone());
1892 let col_before = std::mem::replace(col, new_col);
1893
1894 if col.dtype() == col_before.dtype() {
1895 unsafe { self.set_opt_schema(cached_schema) };
1896 }
1897
1898 Ok(self)
1899 }
1900
1901 pub fn try_apply<F, C>(&mut self, column: &str, f: F) -> PolarsResult<&mut Self>
1944 where
1945 F: FnOnce(&Series) -> PolarsResult<C>,
1946 C: IntoColumn,
1947 {
1948 let idx = self.try_get_column_index(column)?;
1949 self.try_apply_at_idx(idx, |c| f(c.as_materialized_series()))
1950 }
1951
1952 #[must_use]
1982 pub fn slice(&self, offset: i64, length: usize) -> Self {
1983 if offset == 0 && length == self.height() {
1984 return self.clone();
1985 }
1986
1987 if length == 0 {
1988 return self.clear();
1989 }
1990
1991 let cols = self.apply_columns(|s| s.slice(offset, length));
1992
1993 let height = if let Some(fst) = cols.first() {
1994 fst.len()
1995 } else {
1996 let (_, length) = slice_offsets(offset, length, self.height());
1997 length
1998 };
1999
2000 unsafe { DataFrame::_new_unchecked_impl(height, cols).with_schema_from(self) }
2001 }
2002
2003 pub fn split_at(&self, offset: i64) -> (Self, Self) {
2005 let (a, b) = self.columns().iter().map(|s| s.split_at(offset)).unzip();
2006
2007 let (idx, _) = slice_offsets(offset, 0, self.height());
2008
2009 let a = unsafe { DataFrame::new_unchecked(idx, a).with_schema_from(self) };
2010 let b = unsafe { DataFrame::new_unchecked(self.height() - idx, b).with_schema_from(self) };
2011 (a, b)
2012 }
2013
2014 #[must_use]
2015 pub fn clear(&self) -> Self {
2016 let cols = self.columns().iter().map(|s| s.clear()).collect::<Vec<_>>();
2017 unsafe { DataFrame::_new_unchecked_impl(0, cols).with_schema_from(self) }
2018 }
2019
2020 #[must_use]
2021 pub fn slice_par(&self, offset: i64, length: usize) -> Self {
2022 if offset == 0 && length == self.height() {
2023 return self.clone();
2024 }
2025 let columns = self.apply_columns_par(|s| s.slice(offset, length));
2026 unsafe { DataFrame::new_unchecked(length, columns).with_schema_from(self) }
2027 }
2028
2029 #[must_use]
2030 pub fn _slice_and_realloc(&self, offset: i64, length: usize) -> Self {
2031 if offset == 0 && length == self.height() {
2032 return self.clone();
2033 }
2034 let columns = self.apply_columns(|s| {
2036 let mut out = s.slice(offset, length);
2037 out.shrink_to_fit();
2038 out
2039 });
2040 unsafe { DataFrame::new_unchecked(length, columns).with_schema_from(self) }
2041 }
2042
2043 #[must_use]
2077 pub fn head(&self, length: Option<usize>) -> Self {
2078 let new_height = usize::min(self.height(), length.unwrap_or(HEAD_DEFAULT_LENGTH));
2079 let new_cols = self.apply_columns(|c| c.head(Some(new_height)));
2080
2081 unsafe { DataFrame::new_unchecked(new_height, new_cols).with_schema_from(self) }
2082 }
2083
2084 #[must_use]
2115 pub fn tail(&self, length: Option<usize>) -> Self {
2116 let new_height = usize::min(self.height(), length.unwrap_or(TAIL_DEFAULT_LENGTH));
2117 let new_cols = self.apply_columns(|c| c.tail(Some(new_height)));
2118
2119 unsafe { DataFrame::new_unchecked(new_height, new_cols).with_schema_from(self) }
2120 }
2121
2122 pub fn iter_chunks(
2132 &self,
2133 compat_level: CompatLevel,
2134 parallel: bool,
2135 ) -> impl Iterator<Item = RecordBatch> + '_ {
2136 debug_assert!(!self.should_rechunk(), "expected equal chunks");
2137
2138 if self.width() == 0 {
2139 return RecordBatchIterWrap::new_zero_width(self.height());
2140 }
2141
2142 let must_convert = compat_level.0 == 0;
2145 let parallel = parallel
2146 && must_convert
2147 && self.width() > 1
2148 && self
2149 .columns()
2150 .iter()
2151 .any(|s| matches!(s.dtype(), DataType::String | DataType::Binary));
2152
2153 RecordBatchIterWrap::Batches(RecordBatchIter {
2154 df: self,
2155 schema: Arc::new(
2156 self.columns()
2157 .iter()
2158 .map(|c| c.field().to_arrow(compat_level))
2159 .collect(),
2160 ),
2161 idx: 0,
2162 n_chunks: usize::max(1, self.first_col_n_chunks()),
2163 compat_level,
2164 parallel,
2165 })
2166 }
2167
2168 pub fn iter_chunks_physical(&self) -> impl Iterator<Item = RecordBatch> + '_ {
2178 debug_assert!(!self.should_rechunk());
2179
2180 if self.width() == 0 {
2181 return RecordBatchIterWrap::new_zero_width(self.height());
2182 }
2183
2184 RecordBatchIterWrap::PhysicalBatches(PhysRecordBatchIter {
2185 schema: Arc::new(
2186 self.columns()
2187 .iter()
2188 .map(|c| c.field().to_arrow(CompatLevel::newest()))
2189 .collect(),
2190 ),
2191 arr_iters: self
2192 .materialized_column_iter()
2193 .map(|s| s.chunks().iter())
2194 .collect(),
2195 })
2196 }
2197
2198 #[must_use]
2200 pub fn reverse(&self) -> Self {
2201 let new_cols = self.apply_columns(Column::reverse);
2202 unsafe { DataFrame::new_unchecked(self.height(), new_cols).with_schema_from(self) }
2203 }
2204
2205 #[must_use]
2210 pub fn shift(&self, periods: i64) -> Self {
2211 let col = self.apply_columns_par(|s| s.shift(periods));
2212 unsafe { DataFrame::new_unchecked(self.height(), col).with_schema_from(self) }
2213 }
2214
2215 pub fn fill_null(&self, strategy: FillNullStrategy) -> PolarsResult<Self> {
2224 let col = self.try_apply_columns_par(|s| s.fill_null(strategy))?;
2225
2226 Ok(unsafe { DataFrame::new_unchecked(self.height(), col) })
2227 }
2228
2229 pub fn pipe<F, B>(self, f: F) -> PolarsResult<B>
2231 where
2232 F: Fn(DataFrame) -> PolarsResult<B>,
2233 {
2234 f(self)
2235 }
2236
2237 pub fn pipe_mut<F, B>(&mut self, f: F) -> PolarsResult<B>
2239 where
2240 F: Fn(&mut DataFrame) -> PolarsResult<B>,
2241 {
2242 f(self)
2243 }
2244
2245 pub fn pipe_with_args<F, B, Args>(self, f: F, args: Args) -> PolarsResult<B>
2247 where
2248 F: Fn(DataFrame, Args) -> PolarsResult<B>,
2249 {
2250 f(self, args)
2251 }
2252 #[cfg(feature = "algorithm_group_by")]
2286 pub fn unique_stable(
2287 &self,
2288 subset: Option<&[String]>,
2289 keep: UniqueKeepStrategy,
2290 slice: Option<(i64, usize)>,
2291 ) -> PolarsResult<DataFrame> {
2292 self.unique_impl(
2293 true,
2294 subset.map(|v| v.iter().map(|x| PlSmallStr::from_str(x.as_str())).collect()),
2295 keep,
2296 slice,
2297 )
2298 }
2299
2300 #[cfg(feature = "algorithm_group_by")]
2302 pub fn unique<I, S>(
2303 &self,
2304 subset: Option<&[String]>,
2305 keep: UniqueKeepStrategy,
2306 slice: Option<(i64, usize)>,
2307 ) -> PolarsResult<DataFrame> {
2308 self.unique_impl(
2309 false,
2310 subset.map(|v| v.iter().map(|x| PlSmallStr::from_str(x.as_str())).collect()),
2311 keep,
2312 slice,
2313 )
2314 }
2315
2316 #[cfg(feature = "algorithm_group_by")]
2317 pub fn unique_impl(
2318 &self,
2319 maintain_order: bool,
2320 subset: Option<Vec<PlSmallStr>>,
2321 keep: UniqueKeepStrategy,
2322 slice: Option<(i64, usize)>,
2323 ) -> PolarsResult<Self> {
2324 if self.width() == 0 {
2325 let height = usize::min(self.height(), 1);
2326 return Ok(DataFrame::empty_with_height(height));
2327 }
2328
2329 let names = subset.unwrap_or_else(|| self.get_column_names_owned());
2330 let mut df = self.clone();
2331 df.rechunk_mut_par();
2333
2334 let columns = match (keep, maintain_order) {
2335 (UniqueKeepStrategy::First | UniqueKeepStrategy::Any, true) => {
2336 let gb = df.group_by_stable(names)?;
2337 let groups = gb.get_groups();
2338 let (offset, len) = slice.unwrap_or((0, groups.len()));
2339 let groups = groups.slice(offset, len);
2340 df.apply_columns_par(|s| unsafe { s.agg_first(&groups) })
2341 },
2342 (UniqueKeepStrategy::Last, true) => {
2343 let gb = df.group_by_stable(names)?;
2346 let groups = gb.get_groups();
2347
2348 let last_idx: NoNull<IdxCa> = groups
2349 .iter()
2350 .map(|g| match g {
2351 GroupsIndicator::Idx((_first, idx)) => idx[idx.len() - 1],
2352 GroupsIndicator::Slice([first, len]) => first + len - 1,
2353 })
2354 .collect();
2355
2356 let mut last_idx = last_idx.into_inner().sort(false);
2357
2358 if let Some((offset, len)) = slice {
2359 last_idx = last_idx.slice(offset, len);
2360 }
2361
2362 let last_idx = NoNull::new(last_idx);
2363 let out = unsafe { df.take_unchecked(&last_idx) };
2364 return Ok(out);
2365 },
2366 (UniqueKeepStrategy::First | UniqueKeepStrategy::Any, false) => {
2367 let gb = df.group_by(names)?;
2368 let groups = gb.get_groups();
2369 let (offset, len) = slice.unwrap_or((0, groups.len()));
2370 let groups = groups.slice(offset, len);
2371 df.apply_columns_par(|s| unsafe { s.agg_first(&groups) })
2372 },
2373 (UniqueKeepStrategy::Last, false) => {
2374 let gb = df.group_by(names)?;
2375 let groups = gb.get_groups();
2376 let (offset, len) = slice.unwrap_or((0, groups.len()));
2377 let groups = groups.slice(offset, len);
2378 df.apply_columns_par(|s| unsafe { s.agg_last(&groups) })
2379 },
2380 (UniqueKeepStrategy::None, _) => {
2381 let df_part = df.select(names)?;
2382 let mask = df_part.is_unique()?;
2383 let mut filtered = df.filter(&mask)?;
2384
2385 if let Some((offset, len)) = slice {
2386 filtered = filtered.slice(offset, len);
2387 }
2388 return Ok(filtered);
2389 },
2390 };
2391 Ok(unsafe { DataFrame::new_unchecked_infer_height(columns).with_schema_from(self) })
2392 }
2393
2394 #[cfg(feature = "algorithm_group_by")]
2408 pub fn is_unique(&self) -> PolarsResult<BooleanChunked> {
2409 let gb = self.group_by(self.get_column_names_owned())?;
2410 let groups = gb.get_groups();
2411 Ok(is_unique_helper(
2412 groups,
2413 self.height() as IdxSize,
2414 true,
2415 false,
2416 ))
2417 }
2418
2419 #[cfg(feature = "algorithm_group_by")]
2433 pub fn is_duplicated(&self) -> PolarsResult<BooleanChunked> {
2434 let gb = self.group_by(self.get_column_names_owned())?;
2435 let groups = gb.get_groups();
2436 Ok(is_unique_helper(
2437 groups,
2438 self.height() as IdxSize,
2439 false,
2440 true,
2441 ))
2442 }
2443
2444 #[must_use]
2446 pub fn null_count(&self) -> Self {
2447 let cols =
2448 self.apply_columns(|c| Column::new(c.name().clone(), [c.null_count() as IdxSize]));
2449 unsafe { Self::new_unchecked(1, cols) }
2450 }
2451
2452 #[cfg(feature = "row_hash")]
2454 pub fn hash_rows(
2455 &mut self,
2456 hasher_builder: Option<PlSeedableRandomStateQuality>,
2457 ) -> PolarsResult<UInt64Chunked> {
2458 let dfs = split_df(self, RAYON.current_num_threads(), false);
2459 let (cas, _) = _df_rows_to_hashes_threaded_vertical(&dfs, hasher_builder)?;
2460
2461 let mut iter = cas.into_iter();
2462 let mut acc_ca = iter.next().unwrap();
2463 for ca in iter {
2464 acc_ca.append(&ca)?;
2465 }
2466 Ok(acc_ca.rechunk().into_owned())
2467 }
2468
2469 pub fn get_supertype(&self) -> Option<PolarsResult<DataType>> {
2471 self.columns()
2472 .iter()
2473 .map(|s| Ok(s.dtype().clone()))
2474 .reduce(|acc, b| try_get_supertype(&acc?, &b.unwrap()))
2475 }
2476
2477 #[doc(hidden)]
2482 pub unsafe fn _take_unchecked_slice(&self, idx: &[IdxSize], allow_threads: bool) -> Self {
2483 self._take_unchecked_slice_sorted(idx, allow_threads, IsSorted::Not)
2484 }
2485
2486 #[doc(hidden)]
2493 pub unsafe fn _take_unchecked_slice_sorted(
2494 &self,
2495 idx: &[IdxSize],
2496 allow_threads: bool,
2497 sorted: IsSorted,
2498 ) -> Self {
2499 #[cfg(debug_assertions)]
2500 {
2501 if idx.len() > 2 {
2502 use crate::series::IsSorted;
2503
2504 match sorted {
2505 IsSorted::Ascending => {
2506 assert!(idx[0] <= idx[idx.len() - 1]);
2507 },
2508 IsSorted::Descending => {
2509 assert!(idx[0] >= idx[idx.len() - 1]);
2510 },
2511 _ => {},
2512 }
2513 }
2514 }
2515 let mut ca = IdxCa::mmap_slice(PlSmallStr::EMPTY, idx);
2516 ca.set_sorted_flag(sorted);
2517 self.take_unchecked_impl(&ca, allow_threads)
2518 }
2519 #[cfg(all(feature = "partition_by", feature = "algorithm_group_by"))]
2520 #[doc(hidden)]
2521 pub fn _partition_by_impl(
2522 &self,
2523 cols: &[PlSmallStr],
2524 stable: bool,
2525 include_key: bool,
2526 parallel: bool,
2527 ) -> PolarsResult<Vec<DataFrame>> {
2528 let selected_keys = self.select_to_vec(cols.iter().cloned())?;
2529 let groups = self.group_by_with_series(selected_keys, parallel, stable)?;
2530 let groups = groups.into_groups();
2531
2532 let df = if include_key {
2534 self.clone()
2535 } else {
2536 self.drop_many(cols.iter().cloned())
2537 };
2538
2539 if parallel {
2540 RAYON.install(|| {
2543 match groups.as_ref() {
2544 GroupsType::Idx(idx) => {
2545 let mut df = df.clone();
2547 df.rechunk_mut_par();
2548 Ok(idx
2549 .into_par_iter()
2550 .map(|(_, group)| {
2551 unsafe {
2553 df._take_unchecked_slice_sorted(
2554 group,
2555 false,
2556 IsSorted::Ascending,
2557 )
2558 }
2559 })
2560 .collect())
2561 },
2562 GroupsType::Slice { groups, .. } => Ok(groups
2563 .into_par_iter()
2564 .map(|[first, len]| df.slice(*first as i64, *len as usize))
2565 .collect()),
2566 }
2567 })
2568 } else {
2569 match groups.as_ref() {
2570 GroupsType::Idx(idx) => {
2571 let mut df = df;
2573 df.rechunk_mut();
2574 Ok(idx
2575 .into_iter()
2576 .map(|(_, group)| {
2577 unsafe {
2579 df._take_unchecked_slice_sorted(group, false, IsSorted::Ascending)
2580 }
2581 })
2582 .collect())
2583 },
2584 GroupsType::Slice { groups, .. } => Ok(groups
2585 .iter()
2586 .map(|[first, len]| df.slice(*first as i64, *len as usize))
2587 .collect()),
2588 }
2589 }
2590 }
2591
2592 #[cfg(feature = "partition_by")]
2594 pub fn partition_by<I, S>(&self, cols: I, include_key: bool) -> PolarsResult<Vec<DataFrame>>
2595 where
2596 I: IntoIterator<Item = S>,
2597 S: Into<PlSmallStr>,
2598 {
2599 let cols: UnitVec<PlSmallStr> = cols.into_iter().map(Into::into).collect();
2600 self._partition_by_impl(cols.as_slice(), false, include_key, true)
2601 }
2602
2603 #[cfg(feature = "partition_by")]
2606 pub fn partition_by_stable<I, S>(
2607 &self,
2608 cols: I,
2609 include_key: bool,
2610 ) -> PolarsResult<Vec<DataFrame>>
2611 where
2612 I: IntoIterator<Item = S>,
2613 S: Into<PlSmallStr>,
2614 {
2615 let cols: UnitVec<PlSmallStr> = cols.into_iter().map(Into::into).collect();
2616 self._partition_by_impl(cols.as_slice(), true, include_key, true)
2617 }
2618
2619 #[cfg(feature = "dtype-struct")]
2622 pub fn unnest(
2623 &self,
2624 cols: impl IntoIterator<Item = impl Into<PlSmallStr>>,
2625 separator: Option<&str>,
2626 ) -> PolarsResult<DataFrame> {
2627 self.unnest_impl(cols.into_iter().map(Into::into).collect(), separator)
2628 }
2629
2630 #[cfg(feature = "dtype-struct")]
2631 fn unnest_impl(
2632 &self,
2633 cols: PlHashSet<PlSmallStr>,
2634 separator: Option<&str>,
2635 ) -> PolarsResult<DataFrame> {
2636 let mut new_cols = Vec::with_capacity(std::cmp::min(self.width() * 2, self.width() + 128));
2637 let mut count = 0;
2638 for s in self.columns() {
2639 if cols.contains(s.name()) {
2640 let ca = s.struct_()?.clone();
2641 new_cols.extend(ca.fields_as_series().into_iter().map(|mut f| {
2642 if let Some(separator) = &separator {
2643 f.rename(polars_utils::format_pl_smallstr!(
2644 "{}{}{}",
2645 s.name(),
2646 separator,
2647 f.name()
2648 ));
2649 }
2650 Column::from(f)
2651 }));
2652 count += 1;
2653 } else {
2654 new_cols.push(s.clone())
2655 }
2656 }
2657 if count != cols.len() {
2658 let schema = self.schema();
2661 for col in cols {
2662 let _ = schema
2663 .get(col.as_str())
2664 .ok_or_else(|| polars_err!(col_not_found = col))?;
2665 }
2666 }
2667
2668 DataFrame::new(self.height(), new_cols)
2669 }
2670
2671 pub fn append_record_batch(&mut self, rb: RecordBatchT<ArrayRef>) -> PolarsResult<()> {
2672 let df = DataFrame::from(rb);
2675 polars_ensure!(
2676 self.schema() == df.schema(),
2677 SchemaMismatch: "cannot append record batch with different schema\n\n
2678 Got {:?}\nexpected: {:?}", df.schema(), self.schema(),
2679 );
2680 self.vstack_mut_owned_unchecked(df);
2681 Ok(())
2682 }
2683}
2684
2685pub struct RecordBatchIter<'a> {
2686 df: &'a DataFrame,
2687 schema: ArrowSchemaRef,
2688 idx: usize,
2689 n_chunks: usize,
2690 compat_level: CompatLevel,
2691 parallel: bool,
2692}
2693
2694impl Iterator for RecordBatchIter<'_> {
2695 type Item = RecordBatch;
2696
2697 fn next(&mut self) -> Option<Self::Item> {
2698 if self.idx >= self.n_chunks {
2699 return None;
2700 }
2701
2702 let batch_cols: Vec<ArrayRef> = if self.parallel {
2704 let iter = self
2705 .df
2706 .columns()
2707 .par_iter()
2708 .map(Column::as_materialized_series)
2709 .map(|s| s.to_arrow(self.idx, self.compat_level));
2710 RAYON.install(|| iter.collect())
2711 } else {
2712 self.df
2713 .columns()
2714 .iter()
2715 .map(Column::as_materialized_series)
2716 .map(|s| s.to_arrow(self.idx, self.compat_level))
2717 .collect()
2718 };
2719
2720 let length = batch_cols.first().map_or(0, |arr| arr.len());
2721
2722 self.idx += 1;
2723
2724 Some(RecordBatch::new(length, self.schema.clone(), batch_cols))
2725 }
2726
2727 fn size_hint(&self) -> (usize, Option<usize>) {
2728 let n = self.n_chunks - self.idx;
2729 (n, Some(n))
2730 }
2731}
2732
2733pub struct PhysRecordBatchIter<'a> {
2734 schema: ArrowSchemaRef,
2735 arr_iters: Vec<std::slice::Iter<'a, ArrayRef>>,
2736}
2737
2738impl Iterator for PhysRecordBatchIter<'_> {
2739 type Item = RecordBatch;
2740
2741 fn next(&mut self) -> Option<Self::Item> {
2742 let arrs = self
2743 .arr_iters
2744 .iter_mut()
2745 .map(|phys_iter| phys_iter.next().cloned())
2746 .collect::<Option<Vec<_>>>()?;
2747
2748 let length = arrs.first().map_or(0, |arr| arr.len());
2749 Some(RecordBatch::new(length, self.schema.clone(), arrs))
2750 }
2751
2752 fn size_hint(&self) -> (usize, Option<usize>) {
2753 if let Some(iter) = self.arr_iters.first() {
2754 iter.size_hint()
2755 } else {
2756 (0, None)
2757 }
2758 }
2759}
2760
2761pub enum RecordBatchIterWrap<'a> {
2762 ZeroWidth {
2763 remaining_height: usize,
2764 chunk_size: usize,
2765 },
2766 Batches(RecordBatchIter<'a>),
2767 PhysicalBatches(PhysRecordBatchIter<'a>),
2768}
2769
2770impl<'a> RecordBatchIterWrap<'a> {
2771 fn new_zero_width(height: usize) -> Self {
2772 Self::ZeroWidth {
2773 remaining_height: height,
2774 chunk_size: polars_config::config().ideal_morsel_size() as usize,
2775 }
2776 }
2777}
2778
2779impl Iterator for RecordBatchIterWrap<'_> {
2780 type Item = RecordBatch;
2781
2782 fn next(&mut self) -> Option<Self::Item> {
2783 match self {
2784 Self::ZeroWidth {
2785 remaining_height,
2786 chunk_size,
2787 } => {
2788 let n = usize::min(*remaining_height, *chunk_size);
2789 *remaining_height -= n;
2790
2791 (n > 0).then(|| RecordBatch::new(n, ArrowSchemaRef::default(), vec![]))
2792 },
2793 Self::Batches(v) => v.next(),
2794 Self::PhysicalBatches(v) => v.next(),
2795 }
2796 }
2797
2798 fn size_hint(&self) -> (usize, Option<usize>) {
2799 match self {
2800 Self::ZeroWidth {
2801 remaining_height,
2802 chunk_size,
2803 } => {
2804 let n = remaining_height.div_ceil(*chunk_size);
2805 (n, Some(n))
2806 },
2807 Self::Batches(v) => v.size_hint(),
2808 Self::PhysicalBatches(v) => v.size_hint(),
2809 }
2810 }
2811}
2812
2813fn ensure_can_extend(left: &Column, right: &Column) -> PolarsResult<()> {
2815 polars_ensure!(
2816 left.name() == right.name(),
2817 ShapeMismatch: "unable to vstack, column names don't match: {:?} and {:?}",
2818 left.name(), right.name(),
2819 );
2820 Ok(())
2821}
2822
2823#[cfg(test)]
2824mod test {
2825 use super::*;
2826
2827 fn create_frame() -> DataFrame {
2828 let s0 = Column::new("days".into(), [0, 1, 2].as_ref());
2829 let s1 = Column::new("temp".into(), [22.1, 19.9, 7.].as_ref());
2830 DataFrame::new_infer_height(vec![s0, s1]).unwrap()
2831 }
2832
2833 #[test]
2834 #[cfg_attr(miri, ignore)]
2835 fn test_recordbatch_iterator() {
2836 let df = df!(
2837 "foo" => [1, 2, 3, 4, 5]
2838 )
2839 .unwrap();
2840 let mut iter = df.iter_chunks(CompatLevel::newest(), false);
2841 assert_eq!(5, iter.next().unwrap().len());
2842 assert!(iter.next().is_none());
2843 }
2844
2845 #[test]
2846 #[cfg_attr(miri, ignore)]
2847 fn test_select() {
2848 let df = create_frame();
2849 assert_eq!(
2850 df.column("days")
2851 .unwrap()
2852 .as_series()
2853 .unwrap()
2854 .equal(1)
2855 .unwrap()
2856 .sum(),
2857 Some(1)
2858 );
2859 }
2860
2861 #[test]
2862 #[cfg_attr(miri, ignore)]
2863 fn test_filter_broadcast_on_string_col() {
2864 let col_name = "some_col";
2865 let v = vec!["test".to_string()];
2866 let s0 = Column::new(PlSmallStr::from_str(col_name), v);
2867 let mut df = DataFrame::new_infer_height(vec![s0]).unwrap();
2868
2869 df = df
2870 .filter(
2871 &df.column(col_name)
2872 .unwrap()
2873 .as_materialized_series()
2874 .equal("")
2875 .unwrap(),
2876 )
2877 .unwrap();
2878 assert_eq!(
2879 df.column(col_name)
2880 .unwrap()
2881 .as_materialized_series()
2882 .n_chunks(),
2883 1
2884 );
2885 }
2886
2887 #[test]
2888 #[cfg_attr(miri, ignore)]
2889 fn test_filter_broadcast_on_list_col() {
2890 let s1 = Series::new(PlSmallStr::EMPTY, [true, false, true]);
2891 let ll: ListChunked = [&s1].iter().copied().collect();
2892
2893 let mask = BooleanChunked::from_slice(PlSmallStr::EMPTY, &[false]);
2894 let new = ll.filter(&mask).unwrap();
2895
2896 assert_eq!(new.chunks.len(), 1);
2897 assert_eq!(new.len(), 0);
2898 }
2899
2900 #[test]
2901 fn slice() {
2902 let df = create_frame();
2903 let sliced_df = df.slice(0, 2);
2904 assert_eq!(sliced_df.shape(), (2, 2));
2905 }
2906
2907 #[test]
2908 fn rechunk_false() {
2909 let df = create_frame();
2910 assert!(!df.should_rechunk())
2911 }
2912
2913 #[test]
2914 fn rechunk_true() -> PolarsResult<()> {
2915 let mut base = df!(
2916 "a" => [1, 2, 3],
2917 "b" => [1, 2, 3]
2918 )?;
2919
2920 let mut s = Series::new("foo".into(), 0..2);
2922 let s2 = Series::new("bar".into(), 0..1);
2923 s.append(&s2)?;
2924
2925 let out = base.with_column(s.into_column())?;
2927
2928 assert!(out.should_rechunk());
2930 Ok(())
2931 }
2932
2933 #[test]
2934 fn test_duplicate_column() {
2935 let mut df = df! {
2936 "foo" => [1, 2, 3]
2937 }
2938 .unwrap();
2939 assert!(
2941 df.with_column(Column::new("foo".into(), &[1, 2, 3]))
2942 .is_ok()
2943 );
2944 assert!(
2945 df.with_column(Column::new("bar".into(), &[1, 2, 3]))
2946 .is_ok()
2947 );
2948 assert!(df.column("bar").is_ok())
2949 }
2950
2951 #[test]
2952 #[cfg_attr(miri, ignore)]
2953 fn distinct() {
2954 let df = df! {
2955 "flt" => [1., 1., 2., 2., 3., 3.],
2956 "int" => [1, 1, 2, 2, 3, 3, ],
2957 "str" => ["a", "a", "b", "b", "c", "c"]
2958 }
2959 .unwrap();
2960 let df = df
2961 .unique_stable(None, UniqueKeepStrategy::First, None)
2962 .unwrap()
2963 .sort(["flt"], SortMultipleOptions::default())
2964 .unwrap();
2965 let valid = df! {
2966 "flt" => [1., 2., 3.],
2967 "int" => [1, 2, 3],
2968 "str" => ["a", "b", "c"]
2969 }
2970 .unwrap();
2971 assert!(df.equals(&valid));
2972 }
2973
2974 #[test]
2975 fn test_vstack() {
2976 let mut df = df! {
2978 "flt" => [1., 1., 2., 2., 3., 3.],
2979 "int" => [1, 1, 2, 2, 3, 3, ],
2980 "str" => ["a", "a", "b", "b", "c", "c"]
2981 }
2982 .unwrap();
2983
2984 df.vstack_mut(&df.slice(0, 3)).unwrap();
2985 assert_eq!(df.first_col_n_chunks(), 2)
2986 }
2987
2988 #[test]
2989 fn test_vstack_on_empty_dataframe() {
2990 let mut df = DataFrame::empty();
2991
2992 let df_data = df! {
2993 "flt" => [1., 1., 2., 2., 3., 3.],
2994 "int" => [1, 1, 2, 2, 3, 3, ],
2995 "str" => ["a", "a", "b", "b", "c", "c"]
2996 }
2997 .unwrap();
2998
2999 df.vstack_mut(&df_data).unwrap();
3000 assert_eq!(df.height(), 6)
3001 }
3002
3003 #[test]
3004 fn test_unique_keep_none_with_slice() {
3005 let df = df! {
3006 "x" => [1, 2, 3, 2, 1]
3007 }
3008 .unwrap();
3009 let out = df
3010 .unique_stable(
3011 Some(&["x".to_string()][..]),
3012 UniqueKeepStrategy::None,
3013 Some((0, 2)),
3014 )
3015 .unwrap();
3016 let expected = df! {
3017 "x" => [3]
3018 }
3019 .unwrap();
3020 assert!(out.equals(&expected));
3021 }
3022
3023 #[test]
3024 #[cfg(feature = "dtype-i8")]
3025 fn test_apply_result_schema() {
3026 let mut df = df! {
3027 "x" => [1, 2, 3, 2, 1]
3028 }
3029 .unwrap();
3030
3031 let schema_before = df.schema().clone();
3032 df.apply("x", |f| f.cast(&DataType::Int8).unwrap()).unwrap();
3033 assert_ne!(&schema_before, df.schema());
3034 }
3035}