fix CI: update tests for OptionalFilterPhysicalExpr wrap and coalesced parquet ranges

Two CI failures left over from recent commits on this branch:

- test_discover_dynamic_filters_via_expressions_api: `cae5fe605` wrapped
  the probe-side dynamic filter in `OptionalFilterPhysicalExpr`, so the
  test's `downcast_ref::<DynamicFilterPhysicalExpr>` saw only the
  build-side filter (1 instead of 2). Also unwrap the optional wrapper
  before counting.

- query_single_parquet_file_with_single_predicate and
  query_single_parquet_file_multi_row_groups_multiple_predicates: the
  dynamic projection-mask commit (`ada04723f`) coalesces per-page range
  requests into per-column-chunk ranges (same byte totals, fewer
  sub-ranges per GET). Update snapshots to the coalesced ranges; matches
  the pattern already in the no-predicate `query_single_parquet_file`
  snapshot.

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

Author: adriangb

datafusion/core/tests/datasource/object_store_access.rs

@@ -904,7 +904,7 @@ async fn query_single_parquet_file_with_single_predicate() {
     RequestCountingObjectStore()
     Total Requests: 2
     - GET  (opts) path=parquet_table.parquet head=true
-    - GET  (ranges) path=parquet_table.parquet ranges=1064-1481,1481-1594,1594-2011,2011-2124
+    - GET  (ranges) path=parquet_table.parquet ranges=1064-1594,1594-2124
     "
     );
 }
@@ -928,8 +928,8 @@ async fn query_single_parquet_file_multi_row_groups_multiple_predicates() {
     RequestCountingObjectStore()
     Total Requests: 3
     - GET  (opts) path=parquet_table.parquet head=true
-    - GET  (ranges) path=parquet_table.parquet ranges=4-421,421-534,534-951,951-1064
-    - GET  (ranges) path=parquet_table.parquet ranges=1064-1481,1481-1594,1594-2011,2011-2124
+    - GET  (ranges) path=parquet_table.parquet ranges=4-534,534-1064
+    - GET  (ranges) path=parquet_table.parquet ranges=1064-1594,1594-2124
     "
     );
 }

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -2952,14 +2952,24 @@ async fn test_discover_dynamic_filters_via_expressions_api() {
     use datafusion_common::JoinType;
     use datafusion_common::tree_node::TreeNodeRecursion;
     use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
+    use datafusion_physical_expr_common::physical_expr::OptionalFilterPhysicalExpr;
     use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
 
     fn count_dynamic_filters(plan: &Arc<dyn ExecutionPlan>) -> usize {
         let mut count = 0;
 
-        // Check expressions from this node using apply_expressions
+        // Check expressions from this node using apply_expressions. The hash
+        // join wraps the pushed-down dynamic filter in
+        // `OptionalFilterPhysicalExpr`, so peek inside that wrapper too.
         let _ = plan.apply_expressions(&mut |expr| {
-            if let Some(_df) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
+            if expr.downcast_ref::<DynamicFilterPhysicalExpr>().is_some() {
+                count += 1;
+            } else if let Some(opt) = expr.downcast_ref::<OptionalFilterPhysicalExpr>()
+                && opt
+                    .inner()
+                    .downcast_ref::<DynamicFilterPhysicalExpr>()
+                    .is_some()
+            {
                 count += 1;
             }
             Ok(TreeNodeRecursion::Continue)