[GLUTEN-12597][CORE] Migrate nested loop joins from CrossRel to NestedLoopJoinRel (Substrait 0.98)

Substrait 0.98 splits Gluten's overloaded CrossRel (which carried both
`expression` and `type`) into a pure-cartesian CrossRel plus a new
NestedLoopJoinRel (expression=4, type=5) with a paired JoinType enum. This
increment of the proto rebase adopts that split across the vendored proto, the
JVM producer, and both native consumers in lockstep.

The NestedLoopJoinRel.JoinType enum reorders the anti/semi/single values
relative to the old CrossRel.JoinType fork, but Gluten only ever emits and
consumes INNER/OUTER/LEFT/LEFT_SEMI, whose proto numbers (1/2/3/5) are identical
in both, so the split is semantically neutral for Gluten's code paths.

The new rel is parked at a temporary Rel-oneof field number; the final oneof
reconciliation increment relocates it to the official nested_loop_join = 18.
Field numbers are codegen-transparent for Gluten's coupled build (single proto
source, transient plans, producer + both consumers regenerate and ship
together).

Part of #12597.

Generated-by: Claude Code (Claude Opus 4.8)

Author: nielspardon

backends-clickhouse/src/main/java/org/apache/gluten/vectorized/StorageJoinBuilder.java

@@ -83,7 +83,7 @@ public static long build(
       joinType =
           JoinTypeTransform.toSubstraitJoinType(broadcastContext.joinType(), buildRight).ordinal();
     } else {
-      joinType = SubstraitUtil.toCrossRelSubstrait(broadcastContext.joinType()).ordinal();
+      joinType = SubstraitUtil.toNestedLoopJoinSubstrait(broadcastContext.joinType()).ordinal();
     }
 
     return nativeBuild(

backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala

@@ -382,7 +382,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

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));

cpp-ch/local-engine/Common/CHUtil.h

@@ -258,7 +258,7 @@ class JoinUtil
     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);
 };
 
 }

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);

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);
 }
 
 }

cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h

@@ -55,13 +55,13 @@ class CrossRelParser : public RelParser
     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);

cpp/velox/substrait/SubstraitToVeloxPlan.cc

@@ -515,43 +515,44 @@ 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>(
@@ -1691,8 +1692,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()) {

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);

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());