feat(fuzzer): Update makeAlternatePlans 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/CMakeLists.txt

@@ -12,8 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_library(velox_fuzzer_util DuckQueryRunner.cpp PrestoQueryRunner.cpp
-                              FuzzerUtil.cpp ToSQLUtil.cpp)
+add_library(velox_fuzzer_util ReferenceQueryRunner.cpp DuckQueryRunner.cpp
+                            PrestoQueryRunner.cpp FuzzerUtil.cpp ToSQLUtil.cpp)
 
 target_link_libraries(
   velox_fuzzer_util

velox/exec/fuzzer/DuckQueryRunner.cpp

@@ -13,6 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+#include <optional>
+#include <set>
+#include <unordered_map>
+
 #include "velox/exec/fuzzer/DuckQueryRunner.h"
 #include "velox/exec/fuzzer/ToSQLUtil.h"
 #include "velox/exec/tests/utils/QueryAssertions.h"
@@ -104,21 +109,22 @@ DuckQueryRunner::aggregationFunctionDataSpecs() const {
 
 std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
     const std::string& sql,
-    const std::vector<RowVectorPtr>& input,
-    const RowTypePtr& resultType) {
+    const core::PlanNodePtr& plan) {
   DuckDbQueryRunner queryRunner;
-  queryRunner.createTable("tmp", input);
-  return queryRunner.execute(sql, resultType);
+  std::unordered_map<std::string, std::vector<RowVectorPtr>> inputMap =
+      getAllTables(plan);
+  for (const auto& [tableName, input] : inputMap) {
+    queryRunner.createTable(tableName, input);
+  }
+  return queryRunner.execute(sql, plan->outputType());
 }
 
 std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
     const std::string& sql,
-    const std::vector<RowVectorPtr>& probeInput,
-    const std::vector<RowVectorPtr>& buildInput,
+    const std::vector<RowVectorPtr>& input,
     const RowTypePtr& resultType) {
   DuckDbQueryRunner queryRunner;
-  queryRunner.createTable("t", probeInput);
-  queryRunner.createTable("u", buildInput);
+  queryRunner.createTable("tmp", input);
   return queryRunner.execute(sql, resultType);
 }
 
@@ -164,6 +170,11 @@ std::optional<std::string> DuckQueryRunner::toSql(
     return toSql(joinNode);
   }
 
+  if (const auto valuesNode =
+          std::dynamic_pointer_cast<const core::ValuesNode>(plan)) {
+    return toSql(valuesNode);
+  }
+
   VELOX_NYI();
 }
 
