feat(parquet-datasource): always accept pushable filters, run rejected conjuncts post-scan

The parquet scan now always accepts pushable filter conjuncts so the parent
`FilterExec` is removed, and guarantees every accepted conjunct is applied
somewhere:

  - `pushdown_filters = true`  → row-filterable conjuncts via the parquet
    `RowFilter` (during decode); any rejected conjuncts fall through to a
    new in-scan post-scan filter applied on decoded batches.
  - `pushdown_filters = false` → the whole predicate runs as a post-scan
    filter on decoded batches (behaviorally identical to a `FilterExec`).

Also fixes a real correctness bug on main. `build_row_filter` (and its
`RowFilterGenerator` wrapper) silently dropped conjuncts that
`FilterCandidateBuilder::build` rejected (`Ok(None)` was `.flatten()`-ed
away) and swallowed whole-build errors. `ParquetSource::try_pushdown_filters`
had already removed the parent `FilterExec`, so those conjuncts were never
applied anywhere — most commonly when the per-file expr-adapter rewrite
turns a table-schema-pushable predicate into a non-pushable one (schema
evolution / coercion / whole-struct references). `build_row_filter` now
returns `(Option<RowFilter>, Vec<Arc<dyn PhysicalExpr>>)` so the opener
routes rejected conjuncts to the post-scan filter.

Packaging: the projection-widening + rebasing + filter evaluation lives in
a new `post_scan_filter.rs` module exposing `PostScanFilter` and
`DecoderProjection`. The opener calls `DecoderProjection::build(...)` in
place of the prior inline `build_projection_read_plan` /
`reassign_expr_columns` / `make_projector` block — net simplification of
`opener/mod.rs`.

Regression tests:

  - `build_row_filter_surfaces_rejected_struct_conjunct` (`row_filter.rs`)
    asserts the new API contract — `build_row_filter` no longer drops the
    rejected conjunct.
  - `rejected_struct_conjunct_runs_post_scan_not_dropped` (`opener/mod.rs`)
    is an end-to-end test: a struct file with `s IS NOT NULL` and
    `pushdown_filters=true` would return 3 rows on main (conjunct silently
    dropped, predicate relaxed); with this fix it correctly returns 2.

Many tests / snapshots / `.slt` explain outputs are updated to reflect the
new plan shape: the parquet scan now owns the predicate, so the
`FilterExec` is gone from plans (clickbench, push_down_filter_parquet,
projection_pushdown, etc.). Row-count assertions in opener/integration
tests that used to assert "row group not pruned ⇒ all rows returned" now
correctly reflect the matching row count, since the scan applies the
predicate row-level via the post-scan filter.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/core/src/dataframe/parquet.rs

@@ -162,9 +162,14 @@ mod tests {
             .select_columns(&["bool_col", "int_col"])?;
 
         let plan = df.explain(false, false)?.collect().await?;
-        // Filters all the way to Parquet
+        // Filters all the way to Parquet. The parquet scan now accepts the
+        // pushable filter unconditionally so the `FilterExec` is removed —
+        // the predicate appears as `predicate=` on the `DataSourceExec`.
         let formatted = pretty::pretty_format_batches(&plan)?.to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"), "{formatted}");
+        assert!(
+            formatted.contains("predicate=id@0 = 1"),
+            "expected predicate=id@0 = 1 in {formatted}"
+        );
 
         Ok(())
     }

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -751,17 +751,12 @@ mod tests {
             .await
             .unwrap();
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
-        +-----+----+----+
-        | c1  | c3 | c2 |
-        +-----+----+----+
-        |     |    |    |
-        |     | 10 | 1  |
-        |     | 20 |    |
-        |     | 20 | 2  |
-        | Foo | 10 |    |
-        | bar |    |    |
-        +-----+----+----+
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
+        +----+----+----+
+        | c1 | c3 | c2 |
+        +----+----+----+
+        |    | 20 | 2  |
+        +----+----+----+
         ");
     }
 
