You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/04/17 05:48:48 UTC

[GitHub] [spark] c21 opened a new pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

c21 opened a new pull request #32210:
URL: https://github.com/apache/spark/pull/32210


   <!--
   Thanks for sending a pull request!  Here are some tips for you:
     1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
     2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
     3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
     4. Be sure to keep the PR description updated to reflect all changes.
     5. Please write your PR title to summarize what this PR proposes.
     6. If possible, provide a concise example to reproduce the issue for a faster review.
     7. If you want to add a new configuration, please read the guideline first for naming configurations in
        'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
   -->
   
   ### What changes were proposed in this pull request?
   <!--
   Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. 
   If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
     1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
     2. If you fix some SQL features, you can provide some references of other DBMSes.
     3. If there is design documentation, please add the link.
     4. If there is a discussion in the mailing list, please add the link.
   -->
   A major pain point for spark users to stay away from using shuffled hash join is out of memory issue. Shuffled hash join tends to have OOM issue because it allocates in-memory hashed relation (`UnsafeHashedRelation` or `LongHashedRelation`) for build side, and there's no recovery (e.g. fallback/spill) once the size of hashed relation grows and cannot fit in memory. On the other hand, shuffled hash join is more CPU and IO efficient than sort merge join when joining one large table and a small table (but small table is too large to be broadcasted), as SHJ does not sort the large table, but SMJ needs to do that. See historical discussion in https://github.com/apache/spark/pull/11788 for evidence.
   
   To improve the reliability of shuffled hash join, a fallback mechanism can be introduced to avoid shuffled hash join OOM issue automatically. Similarly we already have a fallback to sort-based aggregation for hash aggregate. The idea is:
   
   * Build hashed relation as current, but stop adding rows to hashed relation if there's no enough memory. Do not throw exception, and do not fail the task/query.
   * Sort stream side and build side on join keys if necessary. Note here we need to read all build rows in hashed relation back and destruct the hashed relation on the fly to free memory.
   * Execute sort merge join on sorted stream & build side.
   
   Note:
   
   (1).the fallback is automatic and happened per task, which means task 0 can incur the fallback e.g. if it has a big build side, but task 1,2 don't need to incur the fallback depending on the size of hashed relation.
   
   (2).there's no major code change for SHJ and SMJ. Major change is around `HashedRelation` to introduce some new methods, e.g. `HashedRelation.destructiveValues()` to return an Iterator of build side rows in hashed relation and destruct hashed relation on the fly.
   
   (3).Per this PR, a new config `spark.sql.join.enableShuffledHashJoinFallback` is introduced to enable/disable this feature (disable by default). This PR only supports fallback for non-code-gen execution path. Fallback for code-gen will be added in followup PRs, as it will depend on sort merge join code-gen work (SPARK-34705).
   
   ### Why are the changes needed?
   <!--
   Please clarify why the changes are needed. For instance,
     1. If you propose a new API, clarify the use case for a new API.
     2. If you fix a bug, you can clarify why it is a bug.
   -->
   Shuffled hash join OOM is a huge pain point for users and developers. This is the major reason why people stay away from shuffled hash join. This can improve reliability for shuffled hash join quite a bit.
   
   ### Does this PR introduce _any_ user-facing change?
   <!--
   Note that it means *any* user-facing change including all aspects such as the documentation fix.
   If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
   If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
   If no, write 'No'.
   -->
   No.
   
   ### How was this patch tested?
   <!--
   If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
   If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
   If tests were not added, please describe why they were not added and/or why it was difficult to add.
   -->
   Added unit test in `JoinSuite.scala`.


-- 
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.

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

[GitHub] [spark] c21 edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854356895


   Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   
   At the high level, the feature is:
   
   1. Build the hash table (`HashedRelation`) on build side.
   2. If step 1 succeeds, do shuffled hash join with stream side.
   3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
     3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
     3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. ~~If having matched rows, output and process next row. If no matched rows,~~ For inner/outer/anti join, add the row from stream side to another sorter (`streamedSorter`). For semi join, only add row to sorter if there's no match in hash table.
     3.3. After processing all rows from streamed side, clean up hash table, and sort `buildSorter` and `streamedSorter` and do sort merge join.
   
   In this way, the effort of building hash table is not wasted compared to this PR, and we can control how to do the fallback. Three ways of fallback policy I can think of  - (1).default `SparkOutOfMemoryError` in `BytesToBytesMap`, (2).introduce configurable threshold for key capacity in `BytesToBytesMap` (max number of keys, `BytesToBytesMap.capacity`), (3).introduce configurable threshold for used page size in `BytesToBytesMap` (max size of keys+values, `BytesToBytesMap.used`).
   
   This solution mimics the idea of `hybrid join` in https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf, where it tried to build hash table, and spill rest of rows on disk, and join rest of rows on disk later.
   
   WDYT? @cloud-fan, @maropu and @sigmod, thanks.


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823604225


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42231/
   


-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619980616



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       For example, we have three tasks for a shuffle hash join. If the two tasks have no fallback and the last one has the fallback, the running time of the fallback one (building time of a hash table and sorting time of left/right input rows) can be much longer than the other two non-fallback ones? Or, I misunderstood the current logic?




