[#2362] feat(client): Add support of zstd parallel compression (#2363)

### What changes were proposed in this pull request?

Add support of zstd parallel compression

### Why are the changes needed?

for #2362 

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Existing tests

---------

Co-authored-by: Junfan Zhang <zhangjunfan@qiyi.com>

Author: zuston

common/src/main/java/org/apache/uniffle/common/compression/Codec.java

@@ -23,7 +23,6 @@
 import org.apache.uniffle.common.config.RssConf;
 
 import static org.apache.uniffle.common.config.RssClientConf.COMPRESSION_TYPE;
-import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_LEVEL;
 
 public abstract class Codec {
 
@@ -33,7 +32,7 @@ public static Optional<Codec> newInstance(RssConf rssConf) {
       case NONE:
         return Optional.empty();
       case ZSTD:
-        return Optional.of(ZstdCodec.getInstance(rssConf.get(ZSTD_COMPRESSION_LEVEL)));
+        return Optional.of(ZstdCodec.getInstance(rssConf));
       case SNAPPY:
         return Optional.of(SnappyCodec.getInstance());
       case NOOP:

common/src/main/java/org/apache/uniffle/common/compression/ZstdCodec.java

@@ -20,22 +20,29 @@
 import java.nio.ByteBuffer;
 
 import com.github.luben.zstd.Zstd;
+import com.github.luben.zstd.ZstdCompressCtx;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.uniffle.common.config.RssConf;
 import org.apache.uniffle.common.exception.RssException;
 
+import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_LEVEL;
+import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_WORKER_NUMBER;
+
 public class ZstdCodec extends Codec {
   private static final Logger LOGGER = LoggerFactory.getLogger(ZstdCodec.class);
 
   private int compressionLevel;
+  private int workerNumber;
 
   private static class LazyHolder {
     static final ZstdCodec INSTANCE = new ZstdCodec();
   }
 
-  public static ZstdCodec getInstance(int level) {
-    LazyHolder.INSTANCE.compressionLevel = level;
+  public static ZstdCodec getInstance(RssConf conf) {
+    LazyHolder.INSTANCE.compressionLevel = conf.get(ZSTD_COMPRESSION_LEVEL);
+    LazyHolder.INSTANCE.workerNumber = conf.get(ZSTD_COMPRESSION_WORKER_NUMBER);
     return LazyHolder.INSTANCE;
   }
 
@@ -69,7 +76,16 @@ public void decompress(ByteBuffer src, int uncompressedLen, ByteBuffer dst, int
 
   @Override
   public byte[] compress(byte[] src) {
-    return Zstd.compress(src, compressionLevel);
+    ZstdCompressCtx ctx = new ZstdCompressCtx();
+    try {
+      ctx.setLevel(compressionLevel);
+      if (workerNumber > 0) {
+        ctx.setWorkers(workerNumber);
+      }
+      return ctx.compress(src);
+    } finally {
+      ctx.close();
+    }
   }
 
   @Override

common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java

@@ -52,6 +52,13 @@ public class RssClientConf {
           .defaultValue(3)
           .withDescription("The zstd compression level, the default level is 3");
 
+  public static final ConfigOption<Integer> ZSTD_COMPRESSION_WORKER_NUMBER =
+      ConfigOptions.key("rss.client.io.compression.zstd.workerNumber")
+          .intType()
+          .defaultValue(-1)
+          .withDescription(
+              "Set the parallel compression worker number. This will not enabled by default");
+
   public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE =
       ConfigOptions.key("rss.client.shuffle.data.distribution.type")
           .enumType(ShuffleDataDistributionType.class)

docs/client_guide/client_guide.md

@@ -49,6 +49,7 @@ The important configuration of client is listed as following. These configuratio
 | <client_type>.rss.client.assignment.shuffle.nodes.max           | -1                                     | The number of required assignment shuffle servers. If it is less than 0 or equals to 0 or greater than the coordinator's config of "rss.coordinator.shuffle.nodes.max", it will use the size of "rss.coordinator.shuffle.nodes.max" default                                                                                                                                                                                                                                                  |
 | <client_type>.rss.client.io.compression.codec                   | lz4                                    | The compression codec is used to compress the shuffle data. Default codec is `lz4`. Other options are`ZSTD` and `SNAPPY`.                                                                                                                                                                                                                                                                                                                                                                    |
 | <client_type>.rss.client.io.compression.zstd.level              | 3                                      | The zstd compression level, the default level is 3                                                                                                                                                                                                                                                                                                                                                                                                                                           |
+| <client_type>.rss.client.io.compression.zstd.workerNumber       | -1                                     | Set zstd parallel compression worker number. This will not enabled by default                                                                                                                                                                                                                                                                                                                                                                                                                |
 | <client_type>.rss.client.shuffle.data.distribution.type         | NORMAL                                 | The type of partition shuffle data distribution, including normal and local_order. The default value is normal. Now this config is only valid in Spark3.x                                                                                                                                                                                                                                                                                                                                    |
 | <client_type>.rss.estimate.task.concurrency.dynamic.factor      | 1.0                                    | Between 0 and 1, used to estimate task concurrency, when the client is spark, it represents how likely is this part of the resource between spark.dynamicAllocation.minExecutors and spark.dynamicAllocation.maxExecutors to be allocated, when the client is mr, it represents how likely the resources of map and reduce are satisfied. Effective when <client_type>.rss.estimate.server.assignment.enabled=true or Coordinator's rss.coordinator.select.partition.strategy is CONTINUOUS. |
 | <client_type>.rss.estimate.server.assignment.enabled            | false                                  | Support mr and spark, whether to enable estimation of the number of ShuffleServers that need to be allocated based on the number of concurrent tasks.                                                                                                                                                                                                                                                                                                                                        |