1#![allow(unsafe_op_in_unsafe_fn)]
2use arrow::datatypes::ArrowSchemaRef;
4use polars_row::ArrayRef;
5use polars_utils::UnitVec;
6use polars_utils::itertools::Itertools;
7use rayon::prelude::*;
8
9use crate::chunked_array::flags::StatisticsFlags;
10#[cfg(feature = "algorithm_group_by")]
11use crate::chunked_array::ops::unique::is_unique_helper;
12use crate::prelude::gather::check_bounds_ca;
13use crate::prelude::*;
14#[cfg(feature = "row_hash")]
15use crate::utils::split_df;
16use crate::utils::{Container, NoNull, slice_offsets, try_get_supertype};
17use crate::{HEAD_DEFAULT_LENGTH, TAIL_DEFAULT_LENGTH};
18
19#[cfg(feature = "dataframe_arithmetic")]
20mod arithmetic;
21pub mod builder;
22mod chunks;
23pub use chunks::chunk_df_for_writing;
24mod broadcast;
25pub mod column;
26mod dataframe;
27mod filter;
28mod projection;
29pub use dataframe::DataFrame;
30use filter::filter_zero_width;
31use projection::{AmortizedColumnSelector, LINEAR_SEARCH_LIMIT};
32
33pub mod explode;
34mod from;
35#[cfg(feature = "algorithm_group_by")]
36pub mod group_by;
37pub(crate) mod horizontal;
38#[cfg(any(feature = "rows", feature = "object"))]
39pub mod row;
40mod top_k;
41mod upstream_traits;
42mod validation;
43
44use arrow::record_batch::{RecordBatch, RecordBatchT};
45use polars_utils::pl_str::PlSmallStr;
46#[cfg(feature = "serde")]
47use serde::{Deserialize, Serialize};
48use strum_macros::IntoStaticStr;
49
50#[cfg(feature = "row_hash")]
51use crate::hashing::_df_rows_to_hashes_threaded_vertical;
52use crate::prelude::sort::arg_sort;
53use crate::runtime::RAYON;
54use crate::series::IsSorted;
55
56#[derive(Copy, Clone, Debug, PartialEq, Eq, Default, Hash, IntoStaticStr)]
57#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
58#[cfg_attr(feature = "dsl-schema", derive(schemars::JsonSchema))]
59#[strum(serialize_all = "snake_case")]
60pub enum UniqueKeepStrategy {
61 First,
63 Last,
65 None,
67 #[default]
70 Any,
71}
72
73#[derive(Copy, Clone, Debug, PartialEq, Eq, Default, Hash, IntoStaticStr)]
74#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
75#[cfg_attr(feature = "dsl-schema", derive(schemars::JsonSchema))]
76#[strum(serialize_all = "snake_case")]
77pub enum PivotColumnNaming {
79 Combine,
81 #[default]
84 Auto,
85}
86
87impl DataFrame {
88 pub fn materialized_column_iter(&self) -> impl ExactSizeIterator<Item = &Series> {
89 self.columns().iter().map(Column::as_materialized_series)
90 }
91
92 pub fn estimated_size(&self) -> usize {
105 self.columns().iter().map(Column::estimated_size).sum()
106 }
107
108 pub fn try_apply_columns(
109 &self,
110 func: impl Fn(&Column) -> PolarsResult<Column> + Send + Sync,
111 ) -> PolarsResult<Vec<Column>> {
112 return inner(self, &func);
113
114 fn inner(
115 slf: &DataFrame,
116 func: &(dyn Fn(&Column) -> PolarsResult<Column> + Send + Sync),
117 ) -> PolarsResult<Vec<Column>> {
118 slf.columns().iter().map(func).collect()
119 }
120 }
121
122 pub fn apply_columns(&self, func: impl Fn(&Column) -> Column + Send + Sync) -> Vec<Column> {
123 return inner(self, &func);
124
125 fn inner(slf: &DataFrame, func: &(dyn Fn(&Column) -> Column + Send + Sync)) -> Vec<Column> {
126 slf.columns().iter().map(func).collect()
127 }
128 }
129
130 pub fn try_apply_columns_par(
131 &self,
132 func: impl Fn(&Column) -> PolarsResult<Column> + Send + Sync,
133 ) -> PolarsResult<Vec<Column>> {
134 return inner(self, &func);
135
136 fn inner(
137 slf: &DataFrame,
138 func: &(dyn Fn(&Column) -> PolarsResult<Column> + Send + Sync),
139 ) -> PolarsResult<Vec<Column>> {
140 RAYON.install(|| slf.columns().par_iter().map(func).collect())
141 }
142 }
143
144 pub fn apply_columns_par(&self, func: impl Fn(&Column) -> Column + Send + Sync) -> Vec<Column> {
145 return inner(self, &func);
146
147 fn inner(slf: &DataFrame, func: &(dyn Fn(&Column) -> Column + Send + Sync)) -> Vec<Column> {
148 RAYON.install(|| slf.columns().par_iter().map(func).collect())
149 }
150 }
151
152 pub(crate) fn reserve_chunks(&mut self, additional: usize) {
154 for s in unsafe { self.columns_mut_retain_schema() } {
155 if let Column::Series(s) = s {
156 unsafe { s.chunks_mut().reserve(additional) }
159 }
160 }
161 }
162 pub fn new_from_index(&self, index: usize, height: usize) -> Self {
163 let new_cols = self.apply_columns(|c| c.new_from_index(index, height));
164
165 unsafe { Self::_new_unchecked_impl(height, new_cols).with_schema_from(self) }
166 }
167
168 pub fn full_null(schema: &Schema, height: usize) -> Self {
170 let columns = schema
171 .iter_fields()
172 .map(|f| Column::full_null(f.name().clone(), height, f.dtype()))
173 .collect();
174
175 unsafe { DataFrame::_new_unchecked_impl(height, columns) }
176 }
177
178 pub fn ensure_matches_schema(&mut self, schema: &Schema) -> PolarsResult<()> {
181 let mut did_cast = false;
182 let cached_schema = self.cached_schema().cloned();
183
184 for (col, (name, dt)) in unsafe { self.columns_mut() }.iter_mut().zip(schema.iter()) {
185 polars_ensure!(
186 col.name() == name,
187 SchemaMismatch: "column name mismatch: expected {:?}, found {:?}",
188 name,
189 col.name()
190 );
191
192 let needs_cast = col.dtype().matches_schema_type(dt)?;
193
194 if needs_cast {
195 *col = col.cast(dt)?;
196 did_cast = true;
197 }
198 }
199
200 if !did_cast {
201 unsafe { self.set_opt_schema(cached_schema) };
202 }
203
204 Ok(())
205 }
206
207 pub fn with_row_index(&self, name: PlSmallStr, offset: Option<IdxSize>) -> PolarsResult<Self> {
242 let mut new_columns = Vec::with_capacity(self.width() + 1);
243 let offset = offset.unwrap_or(0);
244
245 if self.get_column_index(&name).is_some() {
246 polars_bail!(duplicate = name)
247 }
248
249 let col = Column::new_row_index(name, offset, self.height())?;
250 new_columns.push(col);
251 new_columns.extend_from_slice(self.columns());
252
253 Ok(unsafe { DataFrame::new_unchecked(self.height(), new_columns) })
254 }
255
256 pub unsafe fn with_row_index_mut(
264 &mut self,
265 name: PlSmallStr,
266 offset: Option<IdxSize>,
267 ) -> &mut Self {
268 debug_assert!(
269 self.get_column_index(&name).is_none(),
270 "with_row_index_mut(): column with name {} already exists",
271 name
272 );
273
274 let offset = offset.unwrap_or(0);
275 let col = Column::new_row_index(name, offset, self.height()).unwrap();
276
277 unsafe { self.columns_mut() }.insert(0, col);
278 self
279 }
280
281 pub fn shrink_to_fit(&mut self) {
283 for s in unsafe { self.columns_mut_retain_schema() } {
285 s.shrink_to_fit();
286 }
287 }
288
289 pub fn rechunk_mut_par(&mut self) -> &mut Self {
292 if self.columns().iter().any(|c| c.n_chunks() > 1) {
293 RAYON.install(|| {
294 unsafe { self.columns_mut_retain_schema() }
295 .par_iter_mut()
296 .for_each(|c| *c = c.rechunk());
297 })
298 }
299
300 self
301 }
302
303 pub fn rechunk_mut(&mut self) -> &mut Self {
305 let columns = unsafe { self.columns_mut() };
307
308 for col in columns.iter_mut().filter(|c| c.n_chunks() > 1) {
309 *col = col.rechunk();
310 }
311
312 self
313 }
314
315 pub fn should_rechunk(&self) -> bool {
317 if !self
320 .columns()
321 .iter()
322 .filter_map(|c| c.as_series().map(|s| s.n_chunks()))
323 .all_equal()
324 {
325 return true;
326 }
327
328 let mut chunk_lengths = self.materialized_column_iter().map(|s| s.chunk_lengths());
330 match chunk_lengths.next() {
331 None => false,
332 Some(first_column_chunk_lengths) => {
333 if first_column_chunk_lengths.size_hint().0 == 1 {
335 return chunk_lengths.any(|cl| cl.size_hint().0 != 1);
336 }
337 let height = self.height();
340 let n_chunks = first_column_chunk_lengths.size_hint().0;
341 if n_chunks > height && !(height == 0 && n_chunks == 1) {
342 return true;
343 }
344 let v: Vec<_> = first_column_chunk_lengths.collect();
346 for cl in chunk_lengths {
347 if cl.enumerate().any(|(idx, el)| Some(&el) != v.get(idx)) {
348 return true;
349 }
350 }
351 false
352 },
353 }
354 }
355
356 pub fn align_chunks_par(&mut self) -> &mut Self {
358 if self.should_rechunk() {
359 self.rechunk_mut_par()
360 } else {
361 self
362 }
363 }
364
365 pub fn align_chunks(&mut self) -> &mut Self {
367 if self.should_rechunk() {
368 self.rechunk_mut()
369 } else {
370 self
371 }
372 }
373
374 pub fn get_column_names(&self) -> Vec<&PlSmallStr> {
385 self.columns().iter().map(|s| s.name()).collect()
386 }
387
388 pub fn get_column_names_owned(&self) -> Vec<PlSmallStr> {
390 self.columns().iter().map(|s| s.name().clone()).collect()
391 }
392
393 pub fn set_column_names<T>(&mut self, new_names: &[T]) -> PolarsResult<()>
405 where
406 T: AsRef<str>,
407 {
408 polars_ensure!(
409 new_names.len() == self.width(),
410 ShapeMismatch: "{} column names provided for a DataFrame of width {}",
411 new_names.len(), self.width()
412 );
413
414 validation::ensure_names_unique(new_names)?;
415
416 *unsafe { self.columns_mut() } = std::mem::take(unsafe { self.columns_mut() })
417 .into_iter()
418 .zip(new_names)
419 .map(|(c, name)| c.with_name(PlSmallStr::from_str(name.as_ref())))
420 .collect();
421
422 Ok(())
423 }
424
425 pub fn dtypes(&self) -> Vec<DataType> {
438 self.columns().iter().map(|s| s.dtype().clone()).collect()
439 }
440
441 pub fn first_col_n_chunks(&self) -> usize {
443 match self.columns().iter().find_map(|col| col.as_series()) {
444 None if self.width() == 0 => 0,
445 None => 1,
446 Some(s) => s.n_chunks(),
447 }
448 }
449
450 pub fn max_n_chunks(&self) -> usize {
452 self.columns()
453 .iter()
454 .map(|s| s.as_series().map(|s| s.n_chunks()).unwrap_or(1))
455 .max()
456 .unwrap_or(0)
457 }
458
459 pub fn fields(&self) -> Vec<Field> {
475 self.columns()
476 .iter()
477 .map(|s| s.field().into_owned())
478 .collect()
479 }
480
481 pub fn hstack(&self, columns: &[Column]) -> PolarsResult<Self> {
515 let mut new_cols = Vec::with_capacity(self.width() + columns.len());
516
517 new_cols.extend(self.columns().iter().cloned());
518 new_cols.extend_from_slice(columns);
519
520 DataFrame::new(self.height(), new_cols)
521 }
522 pub fn vstack(&self, other: &DataFrame) -> PolarsResult<Self> {
563 let mut df = self.clone();
564 df.vstack_mut(other)?;
565 Ok(df)
566 }
567
568 pub fn vstack_mut(&mut self, other: &DataFrame) -> PolarsResult<&mut Self> {
609 if self.width() != other.width() {
610 polars_ensure!(
611 self.shape() == (0, 0),
612 ShapeMismatch:
613 "unable to append to a DataFrame of shape {:?} with a DataFrame of width {}",
614 self.shape(), other.width(),
615 );
616
617 self.clone_from(other);
618
619 return Ok(self);
620 }
621
622 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
623
624 unsafe { self.columns_mut_retain_schema() }
625 .iter_mut()
626 .zip(other.columns())
627 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
628 ensure_can_extend(&*left, right)?;
629 left.append(right)
630 .with_context(|| format!("failed to vstack column '{}'", right.name()))?;
631 Ok(())
632 })?;
633
634 unsafe { self.set_height(new_height) };
635
636 Ok(self)
637 }
638
639 pub fn vstack_mut_owned(&mut self, other: DataFrame) -> PolarsResult<&mut Self> {
640 if self.width() != other.width() {
641 polars_ensure!(
642 self.shape() == (0, 0),
643 ShapeMismatch:
644 "unable to append to a DataFrame of width {} with a DataFrame of width {}",
645 self.width(), other.width(),
646 );
647
648 *self = other;
649
650 return Ok(self);
651 }
652
653 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
654
655 unsafe { self.columns_mut_retain_schema() }
656 .iter_mut()
657 .zip(other.into_columns())
658 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
659 ensure_can_extend(&*left, &right)?;
660 let right_name = right.name().clone();
661 left.append_owned(right)
662 .with_context(|| format!("failed to vstack column '{right_name}'"))?;
663 Ok(())
664 })?;
665
666 unsafe { self.set_height(new_height) };
667
668 Ok(self)
669 }
670
671 pub fn vstack_mut_unchecked(&mut self, other: &DataFrame) -> &mut Self {
678 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
679
680 unsafe { self.columns_mut_retain_schema() }
681 .iter_mut()
682 .zip(other.columns())
683 .for_each(|(left, right)| {
684 left.append(right)
685 .with_context(|| format!("failed to vstack column '{}'", right.name()))
686 .expect("should not fail");
687 });
688
689 unsafe { self.set_height(new_height) };
690
691 self
692 }
693
694 pub fn vstack_mut_owned_unchecked(&mut self, other: DataFrame) -> &mut Self {
701 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
702
703 unsafe { self.columns_mut_retain_schema() }
704 .iter_mut()
705 .zip(other.into_columns())
706 .for_each(|(left, right)| {
707 left.append_owned(right).expect("should not fail");
708 });
709
710 unsafe { self.set_height(new_height) };
711
712 self
713 }
714
715 pub fn extend(&mut self, other: &DataFrame) -> PolarsResult<()> {
730 polars_ensure!(
731 self.width() == other.width(),
732 ShapeMismatch:
733 "unable to extend a DataFrame of width {} with a DataFrame of width {}",
734 self.width(), other.width(),
735 );
736
737 let new_height = usize::checked_add(self.height(), other.height()).unwrap();
738
739 unsafe { self.columns_mut_retain_schema() }
740 .iter_mut()
741 .zip(other.columns())
742 .try_for_each::<_, PolarsResult<_>>(|(left, right)| {
743 ensure_can_extend(&*left, right)?;
744 left.extend(right)
745 .with_context(|| format!("failed to extend column '{}'", right.name()))?;
746 Ok(())
747 })?;
748
749 unsafe { self.set_height(new_height) };
750
751 Ok(())
752 }
753
754 pub fn drop_in_place(&mut self, name: &str) -> PolarsResult<Column> {
771 let idx = self.try_get_column_index(name)?;
772 Ok(unsafe { self.columns_mut() }.remove(idx))
773 }
774
775 pub fn drop_nulls<S>(&self, subset: Option<&[S]>) -> PolarsResult<Self>
804 where
805 for<'a> &'a S: AsRef<str>,
806 {
807 if let Some(v) = subset {
808 let v = self.select_to_vec(v)?;
809 self._drop_nulls_impl(v.as_slice())
810 } else {
811 self._drop_nulls_impl(self.columns())
812 }
813 }
814
815 fn _drop_nulls_impl(&self, subset: &[Column]) -> PolarsResult<Self> {
816 if subset.iter().all(|s| !s.has_nulls()) {
818 return Ok(self.clone());
819 }
820
821 let mut iter = subset.iter();
822
823 let mask = iter
824 .next()
825 .ok_or_else(|| polars_err!(NoData: "no data to drop nulls from"))?;
826 let mut mask = mask.is_not_null();
827
828 for c in iter {
829 mask = mask & c.is_not_null();
830 }
831 self.filter(&mask)
832 }
833
834 pub fn drop(&self, name: &str) -> PolarsResult<Self> {
849 let idx = self.try_get_column_index(name)?;
850 let mut new_cols = Vec::with_capacity(self.width() - 1);
851
852 self.columns().iter().enumerate().for_each(|(i, s)| {
853 if i != idx {
854 new_cols.push(s.clone())
855 }
856 });
857
858 Ok(unsafe { DataFrame::_new_unchecked_impl(self.height(), new_cols) })
859 }
860
861 pub fn drop_many<I, S>(&self, names: I) -> Self
863 where
864 I: IntoIterator<Item = S>,
865 S: Into<PlSmallStr>,
866 {
867 let names: PlHashSet<PlSmallStr> = names.into_iter().map(|s| s.into()).collect();
868 self.drop_many_amortized(&names)
869 }
870
871 pub fn drop_many_amortized(&self, names: &PlHashSet<PlSmallStr>) -> DataFrame {
873 if names.is_empty() {
874 return self.clone();
875 }
876 let mut new_cols = Vec::with_capacity(self.width().saturating_sub(names.len()));
877 self.columns().iter().for_each(|s| {
878 if !names.contains(s.name()) {
879 new_cols.push(s.clone())
880 }
881 });
882
883 unsafe { DataFrame::new_unchecked(self.height(), new_cols) }
884 }
885
886 fn insert_column_no_namecheck(
889 &mut self,
890 index: usize,
891 column: Column,
892 ) -> PolarsResult<&mut Self> {
893 if self.shape() == (0, 0) {
894 unsafe { self.set_height(column.len()) };
895 }
896
897 polars_ensure!(
898 column.len() == self.height(),
899 ShapeMismatch:
900 "unable to add a column of length {} to a DataFrame of height {}",
901 column.len(), self.height(),
902 );
903
904 unsafe { self.columns_mut() }.insert(index, column);
905 Ok(self)
906 }
907
908 pub fn insert_column(&mut self, index: usize, column: Column) -> PolarsResult<&mut Self> {
910 let name = column.name();
911
912 polars_ensure!(
913 self.get_column_index(name).is_none(),
914 Duplicate:
915 "column with name {:?} is already present in the DataFrame", name
916 );
917
918 self.insert_column_no_namecheck(index, column)
919 }
920
921 pub fn with_column(&mut self, mut column: Column) -> PolarsResult<&mut Self> {
924 if self.shape() == (0, 0) {
925 unsafe { self.set_height(column.len()) };
926 }
927
928 if column.len() != self.height() && column.len() == 1 {
929 column = column.new_from_index(0, self.height());
930 }
931
932 polars_ensure!(
933 column.len() == self.height(),
934 ShapeMismatch: "unable to add a column of length {} to a DataFrame of height {}",
935 column.len(), self.height(),
936 );
937
938 if let Some(i) = self.get_column_index(column.name()) {
939 *unsafe { self.columns_mut() }.get_mut(i).unwrap() = column
940 } else {
941 unsafe { self.columns_mut() }.push(column)
942 };
943
944 Ok(self)
945 }
946
947 pub unsafe fn push_column_unchecked(&mut self, column: Column) -> &mut Self {
953 unsafe { self.columns_mut() }.push(column);
954 self
955 }
956
957 pub fn with_columns_mut(
960 &mut self,
961 columns: impl IntoIterator<Item = Column>,
962 output_schema: &Schema,
963 ) -> PolarsResult<()> {
964 let columns = columns.into_iter();
965
966 unsafe {
967 self.columns_mut_retain_schema()
968 .reserve(columns.size_hint().0)
969 }
970
971 for c in columns {
972 self.with_column_and_schema_mut(c, output_schema)?;
973 }
974
975 Ok(())
976 }
977
978 fn with_column_and_schema_mut(
979 &mut self,
980 mut column: Column,
981 output_schema: &Schema,
982 ) -> PolarsResult<&mut Self> {
983 if self.shape() == (0, 0) {
984 unsafe { self.set_height(column.len()) };
985 }
986
987 if column.len() != self.height() && column.len() == 1 {
988 column = column.new_from_index(0, self.height());
989 }
990
991 polars_ensure!(
992 column.len() == self.height(),
993 ShapeMismatch:
994 "unable to add a column of length {} to a DataFrame of height {}",
995 column.len(), self.height(),
996 );
997
998 let i = output_schema
999 .index_of(column.name())
1000 .or_else(|| self.get_column_index(column.name()))
1001 .unwrap_or(self.width());
1002
1003 if i < self.width() {
1004 *unsafe { self.columns_mut() }.get_mut(i).unwrap() = column
1005 } else if i == self.width() {
1006 unsafe { self.columns_mut() }.push(column)
1007 } else {
1008 panic!("{:?}, {}", output_schema, column.name());
1010 }
1011
1012 Ok(self)
1013 }
1014
1015 pub fn get(&self, idx: usize) -> Option<Vec<AnyValue<'_>>> {
1026 (idx < self.height()).then(|| self.columns().iter().map(|c| c.get(idx).unwrap()).collect())
1027 }
1028
1029 pub fn select_at_idx(&self, idx: usize) -> Option<&Column> {
1045 self.columns().get(idx)
1046 }
1047
1048 pub fn get_column_index(&self, name: &str) -> Option<usize> {
1066 if let Some(schema) = self.cached_schema() {
1067 schema.index_of(name)
1068 } else if self.width() <= LINEAR_SEARCH_LIMIT {
1069 self.columns().iter().position(|s| s.name() == name)
1070 } else {
1071 self.schema().index_of(name)
1072 }
1073 }
1074
1075 pub fn try_get_column_index(&self, name: &str) -> PolarsResult<usize> {
1077 self.get_column_index(name)
1078 .ok_or_else(|| polars_err!(col_not_found = name))
1079 }
1080
1081 pub fn column(&self, name: &str) -> PolarsResult<&Column> {
1095 let idx = self.try_get_column_index(name)?;
1096 Ok(self.select_at_idx(idx).unwrap())
1097 }
1098
1099 pub fn select<I, S>(&self, names: I) -> PolarsResult<Self>
1110 where
1111 I: IntoIterator<Item = S>,
1112 S: AsRef<str>,
1113 {
1114 DataFrame::new(self.height(), self.select_to_vec(names)?)
1115 }
1116
1117 pub unsafe fn select_unchecked<I, S>(&self, names: I) -> PolarsResult<Self>
1122 where
1123 I: IntoIterator<Item = S>,
1124 S: AsRef<str>,
1125 {
1126 Ok(unsafe { DataFrame::new_unchecked(self.height(), self.select_to_vec(names)?) })
1127 }
1128
1129 pub fn select_to_vec(
1147 &self,
1148 selection: impl IntoIterator<Item = impl AsRef<str>>,
1149 ) -> PolarsResult<Vec<Column>> {
1150 AmortizedColumnSelector::new(self).select_multiple(selection)
1151 }
1152
1153 pub fn filter(&self, mask: &BooleanChunked) -> PolarsResult<Self> {
1165 if self.width() == 0 {
1166 filter_zero_width(self.height(), mask)
1167 } else if mask.len() == 1 && self.len() >= 1 {
1168 if mask.all() && mask.null_count() == 0 {
1169 Ok(self.clone())
1170 } else {
1171 Ok(self.clear())
1172 }
1173 } else {
1174 let new_columns: Vec<Column> = self.try_apply_columns_par(|s| s.filter(mask))?;
1175 let out = unsafe {
1176 DataFrame::new_unchecked(new_columns[0].len(), new_columns).with_schema_from(self)
1177 };
1178
1179 Ok(out)
1180 }
1181 }
1182
1183 pub fn filter_seq(&self, mask: &BooleanChunked) -> PolarsResult<Self> {
1185 if self.width() == 0 {
1186 filter_zero_width(self.height(), mask)
1187 } else if mask.len() == 1 && mask.null_count() == 0 && self.len() >= 1 {
1188 if mask.all() && mask.null_count() == 0 {
1189 Ok(self.clone())
1190 } else {
1191 Ok(self.clear())
1192 }
1193 } else {
1194 let new_columns: Vec<Column> = self.try_apply_columns(|s| s.filter(mask))?;
1195 let out = unsafe {
1196 DataFrame::new_unchecked(new_columns[0].len(), new_columns).with_schema_from(self)
1197 };
1198
1199 Ok(out)
1200 }
1201 }
1202
1203 pub fn take(&self, indices: &IdxCa) -> PolarsResult<Self> {
1215 check_bounds_ca(indices, self.height().try_into().unwrap_or(IdxSize::MAX))?;
1216
1217 let new_cols = self.apply_columns_par(|c| {
1218 assert_eq!(c.len(), self.height());
1219 unsafe { c.take_unchecked(indices) }
1220 });
1221
1222 Ok(unsafe { DataFrame::new_unchecked(indices.len(), new_cols).with_schema_from(self) })
1223 }
1224
1225 pub unsafe fn take_unchecked(&self, idx: &IdxCa) -> Self {
1228 self.take_unchecked_impl(idx, true)
1229 }
1230
1231 #[cfg(feature = "algorithm_group_by")]
1234 pub unsafe fn gather_group_unchecked(&self, group: &GroupsIndicator) -> Self {
1235 match group {
1236 GroupsIndicator::Idx((_, indices)) => unsafe {
1237 self.take_slice_unchecked_impl(indices.as_slice(), false)
1238 },
1239 GroupsIndicator::Slice([offset, len]) => self.slice(*offset as i64, *len as usize),
1240 }
1241 }
1242
1243 pub unsafe fn take_unchecked_impl(&self, idx: &IdxCa, allow_threads: bool) -> Self {
1246 let cols = if allow_threads && RAYON.current_num_threads() > 1 {
1247 RAYON.install(|| {
1248 if RAYON.current_num_threads() > self.width() {
1249 let stride = usize::max(idx.len().div_ceil(RAYON.current_num_threads()), 256);
1250 if self.height() / stride >= 2 {
1251 self.apply_columns_par(|c| {
1252 let c = if c.dtype().is_nested() {
1255 &c.rechunk()
1256 } else {
1257 c
1258 };
1259
1260 (0..idx.len().div_ceil(stride))
1261 .into_par_iter()
1262 .map(|i| c.take_unchecked(&idx.slice((i * stride) as i64, stride)))
1263 .reduce(
1264 || Column::new_empty(c.name().clone(), c.dtype()),
1265 |mut a, b| {
1266 a.append_owned(b).unwrap();
1267 a
1268 },
1269 )
1270 })
1271 } else {
1272 self.apply_columns_par(|c| c.take_unchecked(idx))
1273 }
1274 } else {
1275 self.apply_columns_par(|c| c.take_unchecked(idx))
1276 }
1277 })
1278 } else {
1279 self.apply_columns(|s| s.take_unchecked(idx))
1280 };
1281
1282 unsafe { DataFrame::new_unchecked(idx.len(), cols).with_schema_from(self) }
1283 }
1284
1285 pub unsafe fn take_slice_unchecked(&self, idx: &[IdxSize]) -> Self {
1288 self.take_slice_unchecked_impl(idx, true)
1289 }
1290
1291 pub unsafe fn take_slice_unchecked_impl(&self, idx: &[IdxSize], allow_threads: bool) -> Self {
1294 let cols = if allow_threads && RAYON.current_num_threads() > 1 {
1295 RAYON.install(|| {
1296 if RAYON.current_num_threads() > self.width() {
1297 let stride = usize::max(idx.len().div_ceil(RAYON.current_num_threads()), 256);
1298 if self.height() / stride >= 2 {
1299 self.apply_columns_par(|c| {
1300 let c = if c.dtype().is_nested() {
1303 &c.rechunk()
1304 } else {
1305 c
1306 };
1307
1308 (0..idx.len().div_ceil(stride))
1309 .into_par_iter()
1310 .map(|i| {
1311 let idx = &idx[i * stride..];
1312 let idx = &idx[..idx.len().min(stride)];
1313 c.take_slice_unchecked(idx)
1314 })
1315 .reduce(
1316 || Column::new_empty(c.name().clone(), c.dtype()),
1317 |mut a, b| {
1318 a.append_owned(b).unwrap();
1319 a
1320 },
1321 )
1322 })
1323 } else {
1324 self.apply_columns_par(|s| s.take_slice_unchecked(idx))
1325 }
1326 } else {
1327 self.apply_columns_par(|s| s.take_slice_unchecked(idx))
1328 }
1329 })
1330 } else {
1331 self.apply_columns(|s| s.take_slice_unchecked(idx))
1332 };
1333 unsafe { DataFrame::new_unchecked(idx.len(), cols).with_schema_from(self) }
1334 }
1335
1336 pub fn rename(&mut self, column: &str, name: PlSmallStr) -> PolarsResult<&mut Self> {
1351 if column == name.as_str() {
1352 return Ok(self);
1353 }
1354 polars_ensure!(
1355 !self.schema().contains(&name),
1356 Duplicate: "column rename attempted with already existing name \"{name}\""
1357 );
1358
1359 self.get_column_index(column)
1360 .and_then(|idx| unsafe { self.columns_mut() }.get_mut(idx))
1361 .ok_or_else(|| polars_err!(col_not_found = column))
1362 .map(|c| c.rename(name))?;
1363
1364 Ok(self)
1365 }
1366
1367 pub fn rename_many<'a>(
1368 &mut self,
1369 renames: impl Iterator<Item = (&'a str, PlSmallStr)>,
1370 ) -> PolarsResult<&mut Self> {
1371 let mut schema_arc = self.schema().clone();
1372 let schema = Arc::make_mut(&mut schema_arc);
1373
1374 for (from, to) in renames {
1375 if from == to.as_str() {
1376 continue;
1377 }
1378
1379 polars_ensure!(
1380 !schema.contains(&to),
1381 Duplicate: "column rename attempted with already existing name \"{to}\""
1382 );
1383
1384 match schema.get_full(from) {
1385 None => polars_bail!(col_not_found = from),
1386 Some((idx, _, _)) => {
1387 let (n, _) = schema.get_at_index_mut(idx).unwrap();
1388 *n = to.clone();
1389 unsafe { self.columns_mut() }
1390 .get_mut(idx)
1391 .unwrap()
1392 .rename(to);
1393 },
1394 }
1395 }
1396
1397 unsafe { self.set_schema(schema_arc) };
1398
1399 Ok(self)
1400 }
1401
1402 pub fn sort_in_place(
1406 &mut self,
1407 by: impl IntoIterator<Item = impl AsRef<str>>,
1408 sort_options: SortMultipleOptions,
1409 ) -> PolarsResult<&mut Self> {
1410 let by_column = self.select_to_vec(by)?;
1411
1412 let mut out = self.sort_impl(by_column, sort_options, None)?;
1413 unsafe { out.set_schema_from(self) };
1414
1415 *self = out;
1416
1417 Ok(self)
1418 }
1419
1420 #[doc(hidden)]
1421 pub fn sort_impl(
1423 &self,
1424 by_column: Vec<Column>,
1425 sort_options: SortMultipleOptions,
1426 slice: Option<(i64, usize)>,
1427 ) -> PolarsResult<Self> {
1428 if by_column.is_empty() {
1429 return if let Some((offset, len)) = slice {
1431 Ok(self.slice(offset, len))
1432 } else {
1433 Ok(self.clone())
1434 };
1435 }
1436
1437 for column in &by_column {
1438 if column.dtype().is_object() {
1439 polars_bail!(
1440 InvalidOperation: "column '{}' has a dtype of '{}', which does not support sorting", column.name(), column.dtype()
1441 )
1442 }
1443 }
1444
1445 let first_descending = sort_options.descending[0];
1450 let first_by_column = by_column[0].name().to_string();
1451
1452 let set_sorted = |df: &mut DataFrame| {
1453 let _ = df.apply(&first_by_column, |s| {
1456 let mut s = s.clone();
1457 if first_descending {
1458 s.set_sorted_flag(IsSorted::Descending)
1459 } else {
1460 s.set_sorted_flag(IsSorted::Ascending)
1461 }
1462 s
1463 });
1464 };
1465
1466 if self.shape_has_zero() {
1467 let mut out = self.clone();
1468 set_sorted(&mut out);
1469 return Ok(out);
1470 }
1471
1472 if let Some((0, k)) = slice {
1473 if k < self.height() {
1474 return self.bottom_k_impl(k, by_column, sort_options);
1475 }
1476 }
1477 #[cfg(feature = "dtype-categorical")]
1481 let is_not_categorical_enum =
1482 !(matches!(by_column[0].dtype(), DataType::Categorical(_, _))
1483 || matches!(by_column[0].dtype(), DataType::Enum(_, _)));
1484
1485 #[cfg(not(feature = "dtype-categorical"))]
1486 #[allow(non_upper_case_globals)]
1487 const is_not_categorical_enum: bool = true;
1488
1489 if by_column.len() == 1 && is_not_categorical_enum {
1490 let required_sorting = if sort_options.descending[0] {
1491 IsSorted::Descending
1492 } else {
1493 IsSorted::Ascending
1494 };
1495 let no_sorting_required = (by_column[0].is_sorted_flag() == required_sorting)
1498 && ((by_column[0].null_count() == 0)
1499 || by_column[0].get(by_column[0].len() - 1).unwrap().is_null()
1500 == sort_options.nulls_last[0]);
1501
1502 if no_sorting_required {
1503 return if let Some((offset, len)) = slice {
1504 Ok(self.slice(offset, len))
1505 } else {
1506 Ok(self.clone())
1507 };
1508 }
1509 }
1510
1511 let has_nested = by_column.iter().any(|s| s.dtype().is_nested());
1512 let allow_threads = sort_options.multithreaded;
1513
1514 let mut df = self.clone();
1516 let df = df.rechunk_mut_par();
1517 let mut take = match (by_column.len(), has_nested) {
1518 (1, false) => {
1519 let s = &by_column[0];
1520 let options = SortOptions {
1521 descending: sort_options.descending[0],
1522 nulls_last: sort_options.nulls_last[0],
1523 multithreaded: sort_options.multithreaded,
1524 maintain_order: sort_options.maintain_order,
1525 limit: sort_options.limit,
1526 };
1527 if df.width() == 1 && df.try_get_column_index(s.name().as_str()).is_ok() {
1531 let mut out = s.sort_with(options)?;
1532 if let Some((offset, len)) = slice {
1533 out = out.slice(offset, len);
1534 }
1535 return Ok(out.into_frame());
1536 }
1537 s.arg_sort(options)
1538 },
1539 _ => arg_sort(&by_column, sort_options)?,
1540 };
1541
1542 if let Some((offset, len)) = slice {
1543 take = take.slice(offset, len);
1544 }
1545
1546 let mut df = unsafe { df.take_unchecked_impl(&take, allow_threads) };
1549 set_sorted(&mut df);
1550 Ok(df)
1551 }
1552
1553 pub fn _to_metadata(&self) -> DataFrame {
1558 let num_columns = self.width();
1559
1560 let mut column_names =
1561 StringChunkedBuilder::new(PlSmallStr::from_static("column_name"), num_columns);
1562 let mut repr_ca = StringChunkedBuilder::new(PlSmallStr::from_static("repr"), num_columns);
1563 let mut sorted_asc_ca =
1564 BooleanChunkedBuilder::new(PlSmallStr::from_static("sorted_asc"), num_columns);
1565 let mut sorted_dsc_ca =
1566 BooleanChunkedBuilder::new(PlSmallStr::from_static("sorted_dsc"), num_columns);
1567 let mut fast_explode_list_ca =
1568 BooleanChunkedBuilder::new(PlSmallStr::from_static("fast_explode_list"), num_columns);
1569 let mut materialized_at_ca =
1570 StringChunkedBuilder::new(PlSmallStr::from_static("materialized_at"), num_columns);
1571
1572 for col in self.columns() {
1573 let flags = col.get_flags();
1574
1575 let (repr, materialized_at) = match col {
1576 Column::Series(s) => ("series", s.materialized_at()),
1577 Column::Scalar(_) => ("scalar", None),
1578 };
1579 let sorted_asc = flags.contains(StatisticsFlags::IS_SORTED_ASC);
1580 let sorted_dsc = flags.contains(StatisticsFlags::IS_SORTED_DSC);
1581 let fast_explode_list = flags.contains(StatisticsFlags::CAN_FAST_EXPLODE_LIST);
1582
1583 column_names.append_value(col.name().clone());
1584 repr_ca.append_value(repr);
1585 sorted_asc_ca.append_value(sorted_asc);
1586 sorted_dsc_ca.append_value(sorted_dsc);
1587 fast_explode_list_ca.append_value(fast_explode_list);
1588 materialized_at_ca.append_option(materialized_at.map(|v| format!("{v:#?}")));
1589 }
1590
1591 unsafe {
1592 DataFrame::new_unchecked(
1593 self.width(),
1594 vec![
1595 column_names.finish().into_column(),
1596 repr_ca.finish().into_column(),
1597 sorted_asc_ca.finish().into_column(),
1598 sorted_dsc_ca.finish().into_column(),
1599 fast_explode_list_ca.finish().into_column(),
1600 materialized_at_ca.finish().into_column(),
1601 ],
1602 )
1603 }
1604 }
1605 pub fn sort(
1643 &self,
1644 by: impl IntoIterator<Item = impl AsRef<str>>,
1645 sort_options: SortMultipleOptions,
1646 ) -> PolarsResult<Self> {
1647 let mut df = self.clone();
1648 df.sort_in_place(by, sort_options)?;
1649 Ok(df)
1650 }
1651
1652 pub fn replace(&mut self, column: &str, new_col: Column) -> PolarsResult<&mut Self> {
1667 self.apply(column, |_| new_col)
1668 }
1669
1670 pub fn replace_column(&mut self, index: usize, new_column: Column) -> PolarsResult<&mut Self> {
1685 polars_ensure!(
1686 index < self.width(),
1687 ShapeMismatch:
1688 "unable to replace at index {}, the DataFrame has only {} columns",
1689 index, self.width(),
1690 );
1691
1692 polars_ensure!(
1693 new_column.len() == self.height(),
1694 ShapeMismatch:
1695 "unable to replace a column, series length {} doesn't match the DataFrame height {}",
1696 new_column.len(), self.height(),
1697 );
1698
1699 unsafe { *self.columns_mut().get_mut(index).unwrap() = new_column };
1700
1701 Ok(self)
1702 }
1703
1704 pub fn apply<F, C>(&mut self, name: &str, f: F) -> PolarsResult<&mut Self>
1745 where
1746 F: FnOnce(&Column) -> C,
1747 C: IntoColumn,
1748 {
1749 let idx = self.try_get_column_index(name)?;
1750 self.apply_at_idx(idx, f)?;
1751 Ok(self)
1752 }
1753
1754 pub fn apply_at_idx<F, C>(&mut self, idx: usize, f: F) -> PolarsResult<&mut Self>
1785 where
1786 F: FnOnce(&Column) -> C,
1787 C: IntoColumn,
1788 {
1789 let df_height = self.height();
1790 let width = self.width();
1791
1792 let cached_schema = self.cached_schema().cloned();
1793
1794 let col = unsafe { self.columns_mut() }.get_mut(idx).ok_or_else(|| {
1795 polars_err!(
1796 ComputeError: "invalid column index: {} for a DataFrame with {} columns",
1797 idx, width
1798 )
1799 })?;
1800
1801 let mut new_col = f(col).into_column();
1802
1803 if new_col.len() != df_height && new_col.len() == 1 {
1804 new_col = new_col.new_from_index(0, df_height);
1805 }
1806
1807 polars_ensure!(
1808 new_col.len() == df_height,
1809 ShapeMismatch:
1810 "apply_at_idx: resulting Series has length {} while the DataFrame has height {}",
1811 new_col.len(), df_height
1812 );
1813
1814 new_col = new_col.with_name(col.name().clone());
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}