@@ -896,7 +891,9 @@ mod tests {
         assert_eq!(get_value(&metrics, "pushdown_rows_matched"), 0);
         assert_eq!(rt.batches.unwrap().len(), 0);
 
-        // Predicate should prune no row groups
+        // Predicate should prune no row groups. The scan now applies the
+        // predicate post-scan (since `pushdown_filters` is off in this test),
+        // so only the matching row survives.
         let filter = col("c1").eq(lit(ScalarValue::Utf8(Some("foo".to_string()))));
         let rt = RoundTrip::new()
             .with_predicate(filter)
@@ -913,7 +910,7 @@ mod tests {
             .iter()
             .map(|b| b.num_rows())
             .sum::<usize>();
-        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+        assert_eq!(read, 1, "Expected 1 row to match the predicate");
     }
 
     #[tokio::test]
@@ -937,7 +934,9 @@ mod tests {
         assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
         assert_eq!(rt.batches.unwrap().len(), 0);
 
-        // Predicate should prune no row groups
+        // Predicate should prune no row groups. The scan now applies the
+        // predicate post-scan (since `pushdown_filters` is off in this test),
+        // so only the matching row survives.
         let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(1))));
         let rt = RoundTrip::new()
             .with_predicate(filter)
@@ -953,7 +952,7 @@ mod tests {
             .iter()
             .map(|b| b.num_rows())
             .sum::<usize>();
-        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+        assert_eq!(read, 1, "Expected 1 row to match the predicate");
     }
 
     #[tokio::test]
@@ -1051,17 +1050,12 @@ mod tests {
         // In a real query where this predicate was pushed down from a filter stage instead of created directly in the `DataSourceExec`,
         // the filter stage would be preserved as a separate execution plan stage so the actual query results would be as expected.
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
-        +-----+----+
-        | c1  | c2 |
-        +-----+----+
-        |     |    |
-        |     |    |
-        |     | 1  |
-        |     | 2  |
-        | Foo |    |
-        | bar |    |
-        +-----+----+
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
+        +----+----+
+        | c1 | c2 |
+        +----+----+
+        |    | 1  |
+        +----+----+
         ");
     }
 
@@ -1147,23 +1141,14 @@ mod tests {
             .round_trip(vec![batch1, batch2, batch3, batch4])
             .await;
 
-        insta::assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()), @r"
-        +------+----+
-        | c1   | c2 |
-        +------+----+
-        |      | 1  |
-        |      | 2  |
-        | Bar  |    |
-        | Bar  | 2  |
-        | Bar  | 2  |
-        | Bar2 |    |
-        | Bar3 |    |
-        | Foo  |    |
-        | Foo  | 1  |
-        | Foo  | 1  |
-        | Foo2 |    |
-        | Foo3 |    |
-        +------+----+
+        insta::assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()), @"
+        +-----+----+
+        | c1  | c2 |
+        +-----+----+
+        |     | 1  |
+        | Foo | 1  |
+        | Foo | 1  |
+        +-----+----+
         ");
         let metrics = rt.parquet_exec.metrics().unwrap();
 
@@ -1232,11 +1217,10 @@ mod tests {
             .await
             .unwrap();
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
         +-----+----+
         | c1  | c2 |
         +-----+----+
-        |     | 2  |
         | Foo | 1  |
         | bar |    |
         +-----+----+
@@ -1756,12 +1740,11 @@ mod tests {
 
         let metrics = rt.parquet_exec.metrics().unwrap();
 
-        assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()),@r"
+        assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()),@"
         +-----+
         | int |
         +-----+
         | 4   |
-        | 5   |
         +-----+
         ");
         let (page_index_rows_pruned, page_index_rows_matched) =

datafusion/core/src/datasource/view_test.rs

@@ -322,11 +322,16 @@ mod tests {
 
         let plan = df.explain(false, false)?.collect().await?;
 
-        // Filters all the way to Parquet
+        // Filters all the way to Parquet. The parquet scan now accepts the
+        // pushable filter unconditionally so the `FilterExec` is removed —
+        // the predicate appears as `predicate=` on the `DataSourceExec`.
         let formatted = arrow::util::pretty::pretty_format_batches(&plan)
             .unwrap()
             .to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"));
+        assert!(
+            formatted.contains("predicate=id@0 = 1"),
+            "expected predicate=id@0 = 1 in {formatted}"
+        );
         Ok(())
     }
 

datafusion/core/tests/parquet/page_pruning.rs

