Pipe: add compression level config for connector ZSTD compressor

iotdb-core/node-commons/pom.xml

@@ -150,6 +150,10 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-jexl3</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.github.luben</groupId>
+            <artifactId>zstd-jni</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -233,6 +233,7 @@ public class CommonConfig {
   private long pipeListeningQueueTransferSnapshotThreshold = 1000;
   private int pipeSnapshotExecutionMaxBatchSize = 1000;
   private double pipeRemainingTimeCommitRateSmoothingFactor = 0.5;
+  private int pipeConnectorRPCCompressionZSTDCompressorLevel = 0;
 
   private long twoStageAggregateMaxCombinerLiveTimeInMs = 8 * 60 * 1000L; // 8 minutes
   private long twoStageAggregateDataRegionInfoCacheTimeInMs = 3 * 60 * 1000L; // 3 minutes
@@ -1022,6 +1023,16 @@ public void setPipeRemainingTimeCommitRateSmoothingFactor(
     this.pipeRemainingTimeCommitRateSmoothingFactor = pipeRemainingTimeCommitRateSmoothingFactor;
   }
 
+  public int getPipeConnectorRPCCompressionZSTDCompressorLevel() {
+    return pipeConnectorRPCCompressionZSTDCompressorLevel;
+  }
+
+  public void setPipeConnectorRPCCompressionZSTDCompressorLevel(
+      int pipeConnectorRPCCompressionZSTDCompressorLevel) {
+    this.pipeConnectorRPCCompressionZSTDCompressorLevel =
+        pipeConnectorRPCCompressionZSTDCompressorLevel;
+  }
+
   public double getPipeAllSinksRateLimitBytesPerSecond() {
     return pipeAllSinksRateLimitBytesPerSecond;
   }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java

@@ -545,6 +545,12 @@ private void loadPipeProps(Properties properties) {
                 "pipe_remaining_time_commit_rate_smoothing_factor",
                 String.valueOf(config.getPipeRemainingTimeCommitRateSmoothingFactor()))));
 
+    config.setPipeConnectorRPCCompressionZSTDCompressorLevel(
+        Integer.parseInt(
+            properties.getProperty(
+                "pipe_connector_rpc_compression_zstd_compressor_level",
+                String.valueOf(config.getPipeConnectorRPCCompressionZSTDCompressorLevel()))));
+
     config.setTwoStageAggregateMaxCombinerLiveTimeInMs(
         Long.parseLong(
             properties.getProperty(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java

@@ -143,6 +143,10 @@ public double getPipeRemainingTimeCommitRateSmoothingFactor() {
     return COMMON_CONFIG.getPipeRemainingTimeCommitRateSmoothingFactor();
   }
 
+  public int getPipeConnectorRPCCompressionZSTDCompressorLevel() {
+    return COMMON_CONFIG.getPipeConnectorRPCCompressionZSTDCompressorLevel();
+  }
+
   /////////////////////////////// Meta Consistency ///////////////////////////////
 
   public boolean isSeperatedPipeHeartbeatEnabled() {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java

@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.commons.pipe.connector.compressor;
 
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -37,7 +39,10 @@ public class PipeCompressorFactory {
     COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_SNAPPY, new PipeSnappyCompressor());
     COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_GZIP, new PipeGZIPCompressor());
     COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_LZ4, new PipeLZ4Compressor());
-    COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_ZSTD, new PipeZSTDCompressor());
+    COMPRESSOR_NAME_TO_INSTANCE.put(
+        CONNECTOR_COMPRESSOR_ZSTD,
+        new PipeZSTDCompressor(
+            PipeConfig.getInstance().getPipeConnectorRPCCompressionZSTDCompressorLevel()));
     COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_LZMA2, new PipeLZMA2Compressor());
     COMPRESSOR_NAME_TO_INSTANCE = Collections.unmodifiableMap(COMPRESSOR_NAME_TO_INSTANCE);
   }
@@ -54,15 +59,20 @@ public static PipeCompressor getCompressor(String name) {
 
   static {
     COMPRESSOR_INDEX_TO_INSTANCE.put(
-        PipeCompressor.PipeCompressionType.SNAPPY.getIndex(), new PipeSnappyCompressor());
+        PipeCompressor.PipeCompressionType.SNAPPY.getIndex(),
+        COMPRESSOR_NAME_TO_INSTANCE.get(CONNECTOR_COMPRESSOR_SNAPPY));
     COMPRESSOR_INDEX_TO_INSTANCE.put(
-        PipeCompressor.PipeCompressionType.GZIP.getIndex(), new PipeGZIPCompressor());
+        PipeCompressor.PipeCompressionType.GZIP.getIndex(),
+        COMPRESSOR_NAME_TO_INSTANCE.get(CONNECTOR_COMPRESSOR_GZIP));
     COMPRESSOR_INDEX_TO_INSTANCE.put(
-        PipeCompressor.PipeCompressionType.LZ4.getIndex(), new PipeLZ4Compressor());
+        PipeCompressor.PipeCompressionType.LZ4.getIndex(),
+        COMPRESSOR_NAME_TO_INSTANCE.get(CONNECTOR_COMPRESSOR_LZ4));
     COMPRESSOR_INDEX_TO_INSTANCE.put(
-        PipeCompressor.PipeCompressionType.ZSTD.getIndex(), new PipeZSTDCompressor());
+        PipeCompressor.PipeCompressionType.ZSTD.getIndex(),
+        COMPRESSOR_NAME_TO_INSTANCE.get(CONNECTOR_COMPRESSOR_ZSTD));
     COMPRESSOR_INDEX_TO_INSTANCE.put(
-        PipeCompressor.PipeCompressionType.LZMA2.getIndex(), new PipeLZMA2Compressor());
+        PipeCompressor.PipeCompressionType.LZMA2.getIndex(),
+        COMPRESSOR_NAME_TO_INSTANCE.get(CONNECTOR_COMPRESSOR_LZMA2));
     COMPRESSOR_INDEX_TO_INSTANCE = Collections.unmodifiableMap(COMPRESSOR_INDEX_TO_INSTANCE);
   }
 

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeZSTDCompressor.java

@@ -19,29 +19,26 @@
 
 package org.apache.iotdb.commons.pipe.connector.compressor;
 
-import org.apache.tsfile.compress.ICompressor;
-import org.apache.tsfile.compress.IUnCompressor;
-import org.apache.tsfile.file.metadata.enums.CompressionType;
+import com.github.luben.zstd.Zstd;
 
 import java.io.IOException;
 
 public class PipeZSTDCompressor extends PipeCompressor {
 
-  private static final ICompressor COMPRESSOR = ICompressor.getCompressor(CompressionType.ZSTD);
-  private static final IUnCompressor DECOMPRESSOR =
-      IUnCompressor.getUnCompressor(CompressionType.ZSTD);
+  private final int compressionLevel;
 
-  public PipeZSTDCompressor() {
+  public PipeZSTDCompressor(int compressionLevel) {
     super(PipeCompressionType.ZSTD);
+    this.compressionLevel = compressionLevel;
   }
 
   @Override
   public byte[] compress(byte[] data) throws IOException {
-    return COMPRESSOR.compress(data);
+    return Zstd.compress(data, compressionLevel);
   }
 
   @Override
-  public byte[] decompress(byte[] byteArray) throws IOException {
-    return DECOMPRESSOR.uncompress(byteArray);
+  public byte[] decompress(byte[] byteArray) {
+    return Zstd.decompress(byteArray, (int) Zstd.decompressedSize(byteArray, 0, byteArray.length));
   }
 }