[SPARK-34089][CORE] HybridRowQueue should respect the configured memory mode

[Ngone51]

What changes were proposed in this pull request?

This PR fixes the HybridRowQueue to respect the configured memory mode.

Besides, this PR also refactored the constructor of MemoryConsumer to accept the memory mode explicitly.

Why are the changes needed?

HybridRowQueue supports both onHeap and offHeap manipulation. But it inherited the wrong MemoryConsumer constructor, which hard-coded the memory mode to onHeap.

Does this PR introduce any user-facing change?

No. (Maybe yes in some cases where users can't complete the job before could complete successfully after the fix because of HybridRowQueue is able to spill under offHeap mode now. )

How was this patch tested?

Updated the existing test to make it test both offHeap and onHeap modes.

[Ngone51]

cc @tgravescs @mridulm @cloud-fan Please take a look, thanks!

[SparkQA]

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

[tgravescs]

so I'm a little confused by this. so I went back to the history. This seems to be used by Spillable, HashRelation, and RowQueue. It looks like at the time those were added, the MemoryConsumer didn't support a memory mode.
It looks like when that memory mode was added spilling was only support to on heap by Spillable (looking at the comment ing): 8fb1d1c
Is that not the case now?
Looking at:
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/collection/Spillable.scala#L111
it seems like its only on heap.

[SparkQA]

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

[SparkQA]

Test build #133972 has finished for PR 31152 at commit 0a3743b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

@tgravescs thanks for providing the context! I looked at the source code and I found one caller of this particular MemoryConsumer constructor, HybridRowQueue, actually supports off-heap.

For MemoryConsumer implementations, I think they are responsible to set the memory mode correctly. We should fix Spillable as it only supports on-heap.

[Ngone51]

It looks like when that memory mode was added spilling was only support to on heap by Spillable

hmm...but seems like Spillable can still spill (by maybeSpill()) under off-heap mode if the spill is triggered by itself.

spark/core/src/main/scala/org/apache/spark/util/collection/Spillable.scala

Line 82 in 3a299aa

protected def maybeSpill(collection: C, currentMemory: Long): Boolean = {

[Ngone51]

kindly ping @tgravescs

[cloud-fan]

hmm...but seems like Spillable can still spill (by maybeSpill()) under off-heap mode if the spill is triggered by itself.

All implementations of Spillable use java collection to store data, so they can't be off-heap.

[tgravescs]

yeah it doesn't make sense to spill if you need off heap and all you have is on heap or vice versa.

[Ngone51]

I see. Thanks for the explanation.

[Ngone51]

Hi @tgravescs @cloud-fan Sorry for the late here. I've updated the PR according to our discussion. To summarize, the latest PR only fixes HybridRowQueue to make it respect the configured memory mode instead of the hard-coded onHeap mode since it supports both modes.

Besides, I've also checked all the other MemoryConsumers which seems correct to me. Please take another look when you get a chance, thanks!

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #136254 has finished for PR 31152 at commit 9c29fbc.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

retest this please

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #136258 has finished for PR 31152 at commit 9c29fbc.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

GA passed, merging to master, thanks!