[CORE] Use SortShuffleManager instance in ColumnarShuffleManager #6022

acvictor · 2024-06-07T15:19:11Z

What changes were proposed in this pull request?

This PR makes a change to introduce an object of SortShuffleManager in ColumnarShuffleManager to reduce code overlap.

How was this patch tested?

SortShuffleManagerSuite has already been enabled against ColumnarShuffleManager #5816

github-actions · 2024-06-07T15:19:27Z

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

Other pull requests

github-actions · 2024-06-07T15:19:42Z

Run Gluten Clickhouse CI

github-actions · 2024-06-07T15:20:41Z

Run Gluten Clickhouse CI

acvictor · 2024-06-08T04:49:02Z

/Benchmark Velox

GlutenPerfBot · 2024-06-08T06:07:11Z

===== Performance report for TPCH SF2000 with Velox backend, for reference only ====

query	log/native_6022_time.csv	log/native_master_06_07_2024_85e8619f12_time.csv	difference	percentage
q1	34.16	34.74	0.584	101.71%
q2	24.59	23.82	-0.776	96.84%
q3	36.75	37.27	0.519	101.41%
q4	33.86	30.13	-3.731	88.98%
q5	69.01	70.45	1.446	102.10%
q6	7.55	7.45	-0.101	98.66%
q7	81.89	83.57	1.676	102.05%
q8	87.32	86.74	-0.576	99.34%
q9	121.00	121.95	0.949	100.78%
q10	45.90	45.15	-0.754	98.36%
q11	23.86	19.51	-4.357	81.74%
q12	26.61	26.76	0.152	100.57%
q13	37.99	39.39	1.408	103.71%
q14	21.04	20.54	-0.499	97.63%
q15	31.63	31.52	-0.105	99.67%
q16	12.95	13.65	0.703	105.43%
q17	104.35	103.56	-0.797	99.24%
q18	144.05	145.21	1.157	100.80%
q19	14.83	14.70	-0.137	99.08%
q20	26.64	28.92	2.279	108.56%
q21	257.09	262.04	4.950	101.93%
q22	13.05	14.56	1.506	111.54%
total	1256.13	1261.63	5.497	100.44%

acvictor · 2024-06-08T06:57:16Z

@marin-ma can you please review this? Thank you!

acvictor · 2024-06-10T05:39:32Z

@rui-mo can you too please review this? Thanks!

github-actions · 2024-06-10T13:03:33Z

Run Gluten Clickhouse CI

marin-ma

LGTM. Thanks!

