[OPPRO-10] Enable hash join in Substrait-to-Velox conversion (#9)

* hash join

* remove extra projection

Author: marin-ma

velox/substrait/SubstraitToVeloxExpr.h

@@ -28,7 +28,7 @@ class SubstraitVeloxExprConverter {
   /// subParser: A Substrait parser used to convert Substrait representations
   /// into recognizable representations. functionMap: A pre-constructed map
   /// storing the relations between the function id and the function name.
-  SubstraitVeloxExprConverter(
+  explicit SubstraitVeloxExprConverter(
       const std::unordered_map<uint64_t, std::string>& functionMap)
       : functionMap_(functionMap) {}
 

velox/substrait/SubstraitToVeloxPlanValidator.cpp

@@ -115,6 +115,75 @@ bool SubstraitToVeloxPlanValidator::validate(
   return false;
 }
 
+bool SubstraitToVeloxPlanValidator::validate(
+    const ::substrait::JoinRel& sJoin) {
+  if (sJoin.has_left() && !validate(sJoin.left())) {
+    return false;
+  }
+  if (sJoin.has_right() && !validate(sJoin.right())) {
+    return false;
+  }
+
+  switch (sJoin.type()) {
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_INNER:
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_OUTER:
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_LEFT:
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_RIGHT:
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_SEMI:
+    case ::substrait::JoinRel_JoinType_JOIN_TYPE_ANTI:
+      break;
+    default:
+      return false;
+  }
+
+  // Validate input types.
+  if (!sJoin.has_advanced_extension()) {
+    std::cout << "Input types are expected in JoinRel." << std::endl;
+    return false;
+  }
+
+  const auto& extension = sJoin.advanced_extension();
+  std::vector<TypePtr> types;
+  if (!validateInputTypes(extension, types)) {
+    std::cout << "Validation failed for input types in JoinRel" << std::endl;
+    return false;
+  }
+
+  int32_t inputPlanNodeId = 0;
+  std::vector<std::string> names;
+  names.reserve(types.size());
+  for (auto colIdx = 0; colIdx < types.size(); colIdx++) {
+    names.emplace_back(subParser_->makeNodeName(inputPlanNodeId, colIdx));
+  }
+  auto rowType = std::make_shared<RowType>(std::move(names), std::move(types));
+
+  if (sJoin.has_expression()) {
+    std::vector<const ::substrait::Expression::FieldReference*> leftExprs,
+        rightExprs;
+    try {
+      planConverter_->extractJoinKeys(
+          sJoin.expression(), leftExprs, rightExprs);
+    } catch (const VeloxException& err) {
+      std::cout << "Validation failed for expression in JoinRel due to:"
+                << err.message() << std::endl;
+      return false;
+    }
+  }
+
+  if (sJoin.has_post_join_filter()) {
+    try {
+      auto expression =
+          exprConverter_->toVeloxExpr(sJoin.post_join_filter(), rowType);
+      exec::ExprSet exprSet({std::move(expression)}, &execCtx_);
+    } catch (const VeloxException& err) {
+      std::cout << "Validation failed for expression in ProjectRel due to:"
+                << err.message() << std::endl;
+      return false;
+    }
+  }
+  return true;
+}
+
 bool SubstraitToVeloxPlanValidator::validate(
     const ::substrait::AggregateRel& sAgg) {
   if (sAgg.has_input() && !validate(sAgg.input())) {
@@ -304,6 +373,9 @@ bool SubstraitToVeloxPlanValidator::validate(const ::substrait::Rel& sRel) {
   if (sRel.has_filter()) {
     return validate(sRel.filter());
   }
+  if (sRel.has_join()) {
+    return validate(sRel.join());
+  }
   if (sRel.has_read()) {
     return validate(sRel.read());
   }

velox/substrait/SubstraitToVeloxPlanValidator.h

@@ -36,6 +36,9 @@ class SubstraitToVeloxPlanValidator {
   /// Used to validate whether the computing of this Filter is supported.
   bool validate(const ::substrait::FilterRel& sFilter);
 
+  /// Used to validate Join.
+  bool validate(const ::substrait::JoinRel& sJoin);
+
   /// Used to validate whether the computing of this Read is supported.
   bool validate(const ::substrait::ReadRel& sRead);