Reduce plan-shape sensitivity for scalar-subquery guard

Broaden derived-relation detection to include projection
wrappers over derived relations. Add regression tests to cover
alias/projection shape changes and ensure mixed-side filters
are preserved. Implement a panic-path robustness test to
confirm that eval mode resets properly, even on closure
panic using catch_unwind.

Author: kosiew

datafusion/optimizer/src/push_down_filter.rs

@@ -305,7 +305,13 @@ fn is_scalar_aggregate_subquery(plan: &LogicalPlan) -> bool {
 }
 
 fn is_derived_relation(plan: &LogicalPlan) -> bool {
-    matches!(plan, LogicalPlan::SubqueryAlias(_))
+    match plan {
+        LogicalPlan::SubqueryAlias(_) => true,
+        LogicalPlan::Projection(projection) => {
+            is_derived_relation(projection.input.as_ref())
+        }
+        _ => false,
+    }
 }
 
 // Keep post-join filters above certain scalar-subquery cross joins to preserve
@@ -2517,6 +2523,60 @@ mod tests {
         )
     }
 
+    #[test]
+    fn window_over_scalar_subquery_cross_join_with_project_wrapper_keeps_filter_above_join(
+    ) -> Result<()> {
+        let left = LogicalPlanBuilder::from(test_table_scan()?)
+            .project(vec![col("a").alias("nation"), col("b").alias("acctbal")])?
+            .alias("s")?
+            .project(vec![col("s.nation"), col("s.acctbal")])?
+            .build()?;
+        let right = LogicalPlanBuilder::from(test_table_scan_with_name("test1")?)
+            .project(vec![col("a").alias("acctbal")])?
+            .aggregate(
+                Vec::<Expr>::new(),
+                vec![avg(col("acctbal")).alias("avg_acctbal")],
+            )?
+            .alias("__scalar_sq_1")?
+            .build()?;
+
+        let window = Expr::from(WindowFunction::new(
+            WindowFunctionDefinition::WindowUDF(
+                datafusion_functions_window::row_number::row_number_udwf(),
+            ),
+            vec![],
+        ))
+        .partition_by(vec![col("s.nation")])
+        .order_by(vec![col("s.acctbal").sort(false, true)])
+        .build()
+        .unwrap();
+
+        let plan = LogicalPlanBuilder::from(left)
+            .cross_join(right)?
+            .filter(col("s.acctbal").gt(col("__scalar_sq_1.avg_acctbal")))?
+            .project(vec![col("s.nation"), col("s.acctbal")])?
+            .window(vec![window])?
+            .build()?;
+
+        assert_optimized_plan_equal!(
+            plan,
+            @r"
+        WindowAggr: windowExpr=[[row_number() PARTITION BY [s.nation] ORDER BY [s.acctbal DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+          Projection: s.nation, s.acctbal
+            Filter: s.acctbal > __scalar_sq_1.avg_acctbal
+              Cross Join:
+                Projection: s.nation, s.acctbal
+                  SubqueryAlias: s
+                    Projection: test.a AS nation, test.b AS acctbal
+                      TableScan: test
+                SubqueryAlias: __scalar_sq_1
+                  Aggregate: groupBy=[[]], aggr=[[avg(acctbal) AS avg_acctbal]]
+                    Projection: test1.a AS acctbal
+                      TableScan: test1
+        "
+        )
+    }
+
     #[test]
     fn cross_join_builder_uses_inner_join_with_no_join_keys() -> Result<()> {
         let plan = LogicalPlanBuilder::from(test_table_scan()?)

datafusion/optimizer/src/utils.rs

@@ -245,6 +245,8 @@ fn coerce(expr: Expr, schema: &DFSchema) -> Result<Expr> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::panic::{AssertUnwindSafe, catch_unwind};
+
     use datafusion_expr::{
         Operator, binary_expr, case, col, in_list, is_null, lit, when,
     };
@@ -512,4 +514,22 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn null_restriction_eval_mode_guard_restores_on_panic() {
+        set_null_restriction_eval_mode_for_test(NullRestrictionEvalMode::Auto);
+
+        let result = catch_unwind(AssertUnwindSafe(|| {
+            with_null_restriction_eval_mode_for_test(
+                NullRestrictionEvalMode::AuthoritativeOnly,
+                || panic!("intentional panic to verify test mode reset"),
+            )
+        }));
+
+        assert!(result.is_err());
+        assert_eq!(
+            null_restriction_eval_mode(),
+            NullRestrictionEvalMode::Auto
+        );
+    }
 }