ulysses-you · 2024-06-11T01:35:18Z

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala

 class ColumnarShuffleManager(conf: SparkConf) extends ShuffleManager with Logging {

  import ColumnarShuffleManager._

-  private lazy val shuffleExecutorComponents = loadShuffleExecutorComponents(conf)
+  private[this] val sortShuffleManager: SortShuffleManager = new SortShuffleManager(conf)


ulysses-you · 2024-06-11T01:41:00Z

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala

-    Option(taskIdMapsForShuffle.remove(shuffleId)).foreach {
-      mapTaskIds =>
-        mapTaskIds.iterator.foreach {
-          mapId => shuffleBlockResolver.removeDataByMap(shuffleId, mapId)
-        }
-    }
-    true
+    sortShuffleManager.unregisterShuffle(shuffleId)


Does it same with before ? The taskIdMapsForShuffle is maintained by ColumnarShuffleManager so sortShuffleManager. unregisterShuffle will do nothing since the taskIdMapsForShuffle is empty.

I missed this thanks, updated shuffleBlockResolver related flow.

ulysses-you · 2024-06-11T01:44:43Z

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala

  }

  /** Shut down this ShuffleManager. */
  override def stop(): Unit = {
-    shuffleBlockResolver.stop()
+    sortShuffleManager.stop()


The shuffleBlockResolver is hold in ColumnarShuffleManager, how can SortShuffleManager stop it?

Updated, thanks!

pending

…r/gluten into acvictor/columnarShuffle

github-actions · 2024-06-11T06:09:24Z

Run Gluten Clickhouse CI

acvictor · 2024-06-11T09:21:36Z

@ulysses-you can you please review again? Thank you!

ulysses-you · 2024-06-11T10:08:48Z

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala

    mapTaskIds.synchronized {
      mapTaskIds.add(context.taskAttemptId())
    }


Shall we move this code in case columnarShuffleHandle: ColumnarShuffleHandle ?

Moved to case columnarShuffleHandle: ColumnarShuffleHandle

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala

ulysses-you · 2024-06-11T10:11:28Z

shall we call unregisterShuffle and stop if it is delegated to sort shuffle manager ? e.g.,

if (taskIdMapsForShuffle.contains(shuffleId)) {
  this.stop
} else {
  delegated.stop
}

github-actions · 2024-06-11T10:32:35Z

Run Gluten Clickhouse CI

acvictor · 2024-06-11T10:33:28Z

shall we call unregisterShuffle and stop if it is delegated to sort shuffle manager ? e.g.,
if (taskIdMapsForShuffle.contains(shuffleId)) {
  this.stop
} else {
  delegated.stop
}

Updated unregisterShuffle and stop to this pattern.

acvictor · 2024-06-11T10:51:56Z

Thank you @ulysses-you for the comments! Updated the PR.

ulysses-you

lgtm, thank you @acvictor , cc @marin-ma

marin-ma · 2024-06-12T02:07:21Z

@ulysses-you Thanks for review!

@acvictor Could you help to add a UT to test the fallback behavior for columnar shuffle? Perhaps, add one test case in FallbackSuite.scala and set spark.gluten.sql.columnar.shuffle=false force the Exchange operator to fall back

github-actions · 2024-06-12T05:18:06Z

Run Gluten Clickhouse CI

marin-ma · 2024-06-12T05:25:44Z

backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala

+        df =>
+          val plan = df.queryExecution.executedPlan
+          val columnarShuffle = find(plan) {
+            case _: ColumnarShuffledJoin => true


We should check the number ofColumnarShuffleExchangeExec and ShuffleExchangeExec. You can use repartition hint to construct the sql https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html#partitioning-hints

e.g. select /*+ REPARTITION(3, c1) */ * FROM tmp1;

Then the check should be like:

assert(collectColumnarShuffleExchange(plan) == 0) assert(collectShuffleExchange(plan) == 1)

For the current query this is the plan

AdaptiveSparkPlan isFinalPlan=true +- == Final Plan == VeloxColumnarToRowExec +- ^(6) HashAggregateTransformer(keys=[c1#14], functions=[count(1)], output=[c1#14, count(1)#26L]) +- ^(6) InputIteratorTransformer[c1#14, count#30L] +- ^(6) InputAdapter +- ^(6) RowToVeloxColumnar +- ^(6) AQEShuffleRead coalesced +- ^(6) ShuffleQueryStage 0 +- Exchange hashpartitioning(c1#14, 5), ENSURE_REQUIREMENTS, [plan_id=281] +- VeloxColumnarToRowExec +- ^(5) HashAggregateTransformer(keys=[c1#14], functions=[partial_count(1)], output=[c1#14, count#30L]) +- ^(5) NativeFileScan parquet spark_catalog.default.tmp11111[c1#14] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/anvicto/src/Gluten/spark-warehouse/org.apache.gluten.execut..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c1:int>

Can I retain it?

It's ok to me. Thanks.

I have raised one small PR here #6055 to enable fallback for shuffle.

github-actions · 2024-06-12T05:41:22Z

Run Gluten Clickhouse CI

github-actions · 2024-06-12T09:21:52Z

Run Gluten Clickhouse CI

acvictor · 2024-06-12T13:01:26Z

@marin-ma added a test.

marin-ma

Thanks!

GlutenPerfBot · 2024-06-13T02:15:01Z

===== Performance report for TPCH SF2000 with Velox backend, for reference only ====

query	log/native_6022_time.csv	log/native_master_06_12_2024_7445f02c9_time.csv	difference	percentage
q1	35.67	33.87	-1.796	94.96%
q2	23.60	26.51	2.905	112.31%
q3	36.84	36.94	0.106	100.29%
q4	32.41	34.65	2.244	106.93%
q5	70.83	69.47	-1.358	98.08%
q6	7.61	5.84	-1.772	76.72%
q7	84.08	82.18	-1.901	97.74%
q8	83.77	85.19	1.414	101.69%
q9	118.64	120.19	1.543	101.30%
q10	44.72	43.54	-1.180	97.36%
q11	22.67	20.47	-2.197	90.31%
q12	26.37	24.09	-2.280	91.35%
q13	37.56	37.22	-0.342	99.09%
q14	19.24	23.00	3.754	119.51%
q15	30.13	30.20	0.073	100.24%
q16	13.74	13.79	0.047	100.34%
q17	103.62	102.85	-0.770	99.26%
q18	143.92	145.02	1.098	100.76%
q19	13.62	16.14	2.518	118.49%
q20	29.19	27.34	-1.848	93.67%
q21	262.53	263.30	0.765	100.29%
q22	12.33	12.08	-0.247	98.00%
total	1253.09	1253.86	0.775	100.06%

Make updates

459bfe5

Fix indentation

65302e4

acvictor marked this pull request as ready for review June 8, 2024 06:56

acvictor changed the title ~~Use SortShuffleManager instance in ColumnarShuffleManager~~ [CORE] Use SortShuffleManager instance in ColumnarShuffleManager Jun 8, 2024

Merge branch 'main' into acvictor/columnarShuffle

cbb10be

marin-ma previously approved these changes Jun 11, 2024

View reviewed changes

ulysses-you reviewed Jun 11, 2024

View reviewed changes

acvictor added 2 commits June 11, 2024 11:38

Retain shuffleBlockResolver code

8290189

Merge branch 'acvictor/columnarShuffle' of https://github.com/acvicto…

c120a02

…r/gluten into acvictor/columnarShuffle

ulysses-you reviewed Jun 11, 2024

View reviewed changes

gluten-core/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala Show resolved Hide resolved

Set taskIdMapsForShuffle only in columnar case

15571b8

ulysses-you previously approved these changes Jun 12, 2024

View reviewed changes

Add UT

1e83dff

acvictor dismissed ulysses-you’s stale review via 1e83dff June 12, 2024 05:17

marin-ma reviewed Jun 12, 2024

View reviewed changes

Update test

82835bd

Merge branch 'main' into acvictor/columnarShuffle

97b61d0

marin-ma approved these changes Jun 13, 2024

View reviewed changes

marin-ma merged commit 2ae80af into apache:main Jun 13, 2024
40 checks passed

jackylee-ch mentioned this pull request Jun 19, 2024

Shuffle cannot be removed expectly. #6140

Open

acvictor mentioned this pull request Jun 19, 2024

Revert SortShuffleManager changes in ColumnarShuffleManager #6149

Merged

[CORE] Use SortShuffleManager instance in ColumnarShuffleManager #6022

[CORE] Use SortShuffleManager instance in ColumnarShuffleManager #6022

Conversation

acvictor commented Jun 7, 2024 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

github-actions bot commented Jun 7, 2024

github-actions bot commented Jun 7, 2024

github-actions bot commented Jun 7, 2024

acvictor commented Jun 8, 2024

GlutenPerfBot commented Jun 8, 2024

acvictor commented Jun 8, 2024

acvictor commented Jun 10, 2024

github-actions bot commented Jun 10, 2024

marin-ma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jun 11, 2024

acvictor commented Jun 11, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ulysses-you commented Jun 11, 2024 • edited Loading

github-actions bot commented Jun 11, 2024

acvictor commented Jun 11, 2024

acvictor commented Jun 11, 2024

ulysses-you left a comment

Choose a reason for hiding this comment

marin-ma commented Jun 12, 2024

github-actions bot commented Jun 12, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jun 12, 2024

github-actions bot commented Jun 12, 2024

acvictor commented Jun 12, 2024

marin-ma left a comment

Choose a reason for hiding this comment

GlutenPerfBot commented Jun 13, 2024

acvictor commented Jun 7, 2024 •

edited

Loading

ulysses-you commented Jun 11, 2024 •

edited

Loading