fix: Fix pushdown conjuncts instead of generating join columns

Author: MBkkt

axiom/optimizer/DerivedTable.cpp

@@ -918,6 +918,19 @@ void DerivedTable::distributeConjuncts() {
         tables[0]->as<DerivedTable>()->setOp.value() ==
             logical_plan::SetOperation::kUnionAll));
 
+  PlanObjectSet noPushdownTables;
+  for (const auto* join : joins) {
+    if (join->leftOptional()) {
+      // No pushdown to the left side of a RIGHT or FULL join.
+      noPushdownTables.add(join->leftTable());
+    }
+    if (join->rightOptional()) {
+      // No pushdown to the right side of a LEFT or FULL join.
+      noPushdownTables.add(join->rightTable());
+    }
+  }
+  VELOX_DCHECK(tables.size() > 1 || noPushdownTables.empty());
+
   for (auto i = 0; i < conjuncts.size(); ++i) {
     // No pushdown of non-deterministic except if only pushdown target is a
     // union all.
@@ -938,6 +951,10 @@ void DerivedTable::distributeConjuncts() {
         continue; // ValuesTable does not have filter pushdown.
       }
 
+      if (noPushdownTables.contains(tables[0])) {
+        continue; // No pushdown if depends on an optional side of a join.
+      }
+
       if (tables[0]->is(PlanType::kUnnestTableNode)) {
         // UnnestTable does not implement filter pushdown yet.
         // TODO: We can push down predicate to left side of unnest if

axiom/optimizer/Optimization.cpp

@@ -1098,53 +1098,6 @@ void Optimization::joinByIndex(
   }
 }
 
-struct ProjectionBuilder {
-  ColumnVector columns;
-  ExprVector exprs;
-
-  // Project 'expr' as 'column'.
-  void add(ColumnCP column, ExprCP expr) {
-    columns.emplace_back(column);
-    exprs.emplace_back(expr);
-  }
-
-  RelationOp* build(RelationOp* input) {
-    return make<Project>(
-        input, exprs, columns, isRedundantProject(input, exprs, columns));
-  }
-
-  ColumnVector inputColumns() const {
-    ColumnVector columns;
-    columns.reserve(exprs.size());
-    for (const auto* expr : exprs) {
-      VELOX_DCHECK(expr->isColumn());
-      columns.emplace_back(expr->as<Column>());
-    }
-
-    return columns;
-  }
-
-  PlanObjectSet outputColumns() const {
-    PlanObjectSet columnSet;
-    columnSet.unionObjects(columns);
-    return columnSet;
-  }
-};
-
-folly::F14FastMap<PlanObjectCP, ExprCP> makeJoinColumnMapping(
-    JoinEdgeP joinEdge) {
-  folly::F14FastMap<PlanObjectCP, ExprCP> mapping;
-  for (auto i = 0; i < joinEdge->leftColumns().size(); ++i) {
-    mapping.emplace(joinEdge->leftColumns()[i], joinEdge->leftExprs()[i]);
-  }
-
-  for (auto i = 0; i < joinEdge->rightColumns().size(); ++i) {
-    mapping.emplace(joinEdge->rightColumns()[i], joinEdge->rightExprs()[i]);
-  }
-
-  return mapping;
-}
-
 namespace {
 
 // Check if 'mark' column produced by a SemiProject join is used only to filter
@@ -1292,38 +1245,23 @@ void Optimization::joinByHash(
   PlanObjectSet probeColumns;
   probeColumns.unionObjects(plan->columns());
 
+  ColumnVector columns;
+  PlanObjectSet columnSet;
   ColumnCP mark = nullptr;
 
-  auto* joinEdge = candidate.join;
-
-  PlanObjectSet joinColumns;
-  joinColumns.unionObjects(joinEdge->leftColumns());
-  joinColumns.unionObjects(joinEdge->rightColumns());
-
-  // Mapping from join output column to probe or build side input.
-  auto joinColumnMapping = makeJoinColumnMapping(joinEdge);
-
-  ProjectionBuilder projectionBuilder;
-  bool needsProjection = false;
-
   state.downstreamColumns().forEach<Column>([&](auto column) {
     if (column == build.markColumn) {
       mark = column;
       return;
     }
 
-    if (joinColumns.contains(column)) {
-      projectionBuilder.add(column, joinColumnMapping.at(column));
-      needsProjection = true;
-      return;
-    }
-
     if ((probeOnly || !buildColumns.contains(column)) &&
         !probeColumns.contains(column)) {
       return;
     }
 
-    projectionBuilder.add(column, column);
+    columnSet.add(column);
+    columns.push_back(column);
   });
 
   if (mark) {
@@ -1335,10 +1273,11 @@ void Optimization::joinByHash(
 
   // If there is an existence flag, it is the rightmost result column.
   if (mark) {
-    projectionBuilder.add(mark, mark);
+    columnSet.add(mark);
+    columns.push_back(mark);
   }
 
-  state.columns = projectionBuilder.outputColumns();
+  state.columns = columnSet;
 
   const auto fanout = fanoutJoinTypeLimit(
       joinType,
@@ -1359,15 +1298,11 @@ void Optimization::joinByHash(
       std::move(buildKeys),
       candidate.join->filter(),
       fanout,
-      projectionBuilder.inputColumns());
+      std::move(columns));
 
   state.addCost(*join);
   state.cost.cost += buildState.cost.cost;
 
-  if (needsProjection) {
-    join = projectionBuilder.build(join);
-  }
-
   state.addNextJoin(&candidate, join, toTry);
 }
 
@@ -1447,38 +1382,23 @@ void Optimization::joinByHashRight(
       rightJoinType == velox::core::JoinType::kRightSemiFilter ||
       rightJoinType == velox::core::JoinType::kRightSemiProject;
 
+  ColumnVector columns;
+  PlanObjectSet columnSet;
   ColumnCP mark = nullptr;
 
-  auto* joinEdge = candidate.join;
-
-  PlanObjectSet joinColumns;
-  joinColumns.unionObjects(joinEdge->leftColumns());
-  joinColumns.unionObjects(joinEdge->rightColumns());
-
-  // Mapping from join output column to probe or build side input.
-  auto joinColumnMapping = makeJoinColumnMapping(joinEdge);
-
-  ProjectionBuilder projectionBuilder;
-  bool needsProjection = false;
-
   state.downstreamColumns().forEach<Column>([&](auto column) {
     if (column == probe.markColumn) {
       mark = column;
       return;
     }
 
-    if (joinColumns.contains(column)) {
-      projectionBuilder.add(column, joinColumnMapping.at(column));
-      needsProjection = true;
-      return;
-    }
-
     if (!buildColumns.contains(column) &&
         (buildOnly || !probeColumns.contains(column))) {
       return;
     }
 
-    projectionBuilder.add(column, column);
+    columnSet.add(column);
+    columns.push_back(column);
   });
 
   if (mark) {
@@ -1489,7 +1409,8 @@ void Optimization::joinByHashRight(
   tryOptimizeSemiProject(rightJoinType, mark, state, negation_);
 
   if (mark) {
-    projectionBuilder.add(mark, mark);
+    columnSet.add(mark);
+    columns.push_back(mark);
   }
 
   const auto fanout = fanoutJoinTypeLimit(
@@ -1500,7 +1421,7 @@ void Optimization::joinByHashRight(
 
   const auto buildCost = state.cost;
 
-  state.columns = projectionBuilder.outputColumns();
+  state.columns = columnSet;
   state.cost = probeState.cost;
   state.cost.cost += buildCost.cost;
 
@@ -1517,13 +1438,9 @@ void Optimization::joinByHashRight(
       std::move(buildKeys),
       candidate.join->filter(),
       fanout,
-      projectionBuilder.inputColumns());
+      std::move(columns));
   state.addCost(*join);
 
-  if (needsProjection) {
-    join = projectionBuilder.build(join);
-  }
-
   state.addNextJoin(&candidate, join, toTry);
 }
 
@@ -1721,6 +1638,19 @@ bool Optimization::placeConjuncts(
   state.dt->singleRowDts.forEach<DerivedTable>(
       [&](auto dt) { columnsAndSingles.unionObjects(dt->columns); });
 
+  PlanObjectSet noPushdownTables;
+  if (!joinsPlaced) {
+    for (const auto* join : state.dt->joins) {
+      if (join->leftOptional()) {
+        // No pushdown to the left side of a RIGHT or FULL join.
+        noPushdownTables.add(join->leftTable());
+      }
+      if (join->rightOptional()) {
+        // No pushdown to the right side of a LEFT or FULL join.
+        noPushdownTables.add(join->rightTable());
+      }
+    }
+  }
   ExprVector filters;
   for (auto& conjunct : state.dt->conjuncts) {
     if (!joinsPlaced && conjunct->containsNonDeterministic()) {
@@ -1729,6 +1659,12 @@ bool Optimization::placeConjuncts(
     if (state.placed.contains(conjunct)) {
       continue;
     }
+    if (!joinsPlaced) {
+      const auto* singleTable = conjunct->singleTable();
+      if (singleTable && noPushdownTables.contains(singleTable)) {
+        continue;
+      }
+    }
     if (conjunct->columns().isSubset(state.columns)) {
       state.columns.add(conjunct);
       filters.push_back(conjunct);

axiom/optimizer/Plan.cpp

@@ -256,11 +256,6 @@ PlanObjectSet PlanState::computeDownstreamColumns(bool includeFilters) const {
     if (addFilter && !join->filter().empty()) {
       addExprs(join->filter());
     }
-
-    if (addFilter) {
-      addExprs(join->leftExprs());
-      addExprs(join->rightExprs());
-    }
   }
 
   // Filters.

axiom/optimizer/QueryGraph.h

@@ -511,20 +511,6 @@ class JoinEdge {
     /// Marker column produced by 'exists' or 'not exists' join.
     ///  If set, the 'joinType' must be kSemi.
     ColumnCP markColumn{nullptr};
-
-    /// Columns produced by the 'left' side of a RIGHT or FULL OUTER join.
-    /// Requires 'leftOptional' to be true.
-    ColumnVector leftColumns;
-
-    /// Input expressions corresponding 1:1 to 'leftColumns'.
-    ExprVector leftExprs;
-
-    /// Columns produced by the 'right' side of a LEFT or FULL OUTER join.
-    /// Requires 'rightOptional' to be true.
-    ColumnVector rightColumns;
-
-    /// Input expressions corresponding 1:1 to 'rightColumns'.
-    ExprVector rightExprs;
   };
 
   /// @param leftTable The left table of the join. May be nullptr if 'leftKeys'
@@ -536,28 +522,14 @@ class JoinEdge {
         rightTable_(rightTable),
         filter_(std::move(spec.filter)),
         joinType_(spec.joinType),
-        markColumn_(spec.markColumn),
-        leftColumns_{spec.leftColumns},
-        leftExprs_{spec.leftExprs},
-        rightColumns_{spec.rightColumns},
-        rightExprs_{spec.rightExprs} {
+        markColumn_(spec.markColumn) {
     // Only left join can have null left table.
     VELOX_DCHECK(leftTable_ || isLeftOuter());
     VELOX_DCHECK_NOT_NULL(rightTable_);
     // filter_ is only for non-inner joins.
     VELOX_DCHECK(filter_.empty() || !isInner());
     // Mark column only for semi joins.
     VELOX_DCHECK(!markColumn_ || isSemi());
-
-    if (!leftColumns_.empty()) {
-      VELOX_DCHECK(leftOptional());
-      VELOX_DCHECK_EQ(leftColumns_.size(), leftExprs_.size());
-    }
-
-    if (!rightColumns_.empty()) {
-      VELOX_DCHECK(rightOptional());
-      VELOX_DCHECK_EQ(rightColumns_.size(), rightExprs_.size());
-    }
   }
 
   static JoinEdge* makeInner(PlanObjectCP leftTable, PlanObjectCP rightTable) {
@@ -642,22 +614,6 @@ class JoinEdge {
     return markColumn_;
   }
 
-  const ColumnVector& leftColumns() const {
-    return leftColumns_;
-  }
-
-  const ExprVector& leftExprs() const {
-    return leftExprs_;
-  }
-
-  const ColumnVector& rightColumns() const {
-    return rightColumns_;
-  }
-
-  const ExprVector& rightExprs() const {
-    return rightExprs_;
-  }
-
   void addEquality(ExprCP left, ExprCP right, bool update = false);
 
   /// True if inner join.
@@ -796,11 +752,6 @@ class JoinEdge {
 
   // Flag to set if right side has a match.
   ColumnCP const markColumn_;
-
-  const ColumnVector leftColumns_;
-  const ExprVector leftExprs_;
-  const ColumnVector rightColumns_;
-  const ExprVector rightExprs_;
 };
 
 using JoinEdgeP = JoinEdge*;