Polars Pushes Predicates Into the Parquet Scanner

            Scan {
                sources,
                file_info,
                hive_parts: scan_hive_parts,
                ref predicate,
                predicate_file_skip_applied,
                scan_type,
                unified_scan_args,
                output_schema,
                maintain_order,
            } => {

                let predicate = combine_predicates(
                    Option::into_iter(predicate.clone()).chain(acc_predicates.into_values()),
                    expr_arena,
                );

                let do_optimization = predicate.is_some()
                    && match &*scan_type {
                        #[cfg(feature = "csv")]
                        FileScanIR::Csv { .. } => unified_scan_args.pre_slice.is_none(),
                        FileScanIR::ExpandedPaths { .. } => false,
                        FileScanIR::Anonymous { function, .. } => {
                            function.allows_predicate_pushdown()
                        },
                        #[allow(unreachable_patterns)]
                        _ => true,
                    };

                let lp = if do_optimization {
                    Scan {
                        sources,
                        file_info,
                        hive_parts,
                        predicate,
                        predicate_file_skip_applied,
                        unified_scan_args,
                        output_schema,
                        scan_type,
                        maintain_order,
                    }
                } else {
                    let lp = Scan {
                        sources,
                        file_info,
                        hive_parts,
                        predicate: None,
                        predicate_file_skip_applied,
                        unified_scan_args,
                        output_schema,
                        scan_type,
                        maintain_order,
                    };
                    if let Some(predicate) = predicate {
                        let input = lp_arena.add(lp);
                        Filter { input, predicate }
                    } else {
                        lp
                    }
                };

                        // col(A) == B -> {
                        //     null_count(A) == 0                              , if B.is_null(),
                        //     null_count(A) == LEN || min(A) > B || max(A) < B, if B.is_not_null(),
                        // }

                            not_null: {
                                let col_min = target.min(arena);
                                let col_max = target.max(arena);

                                let min_is_defined = is_stat_defined(col_min.node(), dtype, arena);
                                let max_is_defined = is_stat_defined(col_max.node(), dtype, arena);

                                let min_gt = col_min.gt(lv_node, arena);
                                let min_gt = min_gt.and(min_is_defined, arena);

                                let max_lt = col_max.lt(lv_node, arena);
                                let max_lt = max_lt.and(max_is_defined, arena);

                                let col_nc = target.null_count(arena);
                                let len = col!(len);
                                let all_nulls = col_nc.eq(len, arena);

                                all_nulls.or(min_gt, arena).or(max_lt, arena).node()
                            }

/// Whether min/max statistics are usable for the given dtype, operator, and literal.
///
/// Rejects nested, null, and categorical types. For floats, Parquet stats exclude NaN
/// but data may contain it. Since NaN is largest under TotalOrd, `col < x` is safe
/// (NaN never matches) but `col > x` is not (NaN always matches).
/// col >= NaN matches NaN == NaN, but stats exclude NaN so max < NaN can't detect them -> unsafe
fn can_use_min_max_stats(
    dtype: &DataType,
    op: Option<&Operator>,
    lv: Option<&LiteralValue>,
) -> bool {
    if dtype.is_nested() || dtype.is_null() || dtype.is_categorical() {
        return false;
    }

    if !dtype.is_float() {
        return true;
    }

    let lv_is_nan = lv.is_some_and(|lv| lv.is_nan());

    use Operator as O;
    match op {
        Some(O::Lt | O::LtEq) => true,
        None | Some(O::Eq | O::EqValidity) => !lv_is_nan && lv.is_some(),
        Some(O::Gt) => lv_is_nan,
        _ => false,
    }
}

            let mut skip_batch_schema = Schema::with_capacity(1 + live_columns.len());

            skip_batch_schema.insert(PlSmallStr::from_static("len"), IDX_DTYPE);
            for (col, dtype) in schema.iter() {
                if !live_columns.contains(col) {
                    continue;
                }

                skip_batch_schema.insert(format_pl_smallstr!("{col}_min"), dtype.clone());
                skip_batch_schema.insert(format_pl_smallstr!("{col}_max"), dtype.clone());
                skip_batch_schema.insert(format_pl_smallstr!("{col}_nc"), null_count_dtype(dtype));
            }

            skip_batch_predicate = Some(create_physical_expr(
                &expr,
                expr_arena,
                &Arc::new(skip_batch_schema),
                state,
            )?);

/// Whether the `min_value` and `max_value` of a leaf bound the values polars decodes
/// as `dtype`: the file declares an order for the leaf, and it is the one polars
/// compares the decoded values with.
fn bounds_are_usable(
    column_order: ColumnOrder,
    physical_type: &PhysicalType,
    dtype: &ArrowDataType,
) -> bool {
    let Some(required) = required_sort_order(dtype, physical_type) else {
        return false;
    };
    match column_order {
        ColumnOrder::TypeDefinedOrder(order) => order == required,
        // Total order agrees with the float comparison on every non-NaN value.
        ColumnOrder::IEEE754TotalOrder => {
            matches!(physical_type, PhysicalType::Float | PhysicalType::Double)
        },
        ColumnOrder::Unsupported | ColumnOrder::Undefined => false,
    }
}

    /// Converts the bounds of a chunk of the leaf. A timestamp the reader
    /// multiplies past `i64` wraps, leaving the chunk's values without order:
    /// such a conversion needs both bounds, and both within range, to hold.
    pub fn convert_bounds<'a>(
        self,
        min: Option<PhysicalBound<'a>>,
        max: Option<PhysicalBound<'a>>,
    ) -> ParquetResult<(Option<ArrowBound<'a>>, Option<ArrowBound<'a>>)> {
        let convert = |bound: Option<_>| bound.map(|b| self.convert(b)).transpose();
        let (min, max) = (convert(min)?.flatten(), convert(max)?.flatten());
        if self.can_wrap() && (min.is_none() || max.is_none()) {
            return Ok((None, None));
        }
        Ok((min, max))
    }

        let mut columns = Vec::with_capacity(1 + skip_batch_columns.len() * 3);

        let lengths: Vec<IdxSize> = row_groups_slice
            .iter()
            .map(|rg| rg.num_rows() as IdxSize)
            .collect();

        columns.push(Column::new("len".into(), lengths));

        for projection in projected_arrow_fields.iter() {
            let c = projection.output_name();

            if !skip_batch_columns.contains(c) {
                continue;
            }

            let mut statistics =
                load_parquet_column_statistics(&metadata, row_group_slice.clone(), projection)?;

            // Note: Order is important here. We re-use the transform for the output column, meaning
            // that it may set the column name.
            statistics.min = projection.apply_transform(statistics.min)?;
            statistics.max = projection.apply_transform(statistics.max)?;
            // The null counts of struct fields have the file's field layout. The transform may
            // reorder, rename or insert fields (e.g. by field ID), and casts the leaves, so it
            // does not apply to them: use unknown null counts.
            if let ArrowFieldProjection::Mapped { output_dtype, .. } = projection
                && matches!(projection.arrow_field().dtype(), ArrowDataType::Struct(_))
            {
                statistics.null_count = Column::full_null(
                    PlSmallStr::EMPTY,
                    statistics.null_count.len(),
                    &null_count_dtype(output_dtype),
                );
            }

            let statistics = statistics.with_base_column_name(c);

            columns.extend([statistics.min, statistics.max, statistics.null_count]);
        }

        if let Some(row_index) = row_index {
            let statistics = build_row_index_statistics(&row_index, row_groups_slice)
                .with_base_column_name(&row_index.name);

            columns.extend([statistics.min, statistics.max, statistics.null_count]);
        }

        let statistics_df = DataFrame::new(num_row_groups, columns)?;

        sbp.evaluate_with_stat_df(&statistics_df)

