Set enableCompression to true by default. Add tablet compression min size.

Author: jt2594838

iotdb-client/session/src/main/java/org/apache/iotdb/session/AbstractSessionBuilder.java

@@ -75,14 +75,16 @@ public abstract class AbstractSessionBuilder {
 
   public String database;
 
-  public Boolean isIoTDBRpcCompressionEnabled = false;
+  public Boolean isIoTDBRpcCompressionEnabled = true;
 
   public Boolean isThriftRpcCompactionEnabled = false;
 
   public CompressionType compressionType = CompressionType.UNCOMPRESSED;
 
   public Map<TSDataType, TSEncoding> columnEncodersMap;
 
+  public int tabletCompressionMinRowSize = 10;
+
   {
     columnEncodersMap = new HashMap<>();
     columnEncodersMap.put(TSDataType.TIMESTAMP, TSEncoding.TS_2DIFF);

iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java

@@ -138,8 +138,9 @@ public class Session implements ISession {
    */
   private long queryTimeoutInMs = -1;
 
-  protected boolean enableRPCCompaction;
-  protected boolean enableRPCCompression;
+  protected boolean enableThriftRpcCompaction;
+  protected boolean enableIoTDBRpcCompression = true;
+  protected int tabletCompressionMinRowSize = 10;
   protected int connectionTimeoutInMs;
   protected ZoneId zoneId;
   protected int thriftDefaultBufferSize;
@@ -454,8 +455,9 @@ public Session(AbstractSessionBuilder builder) {
       this.defaultEndPoint = new TEndPoint(builder.host, builder.rpcPort);
       this.enableQueryRedirection = builder.enableRedirection;
     }
-    this.enableRPCCompaction = builder.isThriftRpcCompactionEnabled;
-    this.enableRPCCompression = builder.isIoTDBRpcCompressionEnabled;
+    this.enableThriftRpcCompaction = builder.isThriftRpcCompactionEnabled;
+    this.enableIoTDBRpcCompression = builder.isIoTDBRpcCompressionEnabled;
+    this.tabletCompressionMinRowSize = builder.tabletCompressionMinRowSize;
     this.compressionType = builder.compressionType;
     this.columnEncodersMap = builder.columnEncodersMap;
     this.enableRedirection = builder.enableRedirection;
@@ -584,7 +586,7 @@ private List<TEndPoint> getNodeUrls() {
 
   @Override
   public synchronized void open(
-      boolean enableRPCCompression,
+      boolean enableThriftRpcCompaction,
       int connectionTimeoutInMs,
       Map<String, TEndPoint> deviceIdToEndpoint,
       INodeSupplier nodesSupplier)
@@ -594,7 +596,7 @@ public synchronized void open(
     }
 
     this.availableNodes = nodesSupplier;
-    this.enableRPCCompression = enableRPCCompression;
+    this.enableThriftRpcCompaction = enableThriftRpcCompaction;
     this.connectionTimeoutInMs = connectionTimeoutInMs;
     setDefaultSessionConnection(constructSessionConnection(this, defaultEndPoint, zoneId));
     getDefaultSessionConnection().setEnableRedirect(enableQueryRedirection);
@@ -609,7 +611,7 @@ public synchronized void open(
 
   @Override
   public synchronized void open(
-      boolean enableRPCCompression,
+      boolean enableThriftRpcCompaction,
       int connectionTimeoutInMs,
       Map<String, TEndPoint> deviceIdToEndpoint,
       Map<IDeviceID, TEndPoint> tableModelDeviceIdToEndpoint,
@@ -620,7 +622,7 @@ public synchronized void open(
     }
 
     this.availableNodes = nodesSupplier;
-    this.enableRPCCompression = enableRPCCompression;
+    this.enableThriftRpcCompaction = enableThriftRpcCompaction;
     this.connectionTimeoutInMs = connectionTimeoutInMs;
     setDefaultSessionConnection(constructSessionConnection(this, defaultEndPoint, zoneId));
     getDefaultSessionConnection().setEnableRedirect(enableQueryRedirection);
@@ -3004,8 +3006,11 @@ private TSInsertTabletReq genTSInsertTabletReq(Tablet tablet, boolean sorted, bo
     request.setPrefixPath(tablet.getDeviceId());
     request.setIsAligned(isAligned);
 
+    boolean trulyEnableRpcCompression =
+        enableIoTDBRpcCompression && tablet.getRowSize() >= tabletCompressionMinRowSize;
+
     List<Byte> encodingTypes;
-    if (enableRPCCompression) {
+    if (trulyEnableRpcCompression) {
       encodingTypes = new ArrayList<>(tablet.getSchemas().size() + 1);
       encodingTypes.add(this.columnEncodersMap.get(TSDataType.INT64).serialize());
       for (IMeasurementSchema measurementSchema : tablet.getSchemas()) {
@@ -3021,11 +3026,11 @@ private TSInsertTabletReq genTSInsertTabletReq(Tablet tablet, boolean sorted, bo
 
     TabletEncoder encoder =
         new TabletEncoder(
-            enableRPCCompression ? this.compressionType : CompressionType.UNCOMPRESSED,
+            trulyEnableRpcCompression ? this.compressionType : CompressionType.UNCOMPRESSED,
             encodingTypes.stream().map(TSEncoding::deserialize).collect(Collectors.toList()));
 
-    request.setIsCompressed(this.enableRPCCompression);
-    if (this.enableRPCCompression) {
+    request.setIsCompressed(trulyEnableRpcCompression);
+    if (trulyEnableRpcCompression) {
       request.setCompressType(compressionType.serialize());
       request.setEncodingTypes(encodingTypes);
     }
@@ -4238,12 +4243,12 @@ protected void setDefaultSessionConnection(SessionConnection defaultSessionConne
     this.defaultSessionConnection = defaultSessionConnection;
   }
 
-  public boolean isEnableRPCCompaction() {
-    return enableRPCCompaction;
+  public boolean isEnableThriftRpcCompaction() {
+    return enableThriftRpcCompaction;
   }
 
-  public void setEnableRPCCompaction(boolean enableRPCCompaction) {
-    this.enableRPCCompaction = enableRPCCompaction;
+  public void setEnableThriftRpcCompaction(boolean enableThriftRpcCompaction) {
+    this.enableThriftRpcCompaction = enableThriftRpcCompaction;
   }
 
   public static class Builder extends AbstractSessionBuilder {

iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java

@@ -207,7 +207,7 @@ private void init(TEndPoint endPoint, boolean useSSL, String trustStore, String
       throw new IoTDBConnectionException(e);
     }
 
-    if (session.enableRPCCompaction) {
+    if (session.enableThriftRpcCompaction) {
       client = new IClientRPCService.Client(new TCompactProtocol(transport));
     } else {
       client = new IClientRPCService.Client(new TBinaryProtocol(transport));

iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java

@@ -369,6 +369,11 @@ public TableSessionBuilder withDateEncoding(TSEncoding tsEncoding) {
     return this;
   }
 
+  public TableSessionBuilder tabletCompressionMinRowSize(int tabletCompressionMinRowSize) {
+    this.tabletCompressionMinRowSize = tabletCompressionMinRowSize;
+    return this;
+  }
+
   /**
    * Builds and returns a configured {@link ITableSession} instance.
    *
@@ -382,7 +387,8 @@ public ITableSession build() throws IoTDBConnectionException {
     }
     this.sqlDialect = TABLE;
     Session newSession = new Session(this);
-    newSession.enableRPCCompression = isIoTDBRpcCompressionEnabled;
+    newSession.enableIoTDBRpcCompression = isIoTDBRpcCompressionEnabled;
+    newSession.tabletCompressionMinRowSize = tabletCompressionMinRowSize;
 
     try {
       newSession.open(isThriftRpcCompactionEnabled, connectionTimeoutInMs);