You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "mbutrovich (via GitHub)" <gi...@apache.org> on 2025/03/07 21:33:02 UTC
[PR] fix: Only instantiate CometTestShuffleMemoryAllocator in testing [datafusion-comet]
mbutrovich opened a new pull request, #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #1438.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
We only want the test memory allocator in tests.
## What changes are included in this PR?
Add check that Comet is in a testing environment when instantiating an allocator and an assertion in the test memory allocator itself.
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
Existing tests.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Rename CometTestShuffleMemoryAllocator since it can be used outside of tests [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1991990566
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -35,43 +35,47 @@
* store serialized rows. This class is simply an implementation of `MemoryConsumer` that delegates
* memory allocation to the `TaskMemoryManager`. This requires that the `TaskMemoryManager` is
* configured with `MemoryMode.OFF_HEAP`, i.e. it is using off-heap memory.
+ *
+ * <p>If the user does not enable off-heap memory then we want to use
+ * CometBoundedShuffleMemoryAllocator. The tests also need to default to using this because off-heap
+ * is not enabled when running the Spark SQL tests. The
+ * COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR flag allows us to use unified memory management
+ * in Comet tests (this does assume that off-heap is enabled).
*/
public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
private static CometShuffleMemoryAllocatorTrait INSTANCE;
/**
* Returns the singleton instance of `CometShuffleMemoryAllocator`. This method should be used
* instead of the constructor to ensure that only one instance of `CometShuffleMemoryAllocator` is
- * created. For Spark tests, this returns `CometTestShuffleMemoryAllocator` which is a test-only
- * allocator that should not be used in production.
+ * created. For Spark tests, this returns `CometBoundedShuffleMemoryAllocator`.
*/
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
- (boolean)
- CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
+ boolean useBoundedAllocator =
+ (boolean) CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useBoundedAllocator) {
Review Comment:
I wonder if renaming `CometShuffleMemoryAllocator` to `CometShuffleUnifiedMemoryAllocator` would help reduce confusion?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Rename CometTestShuffleMemoryAllocator since it can be used outside of tests [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1991982544
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -35,43 +35,47 @@
* store serialized rows. This class is simply an implementation of `MemoryConsumer` that delegates
* memory allocation to the `TaskMemoryManager`. This requires that the `TaskMemoryManager` is
* configured with `MemoryMode.OFF_HEAP`, i.e. it is using off-heap memory.
+ *
+ * <p>If the user does not enable off-heap memory then we want to use
+ * CometBoundedShuffleMemoryAllocator. The tests also need to default to using this because off-heap
+ * is not enabled when running the Spark SQL tests. The
+ * COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR flag allows us to use unified memory management
+ * in Comet tests (this does assume that off-heap is enabled).
*/
public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
private static CometShuffleMemoryAllocatorTrait INSTANCE;
/**
* Returns the singleton instance of `CometShuffleMemoryAllocator`. This method should be used
* instead of the constructor to ensure that only one instance of `CometShuffleMemoryAllocator` is
- * created. For Spark tests, this returns `CometTestShuffleMemoryAllocator` which is a test-only
- * allocator that should not be used in production.
+ * created. For Spark tests, this returns `CometBoundedShuffleMemoryAllocator`.
*/
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
- (boolean)
- CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
+ boolean useBoundedAllocator =
+ (boolean) CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useBoundedAllocator) {
Review Comment:
Also, if we run in on-heap mode, this condition doesn't get triggered so we end up using the unified memory allocator, which we don't want
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Only instantiate CometTestShuffleMemoryAllocator in testing [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1985745025
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometTestShuffleMemoryAllocator.java:
##########
@@ -67,11 +68,10 @@ public final class CometTestShuffleMemoryAllocator extends CometShuffleMemoryAll
private static final int OFFSET_BITS = 51;
private static final long MASK_LONG_LOWER_51_BITS = 0x7FFFFFFFFFFFFL;
- private static CometTestShuffleMemoryAllocator INSTANCE;
Review Comment:
The `isSparkTesting` variable in the other file made me glance at the unused warnings in the files I was touching, and this isn't 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.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Only instantiate CometTestShuffleMemoryAllocator in testing [datafusion-comet]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#issuecomment-2707668585
## [Codecov](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `0%` with `2 lines` in your changes missing coverage. Please review.
> Project coverage is 57.79%. Comparing base [(`f09f8af`)](https://app.codecov.io/gh/apache/datafusion-comet/commit/f09f8af64c6599255e116a376f4f008f2fd63b43?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`7a1820b`)](https://app.codecov.io/gh/apache/datafusion-comet/commit/7a1820b4ab42b9131bb6e41fefc58e70a8e9ac0c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 64 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ark/shuffle/comet/CometShuffleMemoryAllocator.java](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fspark%2Fshuffle%2Fcomet%2FCometShuffleMemoryAllocator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NwYXJrL3NodWZmbGUvY29tZXQvQ29tZXRTaHVmZmxlTWVtb3J5QWxsb2NhdG9yLmphdmE=) | 0.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...shuffle/comet/CometTestShuffleMemoryAllocator.java](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?src=pr&el=tree&filepath=spark%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fspark%2Fshuffle%2Fcomet%2FCometTestShuffleMemoryAllocator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3Bhcmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NwYXJrL3NodWZmbGUvY29tZXQvQ29tZXRUZXN0U2h1ZmZsZU1lbW9yeUFsbG9jYXRvci5qYXZh) | 0.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #1485 +/- ##
============================================
+ Coverage 56.12% 57.79% +1.66%
- Complexity 976 985 +9
============================================
Files 119 122 +3
Lines 11743 12204 +461
Branches 2251 2304 +53
============================================
+ Hits 6591 7053 +462
+ Misses 4012 3966 -46
- Partials 1140 1185 +45
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/datafusion-comet/pull/1485?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary>🚀 New features to boost your workflow: </summary>
- ❄ [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1989931497
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometBoundedShuffleMemoryAllocator.java:
##########
@@ -48,7 +48,7 @@
* Thus, this allocator is used to allocate separate off-heap memory allocation for Comet JVM
* shuffle and execution apart from Spark's on-heap memory configuration.
*/
-public final class CometTestShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
+public final class CometBoundedShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
Review Comment:
I like the new name 👍
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993585782
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -101,14 +101,16 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
* unified memory manager.
*/
private def shouldOverrideMemoryConf(conf: SparkConf): Boolean = {
- conf.getBoolean(CometConf.COMET_ENABLED.key, true) && (
- conf.getBoolean(
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.key,
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.defaultValue.get) ||
- conf.getBoolean(
- CometConf.COMET_EXEC_ENABLED.key,
- CometConf.COMET_EXEC_ENABLED.defaultValue.get)
- ) && CometSparkSessionExtensions.cometUnifiedMemoryManagerEnabled(conf)
+ val comet_enabled = conf.getBoolean(CometConf.COMET_ENABLED.key, true)
Review Comment:
This code was recently updated in main to remove the hard-coded default:
```suggestion
val comet_enabled = conf.getBoolean(CometConf.COMET_ENABLED.key, CometConf.COMET_ENABLED.defaultValue.get)
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1989967146
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometBoundedShuffleMemoryAllocator.java:
##########
@@ -48,7 +48,7 @@
* Thus, this allocator is used to allocate separate off-heap memory allocation for Comet JVM
* shuffle and execution apart from Spark's on-heap memory configuration.
*/
-public final class CometTestShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
+public final class CometBoundedShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
Review Comment:
Could you also rename `CometTestMemoryAllocatorShuffleSuite` to match
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1988084656
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -48,30 +48,30 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
+ boolean useTestAllocator =
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useTestAllocator) {
synchronized (CometShuffleMemoryAllocator.class) {
if (INSTANCE == null) {
// CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
}
}
return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ }
- // CometShuffleMemoryAllocator stores pages in TaskMemoryManager which is not singleton,
- // but one instance per task. So we need to create a new instance for each task.
- return new CometShuffleMemoryAllocator(taskMemoryManager, pageSize);
+ if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
Review Comment:
Oh, the original implementation was renamed to `CometTestShuffleMemoryAllocator`. So maybe we need to rename this.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993587130
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -101,14 +101,16 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
* unified memory manager.
*/
private def shouldOverrideMemoryConf(conf: SparkConf): Boolean = {
- conf.getBoolean(CometConf.COMET_ENABLED.key, true) && (
- conf.getBoolean(
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.key,
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.defaultValue.get) ||
- conf.getBoolean(
- CometConf.COMET_EXEC_ENABLED.key,
- CometConf.COMET_EXEC_ENABLED.defaultValue.get)
- ) && CometSparkSessionExtensions.cometUnifiedMemoryManagerEnabled(conf)
+ val comet_enabled = conf.getBoolean(CometConf.COMET_ENABLED.key, true)
+ val comet_exec_shuffle = conf.getBoolean(
Review Comment:
variable names should be camel 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.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1988082049
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -48,30 +48,30 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
+ boolean useTestAllocator =
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useTestAllocator) {
synchronized (CometShuffleMemoryAllocator.class) {
if (INSTANCE == null) {
// CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
}
}
return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ }
- // CometShuffleMemoryAllocator stores pages in TaskMemoryManager which is not singleton,
- // but one instance per task. So we need to create a new instance for each task.
- return new CometShuffleMemoryAllocator(taskMemoryManager, pageSize);
+ if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
Review Comment:
In Comet 0.3.0 we had a different implementation of CometShuffleMemoryAllocator that managed its own memory separately from Spark:
```
/**
* A simple memory allocator used by `CometShuffleExternalSorter` to allocate memory blocks which
* store serialized rows. We don't rely on Spark memory allocator because we need to allocate
* off-heap memory no matter memory mode is on-heap or off-heap. This allocator is configured with
* fixed size of memory, and it will throw `SparkOutOfMemoryError` if the memory is not enough.
*
* <p>Some methods are copied from `org.apache.spark.unsafe.memory.TaskMemoryManager` with
* modifications. Most modifications are to remove the dependency on the configured memory mode.
*/
public final class CometShuffleMemoryAllocator extends MemoryConsumer {
```
In Comet 0.4.0 we forced use of off-heap memory and in Comet 0.5.0 we reverted that decision but inadvertently started using the test version of the allocator.
Perhaps we must re-instate this earlier version to support the on-heap use case. I am not sure, though. I wonder if @viirya can make any recommendation on this.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993726889
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -19,91 +19,38 @@
package org.apache.spark.shuffle.comet;
-import java.io.IOException;
-
import org.apache.spark.SparkConf;
-import org.apache.spark.memory.MemoryConsumer;
-import org.apache.spark.memory.MemoryMode;
import org.apache.spark.memory.TaskMemoryManager;
-import org.apache.spark.unsafe.memory.MemoryBlock;
-import org.apache.spark.util.Utils;
-import org.apache.comet.CometConf$;
+import org.apache.comet.CometSparkSessionExtensions;
/**
- * A simple memory allocator used by `CometShuffleExternalSorter` to allocate memory blocks which
- * store serialized rows. This class is simply an implementation of `MemoryConsumer` that delegates
- * memory allocation to the `TaskMemoryManager`. This requires that the `TaskMemoryManager` is
- * configured with `MemoryMode.OFF_HEAP`, i.e. it is using off-heap memory.
+ * An interface to instantiate either CometBoundedShuffleMemoryAllocator (on-heap mode) or
+ * CometUnifiedShuffleMemoryAllocator (off-heap mode).
*/
-public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
+public final class CometShuffleMemoryAllocator {
private static CometShuffleMemoryAllocatorTrait INSTANCE;
/**
* Returns the singleton instance of `CometShuffleMemoryAllocator`. This method should be used
* instead of the constructor to ensure that only one instance of `CometShuffleMemoryAllocator` is
- * created. For Spark tests, this returns `CometTestShuffleMemoryAllocator` which is a test-only
- * allocator that should not be used in production.
+ * created. For on-heap mode (Spark tests), this returns `CometBoundedShuffleMemoryAllocator`.
*/
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
- boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
- (boolean)
- CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
-
- if (!useUnifiedMemAllocator) {
- synchronized (CometShuffleMemoryAllocator.class) {
- if (INSTANCE == null) {
- // CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
- INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
- }
- }
- return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ if (CometSparkSessionExtensions.cometUnifiedMemoryManagerEnabled(conf)) {
Review Comment:
We can directly check the memory mode in the provided task manager rather than checking the configs here
```suggestion
if (taskMemoryManager.getTungstenMemoryMode() == MemoryMode.OFF_HEAP) {
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Rename CometTestShuffleMemoryAllocator since it can be used outside of tests [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1991975853
##########
common/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -404,13 +404,15 @@ object CometConf extends ShimCometConf {
"Ensure that Comet shuffle memory overhead factor is a double greater than 0")
.createWithDefault(1.0)
- val COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST: ConfigEntry[Boolean] =
- conf("spark.comet.columnar.shuffle.unifiedMemoryAllocatorTest")
- .doc("Whether to use Spark unified memory allocator for Comet columnar shuffle in tests." +
- "If not configured, Comet will use a test-only memory allocator for Comet columnar " +
- "shuffle when Spark test env detected. The test-ony allocator is proposed to run with " +
- "Spark tests as these tests require on-heap memory configuration. " +
- "By default, this config is false.")
+ val COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR: ConfigEntry[Boolean] =
Review Comment:
I think that we should revert this config change. The original intent of the config was to force use of the unified memory allocator in tests even when running in on-heap mode. I still don't quite understand why we would want to do this, but the current tests are passing with this 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.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1988101908
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -48,30 +48,30 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
+ boolean useTestAllocator =
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useTestAllocator) {
synchronized (CometShuffleMemoryAllocator.class) {
if (INSTANCE == null) {
// CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
}
}
return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ }
- // CometShuffleMemoryAllocator stores pages in TaskMemoryManager which is not singleton,
- // but one instance per task. So we need to create a new instance for each task.
- return new CometShuffleMemoryAllocator(taskMemoryManager, pageSize);
+ if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
Review Comment:
Yea, this memory allocator implementation doesn't change across the versions. It was just renamed to `CometTestShuffleMemoryAllocator` to reflect the decision to use it for test only purpose at the time.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Only instantiate CometTestShuffleMemoryAllocator in testing [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1985782286
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -52,7 +52,7 @@ public static CometShuffleMemoryAllocatorTrait getInstance(
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting && !useUnifiedMemAllocator) {
Review Comment:
I think more changes are required here. If `isSparkTesting` is false and `useUnifiedMemAllocator` is also false, we will fall through to the `else` branch, which will throw an exception because off-heap is not enabled.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993653317
##########
spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala:
##########
@@ -159,23 +159,4 @@ class CometPluginsUnifiedModeOverrideSuite extends CometTestBase {
conf.set("spark.comet.memory.overhead.factor", "0.5")
conf
}
-
- /*
- * Since using unified memory, but not shuffle unified memory
- * executor memory should be overridden by adding comet shuffle memory size
- */
- test("executor memory overhead is correctly overridden") {
Review Comment:
It's now effectively exercising the same path as above (non-override) but I could make the values match. I'd argue that it's a redundant test but I can put it back.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993588356
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -101,14 +101,16 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
* unified memory manager.
*/
private def shouldOverrideMemoryConf(conf: SparkConf): Boolean = {
- conf.getBoolean(CometConf.COMET_ENABLED.key, true) && (
- conf.getBoolean(
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.key,
- CometConf.COMET_EXEC_SHUFFLE_ENABLED.defaultValue.get) ||
- conf.getBoolean(
- CometConf.COMET_EXEC_ENABLED.key,
- CometConf.COMET_EXEC_ENABLED.defaultValue.get)
- ) && CometSparkSessionExtensions.cometUnifiedMemoryManagerEnabled(conf)
+ val comet_enabled = conf.getBoolean(CometConf.COMET_ENABLED.key, true)
+ val comet_exec_shuffle = conf.getBoolean(
+ CometConf.COMET_EXEC_SHUFFLE_ENABLED.key,
+ CometConf.COMET_EXEC_SHUFFLE_ENABLED.defaultValue.get)
+ val comet_exec = conf.getBoolean(
+ CometConf.COMET_EXEC_ENABLED.key,
+ CometConf.COMET_EXEC_ENABLED.defaultValue.get)
+ val unified_memory = CometSparkSessionExtensions.cometUnifiedMemoryManagerEnabled(conf)
+
+ comet_enabled && (comet_exec_shuffle || comet_exec) && !unified_memory
Review Comment:
this is much clearer
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993653317
##########
spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala:
##########
@@ -159,23 +159,4 @@ class CometPluginsUnifiedModeOverrideSuite extends CometTestBase {
conf.set("spark.comet.memory.overhead.factor", "0.5")
conf
}
-
- /*
- * Since using unified memory, but not shuffle unified memory
- * executor memory should be overridden by adding comet shuffle memory size
- */
- test("executor memory overhead is correctly overridden") {
Review Comment:
It's effectively exercising the same path as above now (non-override) but I could make the values match. I'd argue that it's a redundant test but I can put it back.
##########
spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala:
##########
@@ -159,23 +159,4 @@ class CometPluginsUnifiedModeOverrideSuite extends CometTestBase {
conf.set("spark.comet.memory.overhead.factor", "0.5")
conf
}
-
- /*
- * Since using unified memory, but not shuffle unified memory
- * executor memory should be overridden by adding comet shuffle memory size
- */
- test("executor memory overhead is correctly overridden") {
Review Comment:
It's now effectively exercising the same path as above now (non-override) but I could make the values match. I'd argue that it's a redundant test but I can put it back.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1988060641
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -48,30 +48,30 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
+ boolean useTestAllocator =
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useTestAllocator) {
synchronized (CometShuffleMemoryAllocator.class) {
if (INSTANCE == null) {
// CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
}
}
return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ }
- // CometShuffleMemoryAllocator stores pages in TaskMemoryManager which is not singleton,
- // but one instance per task. So we need to create a new instance for each task.
- return new CometShuffleMemoryAllocator(taskMemoryManager, pageSize);
+ if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
Review Comment:
I tested with TPC-H without using off-heap and ran into this:
```
Caused by: java.lang.IllegalArgumentException: CometShuffleMemoryAllocator should be used with off-heap memory mode, but got ON_HEAP
at org.apache.spark.shuffle.comet.CometShuffleMemoryAllocator.getInstance(CometShuffleMemoryAllocator.java:74)
at org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriter.<init>(CometDiskBlockWriter.java:142)
at org.apache.spark.sql.comet.execution.shuffle.CometBypassMergeSortShuffleWriter.write(CometBypassMergeSortShuffleWriter.java:181)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1987723154
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -48,30 +48,30 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
+ boolean useTestAllocator =
(boolean)
CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useTestAllocator) {
synchronized (CometShuffleMemoryAllocator.class) {
if (INSTANCE == null) {
// CometTestShuffleMemoryAllocator handles pages by itself so it can be a singleton.
INSTANCE = new CometTestShuffleMemoryAllocator(conf, taskMemoryManager, pageSize);
}
}
return INSTANCE;
- } else {
- if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
- throw new IllegalArgumentException(
- "CometShuffleMemoryAllocator should be used with off-heap "
- + "memory mode, but got "
- + taskMemoryManager.getTungstenMemoryMode());
- }
+ }
- // CometShuffleMemoryAllocator stores pages in TaskMemoryManager which is not singleton,
- // but one instance per task. So we need to create a new instance for each task.
- return new CometShuffleMemoryAllocator(taskMemoryManager, pageSize);
+ if (taskMemoryManager.getTungstenMemoryMode() != MemoryMode.OFF_HEAP) {
Review Comment:
I'm still confused about this. This seems to imply that we only perform columnar shuffle when off-heap mode is enabled?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Rename CometTestShuffleMemoryAllocator since it can be used outside of tests [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1991981240
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -35,43 +35,47 @@
* store serialized rows. This class is simply an implementation of `MemoryConsumer` that delegates
* memory allocation to the `TaskMemoryManager`. This requires that the `TaskMemoryManager` is
* configured with `MemoryMode.OFF_HEAP`, i.e. it is using off-heap memory.
+ *
+ * <p>If the user does not enable off-heap memory then we want to use
+ * CometBoundedShuffleMemoryAllocator. The tests also need to default to using this because off-heap
+ * is not enabled when running the Spark SQL tests. The
+ * COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR flag allows us to use unified memory management
+ * in Comet tests (this does assume that off-heap is enabled).
*/
public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocatorTrait {
private static CometShuffleMemoryAllocatorTrait INSTANCE;
/**
* Returns the singleton instance of `CometShuffleMemoryAllocator`. This method should be used
* instead of the constructor to ensure that only one instance of `CometShuffleMemoryAllocator` is
- * created. For Spark tests, this returns `CometTestShuffleMemoryAllocator` which is a test-only
- * allocator that should not be used in production.
+ * created. For Spark tests, this returns `CometBoundedShuffleMemoryAllocator`.
*/
public static CometShuffleMemoryAllocatorTrait getInstance(
SparkConf conf, TaskMemoryManager taskMemoryManager, long pageSize) {
boolean isSparkTesting = Utils.isTesting();
- boolean useUnifiedMemAllocator =
- (boolean)
- CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST().get();
+ boolean useBoundedAllocator =
+ (boolean) CometConf$.MODULE$.COMET_COLUMNAR_SHUFFLE_BOUNDED_MEMORY_ALLOCATOR().get();
- if (!useUnifiedMemAllocator) {
+ if (isSparkTesting || useBoundedAllocator) {
Review Comment:
I think this loses the ability for us to force use of the unified memory allocator in tests
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1993639777
##########
spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala:
##########
@@ -159,23 +159,4 @@ class CometPluginsUnifiedModeOverrideSuite extends CometTestBase {
conf.set("spark.comet.memory.overhead.factor", "0.5")
conf
}
-
- /*
- * Since using unified memory, but not shuffle unified memory
- * executor memory should be overridden by adding comet shuffle memory size
- */
- test("executor memory overhead is correctly overridden") {
Review Comment:
Could we update this test rather than remove it?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Simplify CometShuffleMemoryAllocator logic, rename classes, remove config [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Adjust CometTestShuffleMemoryAllocator instantiation [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1485:
URL: https://github.com/apache/datafusion-comet/pull/1485#discussion_r1989943308
##########
spark/src/main/java/org/apache/spark/shuffle/comet/CometShuffleMemoryAllocator.java:
##########
@@ -42,36 +42,35 @@ public final class CometShuffleMemoryAllocator extends CometShuffleMemoryAllocat
/**
* Returns the singleton instance of `CometShuffleMemoryAllocator`. This method should be used
* instead of the constructor to ensure that only one instance of `CometShuffleMemoryAllocator` is
- * created. For Spark tests, this returns `CometTestShuffleMemoryAllocator` which is a test-only
- * allocator that should not be used in production.
+ * created. For Spark tests, this returns `CometBoundedShuffleMemoryAllocator` which is a
+ * test-only allocator that should not be used in production.
*/
Review Comment:
It is no longer true that this is just for testing.
If the user enables Spark's off-heap memory then we want to use unified memory management, which means that we use `CometShuffleMemoryAllocator` which will interact with `TaskMemoryManager` to request memory and this will be sharing an off-heap pool with Spark.
If the user does not enable off-heap memory then we want to use `CometBoundedShuffleMemoryAllocator`. The tests also need to default to using this because off-heap is not enabled when running the Spark SQL tests. The `COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST` flag allows us to use unified memory management in the tests (this does assume that off-heap is enabled). We enable this flag in `CometColumnarShuffleSuite` although I am not sure why.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org