native_support_nested_loop_join

Author: Bhargava Vadlamani

native/core/src/execution/planner.rs

@@ -106,6 +106,7 @@ use arrow::row::{OwnedRow, RowConverter, SortField};
 use datafusion::common::utils::SingleRowListArrayBuilder;
 use datafusion::common::UnnestOptions;
 use datafusion::physical_plan::filter::FilterExec;
+use datafusion::physical_plan::joins::NestedLoopJoinExec;
 use datafusion::physical_plan::limit::GlobalLimitExec;
 use datafusion::physical_plan::unnest::{ListUnnest, UnnestExec};
 use datafusion_comet_proto::spark_expression::ListLiteral;
@@ -134,7 +135,6 @@ use num::{BigInt, ToPrimitive};
 use object_store::path::Path;
 use std::cmp::max;
 use std::{collections::HashMap, sync::Arc};
-use datafusion::physical_plan::joins::NestedLoopJoinExec;
 use url::Url;
 
 // For clippy error on type_complexity.
@@ -1202,12 +1202,12 @@ impl PhysicalPlanner {
             OpStruct::BroadcastNestedLoopJoin(bnlj) => {
                 let (join_params, scans, shuffle_scans) = self.parse_join_parameters(
                     inputs,
-                    &[],
+                    children,
                     &[],
                     &[],
                     bnlj.join_type,
                     &bnlj.condition,
-                    partition_count
+                    partition_count,
                 )?;
 
                 let left = Arc::clone(&join_params.left.native_plan);
@@ -1218,7 +1218,7 @@ impl PhysicalPlanner {
                     right,
                     join_params.join_filter,
                     &join_params.join_type,
-                    None
+                    None,
                 )?);
 
                 Ok((

native/core/src/execution/planner/operator_registry.rs

@@ -151,6 +151,6 @@ fn get_operator_type(spark_operator: &Operator) -> Option<OperatorType> {
         OpStruct::Explode(_) => None, // Not yet in OperatorType enum
         OpStruct::CsvScan(_) => Some(OperatorType::CsvScan),
         OpStruct::ShuffleScan(_) => None, // Not yet in OperatorType enum
-        OpStruct::BroadcastNestedLoopJoin(_) => todo!(),
+        OpStruct::BroadcastNestedLoopJoin(_) => None,
     }
 }

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -254,6 +254,8 @@ object CometConf extends ShimCometConf {
     createExecEnabledConfig("broadcastExchange", defaultValue = true)
   val COMET_EXEC_HASH_JOIN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("hashJoin", defaultValue = true)
+  val COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED: ConfigEntry[Boolean] =
+    createExecEnabledConfig("broadcastNestedLoopJoin", defaultValue = false)
   val COMET_EXEC_SORT_MERGE_JOIN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("sortMergeJoin", defaultValue = true)
   val COMET_EXEC_AGGREGATE_ENABLED: ConfigEntry[Boolean] =

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -2026,8 +2026,11 @@ object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNest
 
 object CometBroadcastHashJoinExec extends CometOperatorSerde[HashJoin] with CometHashJoin {
 
+  override def getSupportLevel(operator: HashJoin): SupportLevel = {
+    Incompatible(Some("Broadcast nested loop join is experimental"))
+  }
   override def enabledConfig: Option[ConfigEntry[Boolean]] =
-    Some(CometConf.COMET_EXEC_HASH_JOIN_ENABLED)
+    Some(CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED)
 
   override def convert(
       join: HashJoin,