feat(rust): Vortex Phase 2 — AExpr-direct filter pushdown (stacked on #2)

crates/polars-plan/src/plans/aexpr/predicates/mod.rs

@@ -1,5 +1,7 @@
 mod column_expr;
 mod skip_batches;
+#[cfg(feature = "vortex")]
+pub mod vortex_convertor;
 
 use std::borrow::Cow;
 

crates/polars-stream/src/nodes/io_sources/vortex/builder.rs

@@ -6,6 +6,7 @@ use polars_io::cloud::CloudOptions;
 use polars_io::metrics::IOMetrics;
 use polars_plan::dsl::ScanSource;
 use polars_vortex::read::VortexSegmentCacheRef;
+use polars_vortex::vortex::expr::Expression as VortexExpression;
 use polars_vortex::{VortexScanOptions, vortex};
 
 use super::VortexFileReader;
@@ -23,6 +24,17 @@ pub struct VortexReaderBuilder {
     /// streaming source falls back to `options.segment_cache.resolve()` (e.g., user-supplied
     /// schema path where no postscript read happened at IR-build).
     pub segment_cache: Option<VortexSegmentCacheRef>,
+    /// AExpr-direct convertor result (PR-13.2, sole pushdown path as of PR-2.6): when
+    /// the predicate translated cleanly via
+    /// `polars_plan::plans::predicates::vortex_convertor::aexpr_to_vortex_expression`
+    /// (the `aexpr` module is `pub(crate)`; the externally-resolvable path goes through
+    /// the `pub use aexpr::*` re-export at `plans/mod.rs`),
+    /// the Vortex `Expression` is captured here at IR-build time (where we still have
+    /// `expr_arena` access). `VortexFileReader::begin_read` uses it directly. The
+    /// multi-scan layer reapplies the full predicate post-decode regardless (we
+    /// advertise `PARTIAL_FILTER`), so it is safe to push only a subset; shapes the
+    /// convertor returns `None` for fall through to no-pushdown + post-decode reapply.
+    pub aexpr_filter: Option<VortexExpression>,
     pub io_metrics: std::sync::OnceLock<Arc<IOMetrics>>,
 }
 
@@ -42,9 +54,12 @@ impl FileReaderBuilder for VortexReaderBuilder {
     fn reader_capabilities(&self) -> ReaderCapabilities {
         use ReaderCapabilities as RC;
         // The multi-scan layer reapplies the full predicate post-decode, so PARTIAL_FILTER
-        // is always safe; FULL_FILTER would require the convertor to consume every shape
-        // it sees (out-of-scope while we lean on `SpecializedColumnPredicate`).
-        // EXTERNAL_FILTER_MASK would need Vortex `Selection` bitmap plumbing.
+        // is always safe; FULL_FILTER would require the AExpr-direct convertor to be a
+        // strict superset of every AExpr predicate Polars constructs (still out of scope:
+        // the convertor returns None for unhandled shapes — Sort/Gather/Filter/Agg/Ternary/
+        // AnonymousFunction/Over/Rolling/temporal extracts/etc. — and the multi-scan
+        // reapply handles them). EXTERNAL_FILTER_MASK would need Vortex `Selection` bitmap
+        // plumbing.
         RC::ROW_INDEX
             | RC::PRE_SLICE
             | RC::NEGATIVE_PRE_SLICE
@@ -78,6 +93,11 @@ impl FileReaderBuilder for VortexReaderBuilder {
             // Threaded resolved cache for the data read; `None` triggers fallback resolve()
             // inside `VortexFileReader::initialize`. Same pattern as `footer` above.
             segment_cache: self.segment_cache.clone(),
+            // AExpr-direct convertor result (sole pushdown path as of PR-2.6). Shared
+            // across all sources in a multi-source scan — the convertor result is
+            // purely a function of the (predicate, schema) pair, both of which are
+            // constant across the scan's sources.
+            aexpr_filter: self.aexpr_filter.clone(),
             io_metrics: OptIOMetrics(self.io_metrics.get().cloned()),
             init_data: None,
         }) as _

crates/polars-stream/src/nodes/io_sources/vortex/mod.rs

@@ -20,7 +20,6 @@ use polars_utils::slice_enum::Slice;
 use polars_vortex::read::array_bridge::{
     ArrowUpstreamSchema, arrow_dtypes_from_schema, record_batch_to_dataframe,
 };
-use polars_vortex::read::predicate::polars_to_vortex_predicate;
 use polars_vortex::read::read_at::local_file_read_at;
 use polars_vortex::read::schema::vortex_dtype_to_schema;
 use polars_vortex::session::session;
@@ -51,6 +50,14 @@ pub struct VortexFileReader {
     /// read. When `None` (user-supplied-schema path; no IR-build postscript read happened),
     /// `initialize()` falls back to `options.segment_cache.resolve()` for a fresh cache.
     pub segment_cache: Option<polars_vortex::read::VortexSegmentCacheRef>,
+    /// AExpr-direct convertor result threaded from IR-build (`FileScanIR::Vortex` →
+    /// `VortexReaderBuilder::aexpr_filter` → here). When `Some`, `begin_read` uses this
+    /// Vortex `Expression` directly. PR-2.6 (Option B → A cutover) deleted the legacy
+    /// `polars_to_vortex_predicate` fallback — the AExpr-direct convertor is now the
+    /// sole filter-pushdown path. When `None` (unhandled AExpr shape, or
+    /// `push_predicate=false`), no filter pushes down; the multi-scan layer reapplies
+    /// the predicate post-decode (we advertise `PARTIAL_FILTER`).
+    pub aexpr_filter: Option<polars_vortex::vortex::expr::Expression>,
     pub io_metrics: OptIOMetrics,
 
     /// Set by `initialize()`.
@@ -248,12 +255,18 @@ impl FileReader for VortexFileReader {
             }
         });
 
-        // Translate the pushable bits of args.predicate into a Vortex `Expression`. We
-        // advertise `PARTIAL_FILTER` capability, so the multi-scan layer keeps the
-        // original predicate around to apply post-decode — pushing only what we can
-        // convert is safe (over-conservative pushdown would drop rows incorrectly).
+        // Use the AExpr-direct convertor result computed at IR-build time
+        // (`physical_plan::lower_ir`). We advertise `PARTIAL_FILTER` capability so the
+        // multi-scan layer keeps the original predicate around to apply post-decode —
+        // pushing only what we can convert is safe.
+        //
+        // PR-2.6 (Option B → A cutover) deleted the legacy `polars_to_vortex_predicate`
+        // (`SpecializedColumnPredicate`-derived) fallback path; the AExpr-direct
+        // convertor is now the sole filter-pushdown path. When `aexpr_filter` is `None`
+        // (unhandled AExpr shape, or `push_predicate=false`), no filter pushes down and
+        // the multi-scan reapply handles correctness.
         let filter_expr = if self.options.push_predicate {
-            args.predicate.as_ref().and_then(polars_to_vortex_predicate)
+            self.aexpr_filter.clone()
         } else {
             None
         };

crates/polars-stream/src/physical_plan/lower_ir.rs

@@ -20,6 +20,8 @@ use polars_plan::dsl::default_values::DefaultFieldValues;
 use polars_plan::dsl::deletion::DeletionFilesList;
 use polars_plan::dsl::{CallbackSinkType, ExtraColumnsPolicy, FileScanIR, SinkTypeIR};
 use polars_plan::plans::expr_ir::{ExprIR, OutputName};
+#[cfg(feature = "vortex")]
+use polars_plan::plans::predicates::vortex_convertor;
 use polars_plan::plans::{AExpr, FunctionIR, IR, IRAggExpr, LiteralValue, write_ir_non_recursive};
 use polars_plan::prelude::*;
 use polars_utils::arena::{Arena, Node};
@@ -767,17 +769,76 @@ pub fn lower_ir(
                         options,
                         metadata: first_metadata,
                         segment_cache,
-                    } => Arc::new(
-                        crate::nodes::io_sources::vortex::builder::VortexReaderBuilder {
-                            options: Arc::new(options.clone()),
-                            first_metadata: first_metadata.clone(),
-                            // Threaded from IR-build (scans.rs::vortex_file_info caller); when
-                            // None (schema-supplied path), the streaming source resolves a
-                            // fresh cache from `options.segment_cache`.
-                            segment_cache: segment_cache.clone(),
-                            io_metrics: std::sync::OnceLock::new(),
-                        },
-                    ) as _,
+                    } => {
+                        // PR-13.2 AExpr-direct pushdown (sole pushdown path as of PR-2.6):
+                        // when a predicate exists and `push_predicate` is on, try the
+                        // convertor. The result rides on the builder; `VortexFileReader::
+                        // begin_read` uses it directly. We are still at IR-build time
+                        // here, so we have `expr_arena` access — which we lose by the
+                        // time `begin_read` runs (where only `Arc<dyn PhysicalIoExpr>`
+                        // survives). When the convertor returns `None` (unhandled AExpr
+                        // shape OR all minterms touched virtual columns), `aexpr_filter`
+                        // stays `None` and no filter pushes down; the multi-scan layer
+                        // reapplies the predicate post-decode via the `PARTIAL_FILTER`
+                        // capability so correctness is preserved.
+                        //
+                        // Per-column file-vs-virtual split (PR-2.8, replaces PR-2.2
+                        // cycle-1 must-fix M1's all-or-nothing guard): `file_info.schema`
+                        // "Does not include logical columns like `include_file_path` and
+                        // row index" but DOES include hive columns (plans/schema.rs:42-43).
+                        // The convertor's `AExpr::Column` arm emits a bare `get_item(name,
+                        // root())` with no schema-membership check, so a predicate
+                        // referencing a hive column / row_index / include_file_paths would
+                        // emit a Vortex reference to a column not in the file's data —
+                        // Vortex would bail at `into_array_stream`.
+                        //
+                        // Build a `virtual_cols` set from hive partition column names
+                        // + row_index name + include_file_paths name. Pass it to
+                        // `vortex_convertor::aexpr_file_minterms_to_vortex_expression`,
+                        // which walks top-level conjuncts via `MintermIter` and converts
+                        // only the file-only minterms; virtual-column-touching minterms
+                        // are left for the multi-scan reapply. Mirrors
+                        // `polars-mem-engine/scan_predicate/functions::create_scan_predicate`'s
+                        // `hive_predicate` extraction. Resolves Deferred work entry
+                        // "Virtual-column-partitioned Vortex scans don't benefit from
+                        // AExpr convertor pushdown".
+                        let aexpr_filter = if options.push_predicate {
+                            predicate.as_ref().and_then(|p| {
+                                let mut virtual_cols: PlHashSet<PlSmallStr> = PlHashSet::default();
+                                if let Some(hp) = hive_parts.as_ref() {
+                                    for name in hp.schema().iter_names() {
+                                        virtual_cols.insert(name.clone());
+                                    }
+                                }
+                                if let Some(ri) = unified_scan_args.row_index.as_ref() {
+                                    virtual_cols.insert(ri.name.clone());
+                                }
+                                if let Some(ifp) = unified_scan_args.include_file_paths.as_ref() {
+                                    virtual_cols.insert(ifp.clone());
+                                }
+                                vortex_convertor::aexpr_file_minterms_to_vortex_expression(
+                                    p.node(),
+                                    expr_arena,
+                                    Some(file_info.schema.as_ref()),
+                                    &virtual_cols,
+                                )
+                            })
+                        } else {
+                            None
+                        };
+                        Arc::new(
+                            crate::nodes::io_sources::vortex::builder::VortexReaderBuilder {
+                                options: Arc::new(options.clone()),
+                                first_metadata: first_metadata.clone(),
+                                // Threaded from IR-build (scans.rs::vortex_file_info caller); when
+                                // None (schema-supplied path), the streaming source resolves a
+                                // fresh cache from `options.segment_cache`.
+                                segment_cache: segment_cache.clone(),
+                                aexpr_filter,
+                                io_metrics: std::sync::OnceLock::new(),
+                            },
+                        ) as _
+                    },
 
                     #[cfg(feature = "csv")]
                     FileScanIR::Csv { options } => {