Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(fuzzer): Support multiple joins in the join node "toSql" methods…
Browse files Browse the repository at this point in the history 
… for reference query runners (facebookincubator#11801)

Summary:

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.

Differential Revision: D66977480
@facebook-github-bot
Daniel Hunte authored and facebook-github-bot committed Dec 16, 2024
1 parent 05c8452 commit 211a0b3
Showing 7 changed files with 397 additions and 101 deletions.
161 changes: 114 additions & 47 deletions velox/exec/fuzzer/DuckQueryRunner.cpp
View file
Original file line number Diff line number Diff line change
@@ -102,6 +102,17 @@ DuckQueryRunner::aggregationFunctionDataSpecs() const {
return kAggregationFunctionDataSpecs;
}

std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
const std::string& sql,
const std::unordered_map<std::string, std::vector<RowVectorPtr>>& inputMap,
const RowTypePtr& resultType) {
DuckDbQueryRunner queryRunner;
for (const auto& [tableName, input] : inputMap) {
queryRunner.createTable(tableName, input);
}
return queryRunner.execute(sql, resultType);
}

std::multiset<std::vector<velox::variant>> DuckQueryRunner::execute(
const std::string& sql,
const std::vector<RowVectorPtr>& input,
@@ -341,38 +352,62 @@ 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();
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 ";
}
if (joinNode->filter()) {
out << " AND " << filterToSql(joinNode->filter());
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 =
fmt::format("t_{}", joinNode->sources()[0]->id());
std::string buildTableName =
fmt::format("t_{}", joinNode->sources()[1]->id());

// If an input to this join is another join, change the table name to a
// sub-query.
for (auto i = 0; i < joinNode->sources().size(); ++i) {
if (joinNode->sources()[i]->name() == "HashJoin" ||
joinNode->sources()[i]->name() == "MergeJoin" ||
joinNode->sources()[i]->name() == "NestedLoopJoin") {
if (auto sql = toSql(joinNode->sources()[i])) {
if (i == 0) {
probeTableName = fmt::format("({})", *sql);
} else {
buildTableName = fmt::format("({})", *sql);
}
} else {
return std::nullopt;
}
}
return out.str();
};
}

const auto& outputNames = joinNode->outputType()->names();

@@ -386,24 +421,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 +450,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 +488,30 @@ std::optional<std::string> DuckQueryRunner::toSql(

std::optional<std::string> DuckQueryRunner::toSql(
const std::shared_ptr<const core::NestedLoopJoinNode>& joinNode) {
std::string probeTableName =
fmt::format("t_{}", joinNode->sources()[0]->id());
std::string buildTableName =
fmt::format("t_{}", joinNode->sources()[1]->id());

// If an input to this join is another join, change the table name to a
// sub-query.
for (auto i = 0; i < joinNode->sources().size(); ++i) {
if (joinNode->sources()[i]->name() == "HashJoin" ||
joinNode->sources()[i]->name() == "MergeJoin" ||
joinNode->sources()[i]->name() == "NestedLoopJoin") {
if (auto sql = toSql(joinNode->sources()[i])) {
if (i == 0) {
probeTableName = fmt::format("({})", *sql);
} else {
buildTableName = fmt::format("({})", *sql);
}
} else {
return std::nullopt;
}
}
}

const auto& outputNames = joinNode->outputType()->names();
std::stringstream sql;
sql << "SELECT " << folly::join(", ", joinNode->outputType()->names());

@@ -458,13 +522,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(
8 changes: 8 additions & 0 deletions velox/exec/fuzzer/DuckQueryRunner.h
View file
Original file line number Diff line number Diff line change
@@ -46,6 +46,14 @@ 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 tables for each entry in 'inputs' and runs 'sql' query. Returns
/// results according to 'resultType' schema.
std::multiset<std::vector<velox::variant>> execute(
const std::string& sql,
const std::unordered_map<std::string, std::vector<RowVectorPtr>>&
inputMap,
const RowTypePtr& resultType) override;

/// Creates 'tmp' table with 'input' data and runs 'sql' query. Returns
/// results according to 'resultType' schema.
std::multiset<std::vector<velox::variant>> execute(
6 changes: 4 additions & 2 deletions velox/exec/fuzzer/JoinFuzzer.cpp
View file
Original file line number Diff line number Diff line change
@@ -680,8 +680,10 @@ std::optional<MaterializedRowMultiset> JoinFuzzer::computeReferenceResults(
}

if (auto sql = referenceQueryRunner_->toSql(plan)) {
return referenceQueryRunner_->execute(
sql.value(), probeInput, buildInput, plan->outputType());
std::unordered_map<std::string, std::vector<RowVectorPtr>> inputs = {
{fmt::format("t_{}", plan->sources()[0]->id()), probeInput},
{fmt::format("t_{}", plan->sources()[1]->id()), buildInput}};
return referenceQueryRunner_->execute(*sql, inputs, plan->outputType());
}

LOG(INFO) << "Query not supported by the reference DB";
Loading

0 comments on commit 211a0b3

Please sign in to comment.