This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 0ee63b2e54 [GLUTEN-12597][CORE] Migrate nested loop joins from
CrossRel to NestedLoopJoinRel (Substrait 0.98) (#12700)
0ee63b2e54 is described below
commit 0ee63b2e5496af2e702a296e9265079180ed4593
Author: Niels Pardon <[email protected]>
AuthorDate: Wed Aug 19 15:26:06 2026 +0200
[GLUTEN-12597][CORE] Migrate nested loop joins from CrossRel to
NestedLoopJoinRel (Substrait 0.98) (#12700)
---
.../gluten/vectorized/StorageJoinBuilder.java | 2 +-
.../org/apache/gluten/metrics/MetricsUtil.scala | 3 +-
cpp-ch/local-engine/Common/CHUtil.cpp | 8 +--
cpp-ch/local-engine/Common/CHUtil.h | 2 +-
cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp | 2 +-
.../Parser/RelParsers/CrossRelParser.cpp | 12 ++--
.../Parser/RelParsers/CrossRelParser.h | 6 +-
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 40 +++++------
cpp/velox/substrait/SubstraitToVeloxPlan.h | 4 +-
.../substrait/SubstraitToVeloxPlanValidator.cc | 38 +++++------
.../substrait/SubstraitToVeloxPlanValidator.h | 4 +-
...rossRelNode.java => NestedLoopJoinRelNode.java} | 26 ++++----
.../apache/gluten/substrait/rel/RelBuilder.java | 6 +-
.../substrait/proto/substrait/algebra.proto | 27 ++++++--
.../BroadcastNestedLoopJoinExecTransformer.scala | 16 ++---
.../CartesianProductExecTransformer.scala | 10 +--
.../org/apache/gluten/execution/JoinUtils.scala | 8 +--
.../org/apache/gluten/utils/SubstraitUtil.scala | 14 ++--
.../gluten/utils/SubstraitUtilJoinTypeSuite.scala | 77 ++++++++++++++++++++++
19 files changed, 198 insertions(+), 107 deletions(-)
diff --git
a/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
b/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
index f3b39865c6..60cbee3997 100644
---
a/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
+++
b/backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java
@@ -83,7 +83,7 @@ public class StorageJoinBuilder {
joinType =
JoinTypeTransform.toSubstraitJoinType(broadcastContext.joinType(),
buildRight).ordinal();
} else {
- joinType =
SubstraitUtil.toCrossRelSubstrait(broadcastContext.joinType()).ordinal();
+ joinType =
SubstraitUtil.toNestedLoopJoinSubstrait(broadcastContext.joinType()).ordinal();
}
return nativeBuild(
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala
b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala
index 779c373543..8bb3386512 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala
@@ -395,7 +395,8 @@ object MetricsUtil extends Logging {
}
smj.updateJoinMetrics(operatorMetrics, singleMetrics, joinParams)
case ju: JoinMetricsUpdaterBase =>
- // JoinRel and CrossRel output two suites of metrics respectively for
build and probe.
+ // JoinRel and NestedLoopJoinRel output two suites of metrics
respectively for build and
+ // probe.
// Therefore, fetch one more suite of metrics here.
operatorMetrics.add(nativeMetrics.get(curMetricsIdx))
curMetricsIdx -= 1
diff --git a/cpp-ch/local-engine/Common/CHUtil.cpp
b/cpp-ch/local-engine/Common/CHUtil.cpp
index 9487b87ef1..f2cde98528 100644
--- a/cpp-ch/local-engine/Common/CHUtil.cpp
+++ b/cpp-ch/local-engine/Common/CHUtil.cpp
@@ -1118,15 +1118,15 @@
JoinUtil::getJoinKindAndStrictness(substrait::JoinRel_JoinType join_type, bool i
}
}
-std::pair<DB::JoinKind, DB::JoinStrictness>
JoinUtil::getCrossJoinKindAndStrictness(substrait::CrossRel_JoinType join_type)
+std::pair<DB::JoinKind, DB::JoinStrictness>
JoinUtil::getCrossJoinKindAndStrictness(substrait::NestedLoopJoinRel_JoinType
join_type)
{
switch (join_type)
{
- case substrait::CrossRel_JoinType_JOIN_TYPE_INNER:
+ case substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_INNER:
return {DB::JoinKind::Cross, DB::JoinStrictness::All};
- case substrait::CrossRel_JoinType_JOIN_TYPE_LEFT:
+ case substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_LEFT:
return {DB::JoinKind::Left, DB::JoinStrictness::All};
- case substrait::CrossRel_JoinType_JOIN_TYPE_OUTER:
+ case substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_OUTER:
return {DB::JoinKind::Full, DB::JoinStrictness::All};
default:
throw Exception(ErrorCodes::UNKNOWN_TYPE, "unsupported join type
{}.", magic_enum::enum_name(join_type));
diff --git a/cpp-ch/local-engine/Common/CHUtil.h
b/cpp-ch/local-engine/Common/CHUtil.h
index 6147790814..7ef23ea524 100644
--- a/cpp-ch/local-engine/Common/CHUtil.h
+++ b/cpp-ch/local-engine/Common/CHUtil.h
@@ -258,7 +258,7 @@ public:
static void adjustJoinOutput(DB::QueryPlan & plan, DB::Names cols);
static std::pair<DB::JoinKind, DB::JoinStrictness>
getJoinKindAndStrictness(substrait::JoinRel_JoinType join_type, bool
is_existence_join);
- static std::pair<DB::JoinKind, DB::JoinStrictness>
getCrossJoinKindAndStrictness(substrait::CrossRel_JoinType join_type);
+ static std::pair<DB::JoinKind, DB::JoinStrictness>
getCrossJoinKindAndStrictness(substrait::NestedLoopJoinRel_JoinType join_type);
};
}
diff --git a/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
index ca9f729184..c17e1200d9 100644
--- a/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
+++ b/cpp-ch/local-engine/Join/BroadcastJoinBuilder.cpp
@@ -173,7 +173,7 @@ std::shared_ptr<StorageJoinFromReadBuffer> buildJoin(
if (is_bhj)
std::tie(kind, strictness) =
JoinUtil::getJoinKindAndStrictness(static_cast<substrait::JoinRel_JoinType>(join_type),
is_existence_join);
else
- std::tie(kind, strictness) =
JoinUtil::getCrossJoinKindAndStrictness(static_cast<substrait::CrossRel_JoinType>(join_type));
+ std::tie(kind, strictness) =
JoinUtil::getCrossJoinKindAndStrictness(static_cast<substrait::NestedLoopJoinRel_JoinType>(join_type));
substrait::NamedStruct substrait_struct;
substrait_struct.ParseFromString(named_struct);
diff --git a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
index 854752c555..2e5af497c9 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
@@ -58,7 +58,7 @@ namespace local_engine
{
using namespace DB;
-std::shared_ptr<DB::TableJoin>
createCrossTableJoin(substrait::CrossRel_JoinType join_type)
+std::shared_ptr<DB::TableJoin>
createCrossTableJoin(substrait::NestedLoopJoinRel_JoinType join_type)
{
auto global_context = QueryContext::globalContext();
auto table_join = std::make_shared<TableJoin>(
@@ -82,7 +82,7 @@ CrossRelParser::parse(DB::QueryPlanPtr /*query_plan*/, const
substrait::Rel & /*
std::vector<const substrait::Rel *> CrossRelParser::getInputs(const
substrait::Rel & rel)
{
- const auto & join = rel.cross();
+ const auto & join = rel.nested_loop_join();
if (!join.has_left() || !join.has_right())
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "left table or right table
is missing.");
@@ -124,7 +124,7 @@ DB::QueryPlanPtr
CrossRelParser::parse(std::vector<DB::QueryPlanPtr> & input_plans_, const
substrait::Rel & rel, std::list<const substrait::Rel *> &)
{
assert(input_plans_.size() == 2);
- const auto & join = rel.cross();
+ const auto & join = rel.nested_loop_join();
std::pair<DB::JoinKind, DB::JoinStrictness> kind_and_strictness =
JoinUtil::getCrossJoinKindAndStrictness(join.type());
if (kind_and_strictness.first != JoinKind::Cross)
addConstJoinKeys(*input_plans_[0], *input_plans_[1]);
@@ -166,7 +166,7 @@ void CrossRelParser::renamePlanColumns(DB::QueryPlan &
left, DB::QueryPlan & rig
left.addStep(std::move(project_step));
}
-DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join,
DB::QueryPlanPtr left, DB::QueryPlanPtr right)
+DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::NestedLoopJoinRel
& join, DB::QueryPlanPtr left, DB::QueryPlanPtr right)
{
google::protobuf::StringValue optimization_info;
optimization_info.ParseFromString(firstOptimizationOrDefault(join.advanced_extension()).value());
@@ -247,7 +247,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const
substrait::CrossRel & join, DB:
}
-void CrossRelParser::addPostFilter(DB::QueryPlan & query_plan, const
substrait::CrossRel & join_rel)
+void CrossRelParser::addPostFilter(DB::QueryPlan & query_plan, const
substrait::NestedLoopJoinRel & join_rel)
{
if (!join_rel.has_expression())
return;
@@ -366,7 +366,7 @@ DB::Names CrossRelParser::collectOutputColumnsName(const
DB::QueryPlan & left, c
void registerCrossRelParser(RelParserFactory & factory)
{
auto builder = [](ParserContextPtr parser_context) { return
std::make_shared<CrossRelParser>(parser_context); };
- factory.registerBuilder(substrait::Rel::RelTypeCase::kCross, builder);
+ factory.registerBuilder(substrait::Rel::RelTypeCase::kNestedLoopJoin,
builder);
}
}
diff --git a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h
b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h
index 0d59905699..8d4f69e829 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h
+++ b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h
@@ -55,13 +55,13 @@ private:
std::vector<DB::QueryPlanPtr> extra_plan_holder;
- DB::QueryPlanPtr parseJoin(const substrait::CrossRel & join,
DB::QueryPlanPtr left, DB::QueryPlanPtr right);
+ DB::QueryPlanPtr parseJoin(const substrait::NestedLoopJoinRel & join,
DB::QueryPlanPtr left, DB::QueryPlanPtr right);
void renamePlanColumns(DB::QueryPlan & left, DB::QueryPlan & right, const
StorageJoinFromReadBuffer & storage_join);
void addConvertStep(DB::TableJoin & table_join, DB::QueryPlan & left,
DB::QueryPlan & right);
- void addPostFilter(DB::QueryPlan & query_plan, const substrait::CrossRel &
join);
+ void addPostFilter(DB::QueryPlan & query_plan, const
substrait::NestedLoopJoinRel & join);
bool applyJoinFilter(
DB::TableJoin & table_join,
- const substrait::CrossRel & join_rel,
+ const substrait::NestedLoopJoinRel & join_rel,
DB::QueryPlan & left,
DB::QueryPlan & right,
bool allow_mixed_condition);
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index 97324de4e1..9195861714 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -516,43 +516,43 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
}
}
-core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const
::substrait::CrossRel& crossRel) {
- // Support basic cross join without any filters
- if (!crossRel.has_left()) {
- VELOX_FAIL("Left Rel is expected in CrossRel.");
+core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const
::substrait::NestedLoopJoinRel& nestedLoopJoinRel) {
+ // Support basic nested loop join without any filters
+ if (!nestedLoopJoinRel.has_left()) {
+ VELOX_FAIL("Left Rel is expected in NestedLoopJoinRel.");
}
- if (!crossRel.has_right()) {
- VELOX_FAIL("Right Rel is expected in CrossRel.");
+ if (!nestedLoopJoinRel.has_right()) {
+ VELOX_FAIL("Right Rel is expected in NestedLoopJoinRel.");
}
- auto leftNode = toVeloxPlan(crossRel.left());
- auto rightNode = toVeloxPlan(crossRel.right());
+ auto leftNode = toVeloxPlan(nestedLoopJoinRel.left());
+ auto rightNode = toVeloxPlan(nestedLoopJoinRel.right());
// Map join type.
core::JoinType joinType;
- switch (crossRel.type()) {
- case ::substrait::CrossRel_JoinType::CrossRel_JoinType_JOIN_TYPE_INNER:
+ switch (nestedLoopJoinRel.type()) {
+ case
::substrait::NestedLoopJoinRel_JoinType::NestedLoopJoinRel_JoinType_JOIN_TYPE_INNER:
joinType = core::JoinType::kInner;
break;
- case ::substrait::CrossRel_JoinType::CrossRel_JoinType_JOIN_TYPE_LEFT:
+ case
::substrait::NestedLoopJoinRel_JoinType::NestedLoopJoinRel_JoinType_JOIN_TYPE_LEFT:
joinType = core::JoinType::kLeft;
break;
- case ::substrait::CrossRel_JoinType::CrossRel_JoinType_JOIN_TYPE_LEFT_SEMI:
- if (crossRel.has_advanced_extension() &&
-
SubstraitParser::configSetInOptimization(crossRel.advanced_extension(),
"isExistenceJoin=")) {
+ case
::substrait::NestedLoopJoinRel_JoinType::NestedLoopJoinRel_JoinType_JOIN_TYPE_LEFT_SEMI:
+ if (nestedLoopJoinRel.has_advanced_extension() &&
+
SubstraitParser::configSetInOptimization(nestedLoopJoinRel.advanced_extension(),
"isExistenceJoin=")) {
joinType = core::JoinType::kLeftSemiProject;
} else {
- VELOX_NYI("Unsupported Join type: {}",
std::to_string(crossRel.type()));
+ VELOX_NYI("Unsupported Join type: {}",
std::to_string(nestedLoopJoinRel.type()));
}
break;
default:
- VELOX_NYI("Unsupported Join type: {}", std::to_string(crossRel.type()));
+ VELOX_NYI("Unsupported Join type: {}",
std::to_string(nestedLoopJoinRel.type()));
}
auto inputRowType = getJoinInputType(leftNode, rightNode);
core::TypedExprPtr joinConditions;
- if (crossRel.has_expression()) {
- joinConditions = exprConverter_->toVeloxExpr(crossRel.expression(),
inputRowType);
+ if (nestedLoopJoinRel.has_expression()) {
+ joinConditions =
exprConverter_->toVeloxExpr(nestedLoopJoinRel.expression(), inputRowType);
}
return std::make_shared<core::NestedLoopJoinNode>(
@@ -1722,8 +1722,8 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
return toVeloxPlan(rel.filter());
} else if (rel.has_join()) {
return toVeloxPlan(rel.join());
- } else if (rel.has_cross()) {
- return toVeloxPlan(rel.cross());
+ } else if (rel.has_nested_loop_join()) {
+ return toVeloxPlan(rel.nested_loop_join());
} else if (rel.has_read()) {
return toVeloxPlan(rel.read());
} else if (rel.has_sort()) {
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.h
b/cpp/velox/substrait/SubstraitToVeloxPlan.h
index 373601916d..5ced0b01d2 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.h
@@ -116,8 +116,8 @@ class SubstraitToVeloxPlanConverter {
/// Used to convert Substrait JoinRel into Velox PlanNode.
core::PlanNodePtr toVeloxPlan(const ::substrait::JoinRel& joinRel);
- /// Used to convert Substrait CrossRel into Velox PlanNode.
- core::PlanNodePtr toVeloxPlan(const ::substrait::CrossRel& crossRel);
+ /// Used to convert Substrait NestedLoopJoinRel into Velox PlanNode.
+ core::PlanNodePtr toVeloxPlan(const ::substrait::NestedLoopJoinRel&
nestedLoopJoinRel);
/// Used to convert Substrait AggregateRel into Velox PlanNode.
core::PlanNodePtr toVeloxPlan(const ::substrait::AggregateRel& aggRel);
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
index f8a7623ecf..0146cae8eb 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
@@ -1097,38 +1097,38 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::JoinRel& joinRel
return true;
}
-bool SubstraitToVeloxPlanValidator::validate(const ::substrait::CrossRel&
crossRel) {
- if (crossRel.has_left() && !validate(crossRel.left())) {
- logValidateMsg("Native validation failed due to: validation fails for
cross join left input. ");
+bool SubstraitToVeloxPlanValidator::validate(const
::substrait::NestedLoopJoinRel& nestedLoopJoinRel) {
+ if (nestedLoopJoinRel.has_left() && !validate(nestedLoopJoinRel.left())) {
+ logValidateMsg("Native validation failed due to: validation fails for
nested loop join left input. ");
return false;
}
- if (crossRel.has_right() && !validate(crossRel.right())) {
- logValidateMsg("Native validation failed due to: validation fails for
cross join right input. ");
+ if (nestedLoopJoinRel.has_right() && !validate(nestedLoopJoinRel.right())) {
+ logValidateMsg("Native validation failed due to: validation fails for
nested loop join right input. ");
return false;
}
// Validate input types.
- if (!crossRel.has_advanced_extension()) {
- logValidateMsg("Native validation failed due to: Input types are expected
in CrossRel.");
+ if (!nestedLoopJoinRel.has_advanced_extension()) {
+ logValidateMsg("Native validation failed due to: Input types are expected
in NestedLoopJoinRel.");
return false;
}
- switch (crossRel.type()) {
- case ::substrait::CrossRel_JoinType_JOIN_TYPE_INNER:
- case ::substrait::CrossRel_JoinType_JOIN_TYPE_LEFT:
- case ::substrait::CrossRel_JoinType_JOIN_TYPE_LEFT_SEMI:
+ switch (nestedLoopJoinRel.type()) {
+ case ::substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_INNER:
+ case ::substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_LEFT:
+ case ::substrait::NestedLoopJoinRel_JoinType_JOIN_TYPE_LEFT_SEMI:
break;
default:
- LOG_VALIDATION_MSG("Unsupported Join type in CrossRel");
+ LOG_VALIDATION_MSG("Unsupported Join type in NestedLoopJoinRel");
return false;
}
- const auto& extension = crossRel.advanced_extension();
+ const auto& extension = nestedLoopJoinRel.advanced_extension();
TypePtr inputRowType;
std::vector<TypePtr> types;
if (!parseVeloxType(extension, inputRowType) ||
!flattenSingleLevel(inputRowType, types)) {
- logValidateMsg("Native validation failed due to: Validation failed for
input types in CrossRel");
+ logValidateMsg("Native validation failed due to: Validation failed for
input types in NestedLoopJoinRel");
return false;
}
@@ -1140,11 +1140,11 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::CrossRel& crossR
}
auto rowType = std::make_shared<RowType>(std::move(names), std::move(types));
- if (crossRel.has_expression()) {
- if (!validateExpression(crossRel.expression(), rowType)) {
+ if (nestedLoopJoinRel.has_expression()) {
+ if (!validateExpression(nestedLoopJoinRel.expression(), rowType)) {
return false;
}
- auto expression = exprConverter_->toVeloxExpr(crossRel.expression(),
rowType);
+ auto expression =
exprConverter_->toVeloxExpr(nestedLoopJoinRel.expression(), rowType);
exec::ExprSet exprSet({std::move(expression)}, execCtx_.get());
}
@@ -1424,8 +1424,8 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::Rel& rel) {
if (rel.has_join()) {
return validate(rel.join());
}
- if (rel.has_cross()) {
- return validate(rel.cross());
+ if (rel.has_nested_loop_join()) {
+ return validate(rel.nested_loop_join());
}
if (rel.has_read()) {
return validate(rel.read());
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
index e9c4f9879e..51c58b6afb 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
@@ -107,8 +107,8 @@ class SubstraitToVeloxPlanValidator {
/// Used to validate Join.
bool validate(const ::substrait::JoinRel& joinRel);
- /// Used to validate Cartesian product.
- bool validate(const ::substrait::CrossRel& crossRel);
+ /// Used to validate nested loop join.
+ bool validate(const ::substrait::NestedLoopJoinRel& nestedLoopJoinRel);
/// Used to validate whether the computing of this Read is supported.
bool validate(const ::substrait::ReadRel& readRel);
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/CrossRelNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/NestedLoopJoinRelNode.java
similarity index 71%
rename from
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/CrossRelNode.java
rename to
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/NestedLoopJoinRelNode.java
index 070e95e37e..2f3ad4db54 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/CrossRelNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/NestedLoopJoinRelNode.java
@@ -19,7 +19,7 @@ package org.apache.gluten.substrait.rel;
import org.apache.gluten.substrait.expression.ExpressionNode;
import org.apache.gluten.substrait.extensions.AdvancedExtensionNode;
-import io.substrait.proto.CrossRel;
+import io.substrait.proto.NestedLoopJoinRel;
import io.substrait.proto.Rel;
import io.substrait.proto.RelCommon;
@@ -27,17 +27,17 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
-public class CrossRelNode implements RelNode, Serializable {
+public class NestedLoopJoinRelNode implements RelNode, Serializable {
private final RelNode left;
private final RelNode right;
- private final CrossRel.JoinType joinType;
+ private final NestedLoopJoinRel.JoinType joinType;
private final ExpressionNode expression;
private final AdvancedExtensionNode extensionNode;
- CrossRelNode(
+ NestedLoopJoinRelNode(
RelNode left,
RelNode right,
- CrossRel.JoinType joinType,
+ NestedLoopJoinRel.JoinType joinType,
ExpressionNode expression,
AdvancedExtensionNode extensionNode) {
this.left = left;
@@ -52,24 +52,24 @@ public class CrossRelNode implements RelNode, Serializable {
RelCommon.Builder relCommonBuilder = RelCommon.newBuilder();
relCommonBuilder.setDirect(RelCommon.Direct.newBuilder());
- CrossRel.Builder crossRelBuilder = CrossRel.newBuilder();
- crossRelBuilder.setCommon(relCommonBuilder.build());
+ NestedLoopJoinRel.Builder nestedLoopJoinRelBuilder =
NestedLoopJoinRel.newBuilder();
+ nestedLoopJoinRelBuilder.setCommon(relCommonBuilder.build());
- crossRelBuilder.setType(joinType);
+ nestedLoopJoinRelBuilder.setType(joinType);
if (left != null) {
- crossRelBuilder.setLeft(left.toProtobuf());
+ nestedLoopJoinRelBuilder.setLeft(left.toProtobuf());
}
if (right != null) {
- crossRelBuilder.setRight(right.toProtobuf());
+ nestedLoopJoinRelBuilder.setRight(right.toProtobuf());
}
if (expression != null) {
- crossRelBuilder.setExpression(expression.toProtobuf());
+ nestedLoopJoinRelBuilder.setExpression(expression.toProtobuf());
}
if (extensionNode != null) {
- crossRelBuilder.setAdvancedExtension(extensionNode.toProtobuf());
+
nestedLoopJoinRelBuilder.setAdvancedExtension(extensionNode.toProtobuf());
}
- return Rel.newBuilder().setCross(crossRelBuilder.build()).build();
+ return
Rel.newBuilder().setNestedLoopJoin(nestedLoopJoinRelBuilder.build()).build();
}
@Override
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
index 4072394624..e7d0640b7d 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
@@ -207,16 +207,16 @@ public class RelBuilder {
left, right, joinType, expression, postJoinFilter, hashTableId,
extensionNode);
}
- public static RelNode makeCrossRel(
+ public static RelNode makeNestedLoopJoinRel(
RelNode left,
RelNode right,
- CrossRel.JoinType joinType,
+ NestedLoopJoinRel.JoinType joinType,
ExpressionNode expression,
AdvancedExtensionNode extensionNode,
SubstraitContext context,
Long operatorId) {
context.registerRelToOperator(operatorId);
- return new CrossRelNode(left, right, joinType, expression, extensionNode);
+ return new NestedLoopJoinRelNode(left, right, joinType, expression,
extensionNode);
}
public static RelNode makeExpandRel(
diff --git
a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
index 6619a0395f..0e1eefdab0 100644
---
a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
+++
b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
@@ -295,11 +295,21 @@ message CrossRel {
RelCommon common = 1;
Rel left = 2;
Rel right = 3;
+
+ substrait.extensions.AdvancedExtension advanced_extension = 10;
+}
+
+// The nested loop join (NLJ) operator will hold the entire right input and
iterate over it using the
+// left input, evaluating the join expression on the Cartesian product of all
rows.
+message NestedLoopJoinRel {
+ RelCommon common = 1;
+ Rel left = 2;
+ Rel right = 3;
+ // optional, defaults to true (a cartesian join)
Expression expression = 4;
JoinType type = 5;
- // TODO -- Remove this unnecessary type.
enum JoinType {
JOIN_TYPE_UNSPECIFIED = 0;
JOIN_TYPE_INNER = 1;
@@ -307,11 +317,13 @@ message CrossRel {
JOIN_TYPE_LEFT = 3;
JOIN_TYPE_RIGHT = 4;
JOIN_TYPE_LEFT_SEMI = 5;
- JOIN_TYPE_LEFT_ANTI = 6;
- JOIN_TYPE_LEFT_SINGLE = 7;
- JOIN_TYPE_RIGHT_SEMI = 8;
- JOIN_TYPE_RIGHT_ANTI = 9;
+ JOIN_TYPE_RIGHT_SEMI = 6;
+ JOIN_TYPE_LEFT_ANTI = 7;
+ JOIN_TYPE_RIGHT_ANTI = 8;
+ JOIN_TYPE_LEFT_SINGLE = 9;
JOIN_TYPE_RIGHT_SINGLE = 10;
+ JOIN_TYPE_LEFT_MARK = 11;
+ JOIN_TYPE_RIGHT_MARK = 12;
}
substrait.extensions.AdvancedExtension advanced_extension = 10;
@@ -562,9 +574,10 @@ message Rel {
ExpandRel expand = 15;
WindowRel window = 16;
GenerateRel generate = 17;
- WriteRel write = 18;
- TopNRel top_n = 19;
+ WriteRel write = 19;
+ TopNRel top_n = 23;
WindowGroupLimitRel windowGroupLimit = 20;
+ NestedLoopJoinRel nested_loop_join = 18;
}
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
index 8c0f7a8d2a..eb7cb0df6c 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BroadcastNestedLoopJoinExecTransformer.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.execution.joins.BaseJoinExec
import org.apache.spark.sql.execution.metric.SQLMetric
import com.google.protobuf.Any
-import io.substrait.proto.CrossRel
+import io.substrait.proto.NestedLoopJoinRel
abstract class BroadcastNestedLoopJoinExecTransformer(
left: SparkPlan,
@@ -51,8 +51,8 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
override def leftKeys: Seq[Expression] = Nil
override def rightKeys: Seq[Expression] = Nil
- private lazy val substraitJoinType: CrossRel.JoinType =
- SubstraitUtil.toCrossRelSubstrait(joinType)
+ private lazy val substraitJoinType: NestedLoopJoinRel.JoinType =
+ SubstraitUtil.toNestedLoopJoinSubstrait(joinType)
// Unique ID for the build side.
lazy val buildBroadcastTableId: String = buildPlan.id.toString
@@ -116,7 +116,7 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
joinParams.isWithCondition = true
}
- val crossRel = JoinUtils.createCrossRel(
+ val nestedLoopJoinRel = JoinUtils.createNestedLoopJoinRel(
substraitJoinType,
condition,
inputStreamedRelNode,
@@ -137,7 +137,7 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
buildPlan.output,
context,
operatorId,
- crossRel,
+ nestedLoopJoinRel,
inputStreamedOutput,
inputBuildOutput
)
@@ -166,7 +166,7 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
}
override protected def doValidateInternal(): ValidationResult = {
- if (substraitJoinType == CrossRel.JoinType.UNRECOGNIZED) {
+ if (substraitJoinType == NestedLoopJoinRel.JoinType.UNRECOGNIZED) {
return ValidationResult.failed(
s"$joinType join is not supported with BroadcastNestedLoopJoin")
}
@@ -178,7 +178,7 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
val substraitContext = new SubstraitContext
- val crossRel = JoinUtils.createCrossRel(
+ val nestedLoopJoinRel = JoinUtils.createNestedLoopJoinRel(
substraitJoinType,
condition,
null,
@@ -190,6 +190,6 @@ abstract class BroadcastNestedLoopJoinExecTransformer(
genJoinParameters(),
validation = true
)
- doNativeValidation(substraitContext, crossRel)
+ doNativeValidation(substraitContext, nestedLoopJoinRel)
}
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/CartesianProductExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/CartesianProductExecTransformer.scala
index 8e6df15934..e954e293ba 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/CartesianProductExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/CartesianProductExecTransformer.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.execution.joins.BaseJoinExec
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.vectorized.ColumnarBatch
-import io.substrait.proto.CrossRel
+import io.substrait.proto.NestedLoopJoinRel
import java.io.{IOException, ObjectOutputStream}
@@ -67,8 +67,8 @@ case class CartesianProductExecTransformer(
override def rightKeys: Seq[Expression] = Nil
- protected lazy val substraitJoinType: CrossRel.JoinType =
- SubstraitUtil.toCrossRelSubstrait(joinType)
+ protected lazy val substraitJoinType: NestedLoopJoinRel.JoinType =
+ SubstraitUtil.toNestedLoopJoinSubstrait(joinType)
// Note: "metrics" is made transient to avoid sending driver-side metrics to
tasks.
@transient override lazy val metrics: Map[String, SQLMetric] =
@@ -102,7 +102,7 @@ case class CartesianProductExecTransformer(
joinParams.isWithCondition = true
}
- val currRel = RelBuilder.makeCrossRel(
+ val currRel = RelBuilder.makeNestedLoopJoinRel(
inputLeftRelNode,
inputRightRelNode,
substraitJoinType,
@@ -139,7 +139,7 @@ case class CartesianProductExecTransformer(
val extensionNode =
JoinUtils.createExtensionNode(left.output ++ right.output, validation =
true)
- val currRel = RelBuilder.makeCrossRel(
+ val currRel = RelBuilder.makeNestedLoopJoinRel(
null,
null,
substraitJoinType,
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala
index eeb6069890..c821d52173 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinUtils.scala
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.types.DataType
import com.google.protobuf.Any
-import io.substrait.proto.{CrossRel, JoinRel}
+import io.substrait.proto.{JoinRel, NestedLoopJoinRel}
import scala.collection.JavaConverters._
@@ -312,8 +312,8 @@ object JoinUtils {
)
}
- def createCrossRel(
- substraitJoinType: CrossRel.JoinType,
+ def createNestedLoopJoinRel(
+ substraitJoinType: NestedLoopJoinRel.JoinType,
condition: Option[Expression],
inputStreamedRelNode: RelNode,
inputBuildRelNode: RelNode,
@@ -335,7 +335,7 @@ object JoinUtils {
val extensionNode =
createJoinExtensionNode(joinParameters, inputStreamedOutput ++
inputBuildOutput)
- RelBuilder.makeCrossRel(
+ RelBuilder.makeNestedLoopJoinRel(
inputStreamedRelNode,
inputBuildRelNode,
substraitJoinType,
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitUtil.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitUtil.scala
index a0fdde31ca..c2032aa592 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitUtil.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitUtil.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute,
Expression}
import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter,
InnerLike, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter}
import com.google.protobuf.{Any, DoubleValue, Int32Value, Int64Value, Message,
StringValue}
-import io.substrait.proto.{CrossRel, JoinRel, NamedStruct, Type}
+import io.substrait.proto.{JoinRel, NamedStruct, NestedLoopJoinRel, Type}
import java.lang.{Double => JDouble, Long => JLong}
import java.util.{Collections, List => JList}
@@ -53,20 +53,20 @@ object SubstraitUtil {
JoinRel.JoinType.UNRECOGNIZED
}
- def toCrossRelSubstrait(sparkJoin: JoinType): CrossRel.JoinType = sparkJoin
match {
+ def toNestedLoopJoinSubstrait(sparkJoin: JoinType):
NestedLoopJoinRel.JoinType = sparkJoin match {
case _: InnerLike =>
- CrossRel.JoinType.JOIN_TYPE_INNER
+ NestedLoopJoinRel.JoinType.JOIN_TYPE_INNER
case LeftOuter | RightOuter =>
// since we always assume build right side in substrait,
// the left and right relations are exchanged and the
// join type is reverted.
- CrossRel.JoinType.JOIN_TYPE_LEFT
+ NestedLoopJoinRel.JoinType.JOIN_TYPE_LEFT
case LeftSemi | ExistenceJoin(_) =>
- CrossRel.JoinType.JOIN_TYPE_LEFT_SEMI
+ NestedLoopJoinRel.JoinType.JOIN_TYPE_LEFT_SEMI
case FullOuter =>
- CrossRel.JoinType.JOIN_TYPE_OUTER
+ NestedLoopJoinRel.JoinType.JOIN_TYPE_OUTER
case _ =>
- CrossRel.JoinType.UNRECOGNIZED
+ NestedLoopJoinRel.JoinType.UNRECOGNIZED
}
def createEnhancement(output: Seq[Attribute]): com.google.protobuf.Any = {
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/utils/SubstraitUtilJoinTypeSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/utils/SubstraitUtilJoinTypeSuite.scala
new file mode 100644
index 0000000000..74ce080b99
--- /dev/null
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/utils/SubstraitUtilJoinTypeSuite.scala
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.utils
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.plans.{Cross, ExistenceJoin, FullOuter,
Inner, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter}
+import org.apache.spark.sql.types.BooleanType
+
+import io.substrait.proto.NestedLoopJoinRel.{JoinType => NLJ}
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Locks the Spark -> Substrait `NestedLoopJoinRel.JoinType` mapping that the
cartesian /
+ * broadcast-nested-loop-join producers rely on. Substrait 0.98 splits the
(previously overloaded)
+ * `CrossRel` into a pure-cartesian `CrossRel` plus a `NestedLoopJoinRel`
whose `JoinType` enum
+ * reorders the anti/semi/single values relative to Gluten's old fork. This
suite pins the invariant
+ * that the reorder is semantically inert for Gluten: the producer only ever
emits the enum values
+ * whose proto number is identical across the fork and 0.98 (INNER=1, OUTER=2,
LEFT=3, LEFT_SEMI=5).
+ */
+class SubstraitUtilJoinTypeSuite extends AnyFunSuite {
+
+ private def convert(joinType: JoinType): NLJ =
+ SubstraitUtil.toNestedLoopJoinSubstrait(joinType)
+
+ private val existsAttr = AttributeReference("exists", BooleanType, nullable
= false)()
+
+ test("toNestedLoopJoinSubstrait maps Spark join types to the expected enum
constants") {
+ assert(convert(Inner) === NLJ.JOIN_TYPE_INNER)
+ assert(convert(Cross) === NLJ.JOIN_TYPE_INNER)
+ // RightOuter is exchanged to LeftOuter by the producer (build side is
always the right side).
+ assert(convert(LeftOuter) === NLJ.JOIN_TYPE_LEFT)
+ assert(convert(RightOuter) === NLJ.JOIN_TYPE_LEFT)
+ assert(convert(LeftSemi) === NLJ.JOIN_TYPE_LEFT_SEMI)
+ assert(convert(ExistenceJoin(existsAttr)) === NLJ.JOIN_TYPE_LEFT_SEMI)
+ assert(convert(FullOuter) === NLJ.JOIN_TYPE_OUTER)
+ }
+
+ test("unsupported join types fall back to UNRECOGNIZED") {
+ // LeftAnti is not supported by the nested-loop-join producers; it must
not silently map to a
+ // valid enum value (which would produce a wrong plan instead of a Spark
fallback).
+ assert(convert(LeftAnti) === NLJ.UNRECOGNIZED)
+ }
+
+ test("emitted enum values keep their stable 0.98 proto numbers") {
+ // These four values are identical between Gluten's old CrossRel fork and
0.98's
+ // NestedLoopJoinRel, so re-pointing the producer at the new message is
semantically neutral.
+ assert(NLJ.JOIN_TYPE_INNER.getNumber === 1)
+ assert(NLJ.JOIN_TYPE_OUTER.getNumber === 2)
+ assert(NLJ.JOIN_TYPE_LEFT.getNumber === 3)
+ assert(NLJ.JOIN_TYPE_LEFT_SEMI.getNumber === 5)
+ }
+
+ test("ordinal equals proto number for every declared JoinType value") {
+ // StorageJoinBuilder passes toNestedLoopJoinSubstrait(...).ordinal() as a
raw int over JNI and
+ // the native side casts it back by proto number. That is only correct
while the enum is
+ // declared densely (0..N with no gaps). Lock it so a future reorder that
introduces a gap
+ // fails here rather than silently corrupting the join type on the native
side.
+ NLJ
+ .values()
+ .filter(_ != NLJ.UNRECOGNIZED)
+ .foreach(v => assert(v.ordinal() === v.getNumber, s"ordinal/number
mismatch for $v"))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
