Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(fuzzer): Update table scan plan functions to generate multi-join plans #11940

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat(fuzzer): Update table scan plan functions to generate multi-join plans #11940

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
571058e
feat(fuzzer): Support multiple joins in the join node "toSql" methods…
Dec 23, 2024
559bb4d
feat(fuzzer): Update "tryFlipJoinSides" functions to handle multi-joi…
Dec 23, 2024
32dabe1
feat(fuzzer): Update "makeDefaultPlan" function to generate multi-joi…
Dec 23, 2024
2274011
feat(fuzzer): Update table scan plan functions to generate multi-join…
Dec 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions velox/core/PlanNode.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
169 changes: 121 additions & 48 deletions velox/exec/fuzzer/DuckQueryRunner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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();

Expand All @@ -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());
}
Expand All @@ -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;
Expand All @@ -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());

Expand All @@ -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(
Expand Down
12 changes: 12 additions & 0 deletions velox/exec/fuzzer/DuckQueryRunner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading
Loading