@@ -340,137 +351,4 @@ 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();
-    }
-    return out.str();
-  };
-
-  const auto filterToSql = [](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());
-    }
-    return out.str();
-  };
-
-  const auto& outputNames = joinNode->outputType()->names();
-
-  std::stringstream sql;
-  if (joinNode->isLeftSemiProjectJoin()) {
-    sql << "SELECT "
-        << folly::join(", ", outputNames.begin(), --outputNames.end());
-  } else {
-    sql << "SELECT " << folly::join(", ", outputNames);
-  }
-
-  switch (joinNode->joinType()) {
-    case core::JoinType::kInner:
-      sql << " FROM t INNER JOIN u ON " << joinConditionAsSql(joinNode);
-      break;
-    case core::JoinType::kLeft:
-      sql << " FROM t LEFT JOIN u ON " << joinConditionAsSql(joinNode);
-      break;
-    case core::JoinType::kFull:
-      sql << " FROM t FULL OUTER JOIN u 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
-      // 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";
-      if (joinNode->filter()) {
-        sql << " WHERE " << filterToSql(joinNode->filter());
-      }
-      sql << ")";
-      break;
-    case core::JoinType::kLeftSemiProject:
-      if (joinNode->isNullAware()) {
-        sql << ", " << joinKeysToSql(joinNode->leftKeys()) << " IN (SELECT "
-            << joinKeysToSql(joinNode->rightKeys()) << " FROM u";
-        if (joinNode->filter()) {
-          sql << " WHERE " << filterToSql(joinNode->filter());
-        }
-        sql << ") FROM t";
-      } else {
-        sql << ", EXISTS (SELECT * FROM u WHERE "
-            << joinConditionAsSql(joinNode);
-        sql << ") FROM t";
-      }
-      break;
-    case core::JoinType::kAnti:
-      if (joinNode->isNullAware()) {
-        sql << " FROM t WHERE " << joinKeysToSql(joinNode->leftKeys())
-            << " NOT IN (SELECT " << joinKeysToSql(joinNode->rightKeys())
-            << " FROM u";
-        if (joinNode->filter()) {
-          sql << " WHERE " << filterToSql(joinNode->filter());
-        }
-        sql << ")";
-      } else {
-        sql << " FROM t WHERE NOT EXISTS (SELECT * FROM u WHERE "
-            << joinConditionAsSql(joinNode);
-        sql << ")";
-      }
-      break;
-    default:
-      VELOX_UNREACHABLE(
-          "Unknown join type: {}", static_cast<int>(joinNode->joinType()));
-  }
-
-  return sql.str();
-}
-
-std::optional<std::string> DuckQueryRunner::toSql(
-    const std::shared_ptr<const core::NestedLoopJoinNode>& joinNode) {
-  std::stringstream sql;
-  sql << "SELECT " << folly::join(", ", joinNode->outputType()->names());
-
-  // Nested loop join without filter.
-  VELOX_CHECK(
-      joinNode->joinCondition() == nullptr,
-      "This code path should be called only for nested loop join without filter");
-  const std::string joinCondition{"(1 = 1)"};
-  switch (joinNode->joinType()) {
-    case core::JoinType::kInner:
-      sql << " FROM t INNER JOIN u ON " << joinCondition;
-      break;
-    case core::JoinType::kLeft:
-      sql << " FROM t LEFT JOIN u ON " << joinCondition;
-      break;
-    case core::JoinType::kFull:
-      sql << " FROM t FULL OUTER JOIN u ON " << joinCondition;
-      break;
-    default:
-      VELOX_UNREACHABLE(
-          "Unknown join type: {}", static_cast<int>(joinNode->joinType()));
-  }
-
-  return sql.str();
-}
 } // namespace facebook::velox::exec::test

velox/exec/fuzzer/DuckQueryRunner.h

@@ -15,6 +15,10 @@
  */
 #pragma once
 
+#include <optional>
+#include <set>
+#include <unordered_map>
+
 #include "velox/exec/fuzzer/ReferenceQueryRunner.h"
 
 namespace facebook::velox::exec::test {
@@ -46,20 +50,21 @@ 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;
 
-  /// Creates 'tmp' table with 'input' data and runs 'sql' query. Returns
-  /// results according to 'resultType' schema.
+  /// Executes SQL query returned by the 'toSql' method based on the plan.
   std::multiset<std::vector<velox::variant>> execute(
       const std::string& sql,
-      const std::vector<RowVectorPtr>& input,
-      const RowTypePtr& resultType) override;
+      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(
       const std::string& sql,
-      const std::vector<RowVectorPtr>& probeInput,
-      const std::vector<RowVectorPtr>& buildInput,
+      const std::vector<RowVectorPtr>& input,
       const RowTypePtr& resultType) override;
 
  private:
+  using ReferenceQueryRunner::toSql;
+
   std::optional<std::string> toSql(
       const std::shared_ptr<const core::AggregationNode>& aggregationNode);
 
@@ -72,12 +77,6 @@ class DuckQueryRunner : public ReferenceQueryRunner {
   std::optional<std::string> toSql(
       const std::shared_ptr<const core::RowNumberNode>& rowNumberNode);
 
-  std::optional<std::string> toSql(
-      const std::shared_ptr<const core::HashJoinNode>& joinNode);
-
-  std::optional<std::string> toSql(
-      const std::shared_ptr<const core::NestedLoopJoinNode>& joinNode);
-
   std::unordered_set<std::string> aggregateFunctionNames_;
 };