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

`ParquetSource::try_pushdown_filters` always returns the per-filter
`Yes` / `No` discriminant from `can_expr_be_pushed_down_with_schemas`,
regardless of the `pushdown_filters` config. The parent `FilterExec` is
always removed for pushable filters, and the scan owns the predicate.

The opener routes the predicate to the post-scan filter when
`pushdown_filters = false`, in addition to the rejected-conjunct path
that already exists for `pushdown_filters = true`:

  - `pushdown_filters = true`  → row-filterable conjuncts via the parquet
    `RowFilter`; any rejected conjuncts via the post-scan filter (the
    correctness fix from the previous commit).
  - `pushdown_filters = false` → the whole predicate runs as a post-scan
    filter on decoded batches (behaviorally identical to a `FilterExec`).

The `pushdown_filters` config keeps its meaning ("build a parquet
`RowFilter`"); doc comments updated.

Plan / test consequences (all results unchanged, plan shape and metrics
change):

  - The `FilterExec` no longer appears above a `DataSourceExec` for
    pushable parquet filters. The predicate appears as `predicate=…` on
    the `DataSourceExec`. Parquet `.slt` files are regenerated to reflect
    this (clickbench, push_down_filter_parquet, projection_pushdown,
    parquet*, etc.). Spurious whitespace churn from `--complete` was
    reverted.
  - Opener / integration tests that asserted "row group not pruned ⇒ all
    rows returned" (e.g. `a = 1` over `[1, 2, 3]` returning 3 rows) are
    updated to reflect the matching-row count, since the scan now applies
    the predicate row-level via the post-scan filter.
  - `FilterExec: id@0 = 1` assertions in DataFrame / view tests become
    `predicate=id@0 = 1` on the `DataSourceExec`.
  - Insta inline snapshots in `parquet.rs` and `explain_analyze.rs` are
    re-accepted (`output_rows=8` → `output_rows=5` plus
    `post_scan_rows_pruned=3`, multi-line plans collapse where the
    `FilterExec`/`RepartitionExec` chain is gone).

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