feat(fuzzer): Support multiple joins in the join node "toSql" methods for reference query runners (facebookincubator#11801)

Summary:
Pull Request resolved: facebookincubator#11801

Currently, the hash join and nested loop join "toSql" methods for all reference query runners only support a single join. This change extends it to support multiple joins, only needing the join node of the last join in the tree. It traverses up the tree and recursively builds the sql query.

Reviewed By: kevinwilfong

Differential Revision: D66977480

fbshipit-source-id: 24355736dfd1a267435d6c36dcb06580c1e62537

Author: Daniel Hunte

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,13 @@
 # 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"
@@ -102,23 +107,39 @@ DuckQueryRunner::aggregationFunctionDataSpecs() const {
   return kAggregationFunctionDataSpecs;
 }
 
-std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
-    const std::string& sql,
-    const std::vector<RowVectorPtr>& input,
-    const RowTypePtr& resultType) {
-  DuckDbQueryRunner queryRunner;
-  queryRunner.createTable("tmp", input);
-  return queryRunner.execute(sql, resultType);
+std::pair<
+    std::optional<std::multiset<std::vector<velox::variant>>>,
+    ReferenceQueryErrorCode>
+DuckQueryRunner::execute(const core::PlanNodePtr& plan) {
+  if (std::optional<std::string> sql = toSql(plan)) {
+    try {
+      DuckDbQueryRunner queryRunner;
+      std::unordered_map<std::string, std::vector<RowVectorPtr>> inputMap =
+          getAllTables(plan);
+      for (const auto& [tableName, input] : inputMap) {
+        queryRunner.createTable(tableName, input);
+      }
+      return std::make_pair(
+          queryRunner.execute(*sql, plan->outputType()),
+          ReferenceQueryErrorCode::kSuccess);
+    } catch (...) {
+      LOG(WARNING) << "Query failed in DuckDB";
+      return std::make_pair(
+          std::nullopt, ReferenceQueryErrorCode::kReferenceQueryFail);
+    }
+  }
+
+  LOG(INFO) << "Query not supported in DuckDB";
+  return std::make_pair(
+      std::nullopt, ReferenceQueryErrorCode::kReferenceQueryUnsupported);
 }
 
 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 +185,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 +366,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,23 @@ 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;
 
+  /// Executes the plan and returns the result along with success or fail error
+  /// code.
+  std::pair<
+      std::optional<std::multiset<std::vector<velox::variant>>>,
+      ReferenceQueryErrorCode>
+  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(
       const std::string& sql,
       const std::vector<RowVectorPtr>& input,
       const RowTypePtr& resultType) override;
 
-  std::multiset<std::vector<velox::variant>> execute(
-      const std::string& sql,
-      const std::vector<RowVectorPtr>& probeInput,
-      const std::vector<RowVectorPtr>& buildInput,
-      const RowTypePtr& resultType) override;
-
  private:
+  using ReferenceQueryRunner::toSql;
+
   std::optional<std::string> toSql(
       const std::shared_ptr<const core::AggregationNode>& aggregationNode);
 
@@ -72,12 +79,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_;
 };
 

velox/exec/fuzzer/FuzzerUtil.h

@@ -131,12 +131,6 @@ void setupMemory(
 void registerHiveConnector(
     const std::unordered_map<std::string, std::string>& hiveConfigs);
 
-enum ReferenceQueryErrorCode {
-  kSuccess,
-  kReferenceQueryFail,
-  kReferenceQueryUnsupported
-};
-
 // Converts 'plan' into an SQL query and runs it on 'input' in the reference DB.
 // Result is returned as a MaterializedRowMultiset with the
 // ReferenceQueryErrorCode::kSuccess if successful, or an std::nullopt with a

velox/exec/fuzzer/JoinFuzzer.cpp

@@ -679,12 +679,12 @@ std::optional<MaterializedRowMultiset> JoinFuzzer::computeReferenceResults(
     VELOX_CHECK(!containsUnsupportedTypes(buildInput[0]->type()));
   }
 
-  if (auto sql = referenceQueryRunner_->toSql(plan)) {
-    return referenceQueryRunner_->execute(
-        sql.value(), probeInput, buildInput, plan->outputType());
+  auto result = referenceQueryRunner_->execute(plan);
+  if (result.first) {
+    return result.first;
   }
 
-  LOG(INFO) << "Query not supported by the reference DB";
+  LOG(INFO) << "Query not supported by or failed in the reference DB";
   return std::nullopt;
 }