polars_io/
predicates.rs

1use std::fmt;
2
3use polars_arrow::array::Array;
4use polars_arrow::bitmap::{Bitmap, BitmapBuilder};
5use polars_arrow::datatypes::ArrowDataType;
6use polars_core::chunked_array::cast::CastOptions;
7use polars_core::prelude::*;
8#[cfg(feature = "parquet")]
9use polars_parquet::read::expr::{ParquetColumnExpr, ParquetScalar, SpecializedParquetColumnExpr};
10use polars_utils::format_pl_smallstr;
11
12pub trait PhysicalIoExpr: Send + Sync {
13    /// Take a [`DataFrame`] and produces a boolean [`Series`] that serves
14    /// as a predicate mask
15    fn evaluate_io(&self, df: &DataFrame) -> PolarsResult<Series>;
16}
17
18#[derive(Debug, Clone)]
19pub enum SpecializedColumnPredicate {
20    Equal(Scalar),
21    /// A closed (inclusive) range.
22    Between(Scalar, Scalar),
23    EqualOneOf(Box<[Scalar]>),
24    StartsWith(Box<[u8]>),
25    EndsWith(Box<[u8]>),
26    RegexMatch(Arc<regex::bytes::Regex>),
27    /// A Categorical or Enum column equal to one of these strings.
28    CategoricalEqualOneOf(Box<[PlSmallStr]>),
29}
30
31/// The strings that an `==` or `is_in` predicate on a Categorical or Enum column matches.
32#[cfg(feature = "parquet")]
33#[derive(Clone)]
34enum DictionaryNeedles {
35    One(Box<str>),
36    Many(polars_utils::aliases::PlHashSet<Box<str>>),
37}
38
39#[derive(Clone)]
40pub struct ColumnPredicateExpr {
41    column_name: PlSmallStr,
42    dtype: DataType,
43    source_arrow_dtype: ArrowDataType,
44    #[cfg(feature = "parquet")]
45    specialized: Option<SpecializedParquetColumnExpr>,
46    /// Lets dictionary values be compared as strings, without adding them to the categories.
47    #[cfg(feature = "parquet")]
48    dictionary_needles: Option<DictionaryNeedles>,
49    expr: Arc<dyn PhysicalIoExpr>,
50}
51
52impl ColumnPredicateExpr {
53    pub fn new(
54        column_name: PlSmallStr,
55        dtype: DataType,
56        source_arrow_dtype: ArrowDataType,
57        expr: Arc<dyn PhysicalIoExpr>,
58        specialized: Option<SpecializedColumnPredicate>,
59    ) -> Self {
60        use SpecializedColumnPredicate as S;
61        #[cfg(feature = "parquet")]
62        use SpecializedParquetColumnExpr as P;
63        #[cfg(feature = "parquet")]
64        let dictionary_needles = if matches!(source_arrow_dtype, ArrowDataType::Dictionary(..))
65            && (dtype.is_categorical() || dtype.is_enum())
66        {
67            let needles: Option<Vec<&str>> = match &specialized {
68                Some(S::Equal(scalar)) => scalar.value().get_str().map(|s| vec![s]),
69                Some(S::EqualOneOf(scalars)) => Some(
70                    scalars
71                        .iter()
72                        .filter_map(|scalar| scalar.value().get_str())
73                        .collect(),
74                ),
75                Some(S::CategoricalEqualOneOf(strings)) => {
76                    Some(strings.iter().map(|s| s.as_str()).collect())
77                },
78                _ => None,
79            };
80            needles.map(|mut needles| {
81                // Values that are not in an Enum are read as null.
82                #[cfg(feature = "dtype-categorical")]
83                if let DataType::Enum(_, mapping) = &dtype {
84                    needles.retain(|s| mapping.get_cat(s).is_some());
85                }
86                match needles.as_slice() {
87                    [needle] => DictionaryNeedles::One((*needle).into()),
88                    _ => DictionaryNeedles::Many(needles.into_iter().map(Box::from).collect()),
89                }
90            })
91        } else {
92            None
93        };
94        // A specialized predicate compares its scalars with the values as the file
95        // stores them, which polars scales for some Arrow types. Dictionary values can
96        // also differ from the column, e.g. an Enum turns unknown values into nulls.
97        #[cfg(feature = "parquet")]
98        let specialized = specialized.and_then(|s| {
99            if DataType::arrow_value_scale(&source_arrow_dtype) != 1
100                || matches!(source_arrow_dtype, ArrowDataType::Dictionary(..))
101            {
102                return None;
103            }
104            Some(match s {
105                S::Equal(s) => P::Equal(cast_to_parquet_scalar(s)?),
106                S::Between(low, high) => {
107                    P::Between(cast_to_parquet_scalar(low)?, cast_to_parquet_scalar(high)?)
108                },
109                S::EqualOneOf(scalars) => P::EqualOneOf(
110                    scalars
111                        .into_iter()
112                        .map(|s| cast_to_parquet_scalar(s).ok_or(()))
113                        .collect::<Result<Box<_>, ()>>()
114                        .ok()?,
115                ),
116                S::StartsWith(s) => P::StartsWith(s),
117                S::EndsWith(s) => P::EndsWith(s),
118                S::RegexMatch(s) => P::RegexMatch(s),
119                S::CategoricalEqualOneOf(_) => return None,
120            })
121        });
122
123        #[allow(unused_mut)]
124        let mut out = Self {
125            column_name,
126            dtype,
127            source_arrow_dtype,
128            #[cfg(feature = "parquet")]
129            specialized,
130            #[cfg(feature = "parquet")]
131            dictionary_needles,
132            expr,
133        };
134        // Values that are not in an Enum are read as null, which this predicate can keep.
135        #[cfg(feature = "parquet")]
136        if out.dictionary_needles.is_some()
137            && out.dtype.is_enum()
138            && out.evaluate_null().unwrap_or(true)
139        {
140            out.dictionary_needles = None;
141        }
142        out
143    }
144}
145
146#[cfg(feature = "parquet")]
147impl ParquetColumnExpr for ColumnPredicateExpr {
148    fn try_evaluate_mut(&self, values: &dyn Array, bm: &mut BitmapBuilder) -> PolarsResult<()> {
149        // We should never evaluate nulls with this.
150        assert!(values.validity().is_none_or(|v| v.set_bits() == 0));
151
152        if let Some(needles) = &self.dictionary_needles
153            && let Some(values) = values
154                .as_any()
155                .downcast_ref::<polars_arrow::array::Utf8ViewArray>()
156        {
157            bm.reserve(values.len());
158            match needles {
159                DictionaryNeedles::One(needle) => {
160                    for value in values.values_iter() {
161                        bm.push(value == needle.as_ref());
162                    }
163                },
164                DictionaryNeedles::Many(needles) => {
165                    for value in values.values_iter() {
166                        bm.push(needles.contains(value));
167                    }
168                },
169            }
170            return Ok(());
171        }
172
173        let series = predicate_values_to_series(
174            self.column_name.clone(),
175            values,
176            &self.dtype,
177            &self.source_arrow_dtype,
178        )?;
179        let column = series.into_column();
180        let df = unsafe { DataFrame::new_unchecked(values.len(), vec![column]) };
181
182        let true_mask = self.expr.evaluate_io(&df)?;
183        let true_mask = true_mask.bool()?;
184
185        bm.reserve(true_mask.len());
186        for chunk in true_mask.downcast_iter() {
187            match chunk.validity() {
188                None => bm.extend_from_bitmap(chunk.values()),
189                Some(v) => bm.extend_from_bitmap(&(chunk.values() & v)),
190            }
191        }
192        Ok(())
193    }
194    fn evaluate_null(&self) -> PolarsResult<bool> {
195        let column = Column::full_null(self.column_name.clone(), 1, &self.dtype);
196        let df = unsafe { DataFrame::new_unchecked(1, vec![column]) };
197
198        let true_mask = self.expr.evaluate_io(&df)?;
199        let true_mask = true_mask.bool()?;
200
201        Ok(true_mask.get(0).unwrap_or(false))
202    }
203
204    fn can_evaluate_null_slots(&self) -> bool {
205        // Evaluating dictionary values adds them to the categories.
206        self.dictionary_needles.is_some()
207            || !matches!(self.source_arrow_dtype, ArrowDataType::Dictionary(..))
208    }
209
210    fn as_specialized(&self) -> Option<&SpecializedParquetColumnExpr> {
211        self.specialized.as_ref()
212    }
213}
214
215#[cfg(feature = "parquet")]
216fn predicate_values_to_series(
217    name: PlSmallStr,
218    values: &dyn Array,
219    dtype: &DataType,
220    source_arrow_dtype: &ArrowDataType,
221) -> PolarsResult<Series> {
222    // Polars stores the values of some Arrow types scaled, e.g. Arrow seconds as
223    // milliseconds, so the predicate series cannot be constructed zero-copy.
224    if DataType::arrow_value_scale(source_arrow_dtype) != 1 {
225        let values = polars_compute::cast::cast(
226            values,
227            source_arrow_dtype,
228            polars_compute::cast::CastOptionsImpl::default(),
229        )?;
230        Series::try_from((name, values))
231    } else if matches!(source_arrow_dtype, ArrowDataType::Dictionary(..)) {
232        // Dictionary columns are evaluated on their string values. Don't insert any of them
233        // if they don't all fit, as the dictionary can hold values no row uses.
234        #[cfg(feature = "dtype-categorical")]
235        if let DataType::Categorical(_, mapping) = dtype
236            && mapping.num_cats_upper_bound() + values.len() > mapping.max_categories()
237        {
238            let values = values
239                .as_any()
240                .downcast_ref::<polars_arrow::array::Utf8ViewArray>()
241                .unwrap();
242            let new_values: polars_utils::aliases::PlHashSet<&str> = values
243                .values_iter()
244                .filter(|v| mapping.get_cat(v).is_none())
245                .collect();
246            polars_ensure!(
247                mapping.num_cats_upper_bound() + new_values.len() <= mapping.max_categories(),
248                ComputeError: "dictionary values do not fit in the categories"
249            );
250        }
251        Series::from_chunk_and_dtype(name, values.to_boxed(), &DataType::String)?.cast(dtype)
252    } else {
253        Series::from_chunk_and_dtype(name, values.to_boxed(), dtype)
254    }
255}
256
257#[cfg(feature = "parquet")]
258fn cast_to_parquet_scalar(scalar: Scalar) -> Option<ParquetScalar> {
259    use AnyValue as A;
260    use ParquetScalar as P;
261
262    Some(match scalar.into_value() {
263        A::Null => P::Null,
264        A::Boolean(v) => P::Boolean(v),
265
266        A::UInt8(v) => P::UInt8(v),
267        A::UInt16(v) => P::UInt16(v),
268        A::UInt32(v) => P::UInt32(v),
269        A::UInt64(v) => P::UInt64(v),
270
271        A::Int8(v) => P::Int8(v),
272        A::Int16(v) => P::Int16(v),
273        A::Int32(v) => P::Int32(v),
274        A::Int64(v) => P::Int64(v),
275
276        #[cfg(feature = "dtype-date")]
277        A::Date(v) => P::Int32(v),
278        #[cfg(feature = "dtype-datetime")]
279        A::Datetime(v, _, _) | A::DatetimeOwned(v, _, _) => P::Int64(v),
280        #[cfg(feature = "dtype-duration")]
281        A::Duration(v, _) => P::Int64(v),
282        #[cfg(feature = "dtype-time")]
283        A::Time(v) => P::Int64(v),
284
285        A::Float32(v) => P::Float32(v),
286        A::Float64(v) => P::Float64(v),
287
288        // @TODO: Cast to string
289        #[cfg(feature = "dtype-categorical")]
290        A::Categorical(_, _) | A::CategoricalOwned(_, _) | A::Enum(_, _) | A::EnumOwned(_, _) => {
291            return None;
292        },
293
294        A::String(v) => P::String(v.into()),
295        A::StringOwned(v) => P::String(v.as_str().into()),
296        A::Binary(v) => P::Binary(v.into()),
297        A::BinaryOwned(v) => P::Binary(v.into()),
298        _ => return None,
299    })
300}
301
302#[cfg(any(feature = "parquet", feature = "ipc"))]
303pub fn apply_predicate(
304    df: &mut DataFrame,
305    predicate: Option<&dyn PhysicalIoExpr>,
306    parallel: bool,
307) -> PolarsResult<()> {
308    if let (Some(predicate), false) = (&predicate, df.columns().is_empty()) {
309        let s = predicate.evaluate_io(df)?;
310        let mask = s.bool().expect("filter predicates was not of type boolean");
311
312        if parallel {
313            *df = df.filter(mask)?;
314        } else {
315            *df = df.filter_seq(mask)?;
316        }
317    }
318    Ok(())
319}
320
321pub struct ColumnStatistics {
322    pub dtype: DataType,
323    pub min: AnyValue<'static>,
324    pub max: AnyValue<'static>,
325    pub null_count: Option<IdxSize>,
326}
327
328pub trait SkipBatchPredicate: Send + Sync {
329    fn schema(&self) -> &SchemaRef;
330
331    fn can_skip_batch(
332        &self,
333        batch_size: IdxSize,
334        live_columns: &PlIndexSet<PlSmallStr>,
335        mut statistics: PlIndexMap<PlSmallStr, ColumnStatistics>,
336    ) -> PolarsResult<bool> {
337        let mut columns = Vec::with_capacity(1 + live_columns.len() * 3);
338
339        columns.push(Column::new_scalar(
340            PlSmallStr::from_static("len"),
341            Scalar::new(IDX_DTYPE, batch_size.into()),
342            1,
343        ));
344
345        for col in live_columns.iter() {
346            let dtype = self.schema().get(col).unwrap();
347            let (min, max, nc) = match statistics.swap_remove(col) {
348                None => (
349                    Scalar::null(dtype.clone()),
350                    Scalar::null(dtype.clone()),
351                    Scalar::null(null_count_dtype(dtype)),
352                ),
353                Some(stat) => (
354                    Scalar::new(dtype.clone(), stat.min),
355                    Scalar::new(dtype.clone(), stat.max),
356                    Scalar::new(
357                        IDX_DTYPE,
358                        stat.null_count.map_or(AnyValue::Null, |nc| nc.into()),
359                    ),
360                ),
361            };
362            columns.extend([
363                Column::new_scalar(format_pl_smallstr!("{col}_min"), min, 1),
364                Column::new_scalar(format_pl_smallstr!("{col}_max"), max, 1),
365                Column::new_scalar(format_pl_smallstr!("{col}_nc"), nc, 1),
366            ]);
367        }
368
369        // SAFETY:
370        // * Each column is length = 1
371        // * We have an IndexSet, so each column name is unique
372        let df = unsafe { DataFrame::new_unchecked(1, columns) };
373        Ok(self.evaluate_with_stat_df(&df)?.get_bit(0))
374    }
375    fn evaluate_with_stat_df(&self, df: &DataFrame) -> PolarsResult<Bitmap>;
376}
377
378/// Dtype of a column's `<col>_nc` statistic: per field for structs (see
379/// `polars_plan::plans::predicates::null_count_dtype`).
380fn null_count_dtype(dtype: &DataType) -> DataType {
381    match dtype {
382        #[cfg(feature = "dtype-struct")]
383        DataType::Struct(fields) => DataType::Struct(
384            fields
385                .iter()
386                .map(|f| Field::new(f.name().clone(), null_count_dtype(f.dtype())))
387                .collect(),
388        ),
389        _ => IDX_DTYPE,
390    }
391}
392
393/// The conjuncts of a row predicate that read one column, conjoined.
394#[derive(Clone)]
395pub struct ColumnPredicate {
396    /// The static conjuncts, conjoined. `None` when the column only has dynamic ones.
397    pub predicate: Option<Arc<dyn PhysicalIoExpr>>,
398    pub specialized: Option<SpecializedColumnPredicate>,
399    /// The conjuncts a producer sets at run time, each on its own.
400    pub dynamic: Vec<DynamicColumnPredicate>,
401}
402
403impl ColumnPredicate {
404    /// Every conjunct, static and dynamic, conjoined.
405    pub fn conjoined(&self) -> Arc<dyn PhysicalIoExpr> {
406        self.predicate
407            .iter()
408            .chain(self.dynamic.iter().map(|d| &d.predicate))
409            .cloned()
410            .reduce(|a, b| Arc::new(AndIoExpr(a, b)))
411            .unwrap()
412    }
413}
414
415/// A conjunct on one column that a producer sets at run time. It keeps every
416/// row until `source` says it filters rows.
417#[derive(Clone)]
418pub struct DynamicColumnPredicate {
419    pub predicate: Arc<dyn PhysicalIoExpr>,
420    pub source: Arc<dyn DynamicPredicateSource>,
421}
422
423/// `a AND b`.
424struct AndIoExpr(Arc<dyn PhysicalIoExpr>, Arc<dyn PhysicalIoExpr>);
425
426impl PhysicalIoExpr for AndIoExpr {
427    fn evaluate_io(&self, df: &DataFrame) -> PolarsResult<Series> {
428        let a = self.0.evaluate_io(df)?;
429        let b = self.1.evaluate_io(df)?;
430        Ok((a.bool()? & b.bool()?).into_series())
431    }
432}
433
434pub struct PhysicalExprWithConstCols<T> {
435    constants: Vec<(PlSmallStr, Scalar)>,
436    child: T,
437}
438
439impl SkipBatchPredicate for PhysicalExprWithConstCols<Arc<dyn SkipBatchPredicate>> {
440    fn schema(&self) -> &SchemaRef {
441        self.child.schema()
442    }
443
444    fn evaluate_with_stat_df(&self, df: &DataFrame) -> PolarsResult<Bitmap> {
445        let mut df = df.clone();
446        for (name, scalar) in self.constants.iter() {
447            df.with_column(Column::new_scalar(
448                name.clone(),
449                scalar.clone(),
450                df.height(),
451            ))?;
452        }
453        self.child.evaluate_with_stat_df(&df)
454    }
455}
456
457impl PhysicalIoExpr for PhysicalExprWithConstCols<Arc<dyn PhysicalIoExpr>> {
458    fn evaluate_io(&self, df: &DataFrame) -> PolarsResult<Series> {
459        let mut df = df.clone();
460        for (name, scalar) in self.constants.iter() {
461            df.with_column(Column::new_scalar(
462                name.clone(),
463                scalar.clone(),
464                df.height(),
465            ))?;
466        }
467
468        self.child.evaluate_io(&df)
469    }
470}
471
472/// The row predicate split into the conjuncts that read a single column and the rest.
473#[derive(Clone)]
474pub struct StagedScanIOPredicate {
475    pub column_predicates: Arc<PlIndexMap<PlSmallStr, ColumnPredicate>>,
476    /// The conjuncts that read no or several columns, conjoined.
477    pub rest: Option<Arc<dyn PhysicalIoExpr>>,
478}
479
480impl StagedScanIOPredicate {
481    /// Every conjunct on a constant column becomes part of `rest`.
482    fn with_constant_columns(&self, constants: &[(PlSmallStr, Scalar)]) -> Self {
483        let mut column_predicates = self.column_predicates.as_ref().clone();
484        let mut rest = self.rest.clone();
485        for (c, _) in constants {
486            if let Some(p) = column_predicates.shift_remove(c) {
487                let p = p.conjoined();
488                rest = Some(match rest {
489                    None => p,
490                    Some(rest) => Arc::new(AndIoExpr(rest, p)),
491                });
492            }
493        }
494        Self {
495            column_predicates: Arc::new(column_predicates),
496            rest: rest.map(|rest| {
497                Arc::new(PhysicalExprWithConstCols {
498                    constants: constants.to_vec(),
499                    child: rest,
500                }) as _
501            }),
502        }
503    }
504}
505
506/// What a producer has published for a column, read once per file: a reader
507/// skips the batches whose statistics fall outside the range.
508#[derive(Clone, Debug)]
509pub enum RuntimeRange {
510    /// Not published yet. Every batch is kept; a later file may see a range.
511    Pending,
512    /// Never published. Every batch is kept.
513    Disabled,
514    /// No value can match. Every batch is skipped.
515    Empty,
516    /// Only values in `lo..=hi` can match.
517    Range { lo: Scalar, hi: Scalar },
518}
519
520/// A reader's view of a predicate that a producer sets at run time.
521pub trait DynamicPredicateSource: Send + Sync {
522    fn runtime_range(&self) -> RuntimeRange;
523
524    /// Whether the producer has published a predicate that rejects rows.
525    fn filters_rows(&self) -> bool;
526
527    /// Whether a reader may stop evaluating the predicate when it rejects too
528    /// little: it stays as it is once set, and the producer checks every row
529    /// again.
530    fn can_bypass(&self) -> bool;
531}
532
533/// A column whose batches a reader may skip by a [`RuntimeRange`]. It is never
534/// evaluated per row.
535#[derive(Clone)]
536pub struct RuntimeRangeHint {
537    pub column: PlSmallStr,
538    pub source: Arc<dyn DynamicPredicateSource>,
539    /// The column's value in this file when it is not stored in the file, such as
540    /// a hive column or a missing column with a default.
541    pub constant: Option<Scalar>,
542}
543
544/// A range bound as `dtype`, or `None` when it does not survive the cast, in
545/// which case it bounds nothing.
546pub fn cast_bound(bound: &Scalar, dtype: &DataType) -> Option<Scalar> {
547    bound
548        .clone()
549        .cast_with_options(dtype, CastOptions::NonStrict)
550        .ok()
551        .filter(|b| !b.is_null())
552}
553
554impl RuntimeRangeHint {
555    /// Whether a file whose column is the constant `value` can hold a match.
556    /// `None` when the range does not settle it.
557    pub fn constant_matches(range: &RuntimeRange, value: &Scalar) -> Option<bool> {
558        match range {
559            RuntimeRange::Pending | RuntimeRange::Disabled => None,
560            RuntimeRange::Empty => Some(false),
561            RuntimeRange::Range { lo, hi } => {
562                if value.is_null() {
563                    return Some(false);
564                }
565                let lo = cast_bound(lo, value.dtype())?;
566                let hi = cast_bound(hi, value.dtype())?;
567                let value = value.value();
568                Some(value >= lo.value() && value <= hi.value())
569            },
570        }
571    }
572
573    /// Bind the hints of column `name` to its constant value in a file.
574    pub fn set_constant(hints: &mut [Self], name: &str, value: &Scalar) {
575        for hint in hints.iter_mut().filter(|h| h.column == name) {
576            hint.constant = Some(value.clone());
577        }
578    }
579}
580
581#[derive(Clone)]
582pub struct ScanIOPredicate {
583    pub predicate: Arc<dyn PhysicalIoExpr>,
584
585    /// `predicate` split for readers that filter while decoding.
586    pub staged: Option<StagedScanIOPredicate>,
587
588    /// Whether `predicate` filters rows at all. False when the scan only has
589    /// runtime ranges to skip batches by.
590    pub filters_rows: bool,
591
592    /// Column names that are used in the predicate.
593    pub live_columns: Arc<PlIndexSet<PlSmallStr>>,
594
595    /// A predicate that gets given statistics and evaluates whether a batch can be skipped.
596    pub skip_batch_predicate: Option<Arc<dyn SkipBatchPredicate>>,
597
598    /// Columns whose batches are skipped by a range published at run time.
599    pub runtime_ranges: Vec<RuntimeRangeHint>,
600
601    /// Predicate parts only referring to hive columns.
602    pub hive_predicate: Option<Arc<dyn PhysicalIoExpr>>,
603
604    pub hive_predicate_is_full_predicate: bool,
605}
606
607impl ScanIOPredicate {
608    /// Whether the predicate or a range hint reads the column.
609    pub fn reads_column(&self, name: &str) -> bool {
610        self.live_columns.contains(name) || self.runtime_ranges.iter().any(|h| h.column == name)
611    }
612
613    pub fn set_external_constant_columns(&mut self, constant_columns: Vec<(PlSmallStr, Scalar)>) {
614        if constant_columns.is_empty() {
615            return;
616        }
617
618        let mut live_columns = self.live_columns.as_ref().clone();
619        for (c, _) in constant_columns.iter() {
620            live_columns.swap_remove(c);
621        }
622        self.live_columns = Arc::new(live_columns);
623
624        for (name, value) in constant_columns.iter() {
625            RuntimeRangeHint::set_constant(&mut self.runtime_ranges, name, value);
626        }
627
628        if let Some(skip_batch_predicate) = self.skip_batch_predicate.take() {
629            let mut sbp_constant_columns = Vec::with_capacity(constant_columns.len() * 3);
630            for (c, v) in constant_columns.iter() {
631                sbp_constant_columns.push((format_pl_smallstr!("{c}_min"), v.clone()));
632                sbp_constant_columns.push((format_pl_smallstr!("{c}_max"), v.clone()));
633                let nc_dtype = null_count_dtype(v.dtype());
634                // Struct null counts are per field, and unknown here.
635                let nc = if v.is_null() || nc_dtype != IDX_DTYPE {
636                    AnyValue::Null
637                } else {
638                    (0 as IdxSize).into()
639                };
640                sbp_constant_columns
641                    .push((format_pl_smallstr!("{c}_nc"), Scalar::new(nc_dtype, nc)));
642            }
643            self.skip_batch_predicate = Some(Arc::new(PhysicalExprWithConstCols {
644                constants: sbp_constant_columns,
645                child: skip_batch_predicate,
646            }));
647        }
648
649        if let Some(staged) = self.staged.as_mut() {
650            *staged = staged.with_constant_columns(&constant_columns);
651        }
652
653        self.predicate = Arc::new(PhysicalExprWithConstCols {
654            constants: constant_columns,
655            child: self.predicate.clone(),
656        });
657    }
658}
659
660impl fmt::Debug for ScanIOPredicate {
661    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
662        f.write_str("scan_io_predicate")
663    }
664}
polars_io/predicates.rs

polars_io/
predicates.rs