pub(super) fn row_groups_to_read(
    metadata: &FileMetadata,
    row_group_slice: Range<usize>,
    row_group_mask: Option<&Bitmap>,
    mut row_offset: usize,
    mut slice_range: Option<Range<usize>>,
) -> VecDeque<RowGroupToRead> {
    let mut row_groups = VecDeque::with_capacity(row_group_slice.len());
    for (i, idx) in row_group_slice.enumerate() {
        let num_rows = metadata.row_groups[idx].num_rows();
        let slice = slice_range.as_mut().map(|slice_range| {
            let rg_row_start = slice_range.start;
            let rg_row_end = slice_range.end.min(num_rows);

            *slice_range = slice_range.start.saturating_sub(num_rows)
                ..slice_range.end.saturating_sub(num_rows);

            (rg_row_start, rg_row_end - rg_row_start)
        });
        if !row_group_mask.is_some_and(|mask| mask.get_bit(i)) {
            row_groups.push_back(RowGroupToRead {
                idx,
                row_offset,
                slice,
            });
        }
        row_offset = row_offset.saturating_add(num_rows);
    }
    row_groups
}

    let mut phys_predicate = create_physical_expr(&predicate, expr_arena, schema, state)?;

    // The hive predicate settles whole files; nothing is left to filter rows by.
    if hive_predicate_is_full_predicate {
        hive_predicate = Some(phys_predicate.clone());
        filters_rows = false;
        let node = expr_arena.add(AExpr::Literal(Scalar::from(true).into()));
        predicate = ExprIR::from_node(node, expr_arena);
        phys_predicate = create_physical_expr(&predicate, expr_arena, schema, state)?;
    }

        let df = if let Some(predicate) = self
            .predicate
            .as_ref()
            .filter(|p| p.filters_rows && !nothing_to_evaluate)
        {
            let mask = predicate.predicate.evaluate_io(&df)?;
            let mask = mask.bool().unwrap();

pub struct PredicateFileSkip {
    /// If `true` the predicate can be skipped at runtime.
    pub no_residual_predicate: bool,
    /// Number of files before skipping
    pub original_len: usize,
}

    let predicate_to_readers = (!can_skip_scan_predicate).then_some(&scan_predicate);

    if let Some(skip_files_mask) = skip_files_mask {
        assert_eq!(skip_files_mask.len(), sources.len());

        let predicate_file_skip = PredicateFileSkip {
            no_residual_predicate: predicate_to_readers.is_none(),
            original_len: sources.len(),
        };

        if verbose {
            eprintln!("apply_scan_predicate_to_scan_ir: {predicate_file_skip:?}");
        }

        *predicate_file_skip_applied = Some(predicate_file_skip);

    if unified_scan_args.has_row_index_or_slice() || predicate_file_skip_applied.is_some() {
        return Ok(());
    }