feat(fuzzer): Update make[merge|NLJ]Plan functions to generate multi-join plans

velox/core/PlanNode.h

@@ -324,6 +324,8 @@ class ValuesNode : public PlanNode {
   const size_t repeatTimes_;
 };
 
+using ValuesNodePtr = std::shared_ptr<const ValuesNode>;
+
 class ArrowStreamNode : public PlanNode {
  public:
   ArrowStreamNode(

velox/exec/fuzzer/DuckQueryRunner.cpp

@@ -102,6 +102,40 @@ DuckQueryRunner::aggregationFunctionDataSpecs() const {
   return kAggregationFunctionDataSpecs;
 }
 
+std::string DuckQueryRunner::getTableName(
+    const core::ValuesNodePtr& valuesNode) {
+  return fmt::format("t_{}", valuesNode->id());
+}
+
+std::unordered_map<std::string, std::vector<velox::RowVectorPtr>>
+DuckQueryRunner::getAllTablesAndNames(const core::PlanNodePtr& plan) {
+  std::unordered_map<std::string, std::vector<velox::RowVectorPtr>> result;
+  if (const auto valuesNode =
+          std::dynamic_pointer_cast<const core::ValuesNode>(plan)) {
+    result.insert({getTableName(valuesNode), valuesNode->values()});
+  } else {
+    for (const auto& source : plan->sources()) {
+      auto tablesAndNames = getAllTablesAndNames(source);
+      result.insert(tablesAndNames.begin(), tablesAndNames.end());
+    }
+  }
+  return result;
+}
+
+std::optional<std::multiset<std::vector<velox::variant>>>
+DuckQueryRunner::execute(const core::PlanNodePtr& plan) {
+  if (const auto sql = toSql(plan)) {
+    DuckDbQueryRunner queryRunner;
+    std::unordered_map<std::string, std::vector<RowVectorPtr>> inputMap =
+        getAllTablesAndNames(plan);
+    for (const auto& [tableName, input] : inputMap) {
+      queryRunner.createTable(tableName, input);
+    }
+    return queryRunner.execute(*sql, plan->outputType());
+  }
+  return std::nullopt;
+}
+
 std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
     const std::string& sql,
     const std::vector<RowVectorPtr>& input,
@@ -341,38 +375,54 @@ std::optional<std::string> DuckQueryRunner::toSql(
   return sql.str();
 }
 
-std::optional<std::string> DuckQueryRunner::toSql(
-    const std::shared_ptr<const core::HashJoinNode>& joinNode) {
-  const auto& joinKeysToSql = [](auto keys) {
-    std::stringstream out;
-    for (auto i = 0; i < keys.size(); ++i) {
-      if (i > 0) {
-        out << ", ";
-      }
-      out << keys[i]->name();
+static const std::string joinKeysToSql(
+    const std::vector<core::FieldAccessTypedExprPtr>& keys) {
+  std::stringstream out;
+  for (auto i = 0; i < keys.size(); ++i) {
+    if (i > 0) {
+      out << ", ";
     }
-    return out.str();
-  };
+    out << keys[i]->name();
+  }
+  return out.str();
+}
 
-  const auto filterToSql = [](core::TypedExprPtr filter) {
-    auto call = std::dynamic_pointer_cast<const core::CallTypedExpr>(filter);
-    return toCallSql(call);
-  };
+static std::string filterToSql(const core::TypedExprPtr& filter) {
+  auto call = std::dynamic_pointer_cast<const core::CallTypedExpr>(filter);
+  return toCallSql(call);
+}
 
-  const auto& joinConditionAsSql = [&](auto joinNode) {
-    std::stringstream out;
-    for (auto i = 0; i < joinNode->leftKeys().size(); ++i) {
-      if (i > 0) {
-        out << " AND ";
-      }
-      out << joinNode->leftKeys()[i]->name() << " = "
-          << joinNode->rightKeys()[i]->name();
-    }
-    if (joinNode->filter()) {
-      out << " AND " << filterToSql(joinNode->filter());
+static std::string joinConditionAsSql(const core::AbstractJoinNode& joinNode) {
+  std::stringstream out;
+  for (auto i = 0; i < joinNode.leftKeys().size(); ++i) {
+    if (i > 0) {
+      out << " AND ";
     }
-    return out.str();
-  };
+    out << joinNode.leftKeys()[i]->name() << " = "
+        << joinNode.rightKeys()[i]->name();
+  }
+  if (joinNode.filter()) {
+    out << " AND " << filterToSql(joinNode.filter());
+  }
+  return out.str();
+}
+
+std::optional<std::string> DuckQueryRunner::toSql(
+    const std::shared_ptr<const core::HashJoinNode>& joinNode) {
+  std::string probeTableName;
+  std::string buildTableName;
+  if (const auto valuesNode = std::dynamic_pointer_cast<const core::ValuesNode>(
+          joinNode->sources()[0])) {
+    probeTableName = getTableName(valuesNode);
+  } else if (const auto subQuery = toSql(joinNode->sources()[0])) {
+    probeTableName = fmt::format("({})", *subQuery);
+  }
+  if (const auto valuesNode = std::dynamic_pointer_cast<const core::ValuesNode>(
+          joinNode->sources()[1])) {
+    buildTableName = getTableName(valuesNode);
+  } else if (const auto subQuery = toSql(joinNode->sources()[1])) {
+    buildTableName = fmt::format("({})", *subQuery);
+  }
 
   const auto& outputNames = joinNode->outputType()->names();
 
@@ -386,24 +436,27 @@ std::optional<std::string> DuckQueryRunner::toSql(
 
   switch (joinNode->joinType()) {
     case core::JoinType::kInner:
-      sql << " FROM t INNER JOIN u ON " << joinConditionAsSql(joinNode);
+      sql << " FROM " << probeTableName << " INNER JOIN " << buildTableName
+          << " ON " << joinConditionAsSql(*joinNode);
       break;
     case core::JoinType::kLeft:
-      sql << " FROM t LEFT JOIN u ON " << joinConditionAsSql(joinNode);
+      sql << " FROM " << probeTableName << " LEFT JOIN " << buildTableName
+          << " ON " << joinConditionAsSql(*joinNode);
       break;
     case core::JoinType::kFull:
-      sql << " FROM t FULL OUTER JOIN u ON " << joinConditionAsSql(joinNode);
+      sql << " FROM " << probeTableName << " FULL OUTER JOIN " << buildTableName
+          << " ON " << joinConditionAsSql(*joinNode);
       break;
     case core::JoinType::kLeftSemiFilter:
       // Multiple columns returned by a scalar subquery is not supported in
-      // DuckDB. A scalar subquery expression is a subquery that returns one
+      // Presto. A scalar subquery expression is a subquery that returns one
       // result row from exactly one column for every input row.
       if (joinNode->leftKeys().size() > 1) {
         return std::nullopt;
       }
-      sql << " FROM t WHERE " << joinKeysToSql(joinNode->leftKeys())
-          << " IN (SELECT " << joinKeysToSql(joinNode->rightKeys())
-          << " FROM u";
+      sql << " FROM " << probeTableName << " WHERE "
+          << joinKeysToSql(joinNode->leftKeys()) << " IN (SELECT "
+          << joinKeysToSql(joinNode->rightKeys()) << " FROM " << buildTableName;
       if (joinNode->filter()) {
         sql << " WHERE " << filterToSql(joinNode->filter());
       }
@@ -412,29 +465,31 @@ std::optional<std::string> DuckQueryRunner::toSql(
     case core::JoinType::kLeftSemiProject:
       if (joinNode->isNullAware()) {
         sql << ", " << joinKeysToSql(joinNode->leftKeys()) << " IN (SELECT "
-            << joinKeysToSql(joinNode->rightKeys()) << " FROM u";
+            << joinKeysToSql(joinNode->rightKeys()) << " FROM "
+            << buildTableName;
         if (joinNode->filter()) {
           sql << " WHERE " << filterToSql(joinNode->filter());
         }
-        sql << ") FROM t";
+        sql << ") FROM " << probeTableName;
       } else {
-        sql << ", EXISTS (SELECT * FROM u WHERE "
-            << joinConditionAsSql(joinNode);
-        sql << ") FROM t";
+        sql << ", EXISTS (SELECT * FROM " << buildTableName << " WHERE "
+            << joinConditionAsSql(*joinNode);
+        sql << ") FROM " << probeTableName;
       }
       break;
     case core::JoinType::kAnti:
       if (joinNode->isNullAware()) {
-        sql << " FROM t WHERE " << joinKeysToSql(joinNode->leftKeys())
-            << " NOT IN (SELECT " << joinKeysToSql(joinNode->rightKeys())
-            << " FROM u";
+        sql << " FROM " << probeTableName << " WHERE "
+            << joinKeysToSql(joinNode->leftKeys()) << " NOT IN (SELECT "
+            << joinKeysToSql(joinNode->rightKeys()) << " FROM "
+            << buildTableName;
         if (joinNode->filter()) {
           sql << " WHERE " << filterToSql(joinNode->filter());
         }
         sql << ")";
       } else {
-        sql << " FROM t WHERE NOT EXISTS (SELECT * FROM u WHERE "
-            << joinConditionAsSql(joinNode);
+        sql << " FROM " << probeTableName << " WHERE NOT EXISTS (SELECT * FROM "
+            << buildTableName << " WHERE " << joinConditionAsSql(*joinNode);
         sql << ")";
       }
       break;
@@ -448,6 +503,21 @@ std::optional<std::string> DuckQueryRunner::toSql(
 
 std::optional<std::string> DuckQueryRunner::toSql(
     const std::shared_ptr<const core::NestedLoopJoinNode>& joinNode) {
+  std::string probeTableName;
+  std::string buildTableName;
+  if (const auto valuesNode = std::dynamic_pointer_cast<const core::ValuesNode>(
+          joinNode->sources()[0])) {
+    probeTableName = getTableName(valuesNode);
+  } else if (const auto subQuery = toSql(joinNode->sources()[0])) {
+    probeTableName = fmt::format("({})", *subQuery);
+  }
+  if (const auto valuesNode = std::dynamic_pointer_cast<const core::ValuesNode>(
+          joinNode->sources()[1])) {
+    buildTableName = getTableName(valuesNode);
+  } else if (const auto subQuery = toSql(joinNode->sources()[1])) {
+    buildTableName = fmt::format("({})", *subQuery);
+  }
+
   std::stringstream sql;
   sql << "SELECT " << folly::join(", ", joinNode->outputType()->names());
 
@@ -458,13 +528,16 @@ std::optional<std::string> DuckQueryRunner::toSql(
   const std::string joinCondition{"(1 = 1)"};
   switch (joinNode->joinType()) {
     case core::JoinType::kInner:
-      sql << " FROM t INNER JOIN u ON " << joinCondition;
+      sql << " FROM " << probeTableName << " INNER JOIN " << buildTableName
+          << " ON " << joinCondition;
       break;
     case core::JoinType::kLeft:
-      sql << " FROM t LEFT JOIN u ON " << joinCondition;
+      sql << " FROM " << probeTableName << " LEFT JOIN " << buildTableName
+          << " ON " << joinCondition;
       break;
     case core::JoinType::kFull:
-      sql << " FROM t FULL OUTER JOIN u ON " << joinCondition;
+      sql << " FROM " << probeTableName << " FULL OUTER JOIN " << buildTableName
+          << " ON " << joinCondition;
       break;
     default:
       VELOX_UNREACHABLE(

velox/exec/fuzzer/DuckQueryRunner.h

@@ -46,6 +46,18 @@ class DuckQueryRunner : public ReferenceQueryRunner {
   /// Assumes that source of AggregationNode or Window Node is 'tmp' table.
   std::optional<std::string> toSql(const core::PlanNodePtr& plan) override;
 
+  /// Returns the name of the values node table in the form t_<id>.
+  std::string getTableName(const core::ValuesNodePtr& valuesNode);
+
+  // Traverses all nodes in the plan and returns all tables and their names.
+  std::unordered_map<std::string, std::vector<velox::RowVectorPtr>>
+  getAllTablesAndNames(const core::PlanNodePtr& plan);
+
+  /// Executes the query based on the plan. Returns std::nullopt if the plan is
+  /// not supported.
+  std::optional<std::multiset<std::vector<velox::variant>>> execute(
+      const core::PlanNodePtr& plan) override;
+
   /// Creates 'tmp' table with 'input' data and runs 'sql' query. Returns
   /// results according to 'resultType' schema.
   std::multiset<std::vector<velox::variant>> execute(