Fix reducing existences

axiom/optimizer/DerivedTable.cpp

@@ -176,11 +176,31 @@ void DerivedTable::setStartTables() {
 
 namespace {
 // Returns a right exists (semijoin) with 'table' on the left and one of
-// 'tables' on the right.
-JoinEdgeP makeExists(PlanObjectCP table, const PlanObjectSet& tables) {
-  for (auto join : joinedBy(table)) {
+// 'tables' on the right. Picks a table on the right that is not a right side of
+// exists inside 'super'.
+JoinEdgeP makeExists(
+    PlanObjectCP table,
+    const PlanObjectSet& tables,
+    const DerivedTable& super) {
+  // True if 'table' is not to the right of some non-commutative join in
+  // 'super'.
+  auto isJoinable = [&](PlanObjectCP table) -> bool {
+    for (auto& join : super.joins) {
+      if ((join->rightExists() || join->rightNotExists()) &&
+          join->rightTable() == table) {
+        return false;
+      }
+    }
+    return true;
+  };
+
+  for (auto join : optimizer::joinedBy(table)) {
+    if (!join->isInner()) {
+      continue;
+    }
     if (join->leftTable() == table) {
-      if (!tables.contains(join->rightTable())) {
+      if (!tables.contains(join->rightTable()) ||
+          !isJoinable(join->rightTable())) {
         continue;
       }
       auto* exists = JoinEdge::makeExists(table, join->rightTable());
@@ -191,7 +211,8 @@ JoinEdgeP makeExists(PlanObjectCP table, const PlanObjectSet& tables) {
     }
 
     if (join->rightTable() == table) {
-      if (!join->leftTable() || !tables.contains(join->leftTable())) {
+      if (!join->leftTable() || !tables.contains(join->leftTable()) ||
+          !isJoinable(join->leftTable())) {
         continue;
       }
 
@@ -323,7 +344,7 @@ void DerivedTable::import(
         // with exists to the main table(s) in the 'this'.
         importedExistences.unionSet(exists);
         auto existsTables = exists.toObjects();
-        auto existsJoin = makeExists(firstTable, exists);
+        auto existsJoin = makeExists(firstTable, exists, super);
         if (existsTables.size() > 1) {
           // There is a join on the right of exists. Needs its own dt.
           auto [existsDt, joinWithDt] = makeExistsDtAndJoin(

axiom/optimizer/Optimization.cpp

@@ -1013,6 +1013,7 @@ void Optimization::joinByIndex(
   }
 }
 
+
 void Optimization::joinByHash(
     const RelationOpPtr& plan,
     const JoinCandidate& candidate,
@@ -1164,7 +1165,6 @@ void Optimization::joinByHash(
       candidate.join->filter(),
       fanout,
       std::move(columns));
-
   state.addCost(*join);
   state.cost.setupCost += buildState.cost.unitCost + buildState.cost.setupCost;
   state.cost.totalBytes += buildState.cost.totalBytes;

axiom/optimizer/Plan.cpp

@@ -202,6 +202,28 @@ const PlanObjectSet& PlanState::downstreamColumns() const {
 
   // Joins.
   for (auto join : dt->joins) {
+    if (join->rightExists() || join->rightNotExists()) {
+      if (placed.contains(join->rightTable())) {
+	continue;
+      }
+      // For an unplaced exists/not exists downstream, we need the left side columns but not the right side since nothing is projected out from the right side.
+      addExprs(join->leftKeys());
+
+      if (!join->filter().empty()) {
+	// If there is a filter, then the filter columns that do not come from the right side are needed.
+	PlanObjectSet filterColumns;
+	for (auto& conjunct : join->filter()) {
+	  filterColumns.unionColumns(conjunct);
+	}
+	PlanObjectSet rightColumns;
+	for (auto& column : tableColumns(join->rightTable())) {
+	  rightColumns.add(column);
+	}
+	filterColumns.except(rightColumns);
+	result.unionSet(filterColumns);
+      }
+      continue;
+    }
     bool addFilter = false;
     if (!placed.contains(join->rightTable())) {
       addFilter = true;
@@ -401,6 +423,22 @@ const JoinEdgeVector& joinedBy(PlanObjectCP table) {
   return table->as<DerivedTable>()->joinedBy;
 }
 
+const ColumnVector& tableColumns(PlanObjectCP table) {
+  if (table->is(PlanType::kTableNode)) {
+    return table->as<BaseTable>()->columns;
+  }
+
+  if (table->is(PlanType::kDerivedTableNode)) {
+    return table->as<DerivedTable>()->columns;
+  }
+
+  if (table->is(PlanType::kValuesTableNode)) {
+    return table->as<ValuesTable>()->columns;
+  }
+
+  VELOX_FAIL("tableColumns expects BaseTable, DerivedTable, or ValuesTable");
+}
+
 std::pair<JoinSide, JoinSide> JoinCandidate::joinSides() const {
   return {join->sideOf(tables[0], false), join->sideOf(tables[0], true)};
 }
@@ -430,10 +468,24 @@ void JoinCandidate::addEdge(PlanState& state, JoinEdgeP edge) {
   for (auto i = 0; i < newPlacedSide.keys.size(); ++i) {
     auto* key = newPlacedSide.keys[i];
     if (!hasEqual(key, tableSide.keys)) {
+      // Make sure to create hyper edge.
       if (!compositeEdge) {
-        compositeEdge = make<JoinEdge>(*join);
+        compositeEdge = JoinEdge::makeInner(nullptr, joined);
+        if (joined == join->rightTable()) {
+          for (auto i = 0; i < join->numKeys(); ++i) {
+            compositeEdge->addEquality(
+                join->leftKeys()[i], join->rightKeys()[i]);
+          }
+        } else {
+          for (auto i = 0; i < join->numKeys(); ++i) {
+            compositeEdge->addEquality(
+                join->rightKeys()[i], join->leftKeys()[i]);
+          }
+        }
+
         join = compositeEdge;
       }
+
       auto [other, preFanout] = join->otherTable(placedSide.table);
       // do not recompute a fanout after adding more equalities. This makes the
       // join edge non-binary and it cannot be sampled.
@@ -488,12 +540,13 @@ std::string JoinCandidate::toString() const {
 }
 
 bool NextJoin::isWorse(const NextJoin& other) const {
-  float shuffle =
-      plan->distribution().isSamePartition(other.plan->distribution())
-      ? 0
-      : plan->cost().fanout * shuffleCost(plan->columns());
-  return cost.unitCost + cost.setupCost + shuffle >
-      other.cost.unitCost + other.cost.setupCost;
+  return false;
+  // float shuffle =
+  //     plan->distribution().isSamePartition(other.plan->distribution())
+  //     ? 0
+  //     : plan->cost().fanout * shuffleCost(plan->columns());
+  // return cost.unitCost + cost.setupCost + shuffle >
+  //     other.cost.unitCost + other.cost.setupCost;
 }
 
 size_t MemoKey::hash() const {

axiom/optimizer/Plan.h

@@ -360,6 +360,9 @@ namespace facebook::axiom::optimizer {
 
 const JoinEdgeVector& joinedBy(PlanObjectCP table);
 
+/// Returns the columns of a BaseTable, DerivedTable, or ValuesTable.
+const ColumnVector& tableColumns(PlanObjectCP table);
+
 /// Returns  the inverse join type, e.g. right outer from left outer.
 /// TODO Move this function to Velox.
 velox::core::JoinType reverseJoinType(velox::core::JoinType joinType);

axiom/optimizer/QueryGraph.h

@@ -552,6 +552,14 @@ class JoinEdge {
     return rightOptional_;
   }
 
+  bool rightExists() const {
+    return rightExists_;
+  }
+
+  bool rightNotExists() const {
+    return rightNotExists_;
+  }
+
   bool directed() const {
     return directed_;
   }
@@ -580,7 +588,7 @@ class JoinEdge {
       return false;
     }
 
-    return !leftTable_ || rightOptional_ || leftOptional_ || rightExists_ ||
+    return rightOptional_ || leftOptional_ || rightExists_ ||
         rightNotExists_ || markColumn_ || directed_;
   }
 

axiom/optimizer/ToVelox.cpp

@@ -24,6 +24,7 @@
 #include "velox/expression/ExprToSubfieldFilter.h"
 #include "velox/expression/ScopedVarSetter.h"
 #include "velox/vector/VariantToVector.h"
+#include "axiom/optimizer/DerivedTablePrinter.h"
 
 namespace facebook::axiom::optimizer {
 
@@ -1526,4 +1527,8 @@ extern std::string planString(const runner::MultiFragmentPlan* plan) {
   return plan->toString(true);
 }
 
+extern std::string dtString(const DerivedTable* dt) {
+  return DerivedTablePrinter::toText(*dt);
+}
+
 } // namespace facebook::axiom::optimizer