Add cpu oom retry split handling to InternalRowToColumnarBatchIterator #10011
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…rator Signed-off-by: Jim Brennan <jimb@nvidia.com>
jbrennan333
added
feature request
|
build |
abellina
reviewed
Dec 11, 2023
| @@ -144,7 +144,7 @@ private class HostAlloc(nonPinnedLimit: Long) extends HostMemoryAllocator with L | |||
| logDebug(s"Targeting host store size of $targetSize bytes") | |||
| // We could not make it work so try and spill enough to make it work | |||
| val maybeAmountSpilled = | |||
| RapidsBufferCatalog.synchronousSpill(RapidsBufferCatalog.getHostStorage, allocSize) | |||
| RapidsBufferCatalog.synchronousSpill(RapidsBufferCatalog.getHostStorage, targetSize) | |||
abellina
reviewed
Dec 11, 2023
| SpillPriorities$.MODULE$.ACTIVE_ON_DECK_PRIORITY(), | ||
| RapidsBufferCatalog$.MODULE$.singleton()); | ||
| hBufs[0] = null; // Was closed by spillable | ||
| hBufs[1] = HostAlloc$.MODULE$.alloc(offsetBytes, true); |
There was a problem hiding this comment.
do we need to wrap this in a try/catch so we can close sBufs[0] if alloc throws?
There was a problem hiding this comment.
oh nm, I think allocBuffersWithRestore takes care of it.
There was a problem hiding this comment.
I think there may be a problem here actually. If we throw with something other than a retry exception, the withRestoreOnRetry won't handle it.
There was a problem hiding this comment.
yeah that would be an issue
Comment on lines
95
to
96
| return (int) Math.max(1, | ||
| Math.min(Integer.MAX_VALUE - 1, targetBytes / sizePerRowEstimate)); |
There was a problem hiding this comment.
nit: casting is forced by targetBytes / sizePerRowEstimate? I'd prefer pushing it down to the origin:
Suggested change
| return (int) Math.max(1, | |
| Math.min(Integer.MAX_VALUE - 1, targetBytes / sizePerRowEstimate)); | |
| return Math.max(1, | |
| Math.min(Integer.MAX_VALUE - 1, (int) (targetBytes / sizePerRowEstimate))); |
sql-plugin/src/main/java/com/nvidia/spark/rapids/InternalRowToColumnarBatchIterator.java
Show resolved
Hide resolved
|
Thanks for the review comments @abellina and @gerashegalov! I believe I have addressed them in this latest commit. |
|
build |
abellina
approved these changes
Dec 12, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This is a followup to #9929 and is part of addressing issue #8887.
This adds handing for
CpuSplitAndRetryOOMinInternalRowToColumnarBatchIterator.For splits, we divide the target number of rows in half via
splitTargetSizeInHalfCpu, and recalculate the buffer sizes from that. It will try that until we hit 1 row.I also included a workaround for #10004, by combining the
withHostBufferWriteLockwith thewithHostBufferReadOnlyblocks. In this case we will hold the write lock during the full operation, but this should allow us to proceed without introducing data corruption until we have a fix for #10004.One other fix is in HostAlloc, where it was passing the wrong argument to
RapidsBufferCatalog.synchronousSpill- it was passingallocSizeinstead oftargetSize, causing us to spill a lot more than needed in some cases.