You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zml1206 (via GitHub)" <gi...@apache.org> on 2024/01/10 05:09:37 UTC
[PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output line by line [spark]
zml1206 opened a new pull request, #44660:
URL: https://github.com/apache/spark/pull/44660
<!--
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'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
Be consistent with closing code-gen, update code-gen for full outer sort merge join output line by line.
When code-gen for full outer sort merge join is true and the parent of SortMergeJoin cannot codegen, full outer sort merge join needs to append the output to this LinkedList. If there are a large number of duplicate keys, tt is likely to cause executor oom.
### Why are the changes needed?
Avoid oom when there are a large number of duplicate keys and the parent of full outer sort merge join cannot code-gen.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing UT.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "wankunde (via GitHub)" <gi...@apache.org>.
wankunde commented on code in PR #44660:
URL: https://github.com/apache/spark/pull/44660#discussion_r1464466836
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala:
##########
@@ -959,38 +963,53 @@ case class SortMergeJoinExec(
// Scan the left and right buffers to find all matched rows.
val matchRowsInBuffer =
s"""
- |int $leftIndex;
- |int $rightIndex;
- |
- |for ($leftIndex = 0; $leftIndex < $leftBuffer.size(); $leftIndex++) {
+ |while ($leftIndex < $leftBuffer.size()) {
| $leftOutputRow = (InternalRow) $leftBuffer.get($leftIndex);
- | for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ | while ($rightIndex < $rightBuffer.size()) {
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $conditionCheck {
| $consumeFullOuterJoinRow();
| $leftMatched.set($leftIndex);
| $rightMatched.set($rightIndex);
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
| }
- |
+ | $rightIndex = 0;
| if (!$leftMatched.get($leftIndex)) {
|
| $rightOutputRow = null;
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $leftIndex++;
+ | return;
+ | }
| }
+ | $leftIndex++;
|}
|
|$leftOutputRow = null;
- |for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ |while ($rightIndex < $rightBuffer.size()) {
| if (!$rightMatched.get($rightIndex)) {
| // The right row has never matched any left row, join it with null row
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
|}
""".stripMargin
s"""
+ |if ($leftIndex < $leftBuffer.size() || $rightIndex < $rightBuffer.size()) {
+ | $matchRowsInBuffer
+ |}
Review Comment:
Could we wrap "$matchRowsInBuffer" as a function? There are too many duplicated code.
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on code in PR #44660:
URL: https://github.com/apache/spark/pull/44660#discussion_r1464928544
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala:
##########
@@ -959,38 +963,53 @@ case class SortMergeJoinExec(
// Scan the left and right buffers to find all matched rows.
val matchRowsInBuffer =
s"""
- |int $leftIndex;
- |int $rightIndex;
- |
- |for ($leftIndex = 0; $leftIndex < $leftBuffer.size(); $leftIndex++) {
+ |while ($leftIndex < $leftBuffer.size()) {
| $leftOutputRow = (InternalRow) $leftBuffer.get($leftIndex);
- | for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ | while ($rightIndex < $rightBuffer.size()) {
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $conditionCheck {
| $consumeFullOuterJoinRow();
| $leftMatched.set($leftIndex);
| $rightMatched.set($rightIndex);
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
| }
- |
+ | $rightIndex = 0;
| if (!$leftMatched.get($leftIndex)) {
|
| $rightOutputRow = null;
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $leftIndex++;
+ | return;
+ | }
| }
+ | $leftIndex++;
|}
|
|$leftOutputRow = null;
- |for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ |while ($rightIndex < $rightBuffer.size()) {
| if (!$rightMatched.get($rightIndex)) {
| // The right row has never matched any left row, join it with null row
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
|}
""".stripMargin
s"""
+ |if ($leftIndex < $leftBuffer.size() || $rightIndex < $rightBuffer.size()) {
+ | $matchRowsInBuffer
+ |}
Review Comment:
Done, generated code in PR description also updated.
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output line by line [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on code in PR #44660:
URL: https://github.com/apache/spark/pull/44660#discussion_r1450325840
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala:
##########
@@ -959,38 +963,53 @@ case class SortMergeJoinExec(
// Scan the left and right buffers to find all matched rows.
val matchRowsInBuffer =
s"""
- |int $leftIndex;
- |int $rightIndex;
- |
- |for ($leftIndex = 0; $leftIndex < $leftBuffer.size(); $leftIndex++) {
+ |while ($leftIndex < $leftBuffer.size()) {
| $leftOutputRow = (InternalRow) $leftBuffer.get($leftIndex);
- | for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ | while ($rightIndex < $rightBuffer.size()) {
Review Comment:
Convenient control, go to +1 in different places.
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on PR #44660:
URL: https://github.com/apache/spark/pull/44660#issuecomment-1909253755
Whenever `processNext` is called, the buffer will be consumed first, and then `findNextJoinRows` will be called to reset the index and buffer and match the new row written into the buffer. @wankunde
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output line by line [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #44660:
URL: https://github.com/apache/spark/pull/44660#discussion_r1450288452
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala:
##########
@@ -959,38 +963,53 @@ case class SortMergeJoinExec(
// Scan the left and right buffers to find all matched rows.
val matchRowsInBuffer =
s"""
- |int $leftIndex;
- |int $rightIndex;
- |
- |for ($leftIndex = 0; $leftIndex < $leftBuffer.size(); $leftIndex++) {
+ |while ($leftIndex < $leftBuffer.size()) {
| $leftOutputRow = (InternalRow) $leftBuffer.get($leftIndex);
- | for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ | while ($rightIndex < $rightBuffer.size()) {
Review Comment:
Why do you need replace for with while?
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on PR #44660:
URL: https://github.com/apache/spark/pull/44660#issuecomment-1905722032
cc @cloud-fan @wankunde @ulysses-you do you have any thought about this? 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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output line by line [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on PR #44660:
URL: https://github.com/apache/spark/pull/44660#issuecomment-1888305778
@cloud-fan Can you help take a look if 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.
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "wankunde (via GitHub)" <gi...@apache.org>.
wankunde commented on PR #44660:
URL: https://github.com/apache/spark/pull/44660#issuecomment-1909242540
I'm sorry where is smj_leftIndex_0 reset to 0 ?
https://gist.github.com/zml1206/a27350b8849951e6efac0fb6088e527f#file-full_outer_sort_merge_join_codegen_after-L280-L305
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on PR #44660:
URL: https://github.com/apache/spark/pull/44660#issuecomment-1909246054
> I'm sorry where is smj_leftIndex_0 reset to 0 ?
>
> https://gist.github.com/zml1206/a27350b8849951e6efac0fb6088e527f#file-full_outer_sort_merge_join_codegen_after-L280-L305
https://gist.github.com/zml1206/a27350b8849951e6efac0fb6088e527f#file-full_outer_sort_merge_join_codegen_after-L126
--
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
Re: [PR] [SPARK-46653][SQL] Code-gen for full outer sort merge join output row by row [spark]
Posted by "zml1206 (via GitHub)" <gi...@apache.org>.
zml1206 commented on code in PR #44660:
URL: https://github.com/apache/spark/pull/44660#discussion_r1464928544
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala:
##########
@@ -959,38 +963,53 @@ case class SortMergeJoinExec(
// Scan the left and right buffers to find all matched rows.
val matchRowsInBuffer =
s"""
- |int $leftIndex;
- |int $rightIndex;
- |
- |for ($leftIndex = 0; $leftIndex < $leftBuffer.size(); $leftIndex++) {
+ |while ($leftIndex < $leftBuffer.size()) {
| $leftOutputRow = (InternalRow) $leftBuffer.get($leftIndex);
- | for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ | while ($rightIndex < $rightBuffer.size()) {
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $conditionCheck {
| $consumeFullOuterJoinRow();
| $leftMatched.set($leftIndex);
| $rightMatched.set($rightIndex);
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
| }
- |
+ | $rightIndex = 0;
| if (!$leftMatched.get($leftIndex)) {
|
| $rightOutputRow = null;
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $leftIndex++;
+ | return;
+ | }
| }
+ | $leftIndex++;
|}
|
|$leftOutputRow = null;
- |for ($rightIndex = 0; $rightIndex < $rightBuffer.size(); $rightIndex++) {
+ |while ($rightIndex < $rightBuffer.size()) {
| if (!$rightMatched.get($rightIndex)) {
| // The right row has never matched any left row, join it with null row
| $rightOutputRow = (InternalRow) $rightBuffer.get($rightIndex);
| $consumeFullOuterJoinRow();
+ | if (shouldStop()) {
+ | $rightIndex++;
+ | return;
+ | }
| }
+ | $rightIndex++;
|}
""".stripMargin
s"""
+ |if ($leftIndex < $leftBuffer.size() || $rightIndex < $rightBuffer.size()) {
+ | $matchRowsInBuffer
+ |}
Review Comment:
Done, generated code of PR description also updated.
--
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