@@ -113,30 +113,20 @@ async fn page_index_filter_one_col() {
     let filter = col("month").eq(lit(1_i32));
 
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // `month = 1` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(3330));
-    //  vec.push(RowSelector::select(339));
-    //  vec.push(RowSelector::skip(3319));
-    // total 651 row
-    assert_eq!(batches[0].num_rows(), 651);
+    // `month = 1` from the page index narrows IO to ~651 candidate rows
+    // (RowSelector skip/select pattern), and the in-scan post-scan filter
+    // then rejects the 31 page-aligned rows that don't actually match.
+    assert_eq!(batches[0].num_rows(), 620);
 
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 620);
 
     // 2. create filter month == 1 or month == 2;
     let filter = col("month").eq(lit(1_i32)).or(col("month").eq(lit(2_i32)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // `month = 1` or `month = 2` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(900));
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(2118));
-    //  vec.push(RowSelector::select(339));
-    //  vec.push(RowSelector::skip(873));
-    //  vec.push(RowSelector::select(318));
-    //  vec.push(RowSelector::skip(2128));
-    assert_eq!(batches[0].num_rows(), 1281);
+    // Page-index pruning leaves a 1281-row candidate set; the scan applies
+    // the predicate (RowFilter or post-scan) to land on 1180 matches.
+    assert_eq!(batches[0].num_rows(), 1180);
 
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 1180);
@@ -154,23 +144,18 @@ async fn page_index_filter_one_col() {
     // 4.create filter 0 < month < 2 ;
     let filter = col("month").gt(lit(0_i32)).and(col("month").lt(lit(2_i32)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // should same with `month = 1`
-    assert_eq!(batches[0].num_rows(), 651);
+    // Same matching set as `month = 1`.
+    assert_eq!(batches[0].num_rows(), 620);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 620);
 
     // 5.create filter date_string_col == "01/01/09"`;
     // Note this test doesn't apply type coercion so the literal must match the actual view type
     let filter = col("date_string_col").eq(lit(ScalarValue::new_utf8view("01/01/09")));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 14);
-
-    // there should only two pages match the filter
-    //                                  min                                        max
-    // page-20                        0  01/01/09                                  01/02/09
-    // page-21                        0  01/01/09                                  01/01/09
-    // each 7 rows
-    assert_eq!(batches[0].num_rows(), 14);
+    // Page index narrows to two pages of 7 rows each (14 rows); the scan
+    // then rejects the 4 page-aligned rows that don't actually match.
+    assert_eq!(batches[0].num_rows(), 10);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 10);
 }
