>(
let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
- let values = get_buffer::(data_ref, block_offset, buffers, num_rows)?.as_ptr();
+ let bytes = get_bytes(data_ref, block_offset, buffers)?;
+ let is_aligned = check_bytes_len_and_is_aligned::
(bytes, num_rows)?;
- Ok(unsafe {
- create_array(
- data,
- num_rows,
- null_count,
- [validity, Some(values)].into_iter(),
- [].into_iter(),
- None,
- None,
- )
- })
+ let out = if is_aligned || std::mem::size_of::() <= 8 {
+ assert!(
+ is_aligned,
+ "primitive type with size <= 8 bytes should have been aligned"
+ );
+ let bytes_ptr = bytes.as_ptr();
+
+ unsafe {
+ create_array(
+ data,
+ num_rows,
+ null_count,
+ [validity, Some(bytes_ptr)].into_iter(),
+ [].into_iter(),
+ None,
+ None,
+ )
+ }
+ } else {
+ let mut values = vec![P::default(); num_rows];
+ unsafe {
+ std::ptr::copy_nonoverlapping(
+ bytes.as_ptr(),
+ values.as_mut_ptr() as *mut u8,
+ bytes.len(),
+ )
+ };
+ // Now we need to keep the new buffer alive
+ let owned_data = Arc::new((
+ // We can drop the original ref if we don't have a validity
+ validity.and(Some(data)),
+ values,
+ ));
+ let bytes_ptr = owned_data.1.as_ptr() as *mut u8;
+
+ unsafe {
+ create_array(
+ owned_data,
+ num_rows,
+ null_count,
+ [validity, Some(bytes_ptr)].into_iter(),
+ [].into_iter(),
+ None,
+ None,
+ )
+ }
+ };
+
+ Ok(out)
}
#[allow(clippy::too_many_arguments)]
@@ -482,7 +529,7 @@ fn get_array>(
match data_type.to_physical_type() {
Null => mmap_null(data, &node, block_offset, buffers),
Boolean => mmap_boolean(data, &node, block_offset, buffers),
- Primitive(p) => with_match_primitive_type!(p, |$T| {
+ Primitive(p) => with_match_primitive_type_full!(p, |$T| {
mmap_primitive::<$T, _>(data, &node, block_offset, buffers)
}),
Utf8 | Binary => mmap_binary::(data, &node, block_offset, buffers),
diff --git a/crates/polars-core/src/chunked_array/from_iterator_par.rs b/crates/polars-core/src/chunked_array/from_iterator_par.rs
index 88b135ad0405..12263053e368 100644
--- a/crates/polars-core/src/chunked_array/from_iterator_par.rs
+++ b/crates/polars-core/src/chunked_array/from_iterator_par.rs
@@ -1,5 +1,6 @@
//! Implementations of upstream traits for [`ChunkedArray`]
use std::collections::LinkedList;
+use std::sync::Mutex;
use arrow::pushable::{NoOption, Pushable};
use rayon::prelude::*;
@@ -139,81 +140,159 @@ where
}
}
-/// From trait
+pub trait FromParIterWithDtype {
+ fn from_par_iter_with_dtype(iter: I, name: &str, dtype: DataType) -> Self
+ where
+ I: IntoParallelIterator- ,
+ Self: Sized;
+}
+
+fn get_value_cap(vectors: &LinkedList>>) -> usize {
+ vectors
+ .iter()
+ .map(|list| {
+ list.iter()
+ .map(|opt_s| opt_s.as_ref().map(|s| s.len()).unwrap_or(0))
+ .sum::()
+ })
+ .sum::()
+}
+
+fn get_dtype(vectors: &LinkedList>>) -> DataType {
+ for v in vectors {
+ for s in v.iter().flatten() {
+ let dtype = s.dtype();
+ if !matches!(dtype, DataType::Null) {
+ return dtype.clone();
+ }
+ }
+ }
+ DataType::Null
+}
+
+fn materialize_list(
+ name: &str,
+ vectors: &LinkedList>>,
+ dtype: DataType,
+ value_capacity: usize,
+ list_capacity: usize,
+) -> ListChunked {
+ match &dtype {
+ #[cfg(feature = "object")]
+ DataType::Object(_, _) => {
+ let s = vectors
+ .iter()
+ .flatten()
+ .find_map(|opt_s| opt_s.as_ref())
+ .unwrap();
+ let mut builder = s.get_list_builder(name, value_capacity, list_capacity);
+
+ for v in vectors {
+ for val in v {
+ builder.append_opt_series(val.as_ref()).unwrap();
+ }
+ }
+ builder.finish()
+ },
+ dtype => {
+ let mut builder = get_list_builder(dtype, value_capacity, list_capacity, name).unwrap();
+ for v in vectors {
+ for val in v {
+ builder.append_opt_series(val.as_ref()).unwrap();
+ }
+ }
+ builder.finish()
+ },
+ }
+}
+
impl FromParallelIterator> for ListChunked {
- fn from_par_iter(iter: I) -> Self
+ fn from_par_iter(par_iter: I) -> Self
+ where
+ I: IntoParallelIterator
- >,
+ {
+ let vectors = collect_into_linked_list_vec(par_iter);
+
+ let list_capacity: usize = get_capacity_from_par_results(&vectors);
+ let value_capacity = get_value_cap(&vectors);
+ let dtype = get_dtype(&vectors);
+ if let DataType::Null = dtype {
+ ListChunked::full_null_with_dtype("", list_capacity, &DataType::Null)
+ } else {
+ materialize_list("", &vectors, dtype, value_capacity, list_capacity)
+ }
+ }
+}
+
+impl FromParIterWithDtype> for ListChunked {
+ fn from_par_iter_with_dtype(iter: I, name: &str, dtype: DataType) -> Self
where
I: IntoParallelIterator
- >,
+ Self: Sized,
{
- let mut dtype = None;
let vectors = collect_into_linked_list_vec(iter);
let list_capacity: usize = get_capacity_from_par_results(&vectors);
- let value_capacity = vectors
- .iter()
- .map(|list| {
- list.iter()
- .map(|opt_s| {
- opt_s
- .as_ref()
- .map(|s| {
- if dtype.is_none() && !matches!(s.dtype(), DataType::Null) {
- dtype = Some(s.dtype().clone())
- }
- s.len()
- })
- .unwrap_or(0)
- })
- .sum::()
- })
- .sum::();
-
- match &dtype {
- #[cfg(feature = "object")]
- Some(DataType::Object(_, _)) => {
- let s = vectors
- .iter()
- .flatten()
- .find_map(|opt_s| opt_s.as_ref())
- .unwrap();
- let mut builder = s.get_list_builder("collected", value_capacity, list_capacity);
-
- for v in vectors {
- for val in v {
- builder.append_opt_series(val.as_ref()).unwrap();
- }
- }
- builder.finish()
- },
- Some(dtype) => {
- let mut builder =
- get_list_builder(dtype, value_capacity, list_capacity, "collected").unwrap();
- for v in &vectors {
- for val in v {
- builder.append_opt_series(val.as_ref()).unwrap();
- }
- }
- builder.finish()
- },
- None => ListChunked::full_null_with_dtype("collected", list_capacity, &DataType::Null),
+ let value_capacity = get_value_cap(&vectors);
+ if let DataType::List(dtype) = dtype {
+ materialize_list(name, &vectors, *dtype, value_capacity, list_capacity)
+ } else {
+ panic!("expected list dtype")
}
}
}
-#[cfg(test)]
-mod test {
- use crate::prelude::*;
+pub trait ChunkedCollectParIterExt: ParallelIterator {
+ fn collect_ca_with_dtype>(
+ self,
+ name: &str,
+ dtype: DataType,
+ ) -> B
+ where
+ Self: Sized,
+ {
+ B::from_par_iter_with_dtype(self, name, dtype)
+ }
+}
- #[test]
- fn test_collect_into_list() {
- let s1 = Series::new("", &[true, false, true]);
- let s2 = Series::new("", &[true, false, true]);
+impl ChunkedCollectParIterExt for I {}
- let ll: ListChunked = [&s1, &s2].iter().copied().collect();
- assert_eq!(ll.len(), 2);
- assert_eq!(ll.null_count(), 0);
- let ll: ListChunked = [None, Some(s2)].into_iter().collect();
- assert_eq!(ll.len(), 2);
- assert_eq!(ll.null_count(), 1);
+// Adapted from rayon
+impl FromParIterWithDtype> for Result
+where
+ C: FromParIterWithDtype,
+ T: Send,
+ E: Send,
+{
+ fn from_par_iter_with_dtype(par_iter: I, name: &str, dtype: DataType) -> Self
+ where
+ I: IntoParallelIterator
- >,
+ {
+ fn ok(saved: &Mutex>) -> impl Fn(Result) -> Option + '_ {
+ move |item| match item {
+ Ok(item) => Some(item),
+ Err(error) => {
+ // We don't need a blocking `lock()`, as anybody
+ // else holding the lock will also be writing
+ // `Some(error)`, and then ours is irrelevant.
+ if let Ok(mut guard) = saved.try_lock() {
+ if guard.is_none() {
+ *guard = Some(error);
+ }
+ }
+ None
+ },
+ }
+ }
+
+ let saved_error = Mutex::new(None);
+ let iter = par_iter.into_par_iter().map(ok(&saved_error)).while_some();
+
+ let collection = C::from_par_iter_with_dtype(iter, name, dtype);
+
+ match saved_error.into_inner().unwrap() {
+ Some(error) => Err(error),
+ None => Ok(collection),
+ }
}
}
diff --git a/crates/polars-core/src/chunked_array/logical/categorical/mod.rs b/crates/polars-core/src/chunked_array/logical/categorical/mod.rs
index 959c1f5ec666..019e9a80a962 100644
--- a/crates/polars-core/src/chunked_array/logical/categorical/mod.rs
+++ b/crates/polars-core/src/chunked_array/logical/categorical/mod.rs
@@ -279,6 +279,10 @@ impl CategoricalChunked {
self
}
+ pub fn _with_fast_unique(self, toggle: bool) -> Self {
+ self.with_fast_unique(toggle)
+ }
+
/// Get a reference to the mapping of categorical types to the string values.
pub fn get_rev_map(&self) -> &Arc {
if let DataType::Categorical(Some(rev_map), _) | DataType::Enum(Some(rev_map), _) =
diff --git a/crates/polars-core/src/chunked_array/ops/mod.rs b/crates/polars-core/src/chunked_array/ops/mod.rs
index f4f81f4b7f42..d139fab377c6 100644
--- a/crates/polars-core/src/chunked_array/ops/mod.rs
+++ b/crates/polars-core/src/chunked_array/ops/mod.rs
@@ -343,10 +343,12 @@ pub trait ChunkCompare {
}
/// Get unique values in a `ChunkedArray`
-pub trait ChunkUnique {
+pub trait ChunkUnique {
// We don't return Self to be able to use AutoRef specialization
/// Get unique values of a ChunkedArray
- fn unique(&self) -> PolarsResult>;
+ fn unique(&self) -> PolarsResult
+ where
+ Self: Sized;
/// Get first index of the unique values in a `ChunkedArray`.
/// This Vec is sorted.
diff --git a/crates/polars-core/src/chunked_array/ops/sort/arg_bottom_k.rs b/crates/polars-core/src/chunked_array/ops/sort/arg_bottom_k.rs
index ab4b82e4b78a..dff7b1fed733 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/arg_bottom_k.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/arg_bottom_k.rs
@@ -26,6 +26,9 @@ impl PartialOrd for CompareRow<'_> {
}
}
+/// Return the indices of the bottom k elements.
+///
+/// Similar to .argsort() then .slice(0, k) but with a more efficient implementation.
pub fn _arg_bottom_k(
k: usize,
by_column: &[Series],
diff --git a/crates/polars-core/src/chunked_array/ops/sort/arg_sort.rs b/crates/polars-core/src/chunked_array/ops/sort/arg_sort.rs
index 3a6ccf312067..3ad18862e0c7 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/arg_sort.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/arg_sort.rs
@@ -48,17 +48,10 @@ where
let idx = if nulls_last {
let mut idx = Vec::with_capacity(len);
idx.extend(iter);
- if descending {
- idx.extend(nulls_idx.into_iter().rev());
- } else {
- idx.extend(nulls_idx);
- }
+ idx.extend(nulls_idx);
idx
} else {
let ptr = nulls_idx.as_ptr() as usize;
- if descending {
- nulls_idx.reverse();
- }
nulls_idx.extend(iter);
// We had a realloc.
debug_assert_eq!(nulls_idx.as_ptr() as usize, ptr);
diff --git a/crates/polars-core/src/chunked_array/ops/sort/mod.rs b/crates/polars-core/src/chunked_array/ops/sort/mod.rs
index effa56a2ec4d..0751e01e3cb7 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/mod.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/mod.rs
@@ -730,7 +730,7 @@ mod test {
});
let idx = idx.cont_slice().unwrap();
// the duplicates are in reverse order of appearance, so we cannot reverse expected
- let expected = [4, 2, 1, 5, 6, 0, 3, 7];
+ let expected = [2, 4, 1, 5, 6, 0, 3, 7];
assert_eq!(idx, expected);
}
diff --git a/crates/polars-core/src/chunked_array/ops/sort/options.rs b/crates/polars-core/src/chunked_array/ops/sort/options.rs
index d9bb5e89a884..49ff2ca52286 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/options.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/options.rs
@@ -196,6 +196,12 @@ impl SortOptions {
self.maintain_order = enabled;
self
}
+
+ /// Reverse the order of sorting.
+ pub fn with_order_reversed(mut self) -> Self {
+ self.descending = !self.descending;
+ self
+ }
}
impl From<&SortOptions> for SortMultipleOptions {
diff --git a/crates/polars-core/src/chunked_array/ops/unique/mod.rs b/crates/polars-core/src/chunked_array/ops/unique/mod.rs
index 648f527b6dbe..84b5a5a96592 100644
--- a/crates/polars-core/src/chunked_array/ops/unique/mod.rs
+++ b/crates/polars-core/src/chunked_array/ops/unique/mod.rs
@@ -45,7 +45,7 @@ pub(crate) fn is_unique_helper(
}
#[cfg(feature = "object")]
-impl ChunkUnique> for ObjectChunked {
+impl ChunkUnique for ObjectChunked {
fn unique(&self) -> PolarsResult>> {
polars_bail!(opq = unique, self.dtype());
}
@@ -79,7 +79,7 @@ macro_rules! arg_unique_ca {
}};
}
-impl ChunkUnique for ChunkedArray
+impl ChunkUnique for ChunkedArray
where
T: PolarsNumericType,
T::Native: TotalHash + TotalEq + ToTotalOrd,
@@ -171,7 +171,7 @@ where
}
}
-impl ChunkUnique for StringChunked {
+impl ChunkUnique for StringChunked {
fn unique(&self) -> PolarsResult {
let out = self.as_binary().unique()?;
Ok(unsafe { out.to_string_unchecked() })
@@ -186,7 +186,7 @@ impl ChunkUnique for StringChunked {
}
}
-impl ChunkUnique for BinaryChunked {
+impl ChunkUnique for BinaryChunked {
fn unique(&self) -> PolarsResult {
match self.null_count() {
0 => {
@@ -234,7 +234,7 @@ impl ChunkUnique for BinaryChunked {
}
}
-impl ChunkUnique for BooleanChunked {
+impl ChunkUnique for BooleanChunked {
fn unique(&self) -> PolarsResult {
// can be None, Some(true), Some(false)
let mut unique = Vec::with_capacity(3);
diff --git a/crates/polars-core/src/datatypes/dtype.rs b/crates/polars-core/src/datatypes/dtype.rs
index 28fe8a09f857..240ae84e70dd 100644
--- a/crates/polars-core/src/datatypes/dtype.rs
+++ b/crates/polars-core/src/datatypes/dtype.rs
@@ -120,6 +120,10 @@ impl PartialEq for DataType {
(Array(left_inner, left_width), Array(right_inner, right_width)) => {
left_width == right_width && left_inner == right_inner
},
+ (Unknown(l), Unknown(r)) => match (l, r) {
+ (UnknownKind::Int(_), UnknownKind::Int(_)) => true,
+ _ => l == r,
+ },
_ => std::mem::discriminant(self) == std::mem::discriminant(other),
}
}
diff --git a/crates/polars-core/src/fmt.rs b/crates/polars-core/src/fmt.rs
index 2ced3591c0a3..5cc4d6be8d00 100644
--- a/crates/polars-core/src/fmt.rs
+++ b/crates/polars-core/src/fmt.rs
@@ -1,6 +1,7 @@
#[cfg(any(feature = "fmt", feature = "fmt_no_tty"))]
use std::borrow::Cow;
use std::fmt::{Debug, Display, Formatter, Write};
+use std::str::FromStr;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::RwLock;
use std::{fmt, str};
@@ -28,7 +29,11 @@ use crate::prelude::*;
// Note: see https://github.com/pola-rs/polars/pull/13699 for the rationale
// behind choosing 10 as the default value for default number of rows displayed
-const LIMIT: usize = 10;
+const DEFAULT_ROW_LIMIT: usize = 10;
+#[cfg(any(feature = "fmt", feature = "fmt_no_tty"))]
+const DEFAULT_COL_LIMIT: usize = 8;
+const DEFAULT_STR_LEN_LIMIT: usize = 30;
+const DEFAULT_LIST_LEN_LIMIT: usize = 3;
#[derive(Copy, Clone)]
#[repr(u8)]
@@ -86,6 +91,40 @@ pub fn set_trim_decimal_zeros(trim: Option) {
TRIM_DECIMAL_ZEROS.store(trim.unwrap_or(false), Ordering::Relaxed)
}
+/// Parses an environment variable value.
+fn parse_env_var(name: &str) -> Option {
+ std::env::var(name).ok().and_then(|v| v.parse().ok())
+}
+/// Parses an environment variable value as a limit or set a default.
+///
+/// Negative values (e.g. -1) are parsed as 'no limit' or [`usize::MAX`].
+fn parse_env_var_limit(name: &str, default: usize) -> usize {
+ parse_env_var(name).map_or(
+ default,
+ |n: i64| {
+ if n < 0 {
+ usize::MAX
+ } else {
+ n as usize
+ }
+ },
+ )
+}
+
+fn get_row_limit() -> usize {
+ parse_env_var_limit(FMT_MAX_ROWS, DEFAULT_ROW_LIMIT)
+}
+#[cfg(any(feature = "fmt", feature = "fmt_no_tty"))]
+fn get_col_limit() -> usize {
+ parse_env_var_limit(FMT_MAX_COLS, DEFAULT_COL_LIMIT)
+}
+fn get_str_len_limit() -> usize {
+ parse_env_var_limit(FMT_STR_LEN, DEFAULT_STR_LEN_LIMIT)
+}
+fn get_list_len_limit() -> usize {
+ parse_env_var_limit(FMT_TABLE_CELL_LIST_LEN, DEFAULT_LIST_LEN_LIMIT)
+}
+
macro_rules! format_array {
($f:ident, $a:expr, $dtype:expr, $name:expr, $array_type:expr) => {{
write!(
@@ -96,43 +135,38 @@ macro_rules! format_array {
$name,
$dtype
)?;
- let truncate = matches!($a.dtype(), DataType::String);
- let truncate_len = if truncate {
- std::env::var(FMT_STR_LEN)
- .as_deref()
- .unwrap_or("")
- .parse()
- .unwrap_or(15)
- } else {
- 15
- };
- let limit: usize = {
- let limit = std::env::var(FMT_MAX_ROWS)
- .as_deref()
- .unwrap_or("")
- .parse()
- .map_or(LIMIT, |n: i64| if n < 0 { $a.len() } else { n as usize });
- std::cmp::min(limit, $a.len())
+
+ let truncate = match $a.dtype() {
+ DataType::String => true,
+ #[cfg(feature = "dtype-categorical")]
+ DataType::Categorical(_, _) | DataType::Enum(_, _) => true,
+ _ => false,
};
+ let truncate_len = if truncate { get_str_len_limit() } else { 0 };
+
let write_fn = |v, f: &mut Formatter| -> fmt::Result {
if truncate {
let v = format!("{}", v);
- let v_trunc = &v[..v
+ let v_no_quotes = &v[1..v.len() - 1];
+ let v_trunc = &v_no_quotes[..v_no_quotes
.char_indices()
.take(truncate_len)
.last()
.map(|(i, c)| i + c.len_utf8())
.unwrap_or(0)];
- if v == v_trunc {
+ if v_no_quotes == v_trunc {
write!(f, "\t{}\n", v)?;
} else {
- write!(f, "\t{}…\n", v_trunc)?;
+ write!(f, "\t\"{}…\n", v_trunc)?;
}
} else {
write!(f, "\t{}\n", v)?;
};
Ok(())
};
+
+ let limit = get_row_limit();
+
if $a.len() > limit {
let half = limit / 2;
let rest = limit % 2;
@@ -166,7 +200,7 @@ fn format_object_array(
) -> fmt::Result {
match object.dtype() {
DataType::Object(inner_type, _) => {
- let limit = std::cmp::min(LIMIT, object.len());
+ let limit = std::cmp::min(DEFAULT_ROW_LIMIT, object.len());
write!(
f,
"shape: ({},)\n{}: '{}' [o][{}]\n[\n",
@@ -232,7 +266,7 @@ where
T: PolarsObject,
{
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
- let limit = std::cmp::min(LIMIT, self.len());
+ let limit = std::cmp::min(DEFAULT_ROW_LIMIT, self.len());
let inner_type = T::type_name();
write!(
f,
@@ -497,14 +531,6 @@ fn fmt_df_shape((shape0, shape1): &(usize, usize)) -> String {
)
}
-fn get_str_width() -> usize {
- std::env::var(FMT_STR_LEN)
- .as_deref()
- .unwrap_or("")
- .parse()
- .unwrap_or(32)
-}
-
impl Display for DataFrame {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
#[cfg(any(feature = "fmt", feature = "fmt_no_tty"))]
@@ -514,19 +540,10 @@ impl Display for DataFrame {
self.columns.iter().all(|s| s.len() == height),
"The column lengths in the DataFrame are not equal."
);
- let str_truncate = get_str_width();
- let max_n_cols = std::env::var(FMT_MAX_COLS)
- .as_deref()
- .unwrap_or("")
- .parse()
- .map_or(8, |n: i64| if n < 0 { self.width() } else { n as usize });
-
- let max_n_rows = std::env::var(FMT_MAX_ROWS)
- .as_deref()
- .unwrap_or("")
- .parse()
- .map_or(LIMIT, |n: i64| if n < 0 { height } else { n as usize });
+ let max_n_cols = get_col_limit();
+ let max_n_rows = get_row_limit();
+ let str_truncate = get_str_len_limit();
let (n_first, n_last) = if self.width() > max_n_cols {
((max_n_cols + 1) / 2, max_n_cols / 2)
@@ -965,7 +982,7 @@ fn format_duration(f: &mut Formatter, v: i64, sizes: &[i64], names: &[&str]) ->
}
fn format_blob(f: &mut Formatter<'_>, bytes: &[u8]) -> fmt::Result {
- let width = get_str_width() * 2;
+ let width = get_str_len_limit() * 2;
write!(f, "b\"")?;
for b in bytes.iter().take(width) {
@@ -1109,11 +1126,7 @@ impl Series {
return "[]".to_owned();
}
- let max_items = std::env::var(FMT_TABLE_CELL_LIST_LEN)
- .as_deref()
- .unwrap_or("")
- .parse()
- .map_or(3, |n: i64| if n < 0 { self.len() } else { n as usize });
+ let max_items = get_list_len_limit();
match max_items {
0 => "[…]".to_owned(),
diff --git a/crates/polars-core/src/series/comparison.rs b/crates/polars-core/src/series/comparison.rs
index 15c891aef935..364e86586938 100644
--- a/crates/polars-core/src/series/comparison.rs
+++ b/crates/polars-core/src/series/comparison.rs
@@ -85,19 +85,19 @@ macro_rules! impl_compare {
fn validate_types(left: &DataType, right: &DataType) -> PolarsResult<()> {
use DataType::*;
- #[cfg(feature = "dtype-categorical")]
- {
- let mismatch = matches!(left, String | Categorical(_, _) | Enum(_, _))
- && right.is_numeric()
- || left.is_numeric() && matches!(right, String | Categorical(_, _) | Enum(_, _));
- polars_ensure!(!mismatch, ComputeError: "cannot compare string with numeric data");
- }
- #[cfg(not(feature = "dtype-categorical"))]
- {
- let mismatch = matches!(left, String) && right.is_numeric()
- || left.is_numeric() && matches!(right, String);
- polars_ensure!(!mismatch, ComputeError: "cannot compare string with numeric data");
- }
+
+ match (left, right) {
+ (String, dt) | (dt, String) if dt.is_numeric() => {
+ polars_bail!(ComputeError: "cannot compare string with numeric type ({})", dt)
+ },
+ #[cfg(feature = "dtype-categorical")]
+ (Categorical(_, _) | Enum(_, _), dt) | (dt, Categorical(_, _) | Enum(_, _))
+ if !(dt.is_categorical() | dt.is_string() | dt.is_enum()) =>
+ {
+ polars_bail!(ComputeError: "cannot compare categorical with {}", dt);
+ },
+ _ => (),
+ };
Ok(())
}
diff --git a/crates/polars-core/src/series/from.rs b/crates/polars-core/src/series/from.rs
index f4389806c8dd..ddb2b05ace74 100644
--- a/crates/polars-core/src/series/from.rs
+++ b/crates/polars-core/src/series/from.rs
@@ -443,11 +443,6 @@ impl Series {
Ok(StructChunked::new_unchecked(name, &fields).into_series())
},
ArrowDataType::FixedSizeBinary(_) => {
- if verbose() {
- eprintln!(
- "Polars does not support decimal types so the 'Series' are read as Float64"
- );
- }
let chunks = cast_chunks(&chunks, &DataType::Binary, true)?;
Ok(BinaryChunked::from_chunks(name, chunks).into_series())
},
diff --git a/crates/polars-core/src/series/implementations/binary_offset.rs b/crates/polars-core/src/series/implementations/binary_offset.rs
index 66da638dc7ee..1a517782d8bd 100644
--- a/crates/polars-core/src/series/implementations/binary_offset.rs
+++ b/crates/polars-core/src/series/implementations/binary_offset.rs
@@ -3,6 +3,7 @@ use crate::chunked_array::comparison::*;
#[cfg(feature = "algorithm_group_by")]
use crate::frame::group_by::*;
use crate::prelude::*;
+use crate::series::private::PrivateSeries;
impl private::PrivateSeries for SeriesWrap {
fn compute_len(&mut self) {
@@ -119,6 +120,12 @@ impl SeriesTrait for SeriesWrap {
self.0.len()
}
+ #[cfg(feature = "algorithm_group_by")]
+ fn n_unique(&self) -> PolarsResult {
+ // Only used by multi-key join validation, doesn't have to be optimal
+ self.group_tuples(true, false).map(|g| g.len())
+ }
+
fn rechunk(&self) -> Series {
self.0.rechunk().into_series()
}
diff --git a/crates/polars-core/src/series/mod.rs b/crates/polars-core/src/series/mod.rs
index f6c7b64e471c..bc73d230f9de 100644
--- a/crates/polars-core/src/series/mod.rs
+++ b/crates/polars-core/src/series/mod.rs
@@ -29,7 +29,9 @@ pub use series_trait::{IsSorted, *};
use crate::chunked_array::Settings;
#[cfg(feature = "zip_with")]
use crate::series::arithmetic::coerce_lhs_rhs;
-use crate::utils::{_split_offsets, handle_casting_failures, split_ca, split_series, Wrap};
+use crate::utils::{
+ _split_offsets, handle_casting_failures, materialize_dyn_int, split_ca, split_series, Wrap,
+};
use crate::POOL;
/// # Series
@@ -309,9 +311,39 @@ impl Series {
/// Cast `[Series]` to another `[DataType]`.
pub fn cast(&self, dtype: &DataType) -> PolarsResult {
- // Best leave as is.
- if !dtype.is_known() || (dtype.is_primitive() && dtype == self.dtype()) {
- return Ok(self.clone());
+ match dtype {
+ DataType::Unknown(kind) => {
+ return match kind {
+ // Best leave as is.
+ UnknownKind::Any => Ok(self.clone()),
+ UnknownKind::Int(v) => {
+ if self.dtype().is_integer() {
+ Ok(self.clone())
+ } else {
+ self.cast(&materialize_dyn_int(*v).dtype())
+ }
+ },
+ UnknownKind::Float => {
+ if self.dtype().is_float() {
+ Ok(self.clone())
+ } else {
+ self.cast(&DataType::Float64)
+ }
+ },
+ UnknownKind::Str => {
+ if self.dtype().is_string() | self.dtype().is_categorical() {
+ Ok(self.clone())
+ } else {
+ self.cast(&DataType::String)
+ }
+ },
+ };
+ },
+ // Best leave as is.
+ dt if dt.is_primitive() && dt == self.dtype() => {
+ return Ok(self.clone());
+ },
+ _ => {},
}
let ret = self.0.cast(dtype);
let len = self.len();
diff --git a/crates/polars-core/src/utils/supertype.rs b/crates/polars-core/src/utils/supertype.rs
index 7181c89d4885..47ecd9c13a4f 100644
--- a/crates/polars-core/src/utils/supertype.rs
+++ b/crates/polars-core/src/utils/supertype.rs
@@ -264,28 +264,43 @@ pub fn get_supertype(l: &DataType, r: &DataType) -> Option {
},
(dt, Unknown(kind)) => {
match kind {
+ // numeric vs float|str -> always float|str
UnknownKind::Float | UnknownKind::Int(_) if dt.is_float() | dt.is_string() => Some(dt.clone()),
- UnknownKind::Float if dt.is_numeric() => Some(Unknown(UnknownKind::Float)),
+ UnknownKind::Float if dt.is_integer() => Some(Unknown(UnknownKind::Float)),
+ // Materialize float
+ UnknownKind::Float if dt.is_float() => Some(dt.clone()),
+ // Materialize str
UnknownKind::Str if dt.is_string() | dt.is_enum() => Some(dt.clone()),
+ // Materialize str
#[cfg(feature = "dtype-categorical")]
UnknownKind::Str if dt.is_categorical() => {
let Categorical(_, ord) = dt else { unreachable!()};
Some(Categorical(None, *ord))
},
+ // Keep unknown
dynam if dt.is_null() => Some(Unknown(*dynam)),
+ // Find integers sizes
UnknownKind::Int(v) if dt.is_numeric() => {
- let smallest_fitting_dtype = if dt.is_unsigned_integer() && v.is_positive() {
- materialize_dyn_int_pos(*v).dtype()
- } else {
- materialize_smallest_dyn_int(*v).dtype()
- };
- match dt {
- UInt64 if smallest_fitting_dtype.is_signed_integer() => {
- // Ensure we don't cast to float when dealing with dynamic literals
- Some(Int64)
- },
- _ => {
- get_supertype(dt, &smallest_fitting_dtype)
+ // Both dyn int
+ if let Unknown(UnknownKind::Int(v_other)) = dt {
+ // Take the maximum value to ensure we bubble up the required minimal size.
+ Some(Unknown(UnknownKind::Int(std::cmp::max(*v, *v_other))))
+ }
+ // dyn int vs number
+ else {
+ let smallest_fitting_dtype = if dt.is_unsigned_integer() && v.is_positive() {
+ materialize_dyn_int_pos(*v).dtype()
+ } else {
+ materialize_smallest_dyn_int(*v).dtype()
+ };
+ match dt {
+ UInt64 if smallest_fitting_dtype.is_signed_integer() => {
+ // Ensure we don't cast to float when dealing with dynamic literals
+ Some(Int64)
+ },
+ _ => {
+ get_supertype(dt, &smallest_fitting_dtype)
+ }
}
}
}
diff --git a/crates/polars-io/src/csv/read/mod.rs b/crates/polars-io/src/csv/read/mod.rs
index 3fad37ce049a..5f5b93948f02 100644
--- a/crates/polars-io/src/csv/read/mod.rs
+++ b/crates/polars-io/src/csv/read/mod.rs
@@ -26,7 +26,7 @@ mod reader;
mod splitfields;
mod utils;
-pub use options::{CommentPrefix, CsvEncoding, CsvParserOptions, NullValues};
+pub use options::{CommentPrefix, CsvEncoding, CsvReaderOptions, NullValues};
pub use parser::count_rows;
pub use read_impl::batched_mmap::{BatchedCsvReaderMmap, OwnedBatchedCsvReaderMmap};
pub use read_impl::batched_read::{BatchedCsvReaderRead, OwnedBatchedCsvReader};
diff --git a/crates/polars-io/src/csv/read/options.rs b/crates/polars-io/src/csv/read/options.rs
index f3b4f26ef1cc..3741bd6d9e47 100644
--- a/crates/polars-io/src/csv/read/options.rs
+++ b/crates/polars-io/src/csv/read/options.rs
@@ -5,11 +5,11 @@ use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
-pub struct CsvParserOptions {
+pub struct CsvReaderOptions {
pub has_header: bool,
pub separator: u8,
- pub comment_prefix: Option,
pub quote_char: Option,
+ pub comment_prefix: Option,
pub eol_char: u8,
pub encoding: CsvEncoding,
pub skip_rows: usize,
@@ -27,13 +27,13 @@ pub struct CsvParserOptions {
pub low_memory: bool,
}
-impl Default for CsvParserOptions {
+impl Default for CsvReaderOptions {
fn default() -> Self {
Self {
has_header: true,
separator: b',',
- comment_prefix: None,
quote_char: Some(b'"'),
+ comment_prefix: None,
eol_char: b'\n',
encoding: CsvEncoding::default(),
skip_rows: 0,
@@ -75,17 +75,22 @@ pub enum CommentPrefix {
impl CommentPrefix {
/// Creates a new `CommentPrefix` for the `Single` variant.
- pub fn new_single(c: u8) -> Self {
- CommentPrefix::Single(c)
+ pub fn new_single(prefix: u8) -> Self {
+ CommentPrefix::Single(prefix)
+ }
+
+ /// Creates a new `CommentPrefix` for the `Multi` variant.
+ pub fn new_multi(prefix: String) -> Self {
+ CommentPrefix::Multi(prefix)
}
- /// Creates a new `CommentPrefix`. If `Multi` variant is used and the string is longer
- /// than 5 characters, it will return `None`.
- pub fn new_multi(s: String) -> Option {
- if s.len() <= 5 {
- Some(CommentPrefix::Multi(s))
+ /// Creates a new `CommentPrefix` from a `&str`.
+ pub fn new_from_str(prefix: &str) -> Self {
+ if prefix.len() == 1 && prefix.chars().next().unwrap().is_ascii() {
+ let c = prefix.as_bytes()[0];
+ CommentPrefix::Single(c)
} else {
- None
+ CommentPrefix::Multi(prefix.to_string())
}
}
}
diff --git a/crates/polars-io/src/csv/read/read_impl/batched_read.rs b/crates/polars-io/src/csv/read/read_impl/batched_read.rs
index 9098d255c6a2..64e165844e7a 100644
--- a/crates/polars-io/src/csv/read/read_impl/batched_read.rs
+++ b/crates/polars-io/src/csv/read/read_impl/batched_read.rs
@@ -7,6 +7,7 @@ use polars_core::frame::DataFrame;
use polars_core::schema::SchemaRef;
use polars_core::POOL;
use polars_error::PolarsResult;
+use polars_utils::sync::SyncPtr;
use polars_utils::IdxSize;
use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
@@ -54,6 +55,8 @@ pub(crate) fn get_offsets(
}
}
+/// Reads bytes from `file` to `buf` and returns pointers into `buf` that can be parsed.
+/// TODO! this can be implemented without copying by pointing in the memmapped file.
struct ChunkReader<'a> {
file: &'a File,
buf: Vec,
@@ -109,18 +112,23 @@ impl<'a> ChunkReader<'a> {
self.buf_end = 0;
}
- fn return_slice(&self, start: usize, end: usize) -> (usize, usize) {
+ fn return_slice(&self, start: usize, end: usize) -> (SyncPtr, usize) {
let slice = &self.buf[start..end];
let len = slice.len();
- (slice.as_ptr() as usize, len)
+ (slice.as_ptr().into(), len)
}
- fn get_buf(&self) -> (usize, usize) {
+ fn get_buf_remaining(&self) -> (SyncPtr, usize) {
let slice = &self.buf[self.buf_end..];
let len = slice.len();
- (slice.as_ptr() as usize, len)
+ (slice.as_ptr().into(), len)
}
+ // Get next `n` offset positions. Where `n` is number of chunks.
+
+ // This returns pointers into slices into `buf`
+ // we must process the slices before the next call
+ // as that will overwrite the slices
fn read(&mut self, n: usize) -> bool {
self.reslice();
@@ -267,7 +275,7 @@ pub struct BatchedCsvReaderRead<'a> {
chunk_size: usize,
finished: bool,
file_chunk_reader: ChunkReader<'a>,
- file_chunks: Vec<(usize, usize)>,
+ file_chunks: Vec<(SyncPtr, usize)>,
projection: Vec,
starting_point_offset: Option,
row_index: Option,
@@ -292,6 +300,7 @@ pub struct BatchedCsvReaderRead<'a> {
}
//
impl<'a> BatchedCsvReaderRead<'a> {
+ /// `n` number of batches.
pub fn next_batches(&mut self, n: usize) -> PolarsResult>> {
if n == 0 || self.finished {
return Ok(None);
@@ -320,7 +329,8 @@ impl<'a> BatchedCsvReaderRead<'a> {
// ensure we process the final slice as well.
if self.file_chunk_reader.finished && self.file_chunks.len() < n {
// get the final slice
- self.file_chunks.push(self.file_chunk_reader.get_buf());
+ self.file_chunks
+ .push(self.file_chunk_reader.get_buf_remaining());
self.finished = true
}
@@ -333,7 +343,7 @@ impl<'a> BatchedCsvReaderRead<'a> {
self.file_chunks
.par_iter()
.map(|(ptr, len)| {
- let chunk = unsafe { std::slice::from_raw_parts(*ptr as *const u8, *len) };
+ let chunk = unsafe { std::slice::from_raw_parts(ptr.get(), *len) };
let stop_at_n_bytes = chunk.len();
let mut df = read_chunk(
chunk,
diff --git a/crates/polars-io/src/csv/read/reader.rs b/crates/polars-io/src/csv/read/reader.rs
index ffe5c1f01353..65fcff4b3c47 100644
--- a/crates/polars-io/src/csv/read/reader.rs
+++ b/crates/polars-io/src/csv/read/reader.rs
@@ -8,7 +8,7 @@ use polars_time::prelude::*;
use rayon::prelude::*;
use super::infer_file_schema;
-use super::options::{CommentPrefix, CsvEncoding, NullValues};
+use super::options::{CommentPrefix, CsvEncoding, CsvReaderOptions, NullValues};
use super::read_impl::batched_mmap::{
to_batched_owned_mmap, BatchedCsvReaderMmap, OwnedBatchedCsvReaderMmap,
};
@@ -42,43 +42,25 @@ pub struct CsvReader<'a, R>
where
R: MmapBytesReader,
{
- /// File or Stream object
+ /// File or Stream object.
reader: R,
+ /// Options for the CSV reader.
+ options: CsvReaderOptions,
/// Stop reading from the csv after this number of rows is reached
n_rows: Option,
- // used by error ignore logic
- max_records: Option,
- skip_rows_before_header: usize,
/// Optional indexes of the columns to project
projection: Option>,
/// Optional column names to project/ select.
columns: Option>,
- separator: Option,
- pub(crate) schema: Option,
- encoding: CsvEncoding,
- n_threads: Option,
path: Option,
- schema_overwrite: Option,
dtype_overwrite: Option<&'a [DataType]>,
sample_size: usize,
chunk_size: usize,
- comment_prefix: Option,
- null_values: Option,
predicate: Option>,
- quote_char: Option,
- skip_rows_after_header: usize,
- try_parse_dates: bool,
row_index: Option,
/// Aggregates chunk afterwards to a single chunk.
rechunk: bool,
- raise_if_empty: bool,
- truncate_ragged_lines: bool,
missing_is_null: bool,
- low_memory: bool,
- has_header: bool,
- ignore_errors: bool,
- eol_char: u8,
- decimal_comma: bool,
}
impl<'a, R> CsvReader<'a, R>
@@ -86,39 +68,62 @@ where
R: 'a + MmapBytesReader,
{
/// Skip these rows after the header
- pub fn with_skip_rows_after_header(mut self, offset: usize) -> Self {
- self.skip_rows_after_header = offset;
+ pub fn with_options(mut self, options: CsvReaderOptions) -> Self {
+ self.options = options;
self
}
- /// Add a row index column.
- pub fn with_row_index(mut self, row_index: Option) -> Self {
- self.row_index = row_index;
+ /// Sets whether the CSV file has headers
+ pub fn has_header(mut self, has_header: bool) -> Self {
+ self.options.has_header = has_header;
self
}
- /// Sets the chunk size used by the parser. This influences performance
- pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
- self.chunk_size = chunk_size;
+ /// Sets the CSV file's column separator as a byte character
+ pub fn with_separator(mut self, separator: u8) -> Self {
+ self.options.separator = separator;
self
}
- /// Set [`CsvEncoding`]
- pub fn with_encoding(mut self, enc: CsvEncoding) -> Self {
- self.encoding = enc;
+ /// Sets the `char` used as quote char. The default is `b'"'`. If set to [`None`], quoting is disabled.
+ pub fn with_quote_char(mut self, quote_char: Option) -> Self {
+ self.options.quote_char = quote_char;
self
}
- /// Try to stop parsing when `n` rows are parsed. During multithreaded parsing the upper bound `n` cannot
- /// be guaranteed.
- pub fn with_n_rows(mut self, num_rows: Option) -> Self {
- self.n_rows = num_rows;
+ /// Sets the comment prefix for this instance. Lines starting with this prefix will be ignored.
+ pub fn with_comment_prefix(mut self, comment_prefix: Option<&str>) -> Self {
+ self.options.comment_prefix = comment_prefix.map(CommentPrefix::new_from_str);
self
}
- /// Continue with next batch when a ParserError is encountered.
- pub fn with_ignore_errors(mut self, ignore: bool) -> Self {
- self.ignore_errors = ignore;
+ /// Sets the comment prefix from `CsvParserOptions` for internal initialization.
+ pub fn _with_comment_prefix(mut self, comment_prefix: Option) -> Self {
+ self.options.comment_prefix = comment_prefix;
+ self
+ }
+
+ /// Set the `char` used as end-of-line char. The default is `b'\n'`.
+ pub fn with_end_of_line_char(mut self, eol_char: u8) -> Self {
+ self.options.eol_char = eol_char;
+ self
+ }
+
+ /// Set [`CsvEncoding`].
+ pub fn with_encoding(mut self, encoding: CsvEncoding) -> Self {
+ self.options.encoding = encoding;
+ self
+ }
+
+ /// Skip the first `n` rows during parsing. The header will be parsed at `n` lines.
+ pub fn with_skip_rows(mut self, n: usize) -> Self {
+ self.options.skip_rows = n;
+ self
+ }
+
+ /// Skip these rows after the header
+ pub fn with_skip_rows_after_header(mut self, n: usize) -> Self {
+ self.options.skip_rows_after_header = n;
self
}
@@ -127,74 +132,111 @@ where
///
/// It is recommended to use [with_dtypes](Self::with_dtypes) instead.
pub fn with_schema(mut self, schema: Option) -> Self {
- self.schema = schema;
+ self.options.schema = schema;
self
}
- /// Skip the first `n` rows during parsing. The header will be parsed at `n` lines.
- pub fn with_skip_rows(mut self, skip_rows: usize) -> Self {
- self.skip_rows_before_header = skip_rows;
+ /// Overwrite the schema with the dtypes in this given Schema. The given schema may be a subset
+ /// of the total schema.
+ pub fn with_dtypes(mut self, schema: Option) -> Self {
+ self.options.schema_overwrite = schema;
self
}
- /// Rechunk the DataFrame to contiguous memory after the CSV is parsed.
- pub fn with_rechunk(mut self, rechunk: bool) -> Self {
- self.rechunk = rechunk;
+ /// Set the CSV reader to infer the schema of the file
+ ///
+ /// # Arguments
+ /// * `n` - Maximum number of rows read for schema inference.
+ /// Setting this to `None` will do a full table scan (slow).
+ pub fn infer_schema(mut self, n: Option) -> Self {
+ // used by error ignore logic
+ self.options.infer_schema_length = n;
self
}
- /// Set whether the CSV file has headers
- pub fn has_header(mut self, has_header: bool) -> Self {
- self.has_header = has_header;
+ /// Automatically try to parse dates/ datetimes and time. If parsing fails, columns remain of dtype `[DataType::String]`.
+ pub fn with_try_parse_dates(mut self, toggle: bool) -> Self {
+ self.options.try_parse_dates = toggle;
self
}
- /// Set the CSV file's column separator as a byte character
- pub fn with_separator(mut self, separator: u8) -> Self {
- self.separator = Some(separator);
+ /// Set values that will be interpreted as missing/null.
+ ///
+ /// Note: any value you set as null value will not be escaped, so if quotation marks
+ /// are part of the null value you should include them.
+ pub fn with_null_values(mut self, null_values: Option) -> Self {
+ self.options.null_values = null_values;
self
}
- /// Set the comment prefix for this instance. Lines starting with this prefix will be ignored.
- pub fn with_comment_prefix(mut self, comment_prefix: Option<&str>) -> Self {
- self.comment_prefix = comment_prefix.map(|s| {
- if s.len() == 1 && s.chars().next().unwrap().is_ascii() {
- CommentPrefix::Single(s.as_bytes()[0])
- } else {
- CommentPrefix::Multi(s.to_string())
- }
- });
+ /// Continue with next batch when a ParserError is encountered.
+ pub fn with_ignore_errors(mut self, toggle: bool) -> Self {
+ self.options.ignore_errors = toggle;
self
}
- /// Sets the comment prefix from `CsvParserOptions` for internal initialization.
- pub fn _with_comment_prefix(mut self, comment_prefix: Option) -> Self {
- self.comment_prefix = comment_prefix;
+ /// Raise an error if CSV is empty (otherwise return an empty frame)
+ pub fn raise_if_empty(mut self, toggle: bool) -> Self {
+ self.options.raise_if_empty = toggle;
self
}
- pub fn with_end_of_line_char(mut self, eol_char: u8) -> Self {
- self.eol_char = eol_char;
+ /// Truncate lines that are longer than the schema.
+ pub fn truncate_ragged_lines(mut self, toggle: bool) -> Self {
+ self.options.truncate_ragged_lines = toggle;
self
}
- /// Set values that will be interpreted as missing/ null. Note that any value you set as null value
- /// will not be escaped, so if quotation marks are part of the null value you should include them.
- pub fn with_null_values(mut self, null_values: Option) -> Self {
- self.null_values = null_values;
+ /// Parse floats with a comma as decimal separator.
+ pub fn with_decimal_comma(mut self, toggle: bool) -> Self {
+ self.options.decimal_comma = toggle;
self
}
- /// Treat missing fields as null.
- pub fn with_missing_is_null(mut self, missing_is_null: bool) -> Self {
- self.missing_is_null = missing_is_null;
+ /// Set the number of threads used in CSV reading. The default uses the number of cores of
+ /// your cpu.
+ ///
+ /// Note that this only works if this is initialized with `CsvReader::from_path`.
+ /// Note that the number of cores is the maximum allowed number of threads.
+ pub fn with_n_threads(mut self, n: Option) -> Self {
+ self.options.n_threads = n;
self
}
- /// Overwrite the schema with the dtypes in this given Schema. The given schema may be a subset
- /// of the total schema.
- pub fn with_dtypes(mut self, schema: Option) -> Self {
- self.schema_overwrite = schema;
+ /// Reduce memory consumption at the expense of performance
+ pub fn low_memory(mut self, toggle: bool) -> Self {
+ self.options.low_memory = toggle;
+ self
+ }
+
+ /// Add a row index column.
+ pub fn with_row_index(mut self, row_index: Option) -> Self {
+ self.row_index = row_index;
+ self
+ }
+
+ /// Sets the chunk size used by the parser. This influences performance
+ pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
+ self.chunk_size = chunk_size;
+ self
+ }
+
+ /// Try to stop parsing when `n` rows are parsed. During multithreaded parsing the upper bound `n` cannot
+ /// be guaranteed.
+ pub fn with_n_rows(mut self, num_rows: Option) -> Self {
+ self.n_rows = num_rows;
+ self
+ }
+
+ /// Rechunk the DataFrame to contiguous memory after the CSV is parsed.
+ pub fn with_rechunk(mut self, rechunk: bool) -> Self {
+ self.rechunk = rechunk;
+ self
+ }
+
+ /// Treat missing fields as null.
+ pub fn with_missing_is_null(mut self, missing_is_null: bool) -> Self {
+ self.missing_is_null = missing_is_null;
self
}
@@ -205,17 +247,6 @@ where
self
}
- /// Set the CSV reader to infer the schema of the file
- ///
- /// # Arguments
- /// * `max_records` - Maximum number of rows read for schema inference.
- /// Setting this to `None` will do a full table scan (slow).
- pub fn infer_schema(mut self, max_records: Option) -> Self {
- // used by error ignore logic
- self.max_records = max_records;
- self
- }
-
/// Set the reader's column projection. This counts from 0, meaning that
/// `vec![0, 4]` would select the 1st and 5th column.
pub fn with_projection(mut self, projection: Option>) -> Self {
@@ -229,16 +260,6 @@ where
self
}
- /// Set the number of threads used in CSV reading. The default uses the number of cores of
- /// your cpu.
- ///
- /// Note that this only works if this is initialized with `CsvReader::from_path`.
- /// Note that the number of cores is the maximum allowed number of threads.
- pub fn with_n_threads(mut self, n: Option) -> Self {
- self.n_threads = n;
- self
- }
-
/// The preferred way to initialize this builder. This allows the CSV file to be memory mapped
/// and thereby greatly increases parsing performance.
pub fn with_path>(mut self, path: Option
) -> Self {
@@ -254,46 +275,10 @@ where
self
}
- /// Raise an error if CSV is empty (otherwise return an empty frame)
- pub fn raise_if_empty(mut self, toggle: bool) -> Self {
- self.raise_if_empty = toggle;
- self
- }
-
- /// Reduce memory consumption at the expense of performance
- pub fn low_memory(mut self, toggle: bool) -> Self {
- self.low_memory = toggle;
- self
- }
-
- /// Set the `char` used as quote char. The default is `b'"'`. If set to `[None]` quoting is disabled.
- pub fn with_quote_char(mut self, quote_char: Option) -> Self {
- self.quote_char = quote_char;
- self
- }
-
- /// Automatically try to parse dates/ datetimes and time. If parsing fails, columns remain of dtype `[DataType::String]`.
- pub fn with_try_parse_dates(mut self, toggle: bool) -> Self {
- self.try_parse_dates = toggle;
- self
- }
-
pub fn with_predicate(mut self, predicate: Option>) -> Self {
self.predicate = predicate;
self
}
-
- /// Truncate lines that are longer than the schema.
- pub fn truncate_ragged_lines(mut self, toggle: bool) -> Self {
- self.truncate_ragged_lines = toggle;
- self
- }
-
- /// Parse floats with decimals.
- pub fn with_decimal_comma(mut self, toggle: bool) -> Self {
- self.decimal_comma = toggle;
- self
- }
}
impl<'a> CsvReader<'a, File> {
@@ -318,34 +303,34 @@ impl<'a, R: MmapBytesReader + 'a> CsvReader<'a, R> {
CoreReader::new(
reader_bytes,
self.n_rows,
- self.skip_rows_before_header,
+ self.options.skip_rows,
std::mem::take(&mut self.projection),
- self.max_records,
- self.separator,
- self.has_header,
- self.ignore_errors,
- self.schema.clone(),
+ self.options.infer_schema_length,
+ Some(self.options.separator),
+ self.options.has_header,
+ self.options.ignore_errors,
+ self.options.schema.clone(),
std::mem::take(&mut self.columns),
- self.encoding,
- self.n_threads,
+ self.options.encoding,
+ self.options.n_threads,
schema,
self.dtype_overwrite,
self.sample_size,
self.chunk_size,
- self.low_memory,
- std::mem::take(&mut self.comment_prefix),
- self.quote_char,
- self.eol_char,
- std::mem::take(&mut self.null_values),
+ self.options.low_memory,
+ std::mem::take(&mut self.options.comment_prefix),
+ self.options.quote_char,
+ self.options.eol_char,
+ std::mem::take(&mut self.options.null_values),
self.missing_is_null,
std::mem::take(&mut self.predicate),
to_cast,
- self.skip_rows_after_header,
+ self.options.skip_rows_after_header,
std::mem::take(&mut self.row_index),
- self.try_parse_dates,
- self.raise_if_empty,
- self.truncate_ragged_lines,
- self.decimal_comma,
+ self.options.try_parse_dates,
+ self.options.raise_if_empty,
+ self.options.truncate_ragged_lines,
+ self.options.decimal_comma,
)
}
@@ -403,26 +388,26 @@ impl<'a, R: MmapBytesReader + 'a> CsvReader<'a, R> {
}
pub fn batched_borrowed_mmap(&'a mut self) -> PolarsResult> {
- if let Some(schema) = self.schema_overwrite.as_deref() {
+ if let Some(schema) = self.options.schema_overwrite.as_deref() {
let (schema, to_cast, has_cat) = self.prepare_schema_overwrite(schema)?;
let schema = Arc::new(schema);
let csv_reader = self.core_reader(Some(schema), to_cast)?;
csv_reader.batched_mmap(has_cat)
} else {
- let csv_reader = self.core_reader(self.schema.clone(), vec![])?;
+ let csv_reader = self.core_reader(self.options.schema.clone(), vec![])?;
csv_reader.batched_mmap(false)
}
}
pub fn batched_borrowed_read(&'a mut self) -> PolarsResult> {
- if let Some(schema) = self.schema_overwrite.as_deref() {
+ if let Some(schema) = self.options.schema_overwrite.as_deref() {
let (schema, to_cast, has_cat) = self.prepare_schema_overwrite(schema)?;
let schema = Arc::new(schema);
let csv_reader = self.core_reader(Some(schema), to_cast)?;
csv_reader.batched_read(has_cat)
} else {
- let csv_reader = self.core_reader(self.schema.clone(), vec![])?;
+ let csv_reader = self.core_reader(self.options.schema.clone(), vec![])?;
csv_reader.batched_read(false)
}
}
@@ -440,20 +425,20 @@ impl<'a> CsvReader<'a, Box> {
let (inferred_schema, _, _) = infer_file_schema(
&reader_bytes,
- self.separator.unwrap_or(b','),
- self.max_records,
- self.has_header,
+ self.options.separator,
+ self.options.infer_schema_length,
+ self.options.has_header,
None,
- &mut self.skip_rows_before_header,
- self.skip_rows_after_header,
- self.comment_prefix.as_ref(),
- self.quote_char,
- self.eol_char,
- self.null_values.as_ref(),
- self.try_parse_dates,
- self.raise_if_empty,
- &mut self.n_threads,
- self.decimal_comma,
+ &mut self.options.skip_rows,
+ self.options.skip_rows_after_header,
+ self.options.comment_prefix.as_ref(),
+ self.options.quote_char,
+ self.options.eol_char,
+ self.options.null_values.as_ref(),
+ self.options.try_parse_dates,
+ self.options.raise_if_empty,
+ &mut self.options.n_threads,
+ self.options.decimal_comma,
)?;
let schema = Arc::new(inferred_schema);
Ok(to_batched_owned_mmap(self, schema))
@@ -471,20 +456,20 @@ impl<'a> CsvReader<'a, Box> {
let (inferred_schema, _, _) = infer_file_schema(
&reader_bytes,
- self.separator.unwrap_or(b','),
- self.max_records,
- self.has_header,
+ self.options.separator,
+ self.options.infer_schema_length,
+ self.options.has_header,
None,
- &mut self.skip_rows_before_header,
- self.skip_rows_after_header,
- self.comment_prefix.as_ref(),
- self.quote_char,
- self.eol_char,
- self.null_values.as_ref(),
- self.try_parse_dates,
- self.raise_if_empty,
- &mut self.n_threads,
- self.decimal_comma,
+ &mut self.options.skip_rows,
+ self.options.skip_rows_after_header,
+ self.options.comment_prefix.as_ref(),
+ self.options.quote_char,
+ self.options.eol_char,
+ self.options.null_values.as_ref(),
+ self.options.try_parse_dates,
+ self.options.raise_if_empty,
+ &mut self.options.n_threads,
+ self.options.decimal_comma,
)?;
let schema = Arc::new(inferred_schema);
Ok(to_batched_owned_read(self, schema))
@@ -501,44 +486,26 @@ where
fn new(reader: R) -> Self {
CsvReader {
reader,
+ options: CsvReaderOptions::default(),
rechunk: true,
n_rows: None,
- max_records: Some(128),
- skip_rows_before_header: 0,
projection: None,
- separator: None,
- has_header: true,
- ignore_errors: false,
- schema: None,
columns: None,
- encoding: CsvEncoding::Utf8,
- n_threads: None,
path: None,
- schema_overwrite: None,
dtype_overwrite: None,
sample_size: 1024,
chunk_size: 1 << 18,
- low_memory: false,
- comment_prefix: None,
- eol_char: b'\n',
- null_values: None,
missing_is_null: true,
predicate: None,
- quote_char: Some(b'"'),
- skip_rows_after_header: 0,
- try_parse_dates: false,
row_index: None,
- raise_if_empty: true,
- truncate_ragged_lines: false,
- decimal_comma: false,
}
}
/// Read the file and create the DataFrame.
fn finish(mut self) -> PolarsResult {
let rechunk = self.rechunk;
- let schema_overwrite = self.schema_overwrite.clone();
- let low_memory = self.low_memory;
+ let schema_overwrite = self.options.schema_overwrite.clone();
+ let low_memory = self.options.low_memory;
#[cfg(feature = "dtype-categorical")]
let mut _cat_lock = None;
@@ -557,6 +524,7 @@ where
#[cfg(feature = "dtype-categorical")]
{
let has_cat = self
+ .options
.schema
.clone()
.map(|schema| {
@@ -569,7 +537,7 @@ where
_cat_lock = Some(polars_core::StringCacheHolder::hold())
}
}
- let mut csv_reader = self.core_reader(self.schema.clone(), vec![])?;
+ let mut csv_reader = self.core_reader(self.options.schema.clone(), vec![])?;
csv_reader.as_df()?
};
@@ -585,7 +553,7 @@ where
#[cfg(feature = "temporal")]
// only needed until we also can parse time columns in place
- if self.try_parse_dates {
+ if self.options.try_parse_dates {
// determine the schema that's given by the user. That should not be changed
let fixed_schema = match (schema_overwrite, self.dtype_overwrite) {
(Some(schema), _) => schema,
diff --git a/crates/polars-lazy/src/dsl/functions.rs b/crates/polars-lazy/src/dsl/functions.rs
index a08559a9d14d..7d401ea76334 100644
--- a/crates/polars-lazy/src/dsl/functions.rs
+++ b/crates/polars-lazy/src/dsl/functions.rs
@@ -17,7 +17,7 @@ pub(crate) fn concat_impl>(
) -> PolarsResult {
let mut inputs = inputs.as_ref().to_vec();
- let mut lf = std::mem::take(
+ let lf = std::mem::take(
inputs
.get_mut(0)
.ok_or_else(|| polars_err!(NoData: "empty container given"))?,
@@ -31,88 +31,24 @@ pub(crate) fn concat_impl>(
..Default::default()
};
- let lf = match &mut lf.logical_plan {
- // reuse the same union
- DslPlan::Union {
- inputs: existing_inputs,
- options: opts,
- } if opts == &options => {
- for lf in &mut inputs[1..] {
- // ensure we enable file caching if any lf has it enabled
- opt_state.file_caching |= lf.opt_state.file_caching;
- let lp = std::mem::take(&mut lf.logical_plan);
- existing_inputs.push(lp)
- }
- lf
- },
- _ => {
- let mut lps = Vec::with_capacity(inputs.len());
- lps.push(lf.logical_plan);
-
- for lf in &mut inputs[1..] {
- // ensure we enable file caching if any lf has it enabled
- opt_state.file_caching |= lf.opt_state.file_caching;
- let lp = std::mem::take(&mut lf.logical_plan);
- lps.push(lp)
- }
+ let mut lps = Vec::with_capacity(inputs.len());
+ lps.push(lf.logical_plan);
- let lp = DslPlan::Union {
- inputs: lps,
- options,
- };
- let mut lf = LazyFrame::from(lp);
- lf.opt_state = opt_state;
+ for lf in &mut inputs[1..] {
+ // ensure we enable file caching if any lf has it enabled
+ opt_state.file_caching |= lf.opt_state.file_caching;
+ let lp = std::mem::take(&mut lf.logical_plan);
+ lps.push(lp)
+ }
- lf
- },
+ let lp = DslPlan::Union {
+ inputs: lps,
+ options,
+ convert_supertypes,
};
-
- if convert_supertypes {
- let DslPlan::Union {
- mut inputs,
- options,
- } = lf.logical_plan
- else {
- unreachable!()
- };
- let mut schema = inputs[0].compute_schema()?.as_ref().clone();
-
- let mut changed = false;
- for input in inputs[1..].iter() {
- changed |= schema.to_supertype(input.compute_schema()?.as_ref())?;
- }
-
- let mut placeholder = DslPlan::default();
- if changed {
- let mut exprs = vec![];
- for input in &mut inputs {
- std::mem::swap(input, &mut placeholder);
- let input_schema = placeholder.compute_schema()?;
-
- exprs.clear();
- let to_cast = input_schema.iter().zip(schema.iter_dtypes()).flat_map(
- |((left_name, left_type), st)| {
- if left_type != st {
- Some(col(left_name.as_ref()).cast(st.clone()))
- } else {
- None
- }
- },
- );
- exprs.extend(to_cast);
- let mut lf = LazyFrame::from(placeholder);
- if !exprs.is_empty() {
- lf = lf.with_columns(exprs.as_slice());
- }
-
- placeholder = lf.logical_plan;
- std::mem::swap(&mut placeholder, input);
- }
- }
- Ok(LazyFrame::from(DslPlan::Union { inputs, options }))
- } else {
- Ok(lf)
- }
+ let mut lf = LazyFrame::from(lp);
+ lf.opt_state = opt_state;
+ Ok(lf)
}
#[cfg(feature = "diagonal_concat")]
diff --git a/crates/polars-lazy/src/dsl/list.rs b/crates/polars-lazy/src/dsl/list.rs
index 9d353a25c052..0b8e6530cf86 100644
--- a/crates/polars-lazy/src/dsl/list.rs
+++ b/crates/polars-lazy/src/dsl/list.rs
@@ -2,6 +2,7 @@ use std::sync::Mutex;
use arrow::array::ValueSize;
use arrow::legacy::utils::CustomIterTools;
+use polars_core::chunked_array::from_iterator_par::ChunkedCollectParIterExt;
use polars_core::prelude::*;
use polars_plan::constants::MAP_LIST_NAME;
use polars_plan::dsl::*;
@@ -72,7 +73,7 @@ fn run_per_sublist(
}
})
})
- .collect();
+ .collect_ca_with_dtype("", output_field.dtype.clone());
err = m_err.into_inner().unwrap();
ca
} else {
diff --git a/crates/polars-lazy/src/frame/mod.rs b/crates/polars-lazy/src/frame/mod.rs
index c7c622f96540..68735a69052b 100644
--- a/crates/polars-lazy/src/frame/mod.rs
+++ b/crates/polars-lazy/src/frame/mod.rs
@@ -31,6 +31,7 @@ pub use ndjson::*;
pub use parquet::*;
use polars_core::prelude::*;
use polars_io::RowIndex;
+use polars_ops::frame::JoinCoalesce;
pub use polars_plan::frame::{AllowedOptimizations, OptState};
use polars_plan::global::FETCH_ROWS;
use smartstring::alias::String as SmartString;
@@ -589,17 +590,22 @@ impl LazyFrame {
Ok(lp_top)
}
- #[allow(unused_mut)]
- fn prepare_collect(
- mut self,
+ fn prepare_collect_post_opt(
+ self,
check_sink: bool,
- ) -> PolarsResult<(ExecutionState, Box, bool)> {
- let mut expr_arena = Arena::with_capacity(256);
- let mut lp_arena = Arena::with_capacity(128);
+ post_opt: P,
+ ) -> PolarsResult<(ExecutionState, Box, bool)>
+ where
+ P: Fn(Node, &mut Arena, &mut Arena) -> PolarsResult<()>,
+ {
+ let mut expr_arena = Arena::with_capacity(16);
+ let mut lp_arena = Arena::with_capacity(16);
let mut scratch = vec![];
let lp_top =
self.optimize_with_scratch(&mut lp_arena, &mut expr_arena, &mut scratch, false)?;
+ post_opt(lp_top, &mut lp_arena, &mut expr_arena)?;
+
// sink should be replaced
let no_file_sink = if check_sink {
!matches!(lp_arena.get(lp_top), IR::Sink { .. })
@@ -612,6 +618,23 @@ impl LazyFrame {
Ok((state, physical_plan, no_file_sink))
}
+ // post_opt: A function that is called after optimization. This can be used to modify the IR jit.
+ pub fn _collect_post_opt(self, post_opt: P) -> PolarsResult
+ where
+ P: Fn(Node, &mut Arena, &mut Arena) -> PolarsResult<()>,
+ {
+ let (mut state, mut physical_plan, _) = self.prepare_collect_post_opt(false, post_opt)?;
+ physical_plan.execute(&mut state)
+ }
+
+ #[allow(unused_mut)]
+ fn prepare_collect(
+ self,
+ check_sink: bool,
+ ) -> PolarsResult<(ExecutionState, Box, bool)> {
+ self.prepare_collect_post_opt(check_sink, |_, _, _| Ok(()))
+ }
+
/// Execute all the lazy operations and collect them into a [`DataFrame`].
///
/// The query is optimized prior to execution.
@@ -630,8 +653,7 @@ impl LazyFrame {
/// }
/// ```
pub fn collect(self) -> PolarsResult {
- let (mut state, mut physical_plan, _) = self.prepare_collect(false)?;
- physical_plan.execute(&mut state)
+ self._collect_post_opt(|_, _, _| Ok(()))
}
/// Profile a LazyFrame.
@@ -1124,7 +1146,7 @@ impl LazyFrame {
other,
[left_on.into()],
[right_on.into()],
- JoinArgs::new(JoinType::Outer { coalesce: false }),
+ JoinArgs::new(JoinType::Outer),
)
}
@@ -1195,6 +1217,7 @@ impl LazyFrame {
.right_on(right_on)
.how(args.how)
.validate(args.validation)
+ .coalesce(args.coalesce)
.join_nulls(args.join_nulls);
if let Some(suffix) = args.suffix {
@@ -1764,6 +1787,7 @@ pub struct JoinBuilder {
force_parallel: bool,
suffix: Option,
validation: JoinValidation,
+ coalesce: JoinCoalesce,
join_nulls: bool,
}
impl JoinBuilder {
@@ -1780,6 +1804,7 @@ impl JoinBuilder {
join_nulls: false,
suffix: None,
validation: Default::default(),
+ coalesce: Default::default(),
}
}
@@ -1851,6 +1876,12 @@ impl JoinBuilder {
self
}
+ /// Whether to coalesce join columns.
+ pub fn coalesce(mut self, coalesce: JoinCoalesce) -> Self {
+ self.coalesce = coalesce;
+ self
+ }
+
/// Finish builder
pub fn finish(self) -> LazyFrame {
let mut opt_state = self.lf.opt_state;
@@ -1865,6 +1896,7 @@ impl JoinBuilder {
suffix: self.suffix,
slice: None,
join_nulls: self.join_nulls,
+ coalesce: self.coalesce,
};
let lp = self
diff --git a/crates/polars-lazy/src/physical_plan/executors/scan/csv.rs b/crates/polars-lazy/src/physical_plan/executors/scan/csv.rs
index 63ad38bcfa12..06277d5b054e 100644
--- a/crates/polars-lazy/src/physical_plan/executors/scan/csv.rs
+++ b/crates/polars-lazy/src/physical_plan/executors/scan/csv.rs
@@ -5,7 +5,7 @@ use super::*;
pub struct CsvExec {
pub path: PathBuf,
pub schema: SchemaRef,
- pub options: CsvParserOptions,
+ pub options: CsvReaderOptions,
pub file_options: FileScanOptions,
pub predicate: Option>,
}
diff --git a/crates/polars-lazy/src/physical_plan/expressions/apply.rs b/crates/polars-lazy/src/physical_plan/expressions/apply.rs
index 1e97e1817bb5..0b75510b6ac6 100644
--- a/crates/polars-lazy/src/physical_plan/expressions/apply.rs
+++ b/crates/polars-lazy/src/physical_plan/expressions/apply.rs
@@ -23,6 +23,7 @@ pub struct ApplyExpr {
allow_threading: bool,
check_lengths: bool,
allow_group_aware: bool,
+ output_dtype: Option,
}
impl ApplyExpr {
@@ -33,6 +34,7 @@ impl ApplyExpr {
options: FunctionOptions,
allow_threading: bool,
input_schema: Option,
+ output_dtype: Option,
) -> Self {
#[cfg(debug_assertions)]
if matches!(options.collect_groups, ApplyOptions::ElementWise) && options.returns_scalar {
@@ -51,6 +53,7 @@ impl ApplyExpr {
allow_threading,
check_lengths: options.check_lengths(),
allow_group_aware: options.allow_group_aware,
+ output_dtype,
}
}
@@ -72,6 +75,7 @@ impl ApplyExpr {
allow_threading: true,
check_lengths: true,
allow_group_aware: true,
+ output_dtype: None,
}
}
@@ -162,13 +166,27 @@ impl ApplyExpr {
};
let ca: ListChunked = if self.allow_threading {
- POOL.install(|| {
- agg.list()
- .unwrap()
- .par_iter()
- .map(f)
- .collect::>()
- })?
+ let dtype = match &self.output_dtype {
+ Some(dtype) if dtype.is_known() && !dtype.is_null() => Some(dtype.clone()),
+ _ => None,
+ };
+
+ let lst = agg.list().unwrap();
+ let iter = lst.par_iter().map(f);
+
+ if let Some(dtype) = dtype {
+ // TODO! uncomment this line and remove debug_assertion after a while.
+ // POOL.install(|| {
+ // iter.collect_ca_with_dtype::>("", DataType::List(Box::new(dtype)))
+ // })?
+ let out: ListChunked = POOL.install(|| iter.collect::>())?;
+
+ debug_assert_eq!(out.dtype(), &DataType::List(Box::new(dtype)));
+
+ out
+ } else {
+ POOL.install(|| iter.collect::>())?
+ }
} else {
agg.list()
.unwrap()
diff --git a/crates/polars-lazy/src/physical_plan/expressions/sortby.rs b/crates/polars-lazy/src/physical_plan/expressions/sortby.rs
index ae1bcadc0e7c..1f41cdec5bdf 100644
--- a/crates/polars-lazy/src/physical_plan/expressions/sortby.rs
+++ b/crates/polars-lazy/src/physical_plan/expressions/sortby.rs
@@ -1,3 +1,4 @@
+use polars_core::chunked_array::from_iterator_par::ChunkedCollectParIterExt;
use polars_core::prelude::*;
use polars_core::POOL;
use polars_utils::idx_vec::IdxVec;
@@ -96,6 +97,7 @@ fn sort_by_groups_no_match_single<'a>(
let mut s_in = s_in.list().unwrap().clone();
let mut s_by = s_by.list().unwrap().clone();
+ let dtype = s_in.dtype().clone();
let ca: PolarsResult = POOL.install(|| {
s_in.par_iter_indexed()
.zip(s_by.par_iter_indexed())
@@ -112,7 +114,7 @@ fn sort_by_groups_no_match_single<'a>(
},
_ => Ok(None),
})
- .collect()
+ .collect_ca_with_dtype("", dtype)
});
let s = ca?.with_name(s_in.name()).into_series();
ac_in.with_series(s, true, Some(expr))?;
diff --git a/crates/polars-lazy/src/physical_plan/planner/expr.rs b/crates/polars-lazy/src/physical_plan/planner/expr.rs
index bdd4785df796..fd7a6aebc653 100644
--- a/crates/polars-lazy/src/physical_plan/planner/expr.rs
+++ b/crates/polars-lazy/src/physical_plan/planner/expr.rs
@@ -529,9 +529,16 @@ fn create_physical_expr_inner(
output_type: _,
options,
} => {
+ let output_dtype = schema.and_then(|schema| {
+ expr_arena
+ .get(expression)
+ .to_dtype(schema, Context::Default, expr_arena)
+ .ok()
+ });
+
let is_reducing_aggregation =
options.returns_scalar && matches!(options.collect_groups, ApplyOptions::GroupWise);
- // will be reset in the function so get that here
+ // Will be reset in the function so get that here.
let has_window = state.local.has_window;
let input = create_physical_expressions_check_state(
&input,
@@ -552,6 +559,7 @@ fn create_physical_expr_inner(
options,
!state.has_cache,
schema.cloned(),
+ output_dtype,
)))
},
Function {
@@ -560,9 +568,15 @@ fn create_physical_expr_inner(
options,
..
} => {
+ let output_dtype = schema.and_then(|schema| {
+ expr_arena
+ .get(expression)
+ .to_dtype(schema, Context::Default, expr_arena)
+ .ok()
+ });
let is_reducing_aggregation =
options.returns_scalar && matches!(options.collect_groups, ApplyOptions::GroupWise);
- // will be reset in the function so get that here
+ // Will be reset in the function so get that here.
let has_window = state.local.has_window;
let input = create_physical_expressions_check_state(
&input,
@@ -583,6 +597,7 @@ fn create_physical_expr_inner(
options,
!state.has_cache,
schema.cloned(),
+ output_dtype,
)))
},
Slice {
diff --git a/crates/polars-lazy/src/scan/csv.rs b/crates/polars-lazy/src/scan/csv.rs
index a81099e9cb51..9999d5219ce5 100644
--- a/crates/polars-lazy/src/scan/csv.rs
+++ b/crates/polars-lazy/src/scan/csv.rs
@@ -13,29 +13,30 @@ pub struct LazyCsvReader {
path: PathBuf,
paths: Arc<[PathBuf]>,
separator: u8,
- has_header: bool,
- ignore_errors: bool,
skip_rows: usize,
n_rows: Option,
- cache: bool,
schema: Option,
schema_overwrite: Option,
- low_memory: bool,
comment_prefix: Option,
quote_char: Option,
eol_char: u8,
null_values: Option,
- missing_is_null: bool,
- truncate_ragged_lines: bool,
infer_schema_length: Option,
rechunk: bool,
skip_rows_after_header: usize,
encoding: CsvEncoding,
row_index: Option,
- try_parse_dates: bool,
- raise_if_empty: bool,
n_threads: Option,
+ cache: bool,
+ has_header: bool,
+ ignore_errors: bool,
+ low_memory: bool,
+ missing_is_null: bool,
+ truncate_ragged_lines: bool,
decimal_comma: bool,
+ try_parse_dates: bool,
+ raise_if_empty: bool,
+ glob: bool,
}
#[cfg(feature = "csv")]
@@ -72,6 +73,7 @@ impl LazyCsvReader {
truncate_ragged_lines: false,
n_threads: None,
decimal_comma: false,
+ glob: true,
}
}
@@ -238,6 +240,13 @@ impl LazyCsvReader {
self
}
+ #[must_use]
+ /// Expand path given via globbing rules.
+ pub fn with_glob(mut self, toggle: bool) -> Self {
+ self.glob = toggle;
+ self
+ }
+
/// Modify a schema before we run the lazy scanning.
///
/// Important! Run this function latest in the builder!
@@ -322,6 +331,10 @@ impl LazyFileListReader for LazyCsvReader {
Ok(lf)
}
+ fn glob(&self) -> bool {
+ self.glob
+ }
+
fn path(&self) -> &Path {
&self.path
}
diff --git a/crates/polars-lazy/src/scan/file_list_reader.rs b/crates/polars-lazy/src/scan/file_list_reader.rs
index ceb36334698a..bc19aea8a7d5 100644
--- a/crates/polars-lazy/src/scan/file_list_reader.rs
+++ b/crates/polars-lazy/src/scan/file_list_reader.rs
@@ -36,6 +36,9 @@ fn polars_glob(pattern: &str, cloud_options: Option<&CloudOptions>) -> PolarsRes
pub trait LazyFileListReader: Clone {
/// Get the final [LazyFrame].
fn finish(self) -> PolarsResult {
+ if !self.glob() {
+ return self.finish_no_glob();
+ }
if let Some(paths) = self.iter_paths()? {
let lfs = paths
.map(|r| {
@@ -89,6 +92,10 @@ pub trait LazyFileListReader: Clone {
/// It is recommended to always use [LazyFileListReader::finish] method.
fn finish_no_glob(self) -> PolarsResult;
+ fn glob(&self) -> bool {
+ true
+ }
+
/// Path of the scanned file.
/// It can be potentially a glob pattern.
fn path(&self) -> &Path;
diff --git a/crates/polars-lazy/src/scan/parquet.rs b/crates/polars-lazy/src/scan/parquet.rs
index fe06d761d137..4e57b25d42bf 100644
--- a/crates/polars-lazy/src/scan/parquet.rs
+++ b/crates/polars-lazy/src/scan/parquet.rs
@@ -10,28 +10,31 @@ use crate::prelude::*;
#[derive(Clone)]
pub struct ScanArgsParquet {
pub n_rows: Option,
- pub cache: bool,
pub parallel: ParallelStrategy,
- pub rechunk: bool,
pub row_index: Option,
- pub low_memory: bool,
pub cloud_options: Option,
- pub use_statistics: bool,
pub hive_options: HiveOptions,
+ pub use_statistics: bool,
+ pub low_memory: bool,
+ pub rechunk: bool,
+ pub cache: bool,
+ /// Expand path given via globbing rules.
+ pub glob: bool,
}
impl Default for ScanArgsParquet {
fn default() -> Self {
Self {
n_rows: None,
- cache: true,
parallel: Default::default(),
- rechunk: false,
row_index: None,
- low_memory: false,
cloud_options: None,
- use_statistics: true,
hive_options: Default::default(),
+ use_statistics: true,
+ rechunk: false,
+ low_memory: false,
+ cache: true,
+ glob: true,
}
}
}
@@ -56,6 +59,9 @@ impl LazyParquetReader {
impl LazyFileListReader for LazyParquetReader {
/// Get the final [LazyFrame].
fn finish(mut self) -> PolarsResult {
+ if !self.args.glob {
+ return self.finish_no_glob();
+ }
if let Some(paths) = self.iter_paths()? {
let paths = paths
.into_iter()
diff --git a/crates/polars-lazy/src/tests/aggregations.rs b/crates/polars-lazy/src/tests/aggregations.rs
index 7df14711c5aa..85a1177b4a63 100644
--- a/crates/polars-lazy/src/tests/aggregations.rs
+++ b/crates/polars-lazy/src/tests/aggregations.rs
@@ -480,40 +480,40 @@ fn take_aggregations() -> PolarsResult<()> {
#[test]
fn test_take_consistency() -> PolarsResult<()> {
let df = fruits_cars();
- // let out = df
- // .clone()
- // .lazy()
- // .select([col("A")
- // .arg_sort(SortOptions {
- // descending: true,
- // nulls_last: false,
- // multithreaded: true,
- // maintain_order: false,
- // })
- // .get(lit(0))])
- // .collect()?;
- //
- // let a = out.column("A")?;
- // let a = a.idx()?;
- // assert_eq!(a.get(0), Some(4));
- //
- // let out = df
- // .clone()
- // .lazy()
- // .group_by_stable([col("cars")])
- // .agg([col("A")
- // .arg_sort(SortOptions {
- // descending: true,
- // nulls_last: false,
- // multithreaded: true,
- // maintain_order: false,
- // })
- // .get(lit(0))])
- // .collect()?;
- //
- // let out = out.column("A")?;
- // let out = out.idx()?;
- // assert_eq!(Vec::from(out), &[Some(3), Some(0)]);
+ let out = df
+ .clone()
+ .lazy()
+ .select([col("A")
+ .arg_sort(SortOptions {
+ descending: true,
+ nulls_last: false,
+ multithreaded: true,
+ maintain_order: false,
+ })
+ .get(lit(0))])
+ .collect()?;
+
+ let a = out.column("A")?;
+ let a = a.idx()?;
+ assert_eq!(a.get(0), Some(4));
+
+ let out = df
+ .clone()
+ .lazy()
+ .group_by_stable([col("cars")])
+ .agg([col("A")
+ .arg_sort(SortOptions {
+ descending: true,
+ nulls_last: false,
+ multithreaded: true,
+ maintain_order: false,
+ })
+ .get(lit(0))])
+ .collect()?;
+
+ let out = out.column("A")?;
+ let out = out.idx()?;
+ assert_eq!(Vec::from(out), &[Some(3), Some(0)]);
let out_df = df
.lazy()
diff --git a/crates/polars-lazy/src/tests/queries.rs b/crates/polars-lazy/src/tests/queries.rs
index d12c703f1ebd..822c830d9d1c 100644
--- a/crates/polars-lazy/src/tests/queries.rs
+++ b/crates/polars-lazy/src/tests/queries.rs
@@ -701,7 +701,7 @@ fn test_lazy_group_by_apply() {
.group_by([col("fruits")])
.agg([col("cars").apply(
|s: Series| Ok(Some(Series::new("", &[s.len() as u32]))),
- GetOutput::same_type(),
+ GetOutput::from_type(DataType::UInt32),
)])
.collect()
.unwrap();
diff --git a/crates/polars-lazy/src/tests/streaming.rs b/crates/polars-lazy/src/tests/streaming.rs
index c320c162b3e2..1c51e480636d 100644
--- a/crates/polars-lazy/src/tests/streaming.rs
+++ b/crates/polars-lazy/src/tests/streaming.rs
@@ -1,3 +1,5 @@
+use polars_ops::frame::JoinCoalesce;
+
use super::*;
fn get_csv_file() -> LazyFrame {
@@ -295,7 +297,8 @@ fn test_streaming_partial() -> PolarsResult<()> {
.left_on([col("a")])
.right_on([col("a")])
.suffix("_foo")
- .how(JoinType::Outer { coalesce: true })
+ .how(JoinType::Outer)
+ .coalesce(JoinCoalesce::CoalesceColumns)
.finish();
let q = q.left_join(
diff --git a/crates/polars-ops/src/chunked_array/strings/namespace.rs b/crates/polars-ops/src/chunked_array/strings/namespace.rs
index 637336f00ea2..099a4953fa17 100644
--- a/crates/polars-ops/src/chunked_array/strings/namespace.rs
+++ b/crates/polars-ops/src/chunked_array/strings/namespace.rs
@@ -457,21 +457,21 @@ pub trait StringNameSpaceImpl: AsString {
fn split_exact(&self, by: &StringChunked, n: usize) -> PolarsResult {
let ca = self.as_string();
- split_to_struct(ca, by, n + 1, |s, by| s.split(by))
+ split_to_struct(ca, by, n + 1, str::split, false)
}
#[cfg(feature = "dtype-struct")]
fn split_exact_inclusive(&self, by: &StringChunked, n: usize) -> PolarsResult {
let ca = self.as_string();
- split_to_struct(ca, by, n + 1, |s, by| s.split_inclusive(by))
+ split_to_struct(ca, by, n + 1, str::split_inclusive, false)
}
#[cfg(feature = "dtype-struct")]
fn splitn(&self, by: &StringChunked, n: usize) -> PolarsResult {
let ca = self.as_string();
- split_to_struct(ca, by, n, |s, by| s.splitn(n, by))
+ split_to_struct(ca, by, n, |s, by| s.splitn(n, by), true)
}
fn split(&self, by: &StringChunked) -> ListChunked {
diff --git a/crates/polars-ops/src/chunked_array/strings/split.rs b/crates/polars-ops/src/chunked_array/strings/split.rs
index d8bd4da5b7fb..3648635f52cf 100644
--- a/crates/polars-ops/src/chunked_array/strings/split.rs
+++ b/crates/polars-ops/src/chunked_array/strings/split.rs
@@ -5,12 +5,61 @@ use polars_core::chunked_array::ops::arity::binary_elementwise_for_each;
use super::*;
+pub struct SplitNChars<'a> {
+ s: &'a str,
+ n: usize,
+ keep_remainder: bool,
+}
+
+impl<'a> Iterator for SplitNChars<'a> {
+ type Item = &'a str;
+
+ fn next(&mut self) -> Option {
+ let single_char_limit = if self.keep_remainder { 2 } else { 1 };
+ if self.n >= single_char_limit {
+ self.n -= 1;
+ let ch = self.s.chars().next()?;
+ let first;
+ (first, self.s) = self.s.split_at(ch.len_utf8());
+ Some(first)
+ } else if self.n == 1 && !self.s.is_empty() {
+ self.n -= 1;
+ Some(self.s)
+ } else {
+ None
+ }
+ }
+}
+
+/// Splits a string into substrings consisting of single characters.
+///
+/// Returns at most n strings, where the last string is the entire remainder
+/// of the string if keep_remainder is True, and just the nth character otherwise.
+#[cfg(feature = "dtype-struct")]
+fn splitn_chars(s: &str, n: usize, keep_remainder: bool) -> SplitNChars<'_> {
+ SplitNChars {
+ s,
+ n,
+ keep_remainder,
+ }
+}
+
+/// Splits a string into substrings consisting of single characters.
+fn split_chars(s: &str) -> SplitNChars<'_> {
+ SplitNChars {
+ s,
+ n: usize::MAX,
+ keep_remainder: false,
+ }
+}
+
#[cfg(feature = "dtype-struct")]
pub fn split_to_struct<'a, F, I>(
ca: &'a StringChunked,
by: &'a StringChunked,
n: usize,
op: F,
+ keep_remainder: bool,
) -> PolarsResult
where
F: Fn(&'a str, &'a str) -> I,
@@ -22,24 +71,43 @@ where
if by.len() == 1 {
if let Some(by) = by.get(0) {
- ca.for_each(|opt_s| match opt_s {
- None => {
- for arr in &mut arrs {
- arr.push_null()
- }
- },
- Some(s) => {
- let mut arr_iter = arrs.iter_mut();
- let split_iter = op(s, by);
- (split_iter)
- .zip(&mut arr_iter)
- .for_each(|(splitted, arr)| arr.push(Some(splitted)));
- // fill the remaining with null
- for arr in arr_iter {
- arr.push_null()
- }
- },
- });
+ if by.is_empty() {
+ ca.for_each(|opt_s| match opt_s {
+ None => {
+ for arr in &mut arrs {
+ arr.push_null()
+ }
+ },
+ Some(s) => {
+ let mut arr_iter = arrs.iter_mut();
+ splitn_chars(s, n, keep_remainder)
+ .zip(&mut arr_iter)
+ .for_each(|(splitted, arr)| arr.push(Some(splitted)));
+ // fill the remaining with null
+ for arr in arr_iter {
+ arr.push_null()
+ }
+ },
+ });
+ } else {
+ ca.for_each(|opt_s| match opt_s {
+ None => {
+ for arr in &mut arrs {
+ arr.push_null()
+ }
+ },
+ Some(s) => {
+ let mut arr_iter = arrs.iter_mut();
+ op(s, by)
+ .zip(&mut arr_iter)
+ .for_each(|(splitted, arr)| arr.push(Some(splitted)));
+ // fill the remaining with null
+ for arr in arr_iter {
+ arr.push_null()
+ }
+ },
+ });
+ }
} else {
for arr in &mut arrs {
arr.push_null()
@@ -49,10 +117,15 @@ where
binary_elementwise_for_each(ca, by, |opt_s, opt_by| match (opt_s, opt_by) {
(Some(s), Some(by)) => {
let mut arr_iter = arrs.iter_mut();
- let split_iter = op(s, by);
- (split_iter)
- .zip(&mut arr_iter)
- .for_each(|(splitted, arr)| arr.push(Some(splitted)));
+ if by.is_empty() {
+ splitn_chars(s, n, keep_remainder)
+ .zip(&mut arr_iter)
+ .for_each(|(splitted, arr)| arr.push(Some(splitted)));
+ } else {
+ op(s, by)
+ .zip(&mut arr_iter)
+ .for_each(|(splitted, arr)| arr.push(Some(splitted)));
+ };
// fill the remaining with null
for arr in arr_iter {
arr.push_null()
@@ -87,13 +160,17 @@ where
let mut builder =
ListStringChunkedBuilder::new(ca.name(), ca.len(), ca.get_values_size());
- ca.for_each(|opt_s| match opt_s {
- Some(s) => {
- let iter = op(s, by);
- builder.append_values_iter(iter)
- },
- _ => builder.append_null(),
- });
+ if by.is_empty() {
+ ca.for_each(|opt_s| match opt_s {
+ Some(s) => builder.append_values_iter(split_chars(s)),
+ _ => builder.append_null(),
+ });
+ } else {
+ ca.for_each(|opt_s| match opt_s {
+ Some(s) => builder.append_values_iter(op(s, by)),
+ _ => builder.append_null(),
+ });
+ }
builder.finish()
} else {
ListChunked::full_null_with_dtype(ca.name(), ca.len(), &DataType::String)
@@ -103,8 +180,11 @@ where
binary_elementwise_for_each(ca, by, |opt_s, opt_by| match (opt_s, opt_by) {
(Some(s), Some(by)) => {
- let iter = op(s, by);
- builder.append_values_iter(iter);
+ if by.is_empty() {
+ builder.append_values_iter(split_chars(s))
+ } else {
+ builder.append_values_iter(op(s, by))
+ }
},
_ => builder.append_null(),
});
diff --git a/crates/polars-ops/src/chunked_array/top_k.rs b/crates/polars-ops/src/chunked_array/top_k.rs
index 873a5e27e042..99e24b912362 100644
--- a/crates/polars-ops/src/chunked_array/top_k.rs
+++ b/crates/polars-ops/src/chunked_array/top_k.rs
@@ -4,54 +4,58 @@ use arrow::array::{BooleanArray, MutableBooleanArray};
use arrow::bitmap::MutableBitmap;
use either::Either;
use polars_core::chunked_array::ops::sort::arg_bottom_k::_arg_bottom_k;
-use polars_core::downcast_as_macro_arg_physical;
use polars_core::prelude::*;
+use polars_core::{downcast_as_macro_arg_physical, POOL};
use polars_utils::total_ord::TotalOrd;
+use rayon::prelude::*;
-fn arg_partition Ordering>(
+fn arg_partition Ordering + Sync>(
v: &mut [T],
k: usize,
- descending: bool,
+ sort_options: SortOptions,
cmp: C,
) -> &[T] {
let (lower, _el, upper) = v.select_nth_unstable_by(k, &cmp);
- if descending {
- lower.sort_unstable_by(cmp);
+ let to_sort = if sort_options.descending {
lower
} else {
- upper.sort_unstable_by(|a, b| cmp(b, a));
upper
- }
-}
-
-fn extract_target_and_k(s: &[Series]) -> PolarsResult<(usize, &Series)> {
- let k_s = &s[1];
-
- polars_ensure!(
- k_s.len() == 1,
- ComputeError: "`k` must be a single value for `top_k`."
- );
-
- let Some(k) = k_s.cast(&IDX_DTYPE)?.idx()?.get(0) else {
- polars_bail!(ComputeError: "`k` must be set for `top_k`")
};
-
- let src = &s[0];
-
- Ok((k as usize, src))
+ let cmp = |a: &T, b: &T| {
+ if sort_options.descending {
+ cmp(a, b)
+ } else {
+ cmp(b, a)
+ }
+ };
+ match (sort_options.multithreaded, sort_options.maintain_order) {
+ (true, true) => POOL.install(|| {
+ to_sort.par_sort_by(cmp);
+ }),
+ (true, false) => POOL.install(|| {
+ to_sort.par_sort_unstable_by(cmp);
+ }),
+ (false, true) => to_sort.sort_by(cmp),
+ (false, false) => to_sort.sort_unstable_by(cmp),
+ };
+ to_sort
}
-fn top_k_num_impl(ca: &ChunkedArray, k: usize, descending: bool) -> ChunkedArray
+fn top_k_num_impl(ca: &ChunkedArray, k: usize, sort_options: SortOptions) -> ChunkedArray
where
T: PolarsNumericType,
ChunkedArray: ChunkSort,
{
if k >= ca.len() {
- return ca.sort(!descending);
+ return ca.sort_with(
+ sort_options
+ .with_maintain_order(false)
+ .with_order_reversed(),
+ );
}
// descending is opposite from sort as top-k returns largest
- let k = if descending {
+ let k = if sort_options.descending {
std::cmp::min(k, ca.len())
} else {
ca.len().saturating_sub(k + 1)
@@ -59,11 +63,21 @@ where
match ca.to_vec_null_aware() {
Either::Left(mut v) => {
- let values = arg_partition(&mut v, k, descending, TotalOrd::tot_cmp);
+ let values = arg_partition(
+ &mut v,
+ k,
+ sort_options.with_maintain_order(false),
+ TotalOrd::tot_cmp,
+ );
ChunkedArray::from_slice(ca.name(), values)
},
Either::Right(mut v) => {
- let values = arg_partition(&mut v, k, descending, TotalOrd::tot_cmp);
+ let values = arg_partition(
+ &mut v,
+ k,
+ sort_options.with_maintain_order(false),
+ TotalOrd::tot_cmp,
+ );
let mut out = ChunkedArray::from_iter(values.iter().copied());
out.rename(ca.name());
out
@@ -74,12 +88,12 @@ where
fn top_k_bool_impl(
ca: &ChunkedArray,
k: usize,
- descending: bool,
+ sort_options: SortOptions,
) -> ChunkedArray {
if ca.null_count() == 0 {
let true_count = ca.sum().unwrap() as usize;
let mut bitmap = MutableBitmap::with_capacity(k);
- if !descending {
+ if !sort_options.descending {
// true first
bitmap.extend_constant(std::cmp::min(k, true_count), true);
bitmap.extend_constant(k.saturating_sub(true_count), false);
@@ -109,11 +123,28 @@ fn top_k_bool_impl(
}
let mut array = MutableBooleanArray::with_capacity(k);
- if !descending {
- // Null -> True -> False
- extend_constant_check_remaining(&mut array, &mut remaining, null_count, None);
- extend_constant_check_remaining(&mut array, &mut remaining, true_count, Some(true));
- extend_constant_check_remaining(&mut array, &mut remaining, false_count, Some(false));
+ if !sort_options.descending {
+ if sort_options.nulls_last {
+ // True -> False -> Null
+ extend_constant_check_remaining(&mut array, &mut remaining, true_count, Some(true));
+ extend_constant_check_remaining(
+ &mut array,
+ &mut remaining,
+ false_count,
+ Some(false),
+ );
+ extend_constant_check_remaining(&mut array, &mut remaining, null_count, None);
+ } else {
+ // Null -> True -> False
+ extend_constant_check_remaining(&mut array, &mut remaining, null_count, None);
+ extend_constant_check_remaining(&mut array, &mut remaining, true_count, Some(true));
+ extend_constant_check_remaining(
+ &mut array,
+ &mut remaining,
+ false_count,
+ Some(false),
+ );
+ }
} else {
// False -> True -> Null
extend_constant_check_remaining(&mut array, &mut remaining, false_count, Some(false));
@@ -129,14 +160,19 @@ fn top_k_bool_impl(
fn top_k_binary_impl(
ca: &ChunkedArray,
k: usize,
- descending: bool,
+ sort_options: SortOptions,
) -> ChunkedArray {
if k >= ca.len() {
- return ca.sort(!descending);
+ return ca.sort_with(
+ sort_options
+ .with_order_reversed()
+ // single series main order is meaningless
+ .with_maintain_order(false),
+ );
}
// descending is opposite from sort as top-k returns largest
- let k = if descending {
+ let k = if sort_options.descending {
std::cmp::min(k, ca.len())
} else {
ca.len().saturating_sub(k + 1)
@@ -147,21 +183,38 @@ fn top_k_binary_impl(
for arr in ca.downcast_iter() {
v.extend(arr.non_null_values_iter());
}
- let values = arg_partition(&mut v, k, descending, TotalOrd::tot_cmp);
+ let values = arg_partition(&mut v, k, sort_options, TotalOrd::tot_cmp);
ChunkedArray::from_slice(ca.name(), values)
} else {
let mut v = Vec::with_capacity(ca.len());
for arr in ca.downcast_iter() {
v.extend(arr.iter());
}
- let values = arg_partition(&mut v, k, descending, TotalOrd::tot_cmp);
+ let values = arg_partition(&mut v, k, sort_options, TotalOrd::tot_cmp);
let mut out = ChunkedArray::from_iter(values.iter().copied());
out.rename(ca.name());
out
}
}
-pub fn top_k(s: &[Series], descending: bool) -> PolarsResult {
+pub fn top_k(s: &[Series], sort_options: SortOptions) -> PolarsResult {
+ fn extract_target_and_k(s: &[Series]) -> PolarsResult<(usize, &Series)> {
+ let k_s = &s[1];
+
+ polars_ensure!(
+ k_s.len() == 1,
+ ComputeError: "`k` must be a single value for `top_k`."
+ );
+
+ let Some(k) = k_s.cast(&IDX_DTYPE)?.idx()?.get(0) else {
+ polars_bail!(ComputeError: "`k` must be set for `top_k`")
+ };
+
+ let src = &s[0];
+
+ Ok((k as usize, src))
+ }
+
let (k, src) = extract_target_and_k(s)?;
if src.is_empty() {
@@ -184,17 +237,19 @@ pub fn top_k(s: &[Series], descending: bool) -> PolarsResult {
let s = src.to_physical_repr();
match s.dtype() {
- DataType::Boolean => Ok(top_k_bool_impl(s.bool().unwrap(), k, descending).into_series()),
+ DataType::Boolean => Ok(top_k_bool_impl(s.bool().unwrap(), k, sort_options).into_series()),
DataType::String => {
- let ca = top_k_binary_impl(&s.str().unwrap().as_binary(), k, descending);
+ let ca = top_k_binary_impl(&s.str().unwrap().as_binary(), k, sort_options);
let ca = unsafe { ca.to_string_unchecked() };
Ok(ca.into_series())
},
- DataType::Binary => Ok(top_k_binary_impl(s.binary().unwrap(), k, descending).into_series()),
+ DataType::Binary => {
+ Ok(top_k_binary_impl(s.binary().unwrap(), k, sort_options).into_series())
+ },
_dt => {
macro_rules! dispatch {
($ca:expr) => {{
- top_k_num_impl($ca, k, descending).into_series()
+ top_k_num_impl($ca, k, sort_options).into_series()
}};
}
unsafe { downcast_as_macro_arg_physical!(&s, dispatch).cast_unchecked(origin_dtype) }
@@ -202,13 +257,52 @@ pub fn top_k(s: &[Series], descending: bool) -> PolarsResult {
}
}
-pub fn top_k_by(
- s: &[Series],
+pub fn top_k_by(s: &[Series], sort_options: SortMultipleOptions) -> PolarsResult {
+ /// Return (k, src, by)
+ fn extract_parameters(s: &[Series]) -> PolarsResult<(usize, &Series, &[Series])> {
+ let k_s = &s[1];
+
+ polars_ensure!(
+ k_s.len() == 1,
+ ComputeError: "`k` must be a single value for `top_k`."
+ );
+
+ let Some(k) = k_s.cast(&IDX_DTYPE)?.idx()?.get(0) else {
+ polars_bail!(ComputeError: "`k` must be set for `top_k`")
+ };
+
+ let src = &s[0];
+
+ let by = &s[2..];
+
+ Ok((k as usize, src, by))
+ }
+
+ let (k, src, by) = extract_parameters(s)?;
+
+ if src.is_empty() {
+ return Ok(src.clone());
+ }
+
+ if by.first().map(|x| x.is_empty()).unwrap_or(false) {
+ return Ok(src.clone());
+ }
+
+ for s in by {
+ if s.len() != src.len() {
+ polars_bail!(ComputeError: "`by` column's ({}) length ({}) should have the same length as the source column length ({}) in `top_k`", s.name(), s.len(), src.len())
+ }
+ }
+
+ top_k_by_impl(k, src, by, sort_options)
+}
+
+fn top_k_by_impl(
+ k: usize,
+ src: &Series,
by: &[Series],
sort_options: SortMultipleOptions,
) -> PolarsResult {
- let (k, src) = extract_target_and_k(s)?;
-
if src.is_empty() {
return Ok(src.clone());
}
diff --git a/crates/polars-ops/src/frame/join/args.rs b/crates/polars-ops/src/frame/join/args.rs
index 51bbbf9d80fe..148c46ce7953 100644
--- a/crates/polars-ops/src/frame/join/args.rs
+++ b/crates/polars-ops/src/frame/join/args.rs
@@ -26,6 +26,36 @@ pub struct JoinArgs {
pub suffix: Option,
pub slice: Option<(i64, usize)>,
pub join_nulls: bool,
+ pub coalesce: JoinCoalesce,
+}
+
+#[derive(Clone, PartialEq, Eq, Debug, Hash, Default)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum JoinCoalesce {
+ #[default]
+ JoinSpecific,
+ CoalesceColumns,
+ KeepColumns,
+}
+
+impl JoinCoalesce {
+ pub fn coalesce(&self, join_type: &JoinType) -> bool {
+ use JoinCoalesce::*;
+ use JoinType::*;
+ match join_type {
+ Left | Inner => {
+ matches!(self, JoinSpecific | CoalesceColumns)
+ },
+ Outer { .. } => {
+ matches!(self, CoalesceColumns)
+ },
+ #[cfg(feature = "asof_join")]
+ AsOf(_) => false,
+ Cross => false,
+ #[cfg(feature = "semi_anti_join")]
+ Semi | Anti => false,
+ }
+ }
}
impl Default for JoinArgs {
@@ -36,6 +66,7 @@ impl Default for JoinArgs {
suffix: None,
slice: None,
join_nulls: false,
+ coalesce: Default::default(),
}
}
}
@@ -48,9 +79,15 @@ impl JoinArgs {
suffix: None,
slice: None,
join_nulls: false,
+ coalesce: Default::default(),
}
}
+ pub fn with_coalesce(mut self, coalesce: JoinCoalesce) -> Self {
+ self.coalesce = coalesce;
+ self
+ }
+
pub fn suffix(&self) -> &str {
self.suffix.as_deref().unwrap_or("_right")
}
@@ -61,9 +98,7 @@ impl JoinArgs {
pub enum JoinType {
Left,
Inner,
- Outer {
- coalesce: bool,
- },
+ Outer,
#[cfg(feature = "asof_join")]
AsOf(AsOfOptions),
Cross,
@@ -73,18 +108,6 @@ pub enum JoinType {
Anti,
}
-impl JoinType {
- pub fn merges_join_keys(&self) -> bool {
- match self {
- Self::Outer { coalesce } => *coalesce,
- // Merges them if they are equal
- #[cfg(feature = "asof_join")]
- Self::AsOf(_) => false,
- _ => true,
- }
- }
-}
-
impl From for JoinArgs {
fn from(value: JoinType) -> Self {
JoinArgs::new(value)
@@ -116,6 +139,19 @@ impl Debug for JoinType {
}
}
+impl JoinType {
+ pub fn is_asof(&self) -> bool {
+ #[cfg(feature = "asof_join")]
+ {
+ matches!(self, JoinType::AsOf(_))
+ }
+ #[cfg(not(feature = "asof_join"))]
+ {
+ false
+ }
+ }
+}
+
#[derive(Copy, Clone, PartialEq, Eq, Default, Hash)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
pub enum JoinValidation {
diff --git a/crates/polars-ops/src/frame/join/hash_join/mod.rs b/crates/polars-ops/src/frame/join/hash_join/mod.rs
index 0be95b1aa1cf..f6b1ca773ee4 100644
--- a/crates/polars-ops/src/frame/join/hash_join/mod.rs
+++ b/crates/polars-ops/src/frame/join/hash_join/mod.rs
@@ -271,9 +271,7 @@ pub trait JoinDispatch: IntoDf {
|| unsafe { other.take_unchecked(&idx_ca_r) },
);
- let JoinType::Outer { coalesce } = args.how else {
- unreachable!()
- };
+ let coalesce = args.coalesce.coalesce(&JoinType::Outer);
let out = _finish_join(df_left, df_right, args.suffix.as_deref());
if coalesce {
Ok(_coalesce_outer_join(
diff --git a/crates/polars-ops/src/frame/join/mod.rs b/crates/polars-ops/src/frame/join/mod.rs
index f3df643de0e8..6a29e2b28c3a 100644
--- a/crates/polars-ops/src/frame/join/mod.rs
+++ b/crates/polars-ops/src/frame/join/mod.rs
@@ -209,9 +209,7 @@ pub trait DataFrameJoinOps: IntoDf {
JoinType::Left => {
left_df._left_join_from_series(other, s_left, s_right, args, _verbose, None)
},
- JoinType::Outer { .. } => {
- left_df._outer_join_from_series(other, s_left, s_right, args)
- },
+ JoinType::Outer => left_df._outer_join_from_series(other, s_left, s_right, args),
#[cfg(feature = "semi_anti_join")]
JoinType::Anti => left_df._semi_anti_join_from_series(
s_left,
@@ -278,13 +276,14 @@ pub trait DataFrameJoinOps: IntoDf {
JoinType::Cross => {
unreachable!()
},
- JoinType::Outer { coalesce } => {
+ JoinType::Outer => {
let names_left = selected_left.iter().map(|s| s.name()).collect::>();
- args.how = JoinType::Outer { coalesce: false };
+ let coalesce = args.coalesce;
+ args.coalesce = JoinCoalesce::KeepColumns;
let suffix = args.suffix.clone();
let out = left_df._outer_join_from_series(other, &lhs_keys, &rhs_keys, args);
- if coalesce {
+ if coalesce.coalesce(&JoinType::Outer) {
Ok(_coalesce_outer_join(
out?,
&names_left,
@@ -411,12 +410,7 @@ pub trait DataFrameJoinOps: IntoDf {
I: IntoIterator- ,
S: AsRef,
{
- self.join(
- other,
- left_on,
- right_on,
- JoinArgs::new(JoinType::Outer { coalesce: false }),
- )
+ self.join(other, left_on, right_on, JoinArgs::new(JoinType::Outer))
}
}
diff --git a/crates/polars-ops/src/series/ops/cut.rs b/crates/polars-ops/src/series/ops/cut.rs
index f1721ac2fd45..aa1025f1ed55 100644
--- a/crates/polars-ops/src/series/ops/cut.rs
+++ b/crates/polars-ops/src/series/ops/cut.rs
@@ -25,6 +25,9 @@ fn map_cats(
PartialOrd::gt
};
+ // Ensure fast unique is only set if all labels were seen.
+ let mut label_has_value = vec![false; 1 + sorted_breaks.len()];
+
if include_breaks {
// This is to replicate the behavior of the old buggy version that only worked on series and
// returned a dataframe. That included a column of the right endpoint of the interval. So we
@@ -33,8 +36,11 @@ fn map_cats(
let mut brk_vals = PrimitiveChunkedBuilder::::new("brk", s.len());
s_iter
.map(|opt| {
- opt.filter(|x| !x.is_nan())
- .map(|x| sorted_breaks.partition_point(|v| op(&x, v)))
+ opt.filter(|x| !x.is_nan()).map(|x| {
+ let pt = sorted_breaks.partition_point(|v| op(&x, v));
+ unsafe { *label_has_value.get_unchecked_mut(pt) = true };
+ pt
+ })
})
.for_each(|idx| match idx {
None => {
@@ -47,17 +53,23 @@ fn map_cats(
},
});
- let outvals = vec![brk_vals.finish().into_series(), bld.finish().into_series()];
+ let outvals = vec![
+ brk_vals.finish().into_series(),
+ bld.finish()
+ ._with_fast_unique(label_has_value.iter().all(bool::clone))
+ .into_series(),
+ ];
Ok(StructChunked::new(&out_name, &outvals)?.into_series())
} else {
Ok(bld
.drain_iter_and_finish(s_iter.map(|opt| {
- opt.filter(|x| !x.is_nan()).map(|x| unsafe {
- labels
- .get_unchecked(sorted_breaks.partition_point(|v| op(&x, v)))
- .as_str()
+ opt.filter(|x| !x.is_nan()).map(|x| {
+ let pt = sorted_breaks.partition_point(|v| op(&x, v));
+ unsafe { *label_has_value.get_unchecked_mut(pt) = true };
+ unsafe { labels.get_unchecked(pt).as_str() }
})
}))
+ ._with_fast_unique(label_has_value.iter().all(bool::clone))
.into_series())
}
}
@@ -145,3 +157,31 @@ pub fn qcut(
map_cats(&s, &cut_labels, &qbreaks, left_closed, include_breaks)
}
+
+mod test {
+ #[test]
+ fn test_map_cats_fast_unique() {
+ // This test is here to check the fast unique flag is set when it can be
+ // as it is not visible to Python.
+ use polars_core::prelude::*;
+
+ use super::map_cats;
+
+ let s = Series::new("x", &[1, 2, 3, 4, 5]);
+
+ let labels = &["a", "b", "c"].map(str::to_owned);
+ let breaks = &[2.0, 4.0];
+ let left_closed = false;
+
+ let include_breaks = false;
+ let out = map_cats(&s, labels, breaks, left_closed, include_breaks).unwrap();
+ let out = out.categorical().unwrap();
+ assert!(out._can_fast_unique());
+
+ let include_breaks = true;
+ let out = map_cats(&s, labels, breaks, left_closed, include_breaks).unwrap();
+ let out = out.struct_().unwrap().fields()[1].clone();
+ let out = out.categorical().unwrap();
+ assert!(out._can_fast_unique());
+ }
+}
diff --git a/crates/polars-pipe/src/executors/sinks/joins/generic_build.rs b/crates/polars-pipe/src/executors/sinks/joins/generic_build.rs
index 864020d1a8a1..1fa7ce58a152 100644
--- a/crates/polars-pipe/src/executors/sinks/joins/generic_build.rs
+++ b/crates/polars-pipe/src/executors/sinks/joins/generic_build.rs
@@ -5,6 +5,7 @@ use hashbrown::hash_map::RawEntryMut;
use polars_core::export::ahash::RandomState;
use polars_core::prelude::*;
use polars_core::utils::{_set_partition_size, accumulate_dataframes_vertical_unchecked};
+use polars_ops::prelude::JoinArgs;
use polars_utils::arena::Node;
use polars_utils::slice::GetSaferUnchecked;
use polars_utils::unitvec;
@@ -34,6 +35,7 @@ pub struct GenericBuild {
materialized_join_cols: Vec>,
suffix: Arc,
hb: RandomState,
+ join_args: JoinArgs,
// partitioned tables that will be used for probing
// stores the key and the chunk_idx, df_idx of the left table
hash_tables: PartitionedMap,
@@ -45,7 +47,6 @@ pub struct GenericBuild {
// amortize allocations
join_columns: Vec,
hashes: Vec,
- join_type: JoinType,
// the join order is swapped to ensure we hash the smaller table
swapped: bool,
join_nulls: bool,
@@ -59,7 +60,7 @@ impl GenericBuild {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
suffix: Arc,
- join_type: JoinType,
+ join_args: JoinArgs,
swapped: bool,
join_columns_left: Arc>>,
join_columns_right: Arc>>,
@@ -76,7 +77,7 @@ impl GenericBuild {
}));
GenericBuild {
chunks: vec![],
- join_type,
+ join_args,
suffix,
hb,
swapped,
@@ -278,7 +279,7 @@ impl Sink for GenericBuild {
fn split(&self, _thread_no: usize) -> Box {
let mut new = Self::new(
self.suffix.clone(),
- self.join_type.clone(),
+ self.join_args.clone(),
self.swapped,
self.join_columns_left.clone(),
self.join_columns_right.clone(),
@@ -317,7 +318,7 @@ impl Sink for GenericBuild {
let mut hashes = std::mem::take(&mut self.hashes);
hashes.clear();
- match self.join_type {
+ match self.join_args.how {
JoinType::Inner | JoinType::Left => {
let probe_operator = GenericJoinProbe::new(
left_df,
@@ -330,13 +331,14 @@ impl Sink for GenericBuild {
self.swapped,
hashes,
context,
- self.join_type.clone(),
+ self.join_args.how.clone(),
self.join_nulls,
);
self.placeholder.replace(Box::new(probe_operator));
Ok(FinalizedSink::Operator)
},
- JoinType::Outer { coalesce } => {
+ JoinType::Outer => {
+ let coalesce = self.join_args.coalesce.coalesce(&JoinType::Outer);
let probe_operator = GenericOuterJoinProbe::new(
left_df,
materialized_join_cols,
diff --git a/crates/polars-pipe/src/executors/sources/csv.rs b/crates/polars-pipe/src/executors/sources/csv.rs
index b2423c832ed8..8d49bcb9ea41 100644
--- a/crates/polars-pipe/src/executors/sources/csv.rs
+++ b/crates/polars-pipe/src/executors/sources/csv.rs
@@ -4,7 +4,7 @@ use std::path::PathBuf;
use polars_core::export::arrow::Either;
use polars_core::POOL;
use polars_io::csv::read::{
- BatchedCsvReaderMmap, BatchedCsvReaderRead, CsvEncoding, CsvParserOptions, CsvReader,
+ BatchedCsvReaderMmap, BatchedCsvReaderRead, CsvEncoding, CsvReader, CsvReaderOptions,
};
use polars_plan::global::_set_n_rows_for_scan;
use polars_plan::prelude::FileScanOptions;
@@ -22,7 +22,7 @@ pub(crate) struct CsvSource {
Option, *mut BatchedCsvReaderRead<'static>>>,
n_threads: usize,
path: Option,
- options: Option,
+ options: Option,
file_options: Option,
verbose: bool,
}
@@ -106,7 +106,7 @@ impl CsvSource {
pub(crate) fn new(
path: PathBuf,
schema: SchemaRef,
- options: CsvParserOptions,
+ options: CsvReaderOptions,
file_options: FileScanOptions,
verbose: bool,
) -> PolarsResult {
diff --git a/crates/polars-pipe/src/pipeline/convert.rs b/crates/polars-pipe/src/pipeline/convert.rs
index 73afa86fea0a..a0e4aee37ee8 100644
--- a/crates/polars-pipe/src/pipeline/convert.rs
+++ b/crates/polars-pipe/src/pipeline/convert.rs
@@ -285,12 +285,12 @@ where
};
match jt {
- join_type @ JoinType::Inner | join_type @ JoinType::Left => {
+ JoinType::Inner | JoinType::Left => {
let (join_columns_left, join_columns_right) = swap_eval();
Box::new(GenericBuild::<()>::new(
Arc::from(options.args.suffix()),
- join_type.clone(),
+ options.args.clone(),
swapped,
join_columns_left,
join_columns_right,
@@ -317,7 +317,7 @@ where
Box::new(GenericBuild::::new(
Arc::from(options.args.suffix()),
- jt.clone(),
+ options.args.clone(),
swapped,
join_columns_left,
join_columns_right,
diff --git a/crates/polars-plan/src/dsl/function_expr/cum.rs b/crates/polars-plan/src/dsl/function_expr/cum.rs
index 8a1aa953479d..74ad6eec596a 100644
--- a/crates/polars-plan/src/dsl/function_expr/cum.rs
+++ b/crates/polars-plan/src/dsl/function_expr/cum.rs
@@ -21,6 +21,7 @@ pub(super) fn cum_max(s: &Series, reverse: bool) -> PolarsResult {
}
pub(super) mod dtypes {
+ use polars_core::utils::materialize_dyn_int;
use DataType::*;
use super::*;
@@ -36,6 +37,11 @@ pub(super) mod dtypes {
UInt64 => UInt64,
Float32 => Float32,
Float64 => Float64,
+ Unknown(kind) => match kind {
+ UnknownKind::Int(v) => cum_sum(&materialize_dyn_int(*v).dtype()),
+ UnknownKind::Float => Float64,
+ _ => dt.clone(),
+ },
_ => Int64,
}
}
diff --git a/crates/polars-plan/src/dsl/function_expr/mod.rs b/crates/polars-plan/src/dsl/function_expr/mod.rs
index 33c57da896dd..e45cb5e86313 100644
--- a/crates/polars-plan/src/dsl/function_expr/mod.rs
+++ b/crates/polars-plan/src/dsl/function_expr/mod.rs
@@ -10,7 +10,7 @@ mod bounds;
#[cfg(feature = "business")]
mod business;
#[cfg(feature = "dtype-categorical")]
-mod cat;
+pub mod cat;
#[cfg(feature = "round_series")]
mod clip;
#[cfg(feature = "dtype-struct")]
@@ -38,13 +38,13 @@ mod nan;
mod peaks;
#[cfg(feature = "ffi_plugin")]
mod plugin;
-mod pow;
+pub mod pow;
#[cfg(feature = "random")]
mod random;
#[cfg(feature = "range")]
mod range;
#[cfg(feature = "rolling_window")]
-mod rolling;
+pub mod rolling;
#[cfg(feature = "round_series")]
mod round;
#[cfg(feature = "row_hash")]
@@ -63,7 +63,7 @@ mod struct_;
#[cfg(any(feature = "temporal", feature = "date_offset"))]
mod temporal;
#[cfg(feature = "trigonometry")]
-mod trigonometry;
+pub mod trigonometry;
mod unique;
use std::fmt::{Display, Formatter};
@@ -88,10 +88,10 @@ pub use self::boolean::BooleanFunction;
#[cfg(feature = "business")]
pub(super) use self::business::BusinessFunction;
#[cfg(feature = "dtype-categorical")]
-pub(crate) use self::cat::CategoricalFunction;
+pub use self::cat::CategoricalFunction;
#[cfg(feature = "temporal")]
pub(super) use self::datetime::TemporalFunction;
-pub(super) use self::pow::PowFunction;
+pub use self::pow::PowFunction;
#[cfg(feature = "range")]
pub(super) use self::range::RangeFunction;
#[cfg(feature = "rolling_window")]
@@ -183,7 +183,13 @@ pub enum FunctionExpr {
#[cfg(feature = "dtype-struct")]
AsStruct,
#[cfg(feature = "top_k")]
- TopK(bool),
+ TopK {
+ sort_options: SortOptions,
+ },
+ #[cfg(feature = "top_k")]
+ TopKBy {
+ sort_options: SortMultipleOptions,
+ },
#[cfg(feature = "cum_agg")]
CumCount {
reverse: bool,
@@ -432,7 +438,7 @@ impl Hash for FunctionExpr {
has_max.hash(state);
},
#[cfg(feature = "top_k")]
- TopK(a) => a.hash(state),
+ TopK { sort_options } => sort_options.hash(state),
#[cfg(feature = "cum_agg")]
CumCount { reverse } => reverse.hash(state),
#[cfg(feature = "cum_agg")]
@@ -551,6 +557,8 @@ impl Hash for FunctionExpr {
#[cfg(feature = "reinterpret")]
Reinterpret(signed) => signed.hash(state),
ExtendConstant => {},
+ #[cfg(feature = "top_k")]
+ TopKBy { sort_options } => sort_options.hash(state),
}
}
}
@@ -623,13 +631,17 @@ impl Display for FunctionExpr {
#[cfg(feature = "dtype-struct")]
AsStruct => "as_struct",
#[cfg(feature = "top_k")]
- TopK(descending) => {
+ TopK {
+ sort_options: SortOptions { descending, .. },
+ } => {
if *descending {
"bottom_k"
} else {
"top_k"
}
},
+ #[cfg(feature = "top_k")]
+ TopKBy { .. } => "top_k_by",
Shift => "shift",
#[cfg(feature = "cum_agg")]
CumCount { .. } => "cum_count",
@@ -950,9 +962,11 @@ impl From for SpecialEq> {
map_as_slice!(coerce::as_struct)
},
#[cfg(feature = "top_k")]
- TopK(descending) => {
- map_as_slice!(top_k, descending)
+ TopK { sort_options } => {
+ map_as_slice!(top_k, sort_options)
},
+ #[cfg(feature = "top_k")]
+ TopKBy { sort_options } => map_as_slice!(top_k_by, sort_options.clone()),
Shift => map_as_slice!(shift_and_fill::shift),
#[cfg(feature = "cum_agg")]
CumCount { reverse } => map!(cum::cum_count, reverse),
diff --git a/crates/polars-plan/src/dsl/function_expr/rolling.rs b/crates/polars-plan/src/dsl/function_expr/rolling.rs
index d3bf85877cd8..f1ae64c5f792 100644
--- a/crates/polars-plan/src/dsl/function_expr/rolling.rs
+++ b/crates/polars-plan/src/dsl/function_expr/rolling.rs
@@ -75,10 +75,6 @@ fn convert<'a>(
let mut by = ss[1].clone();
by = by.rechunk();
- polars_ensure!(
- options.weights.is_none(),
- ComputeError: "`weights` is not supported in 'rolling by' expression"
- );
let (by, tz) = match by.dtype() {
DataType::Datetime(tu, tz) => (by.cast(&DataType::Datetime(*tu, None))?, tz),
DataType::Date => (
@@ -116,12 +112,12 @@ fn convert<'a>(
let options = RollingOptionsImpl {
window_size: options.window_size,
min_periods: options.min_periods,
- weights: None,
+ weights: options.weights,
center: options.center,
by: Some(by_values),
tu: Some(tu),
tz: tz.as_ref(),
- closed_window: options.closed_window.or(Some(ClosedWindow::Right)),
+ closed_window: options.closed_window,
fn_params: options.fn_params.clone(),
};
@@ -130,7 +126,7 @@ fn convert<'a>(
}
pub(super) fn rolling_min(s: &Series, options: RollingOptions) -> PolarsResult {
- s.rolling_min(options.clone().try_into()?)
+ s.rolling_min(options.into())
}
pub(super) fn rolling_min_by(s: &[Series], options: RollingOptions) -> PolarsResult {
@@ -138,7 +134,7 @@ pub(super) fn rolling_min_by(s: &[Series], options: RollingOptions) -> PolarsRes
}
pub(super) fn rolling_max(s: &Series, options: RollingOptions) -> PolarsResult {
- s.rolling_max(options.clone().try_into()?)
+ s.rolling_max(options.into())
}
pub(super) fn rolling_max_by(s: &[Series], options: RollingOptions) -> PolarsResult {
@@ -146,7 +142,7 @@ pub(super) fn rolling_max_by(s: &[Series], options: RollingOptions) -> PolarsRes
}
pub(super) fn rolling_mean(s: &Series, options: RollingOptions) -> PolarsResult {
- s.rolling_mean(options.clone().try_into()?)
+ s.rolling_mean(options.into())
}
pub(super) fn rolling_mean_by(s: &[Series], options: RollingOptions) -> PolarsResult {
@@ -154,7 +150,7 @@ pub(super) fn rolling_mean_by(s: &[Series], options: RollingOptions) -> PolarsRe
}
pub(super) fn rolling_sum(s: &Series, options: RollingOptions) -> PolarsResult {
- s.rolling_sum(options.clone().try_into()?)
+ s.rolling_sum(options.into())
}
pub(super) fn rolling_sum_by(s: &[Series], options: RollingOptions) -> PolarsResult {
@@ -162,7 +158,7 @@ pub(super) fn rolling_sum_by(s: &[Series], options: RollingOptions) -> PolarsRes
}
pub(super) fn rolling_quantile(s: &Series, options: RollingOptions) -> PolarsResult {
- s.rolling_quantile(options.clone().try_into()?)
+ s.rolling_quantile(options.into())
}
pub(super) fn rolling_quantile_by(s: &[Series], options: RollingOptions) -> PolarsResult