This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch yym_rpc_compress
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/yym_rpc_compress by this push:
new e3e9b69ffb6 fix compression
e3e9b69ffb6 is described below
commit e3e9b69ffb662c4a0073922f8c2093a362170415
Author: Tian Jiang <[email protected]>
AuthorDate: Thu Jul 17 14:20:25 2025 +0800
fix compression
---
.../apache/iotdb/it/env/remote/env/RemoteServerEnv.java | 4 ++--
.../apache/iotdb/session/it/IoTDBSessionCompressedIT.java | 5 +++--
.../apache/iotdb/session/rpccompress/TabletEncoder.java | 10 +++++++---
.../java/org/apache/iotdb/session/util/SessionUtils.java | 6 ++++++
.../src/main/java/org/apache/iotdb/util/TabletDecoder.java | 14 ++++++++------
5 files changed, 26 insertions(+), 13 deletions(-)
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
index c2308cfe103..824a2206ce8 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -246,7 +246,7 @@ public class RemoteServerEnv implements BaseEnv {
.maxSize(maxSize)
.fetchSize(SessionConfig.DEFAULT_FETCH_SIZE)
.waitToGetSessionTimeoutInMs(60_000)
- .enableCompression(false)
+ .enableThriftCompression(false)
.zoneId(null)
.enableRedirection(SessionConfig.DEFAULT_REDIRECTION_MODE)
.connectionTimeoutInMs(SessionConfig.DEFAULT_CONNECTION_TIMEOUT_MS)
@@ -267,7 +267,7 @@ public class RemoteServerEnv implements BaseEnv {
.maxSize(maxSize)
.fetchSize(SessionConfig.DEFAULT_FETCH_SIZE)
.waitToGetSessionTimeoutInMs(60_000)
- .enableCompression(false)
+ .enableThriftCompression(false)
.zoneId(null)
.enableRedirection(SessionConfig.DEFAULT_REDIRECTION_MODE)
.connectionTimeoutInMs(SessionConfig.DEFAULT_CONNECTION_TIMEOUT_MS)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionCompressedIT.java
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionCompressedIT.java
index 82d28c34ff6..b04ea0f83c2 100644
---
a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionCompressedIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionCompressedIT.java
@@ -22,7 +22,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.it.env.EnvFactory;
-import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.TableSessionBuilder;
@@ -58,8 +58,9 @@ public class IoTDBSessionCompressedIT {
List<String> nodeUrls =
EnvFactory.getEnv().getDataNodeWrapperList().stream()
- .map(AbstractNodeWrapper::getIpAndPortString)
+ .map(DataNodeWrapper::getIpAndPortString)
.collect(Collectors.toList());
+ // List<String> nodeUrls = Collections.singletonList("127.0.0.1:6667");
session1 =
new TableSessionBuilder()
.nodeUrls(nodeUrls)
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/rpccompress/TabletEncoder.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/rpccompress/TabletEncoder.java
index 7ad02a53bbc..49b88b92180 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/rpccompress/TabletEncoder.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/rpccompress/TabletEncoder.java
@@ -100,15 +100,19 @@ public class TabletEncoder {
private ByteBuffer compressBuffer(ByteBuffer buffer) {
if (compressionType != CompressionType.UNCOMPRESSED) {
ICompressor compressor = ICompressor.getCompressor(compressionType);
- byte[] compressed = new
byte[compressor.getMaxBytesForCompression(buffer.remaining())];
+ int uncompressedSize = buffer.remaining();
+ byte[] compressed = new
byte[compressor.getMaxBytesForCompression(uncompressedSize) + 4];
try {
int compressedLength =
compressor.compress(
buffer.array(),
buffer.arrayOffset() + buffer.position(),
- buffer.remaining(),
+ uncompressedSize,
compressed);
- buffer = ByteBuffer.wrap(compressed, 0, compressedLength);
+ buffer = ByteBuffer.wrap(compressed, 0, compressedLength + 4);
+ buffer.position(compressedLength);
+ buffer.putInt(uncompressedSize);
+ buffer.rewind();
} catch (IOException e) {
throw new IllegalStateException(e);
}
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
index 6867a7134f6..bae3665c10a 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
@@ -37,6 +37,7 @@ import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.util.ArrayList;
@@ -469,6 +470,11 @@ public class SessionUtils {
throw new UnSupportedDataTypeException(
String.format("Data type %s is not supported.", dataType));
}
+ try {
+ encoder.flush(outputStream);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
}
/* Used for table model insert only. */
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/util/TabletDecoder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/util/TabletDecoder.java
index 66afdc3dd03..198444cdc34 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/util/TabletDecoder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/util/TabletDecoder.java
@@ -88,12 +88,14 @@ public class TabletDecoder {
}
try {
int uncompressedLength =
compressedBuffer.getInt(compressedBuffer.limit() - 4);
- ByteBuffer output = ByteBuffer.allocate(uncompressedLength);
- byte[] compressedData = new byte[compressedBuffer.remaining() - 4];
- compressedBuffer.slice().get(compressedData);
- byte[] uncompressedData = unCompressor.uncompress(compressedData);
- output.put(uncompressedData);
- output.flip();
+ byte[] uncompressedBytes = new byte[uncompressedLength];
+ unCompressor.uncompress(
+ compressedBuffer.array(),
+ compressedBuffer.arrayOffset() + compressedBuffer.position(),
+ compressedBuffer.remaining() - 4,
+ uncompressedBytes,
+ 0);
+ ByteBuffer output = ByteBuffer.wrap(uncompressedBytes);
RPCServiceThriftHandlerMetrics.getInstance()
.recordDecompressLatencyTimer(System.nanoTime() -
startUncompressTime);
return output;
