[GitHub] [spark] xingchaozh commented on a diff in pull request #37930: [SPARK-40487][SQL] Make defaultJoin in BroadcastNestedLoopJoinExec running in parallel

[GitBox <gi...@apache.org>]

xingchaozh commented on code in PR #37930:
URL: https://github.com/apache/spark/pull/37930#discussion_r975172041


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala:
##########
@@ -286,21 +286,26 @@ case class BroadcastNestedLoopJoinExec(
    */
   private def defaultJoin(relation: Broadcast[Array[InternalRow]]): RDD[InternalRow] = {
     val streamRdd = streamed.execute()
-    val matchedBroadcastRows = getMatchedBroadcastRowsBitSet(streamRdd, relation)
-    val notMatchedBroadcastRows: Seq[InternalRow] = {
-      val nulls = new GenericInternalRow(streamed.output.size)
-      val buf: CompactBuffer[InternalRow] = new CompactBuffer()
-      val joinedRow = new JoinedRow
-      joinedRow.withLeft(nulls)
-      var i = 0
-      val buildRows = relation.value
-      while (i < buildRows.length) {
-        if (!matchedBroadcastRows.get(i)) {
-          buf += joinedRow.withRight(buildRows(i)).copy()
+    def notMatchedBroadcastRows: RDD[InternalRow] = {
+      getMatchedBroadcastRowsBitSetRDD(streamRdd, relation).
+        repartition(1).
+        mapPartitions(iter => {

Review Comment:
   Thanks for the review! just fixed the code style



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org