Add framework for batch compression of shuffle partitions #487

jlowe · 2020-07-31T19:56:54Z

This adds TableCompressionCodec, an interface for compression codecs that can compress contiguous table buffers. A single trivial copy codec is provided just for testing. Other codecs can be added in the future.

Batch compression is supported and utilized in the partitioning code when using the RAPIDS UCX shuffle plugin which produces contiguous tables. Batch decompression is supported and utilized in the batch coalescing that is inserted into the query after each shuffle. The coalesce code automatically detects compressed shuffle buffers and handles a mix of uncompressed and compressed batches arriving to be coalesced.

If desired I can split this PR into smaller pieces (e.g.: just adding TableCompressor and copy codec, then separate PR to plug it into partitioning and coalescing). I opted to do a single PR initially so it was clear how the table compression APIs were being used in practice.

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-07-31T19:57:09Z

build

jlowe · 2020-08-03T21:26:45Z

build

abellina

This is just a first pass @jlowe I have more than 1/2 to review.

sql-plugin/src/main/scala/com/nvidia/spark/rapids/TableCompressionCodec.scala

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsDeviceMemoryStore.scala

sql-plugin/src/main/java/com/nvidia/spark/rapids/GpuCompressedColumnVector.java

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsBufferStore.scala

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCoalesceBatches.scala

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsBufferStore.scala

sql-plugin/src/main/scala/com/nvidia/spark/rapids/TableCompressionCodec.scala

abellina · 2020-08-04T20:35:51Z

@jlowe I got through the rest of it. Let me know when to re-check.

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-08-04T22:37:03Z

@revans2 @abellina I believe I've addressed all the review comments, so this is ready for another look.

jlowe · 2020-08-04T22:37:08Z

build

jlowe · 2020-08-06T22:29:30Z

This needs to be upmerged to the latest on branch-0.2.

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-08-07T15:03:38Z

build

sql-plugin/src/main/scala/com/nvidia/spark/rapids/MetaUtils.scala

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-08-07T16:18:48Z

build

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-08-10T13:49:08Z

build

jlowe · 2020-08-10T18:27:37Z

@revans2 can you take another look? I needed to upmerge to the latest on branch-0.2.

revans2

Looks good to me. I am fine with it as is, just was curious if it would be cleaner?

revans2 · 2020-08-10T21:59:39Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

