fix: [branch-1.0] read shuffle write buffer, spill limit and off-heap sizes in bytes (#6191) - #6209
Merged
Conversation
…es (apache#6191) * fix: read shuffle write buffer, spill limit and off-heap sizes in bytes spark.comet.shuffle.native.writeBufferSize was declared in MiB but its value was sent to native code as a byte count, so the native shuffle writer ran with a 1-byte write buffer by default. Declare it in bytes with a 1 MiB default; a bare number keeps its meaning. Native code parsed spark.comet.maxTempDirectorySize and the boolean flags it reads from the raw session strings, so a size with a unit or an upper-case boolean silently fell back to the native default. Resolve every config that native code reads on the JVM before it crosses JNI, and document that the spill limit applies per native plan rather than per task. getMemoryConfig read spark.memory.offHeap.size as MiB, while Spark reads a bare number as bytes. Closes apache#6183, apache#6184, apache#6185. * docs: give a real example of a task running several native plans A Spark operator that Comet does not support does not split a stage into two native plans: Comet does not resume native execution above it. The native operators on either side of a union or a coalesce do run as separate plans in the same task. * Update spark/src/main/scala/org/apache/comet/CometConf.scala Co-authored-by: Matt Butrovich <mbutrovich@users.noreply.github.com> --------- Co-authored-by: Matt Butrovich <mbutrovich@users.noreply.github.com> (cherry picked from commit 02e8404) Adapted for branch-1.0: - CI suite lists: added only CometNativeShuffleWriterSuite. The CometNativePositionalRoundRobinSuite and CometDiskBlockWriterSuite neighbours on main come from later commits and do not exist here. - The getMemoryConfig test moved from CometExecIteratorLifecycleSuite, which does not exist on branch-1.0 (it comes from apache#5494), into CometExecSuite next to the PR's other CometExecIterator test. Its body is unchanged. - CometNativeShuffleWriter.buildUnifiedPlan widened from private to private[shuffle], the same change apache#5513 made on main, so the new suite can call it. It still takes separate data and index paths here, so the suite passes both. - Removed the ByteUnit import from CometExecIterator, which the offHeap.size fix leaves unused on branch-1.0. - Updated the spark.comet.maxTempDirectorySize row of the frozen configs.md. branch-1.0 commits the generated config reference; main generates it at publish time.
This file contains hidden or 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
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.
Backport of #6191 to
branch-1.0.Cherry-picked from
02e84047a423a08debe3c6b6daab19fd9233427c. The fixes themselves are unchanged. Five adaptations were needed, described under "What changes are included" below. Two of them touch production code, and neither changes behaviour: a visibility widening and an unused import.Which issue does this PR close?
Closes #6183, #6184 and #6185 on
branch-1.0. Listed in #6201.Rationale for this change
All three bugs ship in 1.0.0, through the same code as on
mainbefore #6191:spark.comet.shuffle.native.writeBufferSizeis declared in MiB but sent to native code as a byte count, so the native shuffle writer runs with a 1-byte write buffer by default. Onbranch-1.0the setting sizes the multi-partition data file'sBufWriterand the flush threshold of everyBufBatchWriter, so each encoded block goes straight to its own write call, for shuffle and spill files alike.spark.comet.maxTempDirectorySizereaches native code as the raw string, and native code parses only a bare integer, so a value with a unit silently falls back to 100 GiB. The boolean flags native code reads have the same problem with anything other than lowercasetrue. Native code onbranch-1.0reads the same configs as onmain.spark.memory.offHeap.sizeis read as MiB when sizing the memory pool, while Spark reads a bare number as bytes, so a bare byte count makes thefair_unifiedper-task cap effectively unlimited.What changes are included in this PR?
The fix is the original one, so see #6191 for the details. The adaptations:
pr_build_linux.ymlandpr_build_macos.yml: added onlyCometNativeShuffleWriterSuite. Its neighbours onmain,CometNativePositionalRoundRobinSuiteandCometDiskBlockWriterSuite, come from later commits and do not exist here.getMemoryConfigtest lives inCometExecSuite, next to this PR's otherCometExecIteratortest, instead of inCometExecIteratorLifecycleSuite. That suite comes from fix: make task-shared memory pool as ref-counted RAII guard #5494, which is not onbranch-1.0. Creating it for one test would mean another suite to register, and an add/add conflict for any later backport of fix: make task-shared memory pool as ref-counted RAII guard #5494. The test body is unchanged, andCometExecSuitegains aSparkConfimport.CometNativeShuffleWriter.buildUnifiedPlanis widened fromprivatetoprivate[shuffle]so the new suite can call it, the same change feat: complete Celeborn map-side shuffle push lifecycle (6/n) #5513 made onmain. Onbranch-1.0it still takes separate data and index paths (perf: spill every shuffle partition of a task into one file #5916 later dropped the index path onmain), so the suite passes both.CometExecIterator.scaladrops itsByteUnitimport. Thespark.memory.offHeap.sizefix removes its last use onbranch-1.0, and the scalafixRemoveUnusedcheck in the Lint Java job rejects unused imports. Onmain, feat: always count native allocations and log executor native memory usage #6162 still uses it.spark.comet.maxTempDirectorySizerow ofdocs/source/user-guide/latest/configs.mdnow has the new description.branch-1.0commits the generated config reference, whilemaingenerates it at publish time, so fix: read shuffle write buffer, spill limit and off-heap sizes in bytes #6191 had no copy to update. The file now matchesGenerateDocsoutput exactly. ThewriteBufferSizerow is unchanged, because the old MiB default already rendered as1048576b.How are these changes tested?
Same tests as the original PR, verified locally on
branch-1.0with Spark 4.1.3, Scala 2.13 and JDK 17 unless noted:CometNativeShuffleWriterSuiteandCometExecSuitepass, 146 tests including the three new ones.branch-1.0, and the tests catch them. With theCometConf.scalaandCometExecIterator.scalachanges reverted and the tests kept, exactly the three new tests fail and the other 143 pass. The plan sent to native code carries a write buffer of1rather than1048576, the serialized configs have nomaxTempDirectorySizeentry, and a bare4294967296off-heap size gives a memory limit of4503599627370496, which is 4 PiB.branch-1.0CI runs: 143 passed, and 3 were cancelled by existing Spark version guards.CometNativeShuffleSuite,CometShuffleSuiteandCometShuffleEncryptionSuitepass, 73 tests, now that the native shuffle writer uses a 1 MiB buffer instead of 1 byte.apache-rat:check,cargo fmt --all -- --check,cargo clippy --all-targets --workspace -- -D warnings,dev/ci/check-suites.py,dev/ci/check-ci-config.py,actionlint, andprettier --checkon the two changed docs all pass.Are there any user-facing changes?
Yes, the same ones as #6191, which are worth weighing for a patch release:
branch-1.0each shuffle-writing task holds up to about 2 MiB of buffers that no memory pool tracks: the data file'sBufWriterand oneBufBatchWriterstaging buffer at a time. That is about half of the 4 MiB onmain, where perf: reuse per-partition scratch in the shuffle write path #5568 and perf: spill every shuffle partition of a task into one file #5916 added buffers of the same size.writeBufferSizevalues with a unit now mean what they say, and a bare number keeps its meaning. A value of 2 GiB or more, which the nativeint32field cannot hold, now fails with a config error. Previously4g, for example, became a 4096-byte buffer.maxTempDirectorySizevalues with a unit, and upper-case booleans for the flags native code reads, now take effect. An invalid value for any of them now fails with the usual config error instead of silently becoming the native default.spark.memory.offHeap.sizenow sizes the Comet memory pool in bytes, as Spark does.