-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826251579


   btw if we worry about if too many sort merge join converted to shuffled hash join if enabling shuffled hash join by default. Please note we will only enable shuffled hash join if the estimated/run-time size of one side is less than `"spark.sql.autoBroadcastJoinThreshold" * "spark.sql.shuffle.partitions"`, and this side is 3x smaller than the other side (https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala#L258-L263). This is somewhat restricted rule, and from out side in practice, enabling shuffled hash join by default only incurs 25% of sort merge join getting converted.


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824550054


   Rebased to latest master.


-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r615212076



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
##########
@@ -107,266 +107,262 @@ case class SortMergeJoinExec(
   override def requiredChildOrdering: Seq[Seq[SortOrder]] =
     requiredOrders(leftKeys) :: requiredOrders(rightKeys) :: Nil
 
-  private def requiredOrders(keys: Seq[Expression]): Seq[SortOrder] = {
-    // This must be ascending in order to agree with the `keyOrdering` defined in `doExecute()`.
-    keys.map(SortOrder(_, Ascending))
-  }
-
   private def createLeftKeyGenerator(): Projection =
     UnsafeProjection.create(leftKeys, left.output)
 
   private def createRightKeyGenerator(): Projection =
     UnsafeProjection.create(rightKeys, right.output)
 
-  private def getSpillThreshold: Int = {
-    sqlContext.conf.sortMergeJoinExecBufferSpillThreshold
-  }
-
-  private def getInMemoryThreshold: Int = {
-    sqlContext.conf.sortMergeJoinExecBufferInMemoryThreshold
-  }
-
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
     val spillThreshold = getSpillThreshold
     val inMemoryThreshold = getInMemoryThreshold
     left.execute().zipPartitions(right.execute()) { (leftIter, rightIter) =>
-      val boundCondition: (InternalRow) => Boolean = {
-        condition.map { cond =>
-          Predicate.create(cond, left.output ++ right.output).eval _
-        }.getOrElse {
-          (r: InternalRow) => true
-        }
+      executeJoinWithIterators(

Review comment:
       Just FYI - the original code in `doExecute()` is moved into a new public method `executeJoinWithIterators()` without any actual change. Not sure why github shows so many code change.




-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824574398


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42301/
   


-- 
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.

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

[GitHub] [spark] cloud-fan commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619309141



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -298,9 +311,83 @@ case class ShuffledHashJoinExec(
     streamResultIter ++ buildResultIter
   }
 
+  private val enableShuffledHashJoinFallback: Boolean =
+    sqlContext.conf.enableShuffledHashJoinFallback
+
+  /**
+   * This is for testing only. We force shuffled hash join to fall back to sort merge join,
+   * once number of rows in build relation exceeds this limit.
+   */
+  private val testFallbackStartsAt: Option[Int] = {
+    sqlContext.getConf("spark.sql.ShuffledHashJoin.testFallbackStartsAt", null) match {
+      case null | "" => None
+      case fallbackStartsAt => Some(fallbackStartsAt.toInt)
+    }
+  }
+
+  /**
+   * Get [[SortExec]] plan for stream side on join keys. This is used for sort-based fallback.
+   * The return value is `Option` as the stream side may be already ordered on join keys.
+   */
+  private def getStreamSortPlan: Option[SortExec] = {
+    val requiredStreamOrdering = requiredOrders(streamedKeys)
+    if (!SortOrder.orderingSatisfies(streamedPlan.outputOrdering, requiredStreamOrdering)) {
+      Some(SortExec(requiredStreamOrdering, global = false, child = streamedPlan))
+    } else {
+      None
+    }
+  }
+
+  /**
+   * Get [[SortExec]] plan for build side on join keys. This is used for sort-based fallback.
+   * The return value is not `Option` as some of build rows may be already read and put into
+   * [[HashedRelation]], so the ordering for build side needs to be forced again.
+   */
+  private def getBuildSortPlan: SortExec = {
+    val requiredBuildOrdering = requiredOrders(buildKeys)
+    SortExec(requiredBuildOrdering, global = false, child = buildPlan)
+  }
+
+  /**
+   * Fallback to sort merge join, from shuffled hash join.
+   *
+   * Sort stream and build side on join keys if necessary, as sort merge join requires both
+   * children to be sorted on join keys. See [[SortMergeJoinExec.requiredChildOrdering]].
+   * After that execute sort merge join on sorted children.
+   */
+  private def joinWithSortFallback(
+      streamIter: Iterator[InternalRow],
+      buildIter: Iterator[InternalRow],
+      relationIter: Iterator[InternalRow],
+      streamSortPlan: Option[SortExec],
+      buildSortPlan: SortExec,
+      sortMergeJoinPlan: SortMergeJoinExec,
+      numOutputRows: SQLMetric,
+      spillThreshold: Int,
+      inMemoryThreshold: Int): Iterator[InternalRow] = {
+
+    // Sort stream and build side on join keys if necessary
+    val streamSortIter = streamSortPlan match {
+      case Some(plan) =>
+        val sorter = plan.createSorter()
+        sorter.sort(streamIter.asInstanceOf[Iterator[UnsafeRow]])
+      case _ => streamIter
+    }
+    val buildSortIter = buildSortPlan.createSorter().sort(
+      (relationIter ++ buildIter).asInstanceOf[Iterator[UnsafeRow]])
+
+    // Fallback to sort merge join
+    buildSide match {
+      case BuildLeft => sortMergeJoinPlan.executeJoinWithIterators(

Review comment:
       It's a bit weird to instantiate a plan and call it method to reuse code. Can we put the shareable code in the base class `ShuffledJoin`?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -298,9 +311,83 @@ case class ShuffledHashJoinExec(
     streamResultIter ++ buildResultIter
   }
 
+  private val enableShuffledHashJoinFallback: Boolean =
+    sqlContext.conf.enableShuffledHashJoinFallback
+
+  /**
+   * This is for testing only. We force shuffled hash join to fall back to sort merge join,
+   * once number of rows in build relation exceeds this limit.
+   */
+  private val testFallbackStartsAt: Option[Int] = {
+    sqlContext.getConf("spark.sql.ShuffledHashJoin.testFallbackStartsAt", null) match {
+      case null | "" => None
+      case fallbackStartsAt => Some(fallbackStartsAt.toInt)
+    }
+  }
+
+  /**
+   * Get [[SortExec]] plan for stream side on join keys. This is used for sort-based fallback.
+   * The return value is `Option` as the stream side may be already ordered on join keys.
+   */
+  private def getStreamSortPlan: Option[SortExec] = {
+    val requiredStreamOrdering = requiredOrders(streamedKeys)
+    if (!SortOrder.orderingSatisfies(streamedPlan.outputOrdering, requiredStreamOrdering)) {
+      Some(SortExec(requiredStreamOrdering, global = false, child = streamedPlan))
+    } else {
+      None
+    }
+  }
+
+  /**
+   * Get [[SortExec]] plan for build side on join keys. This is used for sort-based fallback.
+   * The return value is not `Option` as some of build rows may be already read and put into
+   * [[HashedRelation]], so the ordering for build side needs to be forced again.
+   */
+  private def getBuildSortPlan: SortExec = {
+    val requiredBuildOrdering = requiredOrders(buildKeys)
+    SortExec(requiredBuildOrdering, global = false, child = buildPlan)
+  }
+
+  /**
+   * Fallback to sort merge join, from shuffled hash join.
+   *
+   * Sort stream and build side on join keys if necessary, as sort merge join requires both
+   * children to be sorted on join keys. See [[SortMergeJoinExec.requiredChildOrdering]].
+   * After that execute sort merge join on sorted children.
+   */
+  private def joinWithSortFallback(
+      streamIter: Iterator[InternalRow],
+      buildIter: Iterator[InternalRow],
+      relationIter: Iterator[InternalRow],
+      streamSortPlan: Option[SortExec],
+      buildSortPlan: SortExec,
+      sortMergeJoinPlan: SortMergeJoinExec,
+      numOutputRows: SQLMetric,
+      spillThreshold: Int,
+      inMemoryThreshold: Int): Iterator[InternalRow] = {
+
+    // Sort stream and build side on join keys if necessary
+    val streamSortIter = streamSortPlan match {
+      case Some(plan) =>
+        val sorter = plan.createSorter()
+        sorter.sort(streamIter.asInstanceOf[Iterator[UnsafeRow]])
+      case _ => streamIter
+    }
+    val buildSortIter = buildSortPlan.createSorter().sort(
+      (relationIter ++ buildIter).asInstanceOf[Iterator[UnsafeRow]])
+
+    // Fallback to sort merge join
+    buildSide match {
+      case BuildLeft => sortMergeJoinPlan.executeJoinWithIterators(

Review comment:
       It's a bit weird to instantiate a plan and call its method to reuse code. Can we put the shareable code in the base class `ShuffledJoin`?




-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824721781


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137773/
   


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854396893


   > > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   > 
   > @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.
   
   Ok, it sounds like a working join algorithm that at least does not error out a query because of OOM. 
   
   But the proposal is less efficient compared to hybrid hash join in a few aspects: 
   - (1) HHJ doesn't have to spill all probe rows, because build side has been partitioned - a build side partition is either fully memory-resident or fully spilled -- it's the same for corresponding probe rows;  
   - (2)  It's not friendly to the case that the build side just slightly goes beyond RAM.  E.g., let's say, in memory hash table's size is H before OOM, build side total size is H + h, probe side size is P.  If P >> H >> h, meaning we just slightly go beyond RAM size, we have to sort the entire probe side. 
   
   Given that it's not worse than what the existing HJ does - running slowly is better than an error, I'm fine with merging this new proposal :-)
   


-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854380297






-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854356895






-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619991482



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Yes it is. I explicitly disable code-gen (in `ShuffledHashJoinExec.supportCodegen`) when `SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED` is set to true in this PR.




-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824466760


   **[Test build #137752 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137752/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).
    * This patch **fails Spark unit tests**.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824422057


   **[Test build #137752 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137752/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854380297


   > Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   > 
   > At the high level, the feature is:
   > 
   > 1. Build the hash table (`HashedRelation`) on build side.
   > 2. If step 1 succeeds, do shuffled hash join with stream side.
   > 3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
   >    3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
   >    3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. If having matched rows, output and process next row. If no matched rows, add the row from stream side to another sorter (`streamedSorter`).
   
   Is 3.1 + 3.2 correct?  Let's say, if the build side have two rows `r1` and `r2`, with identical values on join key columns. If r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM.  Now a probe side row `p`, with the same join key column values can only match `r1`.  `p` should match `r2` as well for correctness.  
   


-- 
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.

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

[GitHub] [spark] advancedxy commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
advancedxy commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-828606822


   > > Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map.
   > > How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   > 
   > `Hybrid hash join` was designed to address this issue:
   > 
   > * https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
   > * https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf
   > 
   > In short, it's not to spill an entire, huge hash map onto disk, but combines in-memory hash join with data partitioning/spilling. It's a standard algorithm that been implemented in many query engines. Similarly, you could implement hash aggregation in the same way, which should be more efficient than the runtime fallback approach too.
   > 
   > IMO, either (1) AQE or (2) conservative static QO decisions (e.g., planning shuffled hash join less aggressively) might be some low-hanging fruits, if they can address the issue. But if indeed neither could work for some of your queries, supporting spilling could be viable option to go.
   
   IMO, AQE is the low-hanging fruits for joins with shuffle. We did some benchmark on TPC-DS(SMJ v.s AQE auto converting SMJ-> SHJ), and saw obviously performance gain(~20%). We are launching this strategy in our ad-hoc workload.
   
   For SHJ falling back to SMJ,  I believe it's a viable way.  The SMJ is battle-tested and robust to most cases.


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824746362


   **[Test build #137782 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137782/testReport)** for PR 32210 at commit [`f3e3732`](https://github.com/apache/spark/commit/f3e3732749eaefdd4fa12889b54ecc81cb5b6773).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854414963


   > PS, there's some additional complexity to get left/full-outer joins done right (assuming probe side is on the left) -- for a LOJ/FOJ, it seems that you need to tag each probe side row with "matched" v.s. "not-matched-yet", and then use this information in the customized merge join logic.
   
   @sigmod - this is valid point. For outer join, current SMJ cannot work out of box, need to push the matching info into sorter as well. This does add complexity for SMJ.


-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619976091



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       @c21 This fallback logic has been already deployed in your production? I just want to know that this can work well for real workloads (This can cause high performance penalties if the fallback happens only in a single task).




-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619987605



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > For runtime, yes. The total query run-time is dominated by the last finished task runtime. Just to point it out in case, without this change, this would be task and query failure.
   
   Yea, I basically agree to make the shuffle hash-join more robust (since a user possibly use inappropriate join hints in some cases). What I'm interested in is that; is there any other faster ballback logic than the current approach.




-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-946393564


   @weixiuli - I am planning to reopen the discussion of this feature, after adding codegen for FULL OUTER shuffled hash join, as the other one is more straightforward and it's a prerequisite of this PR.


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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824531538


   The unit test failure is org.apache.spark.sql.catalyst.optimizer.ReplaceNullWithFalseInPredicateSuite, which is not related to this PR. #32157 introduced the failure and #32278 fixed it. So the unit test failure should not block review, thanks.


-- 
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.

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

[GitHub] [spark] SparkQA removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824422057


   **[Test build #137752 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137752/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824442646


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42279/
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824617987


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42311/
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824747758


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137782/
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821772356


   **[Test build #137500 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137500/testReport)** for PR 32210 at commit [`6e81a63`](https://github.com/apache/spark/commit/6e81a63f5fcbb811cb0b13f90a295977d553853f).


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826500384


   > Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map. 
   > How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   `Hybrid hash join` was designed to address this issue: 
   - https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
   - https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf
    
   In short, it's not to spill an entire, huge hash map onto disk, but combines in-memory hash join with data partitioning/spilling. It's a standard algorithm that been implemented in many query engines. Similarly, you could implement hash aggregation in the same way, which should be more efficient than the runtime fallback approach too. 
   
   IMO, either (1) AQE or (2) conservative static QO decisions (e.g., planning shuffled hash join less aggressively) might be some low-hanging fruits, if they can address the issue. But if indeed neither could work for some of your queries, supporting spilling could be viable option to go. 
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824721781


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137773/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826493892


   After skew join handling, the output partitioning is destroyed, but this approach keeps output partitioning. I am just brainstorming the pros and cons, and I think we should pick the direction towards the eventual goal - enabling shuffled hash join by default.


-- 
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.

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

[GitHub] [spark] github-actions[bot] commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821772094


   **[Test build #757821069](https://github.com/c21/spark/actions/runs/757821069)** for PR 32210 at commit [`6e81a63`](https://github.com/c21/spark/commit/6e81a63f5fcbb811cb0b13f90a295977d553853f).


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821804175


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137500/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823503243


   @cloud-fan -
   
   > sorting the stream-side at runtime may lead to slow query plan because the sort is not whole-stage-codegen-ed.
   
   Note this PR is for non-code-gen path only. For code-gen path, I plan to add later as it depends on sort merge join code-gen work (SPARK-34705).Sort operator code-gen is pretty [straightforward](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala#L170), so for code-gen path, shuffled hash join, fallbacked sort, fallbacked sort merge join are all will be code-gen-ed.
   
   > unlike SMJ, the output ordering can't be preserved if we sort the stream-side at runtime.
   
   This is good observation and I noticed this too. I think it might be able to be resolved with propagating `skipSort` from parent operators of `SortExec` (similar to `limitNotReachedCond` for limit). But I need to check closer for how it's feasible. IMO this might not be a blocker, as for sort-based fallback of `HashAggregateExec`, the output ordering is not preserved neither.
   
   > I think the eventual goal is to enable shuffle hash join by default, but I'm not sure adding the fallback can achieve this goal. Do you have some real data to show the benefits?
   
   We enabled shuffled hash join by default with this feature. In our environment, roughly 25% of sort merge join queries are now running with shuffled hash join after enabling shuffled hash join by default. Roughly 20% of shuffled hash join queries has tasks fall-backed. Roughly 10% of shuffled hash join tasks has fall-backed. TLDR is it saves 20% of shuffled hash join queries failure, and 10% of tasks failure.
   
   > Another idea is to pick shuffle hash join in AQE when we know the per-partition size after shuffle.
   
   I think more about this and I assume you are referring to some kind of `HybridJoin` operator, and we can choose to do shuffled hash join vs sort merge join for each task independently, e.g. based on partition size, task1 can do shuffled hash join, and task2 can do sort merge join, etc. This approach has some disadvantages though:
   
   1. Do not work if join child does not have shuffle. E.g., join on bucketed table, multiple join, join after group-by, etc. Do not work for 100% case sounds like a show stopper for me, so we cannot enable shuffled hash join by default.
   2. Similarly the output ordering can't be preserved out of box. We only have a global view of `OutputOrdering` now, which applies to all tasks. We need introduce something else, to accommodate the fact that some tasks preserve the order, but some tasks do not.
   3. Per-partition build size is not 100% same as run-time hash relation size. The hash relation size can be larger with storing the extra keys and underlying `BytesToBytesMap` overhead.


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854386528


   > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   
   @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824566953


   **[Test build #137782 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137782/testReport)** for PR 32210 at commit [`f3e3732`](https://github.com/apache/spark/commit/f3e3732749eaefdd4fa12889b54ecc81cb5b6773).


-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619976792



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       @maropu - yes, more data is in https://github.com/apache/spark/pull/32210#issuecomment-823503243 .
   
   > This can cause high performance penalties if the fallback happens only in a single task
   
   Wondering why it is the case?




-- 
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.

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

[GitHub] [spark] c21 edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826250121


   > Apparently falling back to SMJ wastes the partially-built hash map.
   
   Note the fall back only happens when there's no memory available to add new entry to hash map. So even without enabling shuffled hash join by default for all queries, from my point of view, this is still an improvement for reliability right now if people selectively enable shuffled hash join for certain queries. People do not need to worry about OOM for shuffled hash join. If we think this is not the right direction to go, I think same argument going with hash aggregate fallback. We enable hash aggregate by default, and if the query keeps falling back from hash aggregate to sort-based aggregate, then this is not as efficient as just using sort aggregate for the query.
   
   > If one partition is a bit larger to build the in-memory hash map, I feel spilling the hash map might be a better choice?
   
   Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map. How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   > If one partition is much larger to build the in-memory hash map, seems we can use the same technique of skew join handling, to split the partition into multiple smaller ones so that they can fit in memory.
   
   Again I feel this goes back to the AQE hybrid join discussion. The same disadvantage applies here - only work for join with shuffle. We cannot reliably enable shuffled hash join then.


-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826500384


   > Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map. 
   > How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   `Hybrid hash join` was designed to address this issue: 
   - https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
   - https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf
    
   In short, it's not to spill an entire, huge hash map onto disk, but combines in-memory hash join with data partitioning/spilling. It's a standard algorithm that been implemented in many query engines. Similarly, you could implement hash aggregation in a similar way, which should be more efficient than the runtime fallback approach too. 
   
   IMO, either (1) AQE or (2) conservative static QO decisions (e.g., planning shuffled hash join less aggressively) might be some low-hanging fruits, if they can address the issue. But if indeed neither could work for some of your queries, supporting spilling could be viable option to go. 
   


-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619981709



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > @maropu - yes, more data is in #32210 (comment) .
   
   Ah, thanks. I missed that comment.




-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821800964


   **[Test build #137500 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137500/testReport)** for PR 32210 at commit [`6e81a63`](https://github.com/apache/spark/commit/6e81a63f5fcbb811cb0b13f90a295977d553853f).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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.

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

[GitHub] [spark] maropu commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826501739


   I have the same impression with @sigmod.


-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854403695


   > > > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   > > 
   > > 
   > > @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.
   > 
   > Ok, it sounds like a working join algorithm that at least does not error out a query because of OOM.
   > 
   > But the proposal is less efficient compared to hybrid hash join in a few aspects:
   > 
   > * (1) HHJ doesn't have to spill all probe rows, because build side has been partitioned - a build side partition is either fully memory-resident or fully spilled -- it's the same for corresponding probe rows;
   > * (2)  It's not friendly to the case that the build side just slightly goes beyond RAM.  E.g., let's say, in memory hash table's size is H before OOM, build side total size is H + h, probe side size is P.  If P >> H >> h, meaning we just slightly go beyond RAM size, we have sort the entire probe side.
   > 
   > Given that it's not worse than what the existing HJ does - running slowly is better than an error, I'm fine with merging this new proposal :-)
   
   PS, there's some additional complexity to get left/full-outer joins done right -- for a LOJ/FOJ, it seems that you need to tag each probe side row with "matched" v.s. "not-matched-yet",  and then use this information in the customized merge join logic.  
   


-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619981709



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > @maropu - yes, more data is in #32210 (comment) .
   
   Ah, thanks. I missed that comment. It looks nice.




-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824442646


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42279/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826250121


   > Apparently falling back to SMJ wastes the partially-built hash map.
   
   Note the fall back only happens when there's no memory available to add new entry to hash map. So even without enabling shuffled hash join by default for all queries, from my point of view, this is still an improvement for reliability right now if people selectively enable shuffled hash join for certain queries. People do not need to worry about OOM for shuffled hash join. If we think this is not the right direction to go, I think same argument going with hash aggregate fallback. We enable hash aggregate by default, and if the query keeps falling back from hash aggregate to sort-based aggregate, then this is not as efficient as just using sort aggregate for the query.
   
   > If one partition is a bit larger to build the in-memory hash map, I feel spilling the hash map might be a better choice?
   
   Per my knowledge I don't know any obviously efficient way to do random lookup join will spilled hash map. How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   > If one partition is much larger to build the in-memory hash map, seems we can use the same technique of skew join handling, to split the partition into multiple smaller ones so that they can fit in memory.
   
   Again I feel this goes back to the AQE hybrid join discussion. The same disadvantage applies here - only work for join with shuffle. We cannot reliably enable shuffled hash join then.


-- 
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.

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

[GitHub] [spark] SparkQA removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821772356


   **[Test build #137500 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137500/testReport)** for PR 32210 at commit [`6e81a63`](https://github.com/apache/spark/commit/6e81a63f5fcbb811cb0b13f90a295977d553853f).


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824536722


   Sorry I haven't to rebase to latest master with fix yet, let me rebase in a moment.


-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619912231



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Is this a test for non-codegen path?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>

Review comment:
       How about adding a new SQL metric for #fallbacks then checking it in the test?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
##########
@@ -475,18 +501,89 @@ private[joins] object UnsafeHashedRelation {
           key.getBaseObject, key.getBaseOffset, key.getSizeInBytes,
           row.getBaseObject, row.getBaseOffset, row.getSizeInBytes)
         if (!success) {
-          binaryMap.free()
-          throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          if (allowsFallbackWithNoMemory) {
+            return new UnfinishedUnsafeHashedRelation(numFields, binaryMap, row)
+          } else {
+            // Clean up map and throw exception
+            binaryMap.free()
+            throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          }
         }
       } else if (isNullAware) {
         return HashedRelationWithAllNullKeys
       }
+      i += 1
     }
 
     new UnsafeHashedRelation(key.size, numFields, binaryMap)
   }
 }
 
+/**
+ * An unfinished version of [[UnsafeHashedRelation]].
+ * This is intended to use in sort-based fallback of [[ShuffledHashJoinExec]],
+ * when there is no enough memory to build [[UnsafeHashedRelation]].
+ *
+ * @param numFields Number of fields in each row.
+ * @param binaryMap Backed [[BytesToBytesMap]] to hold keys and rows.
+ * @param pendingRow The row which cannot be added to `binaryMap` due to memory limit.
+ */
+private[joins] class UnfinishedUnsafeHashedRelation(

Review comment:
       Needs tests for this new class.




-- 
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.

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

[GitHub] [spark] c21 edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826250121






-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854396893


   > > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   > 
   > @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.
   
   Ok, it sounds like a working join algorithm that at least does not error out a query because of OOM. 
   
   But the proposal less efficient compared to hybrid hash join in a few aspects: 
   - (1) HHJ doesn't have to spill all probe rows, because build side has been partitioned - a build side partition is either fully memory-resident or fully spilled -- it's the same for corresponding probe rows;  
   - (2)  It's not friendly to the case that the build side just slightly goes beyond RAM.  E.g., let's say, in memory hash table's size is H before OOM, build side total size is H + h, probe side size is P.  If P >> H >> h, meaning we just slightly go beyond RAM size, we have sort the entire probe side. 
   
   Given that it's not worse than what the existing HJ does - running slowly is better than an error, I'm fine with merging this new proposal :-)
   


-- 
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.

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

[GitHub] [spark] SparkQA removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824542377


   **[Test build #137773 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137773/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824466969


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137752/
   


-- 
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.

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

[GitHub] [spark] cloud-fan commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619302789



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
##########
@@ -475,18 +501,89 @@ private[joins] object UnsafeHashedRelation {
           key.getBaseObject, key.getBaseOffset, key.getSizeInBytes,
           row.getBaseObject, row.getBaseOffset, row.getSizeInBytes)
         if (!success) {
-          binaryMap.free()
-          throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          if (allowsFallbackWithNoMemory) {
+            return new UnfinishedUnsafeHashedRelation(numFields, binaryMap, row)
+          } else {
+            // Clean up map and throw exception
+            binaryMap.free()
+            throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          }
         }
       } else if (isNullAware) {
         return HashedRelationWithAllNullKeys
       }
+      i += 1
     }
 
     new UnsafeHashedRelation(key.size, numFields, binaryMap)
   }
 }
 
+/**
+ * An unfinished version of [[UnsafeHashedRelation]].
+ * This is intended to use in sort-based fallback of [[ShuffledHashJoinExec]],
+ * when there is no enough memory to build [[UnsafeHashedRelation]].
+ *
+ * @param numFields Number of fields in each row.
+ * @param binaryMap Backed [[BytesToBytesMap]] to hold keys and rows.
+ * @param pendingRow The row which cannot be added to `binaryMap` due to memory limit.
+ */
+private[joins] class UnfinishedUnsafeHashedRelation(
+    private val numFields: Int,
+    private val binaryMap: BytesToBytesMap,
+    private val pendingRow: UnsafeRow)
+  extends HashedRelation {
+
+  override def destructiveValues(): Iterator[InternalRow] = new Iterator[InternalRow] {

Review comment:
       Do we really need to define this function in the base class? Seems the caller side can just patter match this relation and call `destructiveValues`. It's also safer to do so, as `destructiveValues` in other relations just fail.




-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854380297


   > Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   > 
   > At the high level, the feature is:
   > 
   > 1. Build the hash table (`HashedRelation`) on build side.
   > 2. If step 1 succeeds, do shuffled hash join with stream side.
   > 3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
   >    3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
   >    3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. If having matched rows, output and process next row. If no matched rows, add the row from stream side to another sorter (`streamedSorter`).
   
   Is 3.1 + 3.2 correct?  Let's say, if the build side have two rows `r1` and `r2`, with identical values on join key columns. If r1 is processed before OOM and inserted into the in memory table, while r2 is processed after the OOM.  Now a probe side row `p`, with the same join key column values can only match `r1`.  `p` should match `r2` as well for correctness.  
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824414702


   Thanks @cloud-fan, created SPARK-35179 for the followup on AQE. Resolved conflict for this PR and this is ready for review, thanks.


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824436244






-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619912231



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Is this a test for non-codegen path?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>

Review comment:
       How about adding a new SQL metric for #fallbacks then checking it in the test?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
##########
@@ -475,18 +501,89 @@ private[joins] object UnsafeHashedRelation {
           key.getBaseObject, key.getBaseOffset, key.getSizeInBytes,
           row.getBaseObject, row.getBaseOffset, row.getSizeInBytes)
         if (!success) {
-          binaryMap.free()
-          throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          if (allowsFallbackWithNoMemory) {
+            return new UnfinishedUnsafeHashedRelation(numFields, binaryMap, row)
+          } else {
+            // Clean up map and throw exception
+            binaryMap.free()
+            throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          }
         }
       } else if (isNullAware) {
         return HashedRelationWithAllNullKeys
       }
+      i += 1
     }
 
     new UnsafeHashedRelation(key.size, numFields, binaryMap)
   }
 }
 
+/**
+ * An unfinished version of [[UnsafeHashedRelation]].
+ * This is intended to use in sort-based fallback of [[ShuffledHashJoinExec]],
+ * when there is no enough memory to build [[UnsafeHashedRelation]].
+ *
+ * @param numFields Number of fields in each row.
+ * @param binaryMap Backed [[BytesToBytesMap]] to hold keys and rows.
+ * @param pendingRow The row which cannot be added to `binaryMap` due to memory limit.
+ */
+private[joins] class UnfinishedUnsafeHashedRelation(

Review comment:
       Needs tests for this new class.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       @c21 This fallback logic has been already deployed in your production? I just want to know that this can work well for real workloads (This can cause high performance penalties if the fallback happens only in a single task).

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       For example, we have three tasks for a shuffle hash join. If the two tasks have no fallback and the last one has the fallback, the running time of the fallback one (building time of a hash table and sorting time of left/right input rows) can be much longer than the other two non-fallback ones? Or, I misunderstood the current logic?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > @maropu - yes, more data is in #32210 (comment) .
   
   Ah, thanks. I missed that comment.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > @maropu - yes, more data is in #32210 (comment) .
   
   Ah, thanks. I missed that comment. It looks nice.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > For runtime, yes. The total query run-time is dominated by the last finished task runtime. Just to point it out in case, without this change, this would be task and query failure.
   
   Yea, I basically agree to make the shuffle hash-join more robust (since a user possibly use inappropriate join hints in some cases). What I'm interested in is that; is there any other faster ballback logic than the current approach.

##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Oh, I see.




-- 
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.

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

[GitHub] [spark] maropu commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r620002914



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Oh, I see.




-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824614127


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42311/
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823604225


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42231/
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823601974


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42231/
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824542377


   **[Test build #137773 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137773/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).


-- 
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.

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

[GitHub] [spark] cloud-fan commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826239123


   After more thinking, I'm wondering if this is the right direction to go. Apparently falling back to SMJ wastes the partially-built hash map.
   
   If one partition is a bit larger to build the in-memory hash map, I feel spilling the hash map might be a better choice? If one partition is much larger to build the in-memory hash map, seems we can use the same technique of skew join handling, to split the partition into multiple smaller ones so that they can fit in memory.


-- 
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.

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

[GitHub] [spark] cloud-fan commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824536267


   retest this please


-- 
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.

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

[GitHub] [spark] c21 edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854356895


   Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   
   At the high level, the feature is:
   
   1. Build the hash table (`HashedRelation`) on build side.
   2. If step 1 succeeds, do shuffled hash join with stream side.
   3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
     3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
     3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. ~~If having matched rows, output and process next row. If no matched rows,~~ For inner/outer/anti join, add the row from stream side to another sorter (`streamedSorter`). For semi join, only add row to sorter if there's no match in hash table.
     3.3. After processing all rows from streamed side, clean up hash table, and sort `buildSorter` and `streamedSorter` and do sort merge join.
   
   In this way, the effort of building hash table is not wasted compared to this PR, and we can control how to do the fallback. Three ways of fallback policy I can think of  - (1).default `SparkOutOfMemoryError` in `BytesToBytesMap`, (2).introduce configurable threshold for key capacity in `BytesToBytesMap` (max number of keys, `BytesToBytesMap.capacity`), (3).introduce configurable threshold for used page size in `BytesToBytesMap` (max size of keys+values, `BytesToBytesMap.used`).
   
   This solution mimics the idea of `hybrid join` in https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf, where it tried to build hash table, and spill rest of rows on disk, and join rest of rows on disk later.
   
   WDYT? @cloud-fan, @maropu and @sigmod, thanks.


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854396893


   > > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   > 
   > @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.
   
   Ok, it sounds like a working join algorithm that at least does not error out a query because of OOM. 
   
   But the proposal is less efficient compared to hybrid hash join in a few aspects: 
   - (1) HHJ doesn't have to spill all probe rows, because build side has been partitioned - a build side partition is either fully memory-resident or fully spilled -- it's the same for corresponding probe rows;  
   - (2)  It's not friendly to the case that the build side just slightly goes beyond RAM.  E.g., let's say, in memory hash table's size is H before OOM, build side total size is H + h, probe side size is P.  If P >> H >> h, meaning we just slightly go beyond RAM size, we have sort the entire probe side. 
   
   Given that it's not worse than what the existing HJ does - running slowly is better than an error, I'm fine with merging this new proposal :-)
   


-- 
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.

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

[GitHub] [spark] maropu commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826501739


   I have the same impression with @sigmod.


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854380297






-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824634548


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42311/
   


-- 
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.

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

[GitHub] [spark] cloud-fan commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826239123


   After more thinking, I'm wondering if this is the right direction to go. Apparently falling back to SMJ wastes the partially-built hash map.
   
   If one partition is a bit larger to build the in-memory hash map, I feel spilling the hash map might be a better choice? If one partition is much larger to build the in-memory hash map, seems we can use the same technique of skew join handling, to split the partition into multiple smaller ones so that they can fit in memory.


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854403695


   > > > Is 3.1 + 3.2 correct? Let's say, if the build side have two rows r1 and r2, with identical values on join key columns. r1 is processed before OOM and inserted into the in memory hash table, while r2 is processed after the OOM. Now a probe side row p, with the same join key column values can only match r1. p should match r2 as well for correctness.
   > > 
   > > 
   > > @sigmod - yes you are right. Sorry for missing it. Updated above. For non-semi join, we need to put the rows from stream side to sorter anyway, thanks.
   > 
   > Ok, it sounds like a working join algorithm that at least does not error out a query because of OOM.
   > 
   > But the proposal is less efficient compared to hybrid hash join in a few aspects:
   > 
   > * (1) HHJ doesn't have to spill all probe rows, because build side has been partitioned - a build side partition is either fully memory-resident or fully spilled -- it's the same for corresponding probe rows;
   > * (2)  It's not friendly to the case that the build side just slightly goes beyond RAM.  E.g., let's say, in memory hash table's size is H before OOM, build side total size is H + h, probe side size is P.  If P >> H >> h, meaning we just slightly go beyond RAM size, we have sort the entire probe side.
   > 
   > Given that it's not worse than what the existing HJ does - running slowly is better than an error, I'm fine with merging this new proposal :-)
   
   PS, there's some additional complexity to get left/full-outer joins done right (assuming probe side is on the left) -- for a LOJ/FOJ, it seems that you need to tag each probe side row with "matched" v.s. "not-matched-yet",  and then use this information in the customized merge join logic. 
   


-- 
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.

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

[GitHub] [spark] weixiuli commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-946323634


   @c21 What is the status of this PR ? Thanks.


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

[GitHub] [spark] weixiuli commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-953553547


   @c21 Ok, thanks.


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

[GitHub] [spark] maropu commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-869415873


   @c21 Sorry for my late response. Yea, the new proposal looks fine to me. Btw, one question; we don't need to flush the hash table to disk if OOM happens? Is it safe to keep processing rows while holding the hash table on memory?


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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824719660


   **[Test build #137773 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137773/testReport)** for PR 32210 at commit [`7c04c70`](https://github.com/apache/spark/commit/7c04c70dd8c080cf0064f81dd8abc355ebe708e8).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824747758


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137782/
   


-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619988615



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > is there any other faster ballback logic than the current approach.
   
   @maropu - yes, I am open and welcome any suggestion as well. Thanks for more insights from @sigmod and you @maropu . I will take a deeper look of hybrid join as in documentations you guys provided, and get back to you guys later.




-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-856203715


   > Are you suggesting to add something for matched-or-not info inside `SortMergeJoinExec.condition`?
   
   Nvm. It doesn't seem to work, as the matched rows may show up twice... I don't have a good idea to keep SortMergeJoinExec as it is.
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821777937


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42074/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821772292


   cc @cloud-fan and @maropu could you help take a look when you have time? Thanks.


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824569954


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42301/
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824574398


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42301/
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823604204


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42231/
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821804175


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137500/
   


-- 
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.

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

[GitHub] [spark] SparkQA removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824566953


   **[Test build #137782 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137782/testReport)** for PR 32210 at commit [`f3e3732`](https://github.com/apache/spark/commit/f3e3732749eaefdd4fa12889b54ecc81cb5b6773).


-- 
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.

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

[GitHub] [spark] AmplabJenkins removed a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824466969


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137752/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854356895


   Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   
   At the high level, the feature is:
   
   1. Build the hash table (`HashedRelation`) on build side.
   2. If step 1 succeeds, do shuffled hash join with stream side.
   3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
     3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
     3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. If having matched rows, output and process next row. If no matched rows, add the row from stream side to another sorter (`streamedSorter`).
     3.3. After processing all rows from streamed side, clean up hash table, and sort `buildSorter` and `streamedSorter` and do sort merge join.
   
   In this way, the effort of building hash table is not wasted compared to this PR, and we can control how to do the fallback. Three ways of fallback policy I can think of  - (1).default `SparkOutOfMemoryError` in `BytesToBytesMap`, (2).introduce configurable threshold for key capacity in `BytesToBytesMap` (max number of keys, `BytesToBytesMap.capacity`), (3).introduce configurable threshold for used page size in `BytesToBytesMap` (max size of keys+values, `BytesToBytesMap.used`).
   
   This solution mimics the idea of `hybrid join` in https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf, where it tried to build hash table, and spill rest of rows on disk, and join rest of rows on disk later.
   
   WDYT? @cloud-fan, @maropu and @sigmod, thanks.


-- 
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.

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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-854380297


   > Sorry for the late update. Went through https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf and https://www.db-book.com/db4/slide-dir/ch13.ppt, and I have a new proposal for this feature as followed.
   > 
   > At the high level, the feature is:
   > 
   > 1. Build the hash table (`HashedRelation`) on build side.
   > 2. If step 1 succeeds, do shuffled hash join with stream side.
   > 3. If step 1 fails (either due to OOM `SparkOutOfMemoryError`, or a configurable threshold, more details below)
   >    3.1. Add rest of rows from build side to a sorter (`UnsafeExternalRowSorter.insertRow`, e.g. the sorter called `buildSorter`).
   >    3.2. Process rows from streamed side. Probe hash table first to do shuffled hash join. If having matched rows, output and process next row. If no matched rows, add the row from stream side to another sorter (`streamedSorter`).
   
   Is 3.1 + 3.2 correct?  Let's say, if the build side have two rows `r1` and `r2`, with identical values on join key columns. `r1` is processed before OOM and inserted into the in memory hash table, while `r2` is processed after the OOM.  Now a probe side row `p`, with the same join key column values can only match `r1`.  `p` should match `r2` as well for correctness.  
   


-- 
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.

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

[GitHub] [spark] github-actions[bot] closed pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #32210:
URL: https://github.com/apache/spark/pull/32210


   


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

[GitHub] [spark] github-actions[bot] commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-937337154


   We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
   If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!


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

[GitHub] [spark] github-actions[bot] commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-937337154


   We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
   If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!


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

[GitHub] [spark] sigmod edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826500384


   > Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map. 
   > How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   `Hybrid hash join` was designed to address this issue: 
   - https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
   - https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf
    
   In short, it's not to spill an entire, huge hash map onto disk, but combines in-memory hash join with data partitioning/spilling. It's a standard algorithm that been implemented in many query engines. Similarly, you could implement hash aggregation in the same way, which should be more efficient than the runtime fallback approach too. 
   
   IMO, either (1) AQE or (2) conservative static QO decisions (e.g., planning shuffled hash join less aggressively) might be some low-hanging fruits, if they can address the issue. But if indeed neither could work for some of your queries, supporting spilling could be viable option to go. 
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826250121






-- 
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.

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

[GitHub] [spark] cloud-fan commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619300886



##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -385,6 +385,16 @@ object SQLConf {
     .booleanConf
     .createWithDefault(true)
 
+  val SHUFFLEDHASHJOIN_FALLBACK_ENABLED =
+    buildConf("spark.sql.join.enableShuffledHashJoinFallback")
+      .internal()
+      .doc("When true, enable sort-based fallback for shuffled hash join. " +
+        "The sort-based fallback is to use sort merge join when there is no enough memory " +
+        "to build hash table for shuffled hash join. This can help avoid task OOM automatically.")
+      .version("3.2.0")
+      .booleanConf
+      .createWithDefault(false)

Review comment:
       Seems a bit overkill to add a config for this. People can just flip the existing config to prefer SMJ, then SHJ will never be used.




-- 
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.

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

[GitHub] [spark] c21 edited a comment on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 edited a comment on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826493892


   After skew join handling, the output partitioning is destroyed, but this approach keeps output partitioning. [Skew join handling will not be enabled if it introduces extra shuffle in plan now](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala#L279). But I agree the change in AQE for skew join handling is more incremental and less intrusive. But as we see here, I don't see major intrusive API change here for this PR neither. I am just brainstorming the pros and cons, and I think we should pick the direction towards the eventual goal - enabling shuffled hash join by default.
   
   @cloud-fan - as you mentioned earlier, I agree with that (1). run-time sort-based fallback in shuffled hash join itself & (2). AQE skew join handling / hybrid join features, to be orthogonal with each other. AQE is great to cover a lot of cases, but as we all know it has some limitations (listed some above and here).


-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-826500384


   > Per my knowledge I don't know any obviously efficient way to do random lookup join with spilled hash map. 
   > How do we minimize random disk read for spilled map? Happy to brainstorm more if you have some rough ideas.
   
   `Hybrid hash join` was designed to address this issue: 
   - https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
   - https://cs-people.bu.edu/mathan/reading-groups/papers-classics/join.pdf
    
   In short, it's not to spill an entire, huge hash map onto disk, but combines in-memory hash join with data partitioning/spilling. It's a standard algorithm that been implemented in many query engines. Similarly, you could implement hash aggregation in a similar way, which should be more efficient than the runtime fallback approach too. 
   
   IMO, either (1) AQE or (2) conservative static QO decisions (e.g., planning shuffled hash join less aggressively) might be some low-hanging fruits, if they can address the issue. But if indeed neither could work for some of your queries, supporting spilling could be viable option to go. 
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824574339


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42301/
   


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-856154752


   @cloud-fan and @maropu - sorry to ping again. Wondering how do you think of above approach - https://github.com/apache/spark/pull/32210#issuecomment-854356895? Thanks.


-- 
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.

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

[GitHub] [spark] c21 commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-856186082


   > Option 1: pass an additional condition to check the tag field is "matched" v.s. "not-matched-yet";
   
   @sigmod - can you elaborate more how it can work? Are you suggesting to add something for matched-or-not info inside `SortMergeJoinExec.condition`?
   
   > Option 2: only support this kind of fallback for inner joins, initially.
   
   Yeah I am thinking about it as well, but we have to fix it anyway for outer joins. I am more towards to get consensus from everyone for all join types.
   
   Another option I am thinking of is, to maintain two sorters for stream/probe side when doing outer join. One sorter is for matched rows from hash join (`sorterForMatched`) and another sorter is for non-matched rows from hash join (`sorterForNonmatched`). This still needs some change inside SMJ to merge these two sorter together but it avoids adding matched info to sorter.
   


-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619981541



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       For runtime, yes. The total query run-time is dominated by the last finished task runtime. Just to point it out in case, without this change, this would be task and query failure.




-- 
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.

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

[GitHub] [spark] c21 commented on a change in pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #32210:
URL: https://github.com/apache/spark/pull/32210#discussion_r619976792



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       @maropu - yes, more data is in https://github.com/apache/spark/pull/32210#issuecomment-823503243 .
   
   > This can cause high performance penalties if the fallback happens only in a single task
   
   Wondering why it is the case?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       For runtime, yes. The total query run-time is dominated by the last finished task runtime. Just to point it out in case, without this change, this would be task and query failure.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoinExec.scala
##########
@@ -81,11 +83,22 @@ case class ShuffledHashJoinExec(
 
   protected override def doExecute(): RDD[InternalRow] = {
     val numOutputRows = longMetric("numOutputRows")
+    val spillThreshold = getSpillThreshold
+    val inMemoryThreshold = getInMemoryThreshold
+    val streamSortPlan = getStreamSortPlan
+    val buildSortPlan = getBuildSortPlan
+    val fallbackSMJPlan = SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, left, right)
+
     streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
-      val hashed = buildHashedRelation(buildIter)
-      joinType match {
-        case FullOuter => fullOuterJoin(streamIter, hashed, numOutputRows)
-        case _ => join(streamIter, hashed, numOutputRows)
+      buildHashedRelation(buildIter) match {
+        case r: UnfinishedUnsafeHashedRelation =>
+          joinWithSortFallback(streamIter, buildIter, r.destructiveValues(), streamSortPlan,

Review comment:
       > is there any other faster ballback logic than the current approach.
   
   @maropu - yes, I am open and welcome any suggestion as well. Thanks for more insights from @sigmod and you @maropu . I will take a deeper look of hybrid join as in documentations you guys provided, and get back to you guys later.

##########
File path: sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
##########
@@ -1394,4 +1394,32 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       checkAnswer(fullJoinDF, Row(100))
     }
   }
+
+  test("SPARK-32634: Sort-based fallback for shuffled hash join") {
+    val df1 = spark.range(300).map(_.toString).select($"value".as("k1"))
+    val df2 = spark.range(100).map(_.toString).select($"value".as("k2"))
+
+    val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2")
+    assert(collect(smjDF.queryExecution.executedPlan) {
+      case _: SortMergeJoinExec => true }.size === 1)
+    val smjResult = smjDF.collect()
+
+    Seq(
+      // All tasks fall back
+      0,
+      // Some tasks fall back
+      10,
+      // No task falls back
+      1000
+    ).foreach(fallbackStartsAt =>
+      withSQLConf(SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED.key -> "true",
+        "spark.sql.ShuffledHashJoin.testFallbackStartsAt" -> fallbackStartsAt.toString) {
+        val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+        assert(collect(shjDF.queryExecution.executedPlan) {
+          case _: ShuffledHashJoinExec => true }.size === 1)
+        // Same result between shuffled hash join and sort merge join
+        checkAnswer(shjDF, smjResult)

Review comment:
       Yes it is. I explicitly disable code-gen (in `ShuffledHashJoinExec.supportCodegen`) when `SQLConf.SHUFFLEDHASHJOIN_FALLBACK_ENABLED` is set to true in this PR.




-- 
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.

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

[GitHub] [spark] cloud-fan commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823357554


   I'm a bit worried about this solution:
   1. sorting the stream-side at runtime may lead to slow query plan because the sort is not whole-stage-codegen-ed.
   2. unlike SMJ, the output ordering can't be preserved if we sort the stream-side at runtime.
   
   I think the eventual goal is to enable shuffle hash join by default, but I'm not sure adding the fallback can achieve this goal. Do you have some real data to show the benefits?
   
   Another idea is to pick shuffle hash join in AQE when we know the per-partition size after shuffle.


-- 
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.

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

[GitHub] [spark] cloud-fan commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-823919561


   > We enabled shuffled hash join by default with this feature. In our environment, roughly 25% of sort merge join queries are now running with shuffled hash join after enabling shuffled hash join by default. Roughly 20% of shuffled hash join queries has tasks fall-backed. Roughly 10% of shuffled hash join tasks has fall-backed. TLDR is it saves 20% of shuffled hash join queries failure, and 10% of tasks failure.
   
   This looks great, let's add the fallback then. Picking SHJ in AQE is orthogonal to it and we can have both.


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-824634548


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42311/
   


-- 
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.

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

[GitHub] [spark] AmplabJenkins commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821777937


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42074/
   


-- 
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.

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

[GitHub] [spark] sigmod commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
sigmod commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-856160864


   > > PS, there's some additional complexity to get left/full-outer joins done right (assuming probe side is on the left) -- for a LOJ/FOJ, it seems that you need to tag each probe side row with "matched" v.s. "not-matched-yet", and then use this information in the customized merge join logic.
   > 
   > @sigmod - this is valid point. For outer join, current SMJ cannot work out of box, need to push the matching info into sorter as well. This does add complexity for SMJ.
   
   There seem two ways to keep SMJ unchanged:
   - Option 1:  pass an additional `condition` to check the tag field is "matched" v.s. "not-matched-yet";
   - Option 2:  only support this kind of fallback for inner joins, initially.
   


-- 
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.

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

[GitHub] [spark] SparkQA commented on pull request #32210: [SPARK-32634][SQL] Introduce sort-based fallback for shuffled hash join (non-code-gen path)

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32210:
URL: https://github.com/apache/spark/pull/32210#issuecomment-821776040






-- 
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.

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