@@ -183,11 +168,9 @@ async fn page_index_filter_multi_col() {
     // create filter month == 1 and year = 2009;
     let filter = col("month").eq(lit(1_i32)).and(col("year").eq(lit(2009)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    //  `year = 2009` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(3663));
-    //  vec.push(RowSelector::skip(3642));
-    //  combine with `month = 1` total 333 row
-    assert_eq!(batches[0].num_rows(), 333);
+    // Page index narrows IO to ~333 candidate rows; the scan then rejects
+    // the page-aligned rows that don't actually match, landing on 310.
+    assert_eq!(batches[0].num_rows(), 310);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 310);
 
@@ -197,15 +180,15 @@ async fn page_index_filter_multi_col() {
         .eq(lit(1_i32))
         .and(col("year").eq(lit(2009)).or(col("id").eq(lit(1))));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 651);
+    assert_eq!(batches[0].num_rows(), 310);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 310);
 
     // create filter (year = 2009 or id = 1)
     // this filter use two columns will not push down
     let filter = col("year").eq(lit(2009)).or(col("id").eq(lit(1)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 7300);
+    assert_eq!(batches[0].num_rows(), 3650);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 3650);
 
@@ -218,7 +201,7 @@ async fn page_index_filter_multi_col() {
         .and(col("id").eq(lit(1)))
         .or(col("year").eq(lit(2010)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 7300);
+    assert_eq!(batches[0].num_rows(), 3651);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 3651);
 }

datafusion/core/tests/sql/explain_analyze.rs

@@ -877,8 +877,12 @@ async fn parquet_explain_analyze() {
         .unwrap()
         .to_string();
 
-    // should contain aggregated stats
-    assert_contains!(&formatted, "output_rows=8");
+    // should contain aggregated stats. The scan now applies the predicate
+    // post-scan (the `FilterExec` above it is removed), so `output_rows` is
+    // the matching row count (5) rather than the decoded row count (8).
+    // The 3 rejected rows show up as `post_scan_rows_pruned=3`.
+    assert_contains!(&formatted, "output_rows=5");
+    assert_contains!(&formatted, "post_scan_rows_pruned=3");
     assert_contains!(
         &formatted,
         "row_groups_pruned_bloom_filter=1 total \u{2192} 1 matched"
@@ -1010,14 +1014,10 @@ async fn parquet_recursive_projection_pushdown() -> Result<()> {
 
     assert_snapshot!(
         actual,
-        @r"
+        @"
     SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
       RecursiveQueryExec: name=number_series, is_distinct=false
-        CoalescePartitionsExec
-          ProjectionExec: expr=[id@0 as id, 1 as level]
-            FilterExec: id@0 = 1
-              RepartitionExec: partitioning=RoundRobinBatch(NUM_CORES), input_partitions=1
-                DataSourceExec: file_groups={1 group: [[TMP_DIR/hierarchy.parquet]]}, projection=[id], file_type=parquet, predicate=id@0 = 1, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1 AND 1 <= id_max@1, required_guarantees=[id in (1)]
+        DataSourceExec: file_groups={1 group: [[TMP_DIR/hierarchy.parquet]]}, projection=[id, 1 as level], file_type=parquet, predicate=id@0 = 1, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1 AND 1 <= id_max@1, required_guarantees=[id in (1)]
         CoalescePartitionsExec
           ProjectionExec: expr=[id@0 + 1 as id, level@1 + 1 as level]
             FilterExec: id@0 < 10

datafusion/datasource-parquet/benches/parquet_nested_filter_pushdown.rs

@@ -115,9 +115,9 @@ fn scan_with_predicate(
     let file_metrics = ParquetFileMetrics::new(0, &path.display().to_string(), &metrics);
 
     let builder = if pushdown {
-        if let Some(row_filter) =
-            build_row_filter(predicate, file_schema, &metadata, false, &file_metrics)?
-        {
+        let (row_filter, _rejected) =
+            build_row_filter(predicate, file_schema, &metadata, false, &file_metrics)?;
+        if let Some(row_filter) = row_filter {
             builder.with_row_filter(row_filter)
         } else {
             builder

datafusion/datasource-parquet/benches/parquet_struct_filter_pushdown.rs

@@ -210,9 +210,9 @@ fn scan(
 
     let mut filter_applied = false;
     let builder = if pushdown {
-        if let Some(row_filter) =
-            build_row_filter(predicate, file_schema, &metadata, false, &file_metrics)?
-        {
+        let (row_filter, _rejected) =
+            build_row_filter(predicate, file_schema, &metadata, false, &file_metrics)?;
+        if let Some(row_filter) = row_filter {
             filter_applied = true;
             builder.with_row_filter(row_filter)
         } else {

datafusion/datasource-parquet/src/metrics.rs

@@ -59,6 +59,14 @@ pub struct ParquetFileMetrics {
     pub pushdown_rows_matched: Count,
     /// Total time spent evaluating row-level pushdown filters
     pub row_pushdown_eval_time: Time,
+    /// Total rows filtered out by the in-scan post-scan filter
+    /// (predicate conjuncts that could not be applied as a parquet
+    /// `RowFilter` and were instead evaluated on decoded batches).
+    pub post_scan_rows_pruned: Count,
+    /// Total rows that passed the in-scan post-scan filter.
+    pub post_scan_rows_matched: Count,
+    /// Total time spent evaluating the in-scan post-scan filter.
+    pub post_scan_filter_eval_time: Time,
     /// Total time spent evaluating row group-level statistics filters
     pub statistics_eval_time: Time,
     /// Total time spent evaluating row group Bloom Filters
@@ -167,6 +175,18 @@ impl ParquetFileMetrics {
         let row_pushdown_eval_time = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
             .subset_time("row_pushdown_eval_time", partition);
+
+        let post_scan_rows_pruned = MetricBuilder::new(metrics)
+            .with_new_label("filename", filename.to_string())
+            .with_category(MetricCategory::Rows)
+            .counter("post_scan_rows_pruned", partition);
+        let post_scan_rows_matched = MetricBuilder::new(metrics)
+            .with_new_label("filename", filename.to_string())
+            .with_category(MetricCategory::Rows)
+            .counter("post_scan_rows_matched", partition);
+        let post_scan_filter_eval_time = MetricBuilder::new(metrics)
+            .with_new_label("filename", filename.to_string())
+            .subset_time("post_scan_filter_eval_time", partition);
         let statistics_eval_time = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
             .subset_time("statistics_eval_time", partition);
@@ -202,6 +222,9 @@ impl ParquetFileMetrics {
             pushdown_rows_pruned,
             pushdown_rows_matched,
             row_pushdown_eval_time,
+            post_scan_rows_pruned,
+            post_scan_rows_matched,
+            post_scan_filter_eval_time,
             page_index_rows_pruned,
             page_index_pages_pruned,
             statistics_eval_time,