@@ -590,8 +590,27 @@ object RapidsConf {
    .bytesConf(ByteUnit.BYTE)
    .createWithDefault(50 * 1024)

+  val SHUFFLE_COMPRESSION_ENABLED = conf("spark.rapids.shuffle.compression.enabled")


I am OK with two configs for this, but why not just set the codec to NONE, or something like that to disable it?

I was emulating Spark's shuffle config where the compression codec and compression enable are separate.

Updated this to use a single config for the shuffle codec, with none indicating no shuffle compression codec should be used.

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe · 2020-08-11T14:26:51Z

Holding off on the CI build for now. Waiting on the HashJoin fix for 3.1 and updated cudf with the fix for rapidsai/cudf#5915 which was found during the added unit tests in this PR.

jlowe · 2020-08-11T16:49:16Z

build

* Add framework for batch compression of shuffle partitions Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Extract common row accessor methods to GpuColumnVectorBase Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Address review comments Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Fix handling of degenerate batches Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Fix buffer leak Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Add comment about degenerate batches potentially appearing compressed Signed-off-by: Jason Lowe <jlowe@nvidia.com> * Use a shuffle compression codec of "none" to indicate no compression Signed-off-by: Jason Lowe <jlowe@nvidia.com>

Signed-off-by: spark-rapids automation <70000568+nvauto@users.noreply.github.com> Signed-off-by: spark-rapids automation <70000568+nvauto@users.noreply.github.com>

Add framework for batch compression of shuffle partitions

6ecbabd

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe added SQL part of the SQL/Dataframe plugin shuffle things that impact the shuffle plugin labels Jul 31, 2020

jlowe self-assigned this Jul 31, 2020

abellina self-requested a review August 2, 2020 19:34

sameerz added this to the Aug 3 - Aug 14 milestone Aug 3, 2020

jlowe mentioned this pull request Aug 3, 2020

[FEA] Implement shuffle optimization when AQE is enabled #493

Closed

Merge branch 'branch-0.2' into table-compressor

c08a64a

abellina requested changes Aug 3, 2020

View reviewed changes

abellina mentioned this pull request Aug 3, 2020

[BUG] PartMerge:countDistinct:sum fails sporadically #473

Closed

revans2 reviewed Aug 4, 2020

View reviewed changes

abellina requested changes Aug 4, 2020

View reviewed changes

jlowe mentioned this pull request Aug 4, 2020

[FEA] Connect shuffle table compression to shuffle exec metrics #511

Closed

jlowe added 2 commits August 4, 2020 17:31

Extract common row accessor methods to GpuColumnVectorBase

a15df77

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

Address review comments

33c96d8

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

abellina previously approved these changes Aug 5, 2020

View reviewed changes

Merge branch 'branch-0.2' into table-compressor

e5b8f2a

jlowe marked this pull request as draft August 6, 2020 22:29

jlowe added 3 commits August 7, 2020 09:09

Merge branch 'branch-0.2' into table-compressor

64dcbf6

Fix handling of degenerate batches

9fc1cb3

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

Fix buffer leak

85bcea3

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

abellina reviewed Aug 7, 2020

View reviewed changes

sql-plugin/src/main/scala/com/nvidia/spark/rapids/MetaUtils.scala Show resolved Hide resolved

Add comment about degenerate batches potentially appearing compressed

1c9b220

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

abellina previously approved these changes Aug 8, 2020

View reviewed changes

Merge branch 'branch-0.2' into table-compressor

d6d2305

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

jlowe dismissed abellina’s stale review via d6d2305 August 10, 2020 13:48

revans2 previously approved these changes Aug 10, 2020

View reviewed changes

jlowe added 2 commits August 11, 2020 09:11

Use a shuffle compression codec of "none" to indicate no compression

18e4598

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

Merge branch 'branch-0.2' into table-compressor

30caee3

jlowe dismissed revans2’s stale review via 30caee3 August 11, 2020 14:24

Merge branch 'branch-0.2' into table-compressor

fd865f9

revans2 approved these changes Aug 11, 2020

View reviewed changes

jlowe merged commit 3985f67 into NVIDIA:branch-0.2 Aug 11, 2020

jlowe mentioned this pull request Aug 21, 2020

Fix data size metric always 0 when using RAPIDS shuffle #603

Merged

jlowe deleted the table-compressor branch September 10, 2021 15:31

pxLi pushed a commit to pxLi/spark-rapids that referenced this pull request May 12, 2022

sync clients in runner_proess. (NVIDIA#487)

3c35e87

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add framework for batch compression of shuffle partitions #487

Add framework for batch compression of shuffle partitions #487

jlowe commented Jul 31, 2020

jlowe commented Jul 31, 2020

jlowe commented Aug 3, 2020

abellina left a comment

abellina commented Aug 4, 2020

jlowe commented Aug 4, 2020

jlowe commented Aug 4, 2020

jlowe commented Aug 6, 2020

jlowe commented Aug 7, 2020

jlowe commented Aug 7, 2020

jlowe commented Aug 10, 2020

jlowe commented Aug 10, 2020

revans2 left a comment

revans2 Aug 10, 2020

jlowe Aug 10, 2020

jlowe Aug 11, 2020

jlowe commented Aug 11, 2020

jlowe commented Aug 11, 2020

Add framework for batch compression of shuffle partitions #487

Add framework for batch compression of shuffle partitions #487

Conversation

jlowe commented Jul 31, 2020

jlowe commented Jul 31, 2020

jlowe commented Aug 3, 2020

abellina left a comment

Choose a reason for hiding this comment

abellina commented Aug 4, 2020

jlowe commented Aug 4, 2020

jlowe commented Aug 4, 2020

jlowe commented Aug 6, 2020

jlowe commented Aug 7, 2020

jlowe commented Aug 7, 2020

jlowe commented Aug 10, 2020

jlowe commented Aug 10, 2020

revans2 left a comment

Choose a reason for hiding this comment

revans2 Aug 10, 2020

Choose a reason for hiding this comment

jlowe Aug 10, 2020

Choose a reason for hiding this comment

jlowe Aug 11, 2020

Choose a reason for hiding this comment

jlowe commented Aug 11, 2020

jlowe commented Aug 11, 2020