fix(parquet-datasource): never drop conjuncts the RowFilter cannot place

`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. By the time
`build_row_filter` runs, `ParquetSource::try_pushdown_filters` has already
accepted the filter and the parent `FilterExec` has been removed, so those
dropped conjuncts were never applied anywhere — wrong results.

Most reproducible trigger: the per-file expr adapter rewrites a predicate
that was pushable at *table schema* time into something the
`PushdownChecker` rejects at *physical file schema* time (schema
evolution / coercion / whole-struct references introduced by the rewrite).

Surface the rejected conjuncts instead of dropping them:

  - `build_row_filter` now returns
    `Result<(Option<RowFilter>, Vec<Arc<dyn PhysicalExpr>>)>`. The second
    element is the conjuncts it could not place. Bench / in-file test call
    sites updated.
  - `RowFilterGenerator` exposes `rejected_conjuncts()`. On a whole-file
    build error it routes every conjunct through that list, so an error no
    longer relaxes the predicate.
  - `DecoderProjection::build` grows a `post_scan_conjuncts` parameter and
    a `post_scan_filter: Option<PostScanFilter>` field. When non-empty it
    widens the decoder mask (over the user projection ∪ post-scan filter
    columns), rebases the conjuncts onto the stream schema, and returns a
    `PostScanFilter` that the stream applies to every decoded batch with
    SQL `WHERE` semantics (mirroring `FilterExec`'s `batch_filter`).
  - `PushDecoderStreamState` carries the optional `PostScanFilter` and
    applies it in the `DecodeResult::Data` arm, skipping empty batches.
  - The decoder-local LIMIT is unsafe with a post-scan filter (the decoder
    would short-circuit before the filter rejects enough rows), so the
    opener routes the limit to `remaining_limit` whenever a post-scan
    filter is present.
  - New `post_scan_rows_pruned` / `post_scan_rows_matched` counters and
    `post_scan_filter_eval_time` `Time` on `ParquetFileMetrics`, mirroring
    the existing `pushdown_rows_*` / `row_pushdown_eval_time` so
    `EXPLAIN ANALYZE` keeps surfacing filter cost.

Two regression tests:

  - `build_row_filter_surfaces_rejected_struct_conjunct` (`row_filter.rs`)
    asserts the new API contract directly — the rejected struct conjunct
    is returned, not dropped.
  - `rejected_struct_conjunct_runs_post_scan_not_dropped` (`opener/mod.rs`)
    is end-to-end: with `pushdown_filters=true` and a `s IS NOT NULL`
    predicate over a struct column where one row is NULL, `main` returns 3
    rows (conjunct silently dropped, predicate relaxed); after this fix it
    correctly returns 2.

The `pushdown_filters = false` path is intentionally unchanged in this
commit — `try_pushdown_filters` still leaves the `FilterExec` above the
scan in that case. Always-accepting filters and removing the `FilterExec`
unconditionally is a separate behaviour change in a follow-up commit.

`push_down_filter_parquet.slt` updated for the new `post_scan_rows_*`
metric lines on `EXPLAIN ANALYZE` output.

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

Author: adriangb

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,

datafusion/datasource-parquet/src/opener/mod.rs

@@ -1149,31 +1149,38 @@ impl RowGroupsPrunedParquetOpen {
 
         let arrow_reader_metrics = ArrowReaderMetrics::enabled();
 
-        // Build the decoder projection (mask + projector + replace_schema) in
-        // a single call.  Encapsulating it behind `DecoderProjection` keeps
-        // the opener's orchestration body focused on filter / decoder /
-        // stream wiring, and gives a clean seam for the in-scan post-scan
-        // filter introduced in a later change.
+        // Build the `RowFilter` first so we can collect any conjuncts it
+        // could not place. Those rejected conjuncts must still be applied
+        // (`try_pushdown_filters` already removed the parent `FilterExec`),
+        // so they fall through to the in-scan post-scan filter.
+        let pushdown_predicate = prepared
+            .pushdown_filters
+            .then_some(prepared.predicate.as_ref())
+            .flatten();
+        let mut row_filter_generator = RowFilterGenerator::new(
+            pushdown_predicate,
+            &prepared.physical_file_schema,
+            file_metadata.as_ref(),
+            prepared.reorder_predicates,
+            &prepared.file_metrics,
+        );
+        let post_scan_conjuncts: Vec<Arc<dyn PhysicalExpr>> = if prepared.pushdown_filters
+        {
+            row_filter_generator.rejected_conjuncts().to_vec()
+        } else {
+            Vec::new()
+        };
+
         let decoder_projection = DecoderProjection::build(
             &prepared.projection,
+            &post_scan_conjuncts,
             &prepared.physical_file_schema,
             reader_metadata.parquet_schema(),
             &prepared.output_schema,
+            &prepared.file_metrics,
         )?;
 
         let (decoder, pending_decoders, remaining_limit) = {
-            let pushdown_predicate = prepared
-                .pushdown_filters
-                .then_some(prepared.predicate.as_ref())
-                .flatten();
-            let mut row_filter_generator = RowFilterGenerator::new(
-                pushdown_predicate,
-                &prepared.physical_file_schema,
-                file_metadata.as_ref(),
-                prepared.reorder_predicates,
-                &prepared.file_metrics,
-            );
-
             // Split into consecutive runs of row groups that share the same filter
             // requirement. Fully matched row groups skip the RowFilter; others need it.
             // Reverse the run order for reverse scans so the combined decoder stream
@@ -1183,8 +1190,14 @@ impl RowGroupsPrunedParquetOpen {
                 runs.reverse();
             }
             let run_count = runs.len();
-            let decoder_limit = prepared.limit.filter(|_| run_count == 1);
-            let remaining_limit = prepared.limit.filter(|_| run_count > 1);
+            // Decoder-local limits are only safe when no post-decode work can
+            // reject rows. A post-scan filter can — so when one is present
+            // the limit must be enforced at the stream level via
+            // `remaining_limit`.
+            let single_no_post_scan =
+                run_count == 1 && decoder_projection.post_scan_filter.is_none();
+            let decoder_limit = prepared.limit.filter(|_| single_no_post_scan);
+            let remaining_limit = prepared.limit.filter(|_| !single_no_post_scan);
 
             let decoder_config = DecoderBuilderConfig {
                 projection_mask: &decoder_projection.projection_mask,
@@ -1229,6 +1242,7 @@ impl RowGroupsPrunedParquetOpen {
             projection_mask: _,
             projector,
             replace_schema,
+            post_scan_filter,
         } = decoder_projection;
         let output_schema = Arc::clone(&prepared.output_schema);
         let files_ranges_pruned_statistics =
@@ -1245,6 +1259,7 @@ impl RowGroupsPrunedParquetOpen {
             predicate_cache_inner_records,
             predicate_cache_records,
             baseline_metrics: prepared.baseline_metrics,
+            post_scan_filter,
         }
         .into_stream();
 
@@ -2673,4 +2688,96 @@ mod test {
         assert!(runs[2].needs_filter);
         assert_eq!(runs[2].access_plan.row_group_indexes(), vec![3]);
     }
+
+    /// End-to-end regression test for the "drop-on-floor" bug fixed by
+    /// `build_row_filter` now returning rejected conjuncts and the opener
+    /// routing them to the post-scan filter.
+    ///
+    /// Setup: a parquet file with a struct column where some rows have a NULL
+    /// struct. Predicate `s IS NOT NULL` is set on the source with
+    /// `pushdown_filters = true`. `ParquetSource::try_pushdown_filters` would
+    /// have already removed the parent `FilterExec` (the conjunct is pushable
+    /// at table schema level). Inside `build_row_filter`,
+    /// `FilterCandidateBuilder::build` rejects the whole-struct reference as
+    /// non-primitive.
+    ///
+    /// Before the fix the rejected conjunct was silently dropped, leaving the
+    /// scan with no `RowFilter` and no post-scan filter, so every row was
+    /// returned — i.e. the predicate was relaxed and the query returned wrong
+    /// results. After the fix the conjunct is surfaced and applied as a
+    /// post-scan filter, so only the rows with a non-null struct survive.
+    #[tokio::test]
+    async fn rejected_struct_conjunct_runs_post_scan_not_dropped() {
+        use arrow::array::{Int32Array, StringArray, StructArray};
+        use arrow::buffer::NullBuffer;
+        use arrow::datatypes::Fields;
+
+        let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+
+        // Schema: id (Int32), s (Struct{value: Int32, label: Utf8}).
+        let struct_fields: Fields = vec![
+            Arc::new(Field::new("value", DataType::Int32, true)),
+            Arc::new(Field::new("label", DataType::Utf8, true)),
+        ]
+        .into();
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("s", DataType::Struct(struct_fields.clone()), true),
+        ]));
+
+        // Data: rows 0 and 2 have a non-null struct, row 1 is null.
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2, 3])),
+                Arc::new(StructArray::new(
+                    struct_fields,
+                    vec![
+                        Arc::new(Int32Array::from(vec![Some(10), None, Some(30)])) as _,
+                        Arc::new(StringArray::from(vec![Some("a"), None, Some("c")]))
+                            as _,
+                    ],
+                    Some(NullBuffer::from(vec![true, false, true])),
+                )),
+            ],
+        )
+        .unwrap();
+
+        let data_size = write_parquet_batches(
+            Arc::clone(&store),
+            "rejected.parquet",
+            vec![batch],
+            None,
+        )
+        .await;
+
+        let file =
+            PartitionedFile::new("rejected.parquet".to_string(), data_size as u64);
+
+        // `s IS NOT NULL` references a whole struct, which `PushdownChecker`
+        // flags as non-primitive — `FilterCandidateBuilder::build` returns
+        // `Ok(None)` and the conjunct lands in `rejected`.
+        let predicate = logical2physical(&col("s").is_not_null(), &schema);
+
+        let morselizer = ParquetMorselizerBuilder::new()
+            .with_store(Arc::clone(&store))
+            .with_schema(Arc::clone(&schema))
+            .with_predicate(predicate)
+            // The RowFilter path: emulates the post-`try_pushdown_filters`
+            // state where the parent `FilterExec` has already been removed
+            // and the scan owns the conjunct.
+            .with_pushdown_filters(true)
+            .build();
+
+        let stream = open_file(&morselizer, file).await.unwrap();
+        let (_, rows) = count_batches_and_rows(stream).await;
+
+        // 2 rows have a non-null struct. Before the fix this returned 3
+        // (the conjunct was silently dropped).
+        assert_eq!(
+            rows, 2,
+            "expected 2 rows with non-null struct; the rejected conjunct must \
+             be applied post-scan, not silently dropped"
+        );
+    }
 }