fix(optimizer): Stop dropping correlated subquery filters

axiom/optimizer/DerivedTable.cpp

@@ -2405,6 +2405,25 @@ bool DerivedTable::addFilter(ExprCP conjunct) {
   return true;
 }
 
+namespace {
+
+// True if 'table' is the null-producing side of an outer join in 'dt'. Such a
+// table cannot take on further join edges: any edge added to it would have to
+// be applied after the outer join, which the join order search cannot express.
+bool isNullProducingSide(const DerivedTableP dt, PlanObjectCP table) {
+  for (const auto* join : dt->joins) {
+    if (join->rightOptional() && join->rightTable() == table) {
+      return true;
+    }
+    if (join->leftOptional() && join->leftTable() == table) {
+      return true;
+    }
+  }
+  return false;
+}
+
+} // namespace
+
 void DerivedTable::distributeConjuncts() {
   if (!having.empty()) {
     VELOX_CHECK_NOT_NULL(aggregation);
@@ -2503,6 +2522,14 @@ void DerivedTable::distributeConjuncts() {
         continue;
       }
 
+      if (isNullProducingSide(this, tables[0]) ||
+          isNullProducingSide(this, tables[1])) {
+        // The conjunct constrains a table whose join order position is already
+        // pinned by an outer join. Leave it in place to evaluate above that
+        // join.
+        continue;
+      }
+
       ExprCP left = nullptr;
       ExprCP right = nullptr;
       // expr depends on 2 tables. If it is left = right or right = left and

axiom/optimizer/tests/TpchPlanTest.cpp

@@ -149,13 +149,13 @@ TEST_F(TpchPlanTest, q01) {
 TEST_F(TpchPlanTest, q02) {
   // (
   //   ((partsupp INNER part) INNER (supplier INNER (nation INNER region)))
-  //   INNER
+  //   LEFT
   //   agg((
   //     (partsupp LEFT SEMI (FILTER) part)
   //     INNER
   //     (supplier INNER (nation INNER region))
   //   ))
-  // )
+  // ) FILTER ps_supplycost = min
   auto matcher =
       matchScan("partsupp")
           .hashJoinInner(
@@ -164,7 +164,7 @@ TEST_F(TpchPlanTest, q02) {
               matchScan("supplier")
                   .hashJoinInner(matchScan("nation").hashJoinInner(
                       matchScan("region").filter("r_name = 'EUROPE'"))))
-          .hashJoinInner(
+          .hashJoinLeft(
               matchScan("partsupp")
                   .hashJoinLeftSemiFilter(matchScan("part").filter(
                       "p_size = 15 and p_type like '%BRASS'"))
@@ -176,6 +176,7 @@ TEST_F(TpchPlanTest, q02) {
                                   .filter("region_name = 'EUROPE'"))))
                   .aggregation()
                   .project())
+          .filter("ps_supplycost = min")
           .topN()
           .project()
           .build();

axiom/optimizer/tests/sql/subquery.sql

@@ -736,6 +736,20 @@ LEFT JOIN (VALUES (1, 'x')) AS u(k, b) ON t.a = u.k
 INNER JOIN (VALUES (1)) AS v(c)
   ON u.b IN (SELECT 'x' FROM (VALUES (1)) AS w(d) WHERE d = v.c)
 ----
+-- A correlated scalar aggregate attaches to p through a LEFT join during
+-- decorrelation. The comparison with sibling ps must remain a filter above
+-- that join instead of becoming a second join edge on the null-producing
+-- aggregate side.
+SELECT ps.k, ps.cost
+FROM (VALUES (1), (2)) AS p(k),
+     (VALUES (1, 100.0), (1, 200.0), (2, 50.0), (2, 30.0)) AS ps(k, cost)
+WHERE p.k = ps.k
+  AND ps.cost = (
+      SELECT min(ps2.cost)
+      FROM (VALUES (1, 100.0), (1, 200.0), (2, 50.0), (2, 30.0)) AS ps2(k, cost)
+      WHERE ps2.k = p.k)
+ORDER BY ps.k
+----
 -- Shared CTE with a nested-IN filter, referenced from both UNION legs,
 -- second leg wrapping it in GROUP BY.
 WITH s AS (