configurable uncompressed buffer

Author: chernser

client-v2/src/main/java/com/clickhouse/client/api/Client.java

@@ -17,6 +17,7 @@
 import com.clickhouse.client.api.insert.InsertSettings;
 import com.clickhouse.client.api.insert.POJOSerializer;
 import com.clickhouse.client.api.insert.SerializerNotFoundException;
+import com.clickhouse.client.api.internal.ClickHouseLZ4OutputStream;
 import com.clickhouse.client.api.internal.ClientStatisticsHolder;
 import com.clickhouse.client.api.internal.ClientV1AdaptorHelper;
 import com.clickhouse.client.api.internal.HttpAPIClientHelper;
@@ -410,11 +411,32 @@ public Builder compressClientRequest(boolean enabled) {
             return this;
         }
 
+        /**
+         * Configures the client to use HTTP compression. In this case compression is controlled by
+         * http headers. Client compression will set {@code Content-Encoding: lz4} header and server
+         * compression will set {@code Accept-Encoding: lz4} header.
+         *
+         * @param enabled - indicates if http compression is enabled
+         * @return
+         */
         public Builder useHttpCompression(boolean enabled) {
             this.configuration.put("client.use_http_compression", String.valueOf(enabled));
             return this;
         }
 
+        /**
+         * Sets buffer size for uncompressed data in LZ4 compression.
+         * For outgoing data it is the size of a buffer that will be compressed.
+         * For incoming data it is the size of a buffer that will be decompressed.
+         *
+         * @param size - size of the buffer in bytes
+         * @return
+         */
+        public Builder setLZ4UncompressedBufferSize(int size) {
+            this.configuration.put("compression.lz4.uncompressed_buffer_size", String.valueOf(size));
+            return this;
+        }
+
         /**
          * Sets the default database name that will be used by operations if not specified.
          * @param database - actual default database name.
@@ -573,6 +595,10 @@ private Map<String, String> setDefaults(Map<String, String> userConfig) {
                         String.valueOf(ClickHouseClientOption.MAX_THREADS_PER_CLIENT.getDefaultValue()));
             }
 
+            if (!userConfig.containsKey("compression.lz4.uncompressed_buffer_size")) {
+                userConfig.put("compression.lz4.uncompressed_buffer_size",
+                        String.valueOf(ClickHouseLZ4OutputStream.UNCOMPRESSED_BUFF_SIZE));
+            }
             return userConfig;
         }
     }

client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStream.java

@@ -11,7 +11,6 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.StreamCorruptedException;
 import java.nio.ByteBuffer;
 
 public class ClickHouseLZ4InputStream extends InputStream {
@@ -26,11 +25,11 @@ public class ClickHouseLZ4InputStream extends InputStream {
     private byte[] tmpBuffer = new byte[1];
 
 
-    public ClickHouseLZ4InputStream(InputStream in, LZ4FastDecompressor decompressor) {
+    public ClickHouseLZ4InputStream(InputStream in, LZ4FastDecompressor decompressor, int bufferSize) {
         super();
         this.decompressor = decompressor;
         this.in = in;
-        this.buffer = ByteBuffer.allocate(8192);
+        this.buffer = ByteBuffer.allocate(bufferSize);
         this.buffer.limit(0);
     }
 

client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4OutputStream.java

@@ -1,18 +1,16 @@
 package com.clickhouse.client.api.internal;
 
-import com.clickhouse.data.ClickHouseByteUtils;
 import com.clickhouse.data.ClickHouseCityHash;
-import com.clickhouse.data.stream.Lz4InputStream;
 import net.jpountz.lz4.LZ4Compressor;
-import net.jpountz.lz4.LZ4FastDecompressor;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
-import java.nio.MappedByteBuffer;
 
 public class ClickHouseLZ4OutputStream extends OutputStream {
 
+    public static final int UNCOMPRESSED_BUFF_SIZE = 8192;
+
     private final ByteBuffer buffer;
 
     private final OutputStream out;
@@ -26,9 +24,9 @@ public class ClickHouseLZ4OutputStream extends OutputStream {
     private static int HEADER_LEN = 15; // 9 bytes for header, 6 bytes for checksum
 
 
-    public ClickHouseLZ4OutputStream(OutputStream out, LZ4Compressor compressor) {
+    public ClickHouseLZ4OutputStream(OutputStream out, LZ4Compressor compressor, int bufferSize) {
         super();
-        this.buffer = ByteBuffer.allocate(8192);
+        this.buffer = ByteBuffer.allocate(bufferSize);
         this.out = out;
         this.compressor = compressor;
         this.compressedBuffer = ByteBuffer.allocate(compressor.maxCompressedLength(buffer.capacity()) + HEADER_LEN);

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -329,7 +329,8 @@ private HttpEntity wrapEntity(HttpEntity httpEntity) {
         boolean clientCompression = chConfiguration.getOrDefault(ClickHouseClientOption.DECOMPRESS.getKey(), "false").equalsIgnoreCase("true");
         boolean useHttpCompression = chConfiguration.getOrDefault("client.use_http_compression", "false").equalsIgnoreCase("true");
         if (serverCompression || clientCompression) {
-            return new LZ4Entity(httpEntity, useHttpCompression, serverCompression, clientCompression);
+            return new LZ4Entity(httpEntity, useHttpCompression, serverCompression, clientCompression,
+                    MapUtils.getInt(chConfiguration, "compression.lz4.uncompressed_buffer_size"));
         } else {
             return httpEntity;
         }

client-v2/src/main/java/com/clickhouse/client/api/internal/LZ4Entity.java

@@ -17,16 +17,19 @@ class LZ4Entity implements HttpEntity {
 
     private HttpEntity httpEntity;
 
-    private boolean useHttpCompression;
-    private boolean serverCompression; 
-    private boolean clientCompression;
+    private final boolean useHttpCompression;
+    private final boolean serverCompression;
+    private final boolean clientCompression;
 
-    
-    LZ4Entity(HttpEntity httpEntity, boolean useHttpCompression, boolean serverCompression, boolean clientCompression) {
+    private final int bufferSize;
+
+    LZ4Entity(HttpEntity httpEntity, boolean useHttpCompression, boolean serverCompression, boolean clientCompression,
+              int bufferSize) {
         this.httpEntity = httpEntity;
         this.useHttpCompression = useHttpCompression;
         this.serverCompression = serverCompression;
         this.clientCompression = clientCompression;
+        this.bufferSize = bufferSize;
     }
 
     @Override
@@ -48,7 +51,8 @@ public InputStream getContent() throws IOException, UnsupportedOperationExceptio
                 return content;
             }
         } else if (serverCompression) {
-            return new ClickHouseLZ4InputStream(httpEntity.getContent(), LZ4Factory.fastestInstance().fastDecompressor());
+            return new ClickHouseLZ4InputStream(httpEntity.getContent(), LZ4Factory.fastestInstance().fastDecompressor(),
+                    bufferSize);
         } else {
             return httpEntity.getContent();
         }
@@ -59,7 +63,8 @@ public void writeTo(OutputStream outStream) throws IOException {
         if (clientCompression && useHttpCompression) {
             httpEntity.writeTo(new FramedLZ4CompressorOutputStream(outStream));
         } else if (clientCompression) {
-            httpEntity.writeTo(new ClickHouseLZ4OutputStream(outStream, LZ4Factory.fastestInstance().fastCompressor()));
+            httpEntity.writeTo(new ClickHouseLZ4OutputStream(outStream, LZ4Factory.fastestInstance().fastCompressor(),
+                    bufferSize));
         } else {
             httpEntity.writeTo(outStream);
         }

client-v2/src/main/java/com/clickhouse/client/api/internal/MapUtils.java

@@ -29,4 +29,16 @@ public static void applyInt(Map<String, String> map, String key, Consumer<Intege
             }
         }
     }
+
+    public static int getInt(Map<String, String> map, String key) {
+        String val = map.get(key);
+        if (val != null) {
+            try {
+                return Integer.parseInt(val);
+            } catch (NumberFormatException e) {
+                throw new RuntimeException("Invalid value for key " + key + ": " + val, e);
+            }
+        }
+        return 0;
+    }
 }