feat: Enable columnar shuffle by default #250

viirya · 2024-04-09T21:32:04Z

Which issue does this PR close?

Closes #95.

Rationale for this change

What changes are included in this PR?

How are these changes tested?

spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala

codecov-commenter · 2024-04-10T19:59:26Z

Codecov Report

Attention: Patch coverage is 50.00000% with 4 lines in your changes are missing coverage. Please review.

Project coverage is 33.50%. Comparing base (9ab6c75) to head (ace91fe).
Report is 9 commits behind head on main.

Files	Patch %	Lines
.../scala/org/apache/spark/sql/comet/util/Utils.scala	0.00%	2 Missing ⚠️
...ain/scala/org/apache/comet/vector/NativeUtil.scala	0.00%	1 Missing ⚠️
.../scala/org/apache/comet/serde/QueryPlanSerde.scala	0.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #250      +/-   ##
============================================
+ Coverage     33.47%   33.50%   +0.03%     
- Complexity      795      798       +3     
============================================
  Files           110      110              
  Lines         37533    37541       +8     
  Branches       8215     8217       +2     
============================================
+ Hits          12563    12579      +16     
+ Misses        22322    22321       -1     
+ Partials       2648     2641       -7

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

viirya · 2024-04-10T20:10:58Z

Hmm, all tests in TPCDSQuerySuite` are passed locally with this PR. I need to look at the CI failure.

viirya · 2024-04-11T20:03:06Z

Observed several Spark SQL test failures regarding aggregation: #260

viirya · 2024-04-11T20:06:41Z

dev/diffs/3.4.2.diff

@@ -1414,6 +1424,7 @@ index ed2e309fa07..4cfe0093da7 100644
 +          .set("spark.shuffle.manager",
 +            "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
 +          .set("spark.comet.exec.shuffle.enabled", "true")
+          .set("spark.comet.memoryOverhead", "10g")


Observed that Comet is unable to acquire enough memory for columnar shuffle when doing Spark SQL tests:

For example, DatasetPrimitiveSuite:

Cause: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 37.0 failed 1 times, most recent failure: Lost task 1.0 in stage 37.0 (TID 75) (e4773b5abe7e executor driver): org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 67108848 bytes of memory, got 96 bytes. Available: 96 [info] at org.apache.spark.shuffle.comet.CometShuffleMemoryAllocator.allocate(CometShuffleMemoryAllocator.java:132) [info] at org.apache.spark.shuffle.comet.CometShuffleMemoryAllocator.allocatePage(CometShuffleMemoryAllocator.java:119) [info] at org.apache.spark.sql.comet.execution.shuffle.SpillWriter.initialCurrentPage(SpillWriter.java:158) [info] at org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriter.insertRow(CometDiskBlockWriter.java:284)

Increased Comet memoryoverhead to overcome it.

…fault

viirya · 2024-05-05T16:49:01Z

I fixed all Spark SQL test failures. Now waiting for #380 to be merged.

…fault

viirya · 2024-05-06T19:40:44Z

cc @sunchao @andygrove

viirya · 2024-05-06T19:42:19Z

spark/src/main/scala/org/apache/spark/sql/comet/plans/AliasAwareOutputExpression.scala

+  protected val aliasCandidateLimit: Int =
+    conf.getConfString("spark.sql.optimizer.expressionProjectionCandidateLimit", "100").toInt


Some Spark tests tune this config. So we need to get the configured value.

viirya · 2024-05-08T02:19:56Z

@andygrove @sunchao Could you take a look and see if you have any comments on this? Thanks.

core/src/execution/datafusion/planner.rs

Co-authored-by: Andy Grove <[email protected]>

core/src/execution/datafusion/planner.rs

Co-authored-by: Andy Grove <[email protected]>

andygrove · 2024-05-08T18:33:55Z

dev/diffs/3.4.2.diff

+     assert(
+-      collect(df.queryExecution.executedPlan) { case e: ShuffleExchangeExec => e }.size == expected)
+      collect(df.queryExecution.executedPlan) {
+        case _: ShuffleExchangeExec | _: CometShuffleExchangeExec => 1 }.size == expected)


Could we just check for ShuffleExchangeLike instead (and even push that change upstream?)

We could. For upstream, not sure if it is accepted, but we can try.

I changed to check ShuffleExchangeLike instead for the places which is possible.

andygrove

I am not familiar with all of the Spark tests in the patch, but the changes to Comet LGTM

viirya · 2024-05-08T20:49:28Z

Thank you @andygrove

viirya · 2024-05-09T00:38:51Z

Merged. Thanks @andygrove for review.

* feat: Enable columnar shuffle by default * Update plan stability * Fix * Update diff * Add Comet memoryoverhead for Spark SQL tests * Update plan stability * Update diff * Update more diff * Update DataFusion commit * Update diff * Update diff * Update diff * Update diff * Update diff * Fix more tests * Fix more * Fix * Fix more * Fix more * Fix more * Fix more * Fix more * Update diff * Fix memory leak * Update plan stability * Restore diff * Update core/src/execution/datafusion/planner.rs Co-authored-by: Andy Grove <[email protected]> * Update core/src/execution/datafusion/planner.rs Co-authored-by: Andy Grove <[email protected]> * Fix style * Use ShuffleExchangeLike instead --------- Co-authored-by: Andy Grove <[email protected]>

andygrove reviewed Apr 10, 2024

View reviewed changes

spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala Show resolved Hide resolved

viirya closed this Apr 10, 2024

viirya reopened this Apr 10, 2024

viirya force-pushed the columnar_shuffle_default branch from be83771 to ef013c1 Compare April 11, 2024 19:29

viirya commented Apr 11, 2024

View reviewed changes

viirya mentioned this pull request Apr 11, 2024

fix: Average expression in Comet Final should handle all null inputs from partial Spark aggregation #261

Merged

viirya force-pushed the columnar_shuffle_default branch from 3e3daea to 77e5604 Compare April 12, 2024 02:02

viirya mentioned this pull request Apr 12, 2024

Only trigger Comet Final aggregation on Comet partial aggregation #262

Closed

viirya force-pushed the columnar_shuffle_default branch from 77e5604 to e3d861c Compare April 13, 2024 06:30

This was referenced Apr 14, 2024

Scalar function in DataFusion cannot coerce dictionary type inputs #265

Closed

Different behavior of distinct count on null inputs #267

Closed

Got arrow error when sorting on empty batch #270

Closed

viirya force-pushed the columnar_shuffle_default branch 2 times, most recently from 54da021 to 7465384 Compare April 16, 2024 18:55

viirya mentioned this pull request Apr 16, 2024

Comet should not translate try_sum to native sum expression #276

Closed

viirya force-pushed the columnar_shuffle_default branch from 7465384 to ace91fe Compare April 18, 2024 00:22

viirya mentioned this pull request Apr 18, 2024

Aggregate expression with filter is incorrectly translated to Comet aggregate native aggregation expression #283

Closed

viirya force-pushed the columnar_shuffle_default branch from ace91fe to a6e2d16 Compare April 18, 2024 03:15

viirya mentioned this pull request Apr 18, 2024

Comet fails on limit operator with negative limit parameter #287

Closed

viirya force-pushed the columnar_shuffle_default branch from a6e2d16 to b145499 Compare April 19, 2024 05:19

viirya mentioned this pull request Apr 20, 2024

Repeated shuffle should not trigger Comet columnar shuffle #295

Closed

viirya force-pushed the columnar_shuffle_default branch from dcac8f9 to 30043e4 Compare April 21, 2024 17:11

viirya mentioned this pull request Apr 22, 2024

CometExec's outputPartitioning might not be same as Spark expects after AQE interferes #298

Closed

viirya force-pushed the columnar_shuffle_default branch from e91bac1 to 9767acc Compare April 23, 2024 16:41

This was referenced Apr 25, 2024

CometShuffleExchangeExec logical link is different to Spark ShuffleExchangeExec #323

Closed

Switch back to released version of DataFusion and arrow-rs after Arrow Java 16 is released #248

Closed

viirya force-pushed the columnar_shuffle_default branch from 9767acc to edfce1f Compare April 30, 2024 02:31

viirya added 6 commits May 1, 2024 17:36

Fix more

b342b66

Fix more

6d27a5f

Fix more

508cb1a

Fix more

5e03881

Update diff

1e560d3

Merge remote-tracking branch 'upstream/main' into columnar_shuffle_de…

126675e

…fault

viirya mentioned this pull request May 4, 2024

Detected memory leak on Comet columnar shuffle when AQE coalesce partitions enabled #381

Closed

Fix memory leak

b37070d

viirya force-pushed the columnar_shuffle_default branch from 37c8186 to b37070d Compare May 4, 2024 21:59

viirya added 2 commits May 4, 2024 21:17

Update plan stability

557b753

Restore diff

65a9515

Merge remote-tracking branch 'upstream/main' into columnar_shuffle_de…

70c53b4

…fault

viirya commented May 6, 2024

View reviewed changes

viirya mentioned this pull request May 7, 2024

[EPIC] Improve performance of TPC-H queries #391

Open

14 tasks

viirya mentioned this pull request May 8, 2024

build: Switch back to official DataFusion repo and arrow-rs after Arrow Java 16 is released #403

Merged

andygrove reviewed May 8, 2024

View reviewed changes

core/src/execution/datafusion/planner.rs Outdated Show resolved Hide resolved

Update core/src/execution/datafusion/planner.rs

c9633ba

Co-authored-by: Andy Grove <[email protected]>

andygrove reviewed May 8, 2024

View reviewed changes

core/src/execution/datafusion/planner.rs Outdated Show resolved Hide resolved

Update core/src/execution/datafusion/planner.rs

a61dc70

Co-authored-by: Andy Grove <[email protected]>

andygrove reviewed May 8, 2024

View reviewed changes

viirya added 2 commits May 8, 2024 11:43

Fix style

cceb552

Use ShuffleExchangeLike instead

cbc5305

andygrove approved these changes May 8, 2024

View reviewed changes

viirya merged commit 14494d3 into apache:main May 9, 2024
40 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Enable columnar shuffle by default #250

feat: Enable columnar shuffle by default #250

viirya commented Apr 9, 2024

codecov-commenter commented Apr 10, 2024 •

edited

Loading

viirya commented Apr 10, 2024

viirya commented Apr 11, 2024

viirya Apr 11, 2024

viirya Apr 11, 2024

viirya commented May 5, 2024

viirya commented May 6, 2024

viirya May 6, 2024

viirya commented May 8, 2024

andygrove May 8, 2024

viirya May 8, 2024

viirya May 8, 2024

andygrove left a comment

viirya commented May 8, 2024

viirya commented May 9, 2024

		protected val aliasCandidateLimit: Int =
		conf.getConfString("spark.sql.optimizer.expressionProjectionCandidateLimit", "100").toInt

feat: Enable columnar shuffle by default #250

feat: Enable columnar shuffle by default #250

Conversation

viirya commented Apr 9, 2024

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

codecov-commenter commented Apr 10, 2024 • edited Loading

Codecov Report

viirya commented Apr 10, 2024

viirya commented Apr 11, 2024

viirya Apr 11, 2024

Choose a reason for hiding this comment

viirya Apr 11, 2024

Choose a reason for hiding this comment

viirya commented May 5, 2024

viirya commented May 6, 2024

viirya May 6, 2024

Choose a reason for hiding this comment

viirya commented May 8, 2024

andygrove May 8, 2024

Choose a reason for hiding this comment

viirya May 8, 2024

Choose a reason for hiding this comment

viirya May 8, 2024

Choose a reason for hiding this comment

andygrove left a comment

Choose a reason for hiding this comment

viirya commented May 8, 2024

viirya commented May 9, 2024

codecov-commenter commented Apr 10, 2024 •

edited

Loading