From a76794568e22e953b10cf13b2aeb439cd0038c81 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E5=AD=90=E5=9D=A4?= <55695098+DanielWang2035@users.noreply.github.com> Date: Fri, 31 May 2024 15:54:27 +0800 Subject: [PATCH 1/9] Pipe: add compression level config for connector ZSTD compressor --- .../iotdb/commons/conf/CommonConfig.java | 11 ++++++++++ .../iotdb/commons/conf/CommonDescriptor.java | 6 +++++ .../iotdb/commons/pipe/config/PipeConfig.java | 4 ++++ .../compressor/PipeCompressorFactory.java | 22 ++++++++++++++----- .../compressor/PipeZSTDCompressor.java | 17 ++++++-------- 5 files changed, 44 insertions(+), 16 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 6e8c943400674..63c9b289116f2 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/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; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 5979a6f56e5e0..27d242767deee 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/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( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index 48debcce40816..123c442face28 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/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() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index 14ae972fc4f04..cc2b43598db00 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/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); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeZSTDCompressor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeZSTDCompressor.java index 72782353d689f..50e2e1f845cf7 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeZSTDCompressor.java +++ b/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)); } } From 104b841eb319b6990b84be18837c2c82846f9806 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E5=AD=90=E5=9D=A4?= <55695098+DanielWang2035@users.noreply.github.com> Date: Fri, 31 May 2024 16:16:53 +0800 Subject: [PATCH 2/9] update dependency --- iotdb-core/node-commons/pom.xml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/iotdb-core/node-commons/pom.xml b/iotdb-core/node-commons/pom.xml index c19924c2870df..594904c360c07 100644 --- a/iotdb-core/node-commons/pom.xml +++ b/iotdb-core/node-commons/pom.xml @@ -150,6 +150,10 @@ org.apache.commons commons-jexl3 + + com.github.luben + zstd-jni + org.mockito mockito-core From a26b6e3e48016cf8b46cd560de8c8a707b8f0a64 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E5=AD=90=E5=9D=A4?= <55695098+DanielWang2035@users.noreply.github.com> Date: Tue, 4 Jun 2024 18:19:00 +0800 Subject: [PATCH 3/9] make zstd level connector-specific --- .../iotdb/commons/conf/CommonConfig.java | 11 ----- .../iotdb/commons/conf/CommonDescriptor.java | 6 --- .../iotdb/commons/pipe/config/PipeConfig.java | 4 -- .../constant/PipeConnectorConstant.java | 10 +++++ .../compressor/PipeCompressorConfig.java | 39 ++++++++++++++++ .../compressor/PipeCompressorFactory.java | 44 ++++++++++++++----- .../request/PipeTransferCompressedReq.java | 3 +- .../connector/protocol/IoTDBConnector.java | 25 ++++++++++- 8 files changed, 109 insertions(+), 33 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorConfig.java diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index c4682438c41d3..b78f7dcb6fbf0 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -236,7 +236,6 @@ 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 @@ -1026,16 +1025,6 @@ public void setPipeRemainingTimeCommitRateSmoothingFactor( this.pipeRemainingTimeCommitRateSmoothingFactor = pipeRemainingTimeCommitRateSmoothingFactor; } - public int getPipeConnectorRPCCompressionZSTDCompressorLevel() { - return pipeConnectorRPCCompressionZSTDCompressorLevel; - } - - public void setPipeConnectorRPCCompressionZSTDCompressorLevel( - int pipeConnectorRPCCompressionZSTDCompressorLevel) { - this.pipeConnectorRPCCompressionZSTDCompressorLevel = - pipeConnectorRPCCompressionZSTDCompressorLevel; - } - public double getPipeAllSinksRateLimitBytesPerSecond() { return pipeAllSinksRateLimitBytesPerSecond; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 27d242767deee..5979a6f56e5e0 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -545,12 +545,6 @@ 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( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index 123c442face28..48debcce40816 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java @@ -143,10 +143,6 @@ public double getPipeRemainingTimeCommitRateSmoothingFactor() { return COMMON_CONFIG.getPipeRemainingTimeCommitRateSmoothingFactor(); } - public int getPipeConnectorRPCCompressionZSTDCompressorLevel() { - return COMMON_CONFIG.getPipeConnectorRPCCompressionZSTDCompressorLevel(); - } - /////////////////////////////// Meta Consistency /////////////////////////////// public boolean isSeperatedPipeHeartbeatEnabled() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java index 3d0bd4cab7468..fab31efb74753 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java @@ -22,6 +22,8 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import com.github.luben.zstd.Zstd; + import java.io.File; import java.util.Arrays; import java.util.Collections; @@ -177,6 +179,14 @@ public class PipeConnectorConstant { CONNECTOR_COMPRESSOR_ZSTD, CONNECTOR_COMPRESSOR_LZMA2))); + public static final String CONNECTOR_COMPRESSOR_ZSTD_LEVEL_KEY = + "connector.compressor.zstd.level"; + public static final String SINK_COMPRESSOR_ZSTD_LEVEL_KEY = "sink.compressor.zstd.level"; + public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE = + Zstd.defaultCompressionLevel(); + public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE = Zstd.maxCompressionLevel(); + public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE = 1; + public static final String CONNECTOR_RATE_LIMIT_KEY = "connector.rate-limit-bytes-per-second"; public static final String SINK_RATE_LIMIT_KEY = "sink.rate-limit-bytes-per-second"; public static final double CONNECTOR_RATE_LIMIT_DEFAULT_VALUE = -1; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorConfig.java new file mode 100644 index 0000000000000..c13028c4d31f0 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorConfig.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.pipe.connector.compressor; + +public class PipeCompressorConfig { + + private final String name; + private final int zstdCompressionLevel; + + public PipeCompressorConfig(String name, int zstdCompressionLevel) { + this.name = name; + this.zstdCompressionLevel = zstdCompressionLevel; + } + + public String getName() { + return name; + } + + public int getZstdCompressionLevel() { + return zstdCompressionLevel; + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index cc2b43598db00..4c2278be88aa3 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java @@ -19,20 +19,25 @@ package org.apache.iotdb.commons.pipe.connector.compressor; -import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Objects; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_GZIP; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_LZ4; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_LZMA2; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_SNAPPY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE; public class PipeCompressorFactory { + private static final Logger LOGGER = LoggerFactory.getLogger(PipeCompressorFactory.class); + private static Map COMPRESSOR_NAME_TO_INSTANCE = new HashMap<>(); static { @@ -41,18 +46,37 @@ public class PipeCompressorFactory { COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_LZ4, new PipeLZ4Compressor()); COMPRESSOR_NAME_TO_INSTANCE.put( CONNECTOR_COMPRESSOR_ZSTD, - new PipeZSTDCompressor( - PipeConfig.getInstance().getPipeConnectorRPCCompressionZSTDCompressorLevel())); + new PipeZSTDCompressor(CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE)); COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_LZMA2, new PipeLZMA2Compressor()); - COMPRESSOR_NAME_TO_INSTANCE = Collections.unmodifiableMap(COMPRESSOR_NAME_TO_INSTANCE); } - public static PipeCompressor getCompressor(String name) { - final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(name); - if (compressor == null) { - throw new UnsupportedOperationException("PipeCompressor not found for name: " + name); + public static PipeCompressor getCompressor(PipeCompressorConfig config) { + if (config == null) { + return null; + } + + if (Objects.equals(config.getName(), CONNECTOR_COMPRESSOR_ZSTD)) { + // For ZSTD compressor, we need to consider the compression level + String compressorKey = CONNECTOR_COMPRESSOR_ZSTD + "_" + config.getZstdCompressionLevel(); + if (COMPRESSOR_NAME_TO_INSTANCE.containsKey(compressorKey)) { + return COMPRESSOR_NAME_TO_INSTANCE.get(compressorKey); + } + + LOGGER.info("Create new ZSTD compressor with level: {}", config.getZstdCompressionLevel()); + final PipeZSTDCompressor newZstdCompressor = + new PipeZSTDCompressor(config.getZstdCompressionLevel()); + COMPRESSOR_NAME_TO_INSTANCE.put(compressorKey, newZstdCompressor); + return newZstdCompressor; + } else { + // For other compressors, we can directly get the instance by name + final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(config.getName()); + if (compressor != null) { + return compressor; + } + + throw new UnsupportedOperationException( + "PipeCompressor not found for name: " + config.getName()); } - return compressor; } private static Map COMPRESSOR_INDEX_TO_INSTANCE = new HashMap<>(); @@ -76,7 +100,7 @@ public static PipeCompressor getCompressor(String name) { COMPRESSOR_INDEX_TO_INSTANCE = Collections.unmodifiableMap(COMPRESSOR_INDEX_TO_INSTANCE); } - public static PipeCompressor getCompressor(byte index) { + public static PipeCompressor deserializeCompressorFromIndex(byte index) { final PipeCompressor compressor = COMPRESSOR_INDEX_TO_INSTANCE.get(index); if (compressor == null) { throw new UnsupportedOperationException("PipeCompressor not found for index: " + index); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java index f5e114949bd75..89261daf026c3 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java @@ -87,7 +87,8 @@ public static TPipeTransferReq fromTPipeTransferReq(final TPipeTransferReq trans final int compressorsSize = ReadWriteIOUtils.readByte(compressedBuffer); for (int i = 0; i < compressorsSize; ++i) { compressors.add( - PipeCompressorFactory.getCompressor(ReadWriteIOUtils.readByte(compressedBuffer))); + PipeCompressorFactory.deserializeCompressorFromIndex( + ReadWriteIOUtils.readByte(compressedBuffer))); } byte[] body = new byte[compressedBuffer.remaining()]; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java index f65daf86d4ff8..24b461aa068ac 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java @@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.pipe.connector.PipeReceiverStatusHandler; import org.apache.iotdb.commons.pipe.connector.compressor.PipeCompressor; +import org.apache.iotdb.commons.pipe.connector.compressor.PipeCompressorConfig; import org.apache.iotdb.commons.pipe.connector.compressor.PipeCompressorFactory; import org.apache.iotdb.commons.pipe.connector.limiter.GlobalRateLimiter; import org.apache.iotdb.commons.pipe.connector.limiter.PipeEndPointRateLimiter; @@ -50,6 +51,10 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_SET; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD_LEVEL_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXCEPTION_CONFLICT_RECORD_IGNORED_DATA_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXCEPTION_CONFLICT_RECORD_IGNORED_DATA_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXCEPTION_CONFLICT_RESOLVE_STRATEGY_DEFAULT_VALUE; @@ -72,6 +77,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_RATE_LIMIT_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_RATE_LIMIT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_COMPRESSOR_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_COMPRESSOR_ZSTD_LEVEL_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_EXCEPTION_CONFLICT_RECORD_IGNORED_DATA_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_EXCEPTION_CONFLICT_RESOLVE_STRATEGY_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_EXCEPTION_CONFLICT_RETRY_MAX_TIME_SECONDS_KEY; @@ -151,6 +157,21 @@ public void validate(PipeParameterValidator validator) throws Exception { CONNECTOR_LOAD_BALANCE_STRATEGY_SET, loadBalanceStrategy), loadBalanceStrategy); + final int zstdCompressionLevel = + parameters.getIntOrDefault( + Arrays.asList(CONNECTOR_COMPRESSOR_ZSTD_LEVEL_KEY, SINK_COMPRESSOR_ZSTD_LEVEL_KEY), + CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE); + validator.validate( + arg -> + zstdCompressionLevel >= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE + && zstdCompressionLevel <= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE, + String.format( + "Zstd compression level should be in the range [%d, %d], but got %d.", + CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE, + CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE, + zstdCompressionLevel), + zstdCompressionLevel); + final String compressionTypes = parameters .getStringOrDefault( @@ -170,7 +191,9 @@ public void validate(PipeParameterValidator validator) throws Exception { "Compressor should be one of %s, but got %s.", CONNECTOR_COMPRESSOR_SET, trimmedCompressionType), trimmedCompressionType); - compressors.add(PipeCompressorFactory.getCompressor(trimmedCompressionType)); + compressors.add( + PipeCompressorFactory.getCompressor( + new PipeCompressorConfig(trimmedCompressionType, zstdCompressionLevel))); } } validator.validate( From a1fc078ba0b38ad43bb7c7ece8bc9f80c2a0ff73 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Wed, 5 Jun 2024 21:08:12 +0800 Subject: [PATCH 4/9] refactor --- .../constant/PipeConnectorConstant.java | 2 +- .../compressor/PipeCompressorFactory.java | 20 +++++++++---------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java index fab31efb74753..8d6879ae8c23b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java @@ -184,8 +184,8 @@ public class PipeConnectorConstant { public static final String SINK_COMPRESSOR_ZSTD_LEVEL_KEY = "sink.compressor.zstd.level"; public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE = Zstd.defaultCompressionLevel(); + public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE = Zstd.minCompressionLevel(); public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE = Zstd.maxCompressionLevel(); - public static final int CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE = 1; public static final String CONNECTOR_RATE_LIMIT_KEY = "connector.rate-limit-bytes-per-second"; public static final String SINK_RATE_LIMIT_KEY = "sink.rate-limit-bytes-per-second"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index 4c2278be88aa3..f87fddf15f689 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java @@ -38,7 +38,7 @@ public class PipeCompressorFactory { private static final Logger LOGGER = LoggerFactory.getLogger(PipeCompressorFactory.class); - private static Map COMPRESSOR_NAME_TO_INSTANCE = new HashMap<>(); + private static final Map COMPRESSOR_NAME_TO_INSTANCE = new HashMap<>(); static { COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_SNAPPY, new PipeSnappyCompressor()); @@ -52,7 +52,7 @@ public class PipeCompressorFactory { public static PipeCompressor getCompressor(PipeCompressorConfig config) { if (config == null) { - return null; + throw new IllegalArgumentException("PipeCompressorConfig is null"); } if (Objects.equals(config.getName(), CONNECTOR_COMPRESSOR_ZSTD)) { @@ -67,16 +67,16 @@ public static PipeCompressor getCompressor(PipeCompressorConfig config) { new PipeZSTDCompressor(config.getZstdCompressionLevel()); COMPRESSOR_NAME_TO_INSTANCE.put(compressorKey, newZstdCompressor); return newZstdCompressor; - } else { - // For other compressors, we can directly get the instance by name - final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(config.getName()); - if (compressor != null) { - return compressor; - } + } - throw new UnsupportedOperationException( - "PipeCompressor not found for name: " + config.getName()); + // For other compressors, we can directly get the instance by name + final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(config.getName()); + if (compressor != null) { + return compressor; } + + throw new UnsupportedOperationException( + "PipeCompressor not found for name: " + config.getName()); } private static Map COMPRESSOR_INDEX_TO_INSTANCE = new HashMap<>(); From 35c3c27fc12a5dfd29dcab72210133f22744f7c2 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Wed, 5 Jun 2024 21:45:21 +0800 Subject: [PATCH 5/9] refactor --- .../compressor/PipeCompressorFactory.java | 33 ++++++++++--------- .../request/PipeTransferCompressedReq.java | 3 +- 2 files changed, 19 insertions(+), 17 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index f87fddf15f689..5d099efcd1640 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java @@ -25,7 +25,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; -import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_GZIP; import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_COMPRESSOR_LZ4; @@ -38,7 +38,8 @@ public class PipeCompressorFactory { private static final Logger LOGGER = LoggerFactory.getLogger(PipeCompressorFactory.class); - private static final Map COMPRESSOR_NAME_TO_INSTANCE = new HashMap<>(); + private static final Map COMPRESSOR_NAME_TO_INSTANCE = + new ConcurrentHashMap<>(); static { COMPRESSOR_NAME_TO_INSTANCE.put(CONNECTOR_COMPRESSOR_SNAPPY, new PipeSnappyCompressor()); @@ -54,19 +55,21 @@ public static PipeCompressor getCompressor(PipeCompressorConfig config) { if (config == null) { throw new IllegalArgumentException("PipeCompressorConfig is null"); } + if (config.getName() == null) { + throw new IllegalArgumentException("PipeCompressorConfig.getName() is null"); + } - if (Objects.equals(config.getName(), CONNECTOR_COMPRESSOR_ZSTD)) { - // For ZSTD compressor, we need to consider the compression level - String compressorKey = CONNECTOR_COMPRESSOR_ZSTD + "_" + config.getZstdCompressionLevel(); - if (COMPRESSOR_NAME_TO_INSTANCE.containsKey(compressorKey)) { - return COMPRESSOR_NAME_TO_INSTANCE.get(compressorKey); - } - - LOGGER.info("Create new ZSTD compressor with level: {}", config.getZstdCompressionLevel()); - final PipeZSTDCompressor newZstdCompressor = - new PipeZSTDCompressor(config.getZstdCompressionLevel()); - COMPRESSOR_NAME_TO_INSTANCE.put(compressorKey, newZstdCompressor); - return newZstdCompressor; + final String compressorName = config.getName(); + + // For ZSTD compressor, we need to consider the compression level + if (compressorName.equals(CONNECTOR_COMPRESSOR_ZSTD)) { + final int zstdCompressionLevel = config.getZstdCompressionLevel(); + return COMPRESSOR_NAME_TO_INSTANCE.computeIfAbsent( + CONNECTOR_COMPRESSOR_ZSTD + "_" + zstdCompressionLevel, + key -> { + LOGGER.info("Create new PipeZSTDCompressor with level: {}", zstdCompressionLevel); + return new PipeZSTDCompressor(zstdCompressionLevel); + }); } // For other compressors, we can directly get the instance by name @@ -100,7 +103,7 @@ public static PipeCompressor getCompressor(PipeCompressorConfig config) { COMPRESSOR_INDEX_TO_INSTANCE = Collections.unmodifiableMap(COMPRESSOR_INDEX_TO_INSTANCE); } - public static PipeCompressor deserializeCompressorFromIndex(byte index) { + public static PipeCompressor getCompressor(byte index) { final PipeCompressor compressor = COMPRESSOR_INDEX_TO_INSTANCE.get(index); if (compressor == null) { throw new UnsupportedOperationException("PipeCompressor not found for index: " + index); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java index 89261daf026c3..f5e114949bd75 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java @@ -87,8 +87,7 @@ public static TPipeTransferReq fromTPipeTransferReq(final TPipeTransferReq trans final int compressorsSize = ReadWriteIOUtils.readByte(compressedBuffer); for (int i = 0; i < compressorsSize; ++i) { compressors.add( - PipeCompressorFactory.deserializeCompressorFromIndex( - ReadWriteIOUtils.readByte(compressedBuffer))); + PipeCompressorFactory.getCompressor(ReadWriteIOUtils.readByte(compressedBuffer))); } byte[] body = new byte[compressedBuffer.remaining()]; From 3d0e5ccceea3fc1cfd15c883553139b76bc7bf8c Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Wed, 5 Jun 2024 21:47:48 +0800 Subject: [PATCH 6/9] Update PipeCompressorFactory.java --- .../pipe/connector/compressor/PipeCompressorFactory.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index 5d099efcd1640..b565c583730d2 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java @@ -73,13 +73,13 @@ public static PipeCompressor getCompressor(PipeCompressorConfig config) { } // For other compressors, we can directly get the instance by name - final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(config.getName()); + final PipeCompressor compressor = COMPRESSOR_NAME_TO_INSTANCE.get(compressorName); if (compressor != null) { return compressor; } throw new UnsupportedOperationException( - "PipeCompressor not found for name: " + config.getName()); + "PipeCompressor not found for name: " + compressorName); } private static Map COMPRESSOR_INDEX_TO_INSTANCE = new HashMap<>(); From c69b8f7ff59ef8130be4e29df07de8337d6a9bdc Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Wed, 5 Jun 2024 21:52:27 +0800 Subject: [PATCH 7/9] Update PipeCompressorFactory.java --- .../pipe/connector/compressor/PipeCompressorFactory.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java index b565c583730d2..bcee9b85aebe5 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/compressor/PipeCompressorFactory.java @@ -78,8 +78,7 @@ public static PipeCompressor getCompressor(PipeCompressorConfig config) { return compressor; } - throw new UnsupportedOperationException( - "PipeCompressor not found for name: " + compressorName); + throw new UnsupportedOperationException("PipeCompressor not found for name: " + compressorName); } private static Map COMPRESSOR_INDEX_TO_INSTANCE = new HashMap<>(); From 63a272ef434aeb3906c8a252f6f2b28578658069 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E5=AD=90=E5=9D=A4?= <55695098+DanielWang2035@users.noreply.github.com> Date: Thu, 6 Jun 2024 11:53:42 +0800 Subject: [PATCH 8/9] add IT --- .../IoTDBPipeConnectorCompressionIT.java | 126 ++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeConnectorCompressionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeConnectorCompressionIT.java index be4aa458d9fb0..f7c0c63b842e8 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeConnectorCompressionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeConnectorCompressionIT.java @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.it.env.MultiEnvFactory; @@ -36,11 +38,17 @@ import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import static org.junit.Assert.fail; + @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT2AutoCreateSchema.class}) public class IoTDBPipeConnectorCompressionIT extends AbstractPipeDualAutoIT { @@ -179,4 +187,122 @@ private void doTest( Collections.singleton("8,")); } } + + @Test + public void testZstdCompressorLevel() throws Exception { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + + final String receiverIp = receiverDataNode.getIp(); + final int receiverPort = receiverDataNode.getPort(); + + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( + "insert into root.db.d1(time, s1) values (1, 1)", + "insert into root.db.d1(time, s2) values (1, 1)", + "insert into root.db.d1(time, s3) values (1, 1)", + "insert into root.db.d1(time, s4) values (1, 1)", + "insert into root.db.d1(time, s5) values (1, 1)", + "flush"))) { + return; + } + + // Create 5 pipes with different zstd compression levels, p4 and p5 should fail. + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "create pipe p1" + + " with extractor ('extractor.pattern'='root.db.d1.s1')" + + " with connector (" + + "'connector.ip'='%s'," + + "'connector.port'='%s'," + + "'connector.compressor'='zstd, zstd'," + + "'connector.compressor.zstd.level'='3')", + receiverIp, receiverPort)); + } catch (SQLException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "create pipe p2" + + " with extractor ('extractor.pattern'='root.db.d1.s2')" + + " with connector (" + + "'connector.ip'='%s'," + + "'connector.port'='%s'," + + "'connector.compressor'='zstd, zstd'," + + "'connector.compressor.zstd.level'='22')", + receiverIp, receiverPort)); + } catch (SQLException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "create pipe p3" + + " with extractor ('extractor.pattern'='root.db.d1.s3')" + + " with connector (" + + "'connector.ip'='%s'," + + "'connector.port'='%s'," + + "'connector.compressor'='zstd, zstd'," + + "'connector.compressor.zstd.level'='-131072')", + receiverIp, receiverPort)); + } catch (SQLException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "create pipe p4" + + " with extractor ('extractor.pattern'='root.db.d1.s4')" + + " with connector (" + + "'connector.ip'='%s'," + + "'connector.port'='%s'," + + "'connector.compressor'='zstd, zstd'," + + "'connector.compressor.zstd.level'='-131073')", + receiverIp, receiverPort)); + fail(); + } catch (SQLException e) { + // Make sure the error message in IoTDBConnector.java is returned + Assert.assertTrue(e.getMessage().contains("Zstd compression level should be in the range")); + } + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "create pipe p5" + + " with extractor ('extractor.pattern'='root.db.d1.s5')" + + " with connector (" + + "'connector.ip'='%s'," + + "'connector.port'='%s'," + + "'connector.compressor'='zstd, zstd'," + + "'connector.compressor.zstd.level'='23')", + receiverIp, receiverPort)); + fail(); + } catch (SQLException e) { + // Make sure the error message in IoTDBConnector.java is returned + Assert.assertTrue(e.getMessage().contains("Zstd compression level should be in the range")); + } + + final List showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList; + Assert.assertEquals(3, showPipeResult.size()); + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton("3,")); + } + } } From 6f6372f015cbd742d096b1c66505f2f9feb15939 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E5=AD=90=E5=9D=A4?= <55695098+DanielWang2035@users.noreply.github.com> Date: Thu, 6 Jun 2024 12:57:45 +0800 Subject: [PATCH 9/9] fix --- .../iotdb/commons/pipe/connector/protocol/IoTDBConnector.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java index 24b461aa068ac..ad8e328e45e03 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java @@ -163,8 +163,8 @@ public void validate(PipeParameterValidator validator) throws Exception { CONNECTOR_COMPRESSOR_ZSTD_LEVEL_DEFAULT_VALUE); validator.validate( arg -> - zstdCompressionLevel >= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE - && zstdCompressionLevel <= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE, + (int) arg >= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE + && (int) arg <= CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MAX_VALUE, String.format( "Zstd compression level should be in the range [%d, %d], but got %d.", CONNECTOR_COMPRESSOR_ZSTD_LEVEL_MIN_VALUE,