optimized buffer allocations for lz4input stream

Author: chernser

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.nio.ByteBuffer;
 
 public class ClickHouseLZ4InputStream extends InputStream {
 
@@ -20,7 +19,13 @@ public class ClickHouseLZ4InputStream extends InputStream {
 
     private final InputStream in;
 
-    private ByteBuffer buffer;
+    private byte[] outBuffer;
+
+    private int position = 0;
+
+    private int limit = 0;
+
+    private byte[] inBuffer;
 
     private byte[] tmpBuffer = new byte[1];
 
@@ -29,8 +34,8 @@ public ClickHouseLZ4InputStream(InputStream in, LZ4FastDecompressor decompressor
         super();
         this.decompressor = decompressor;
         this.in = in;
-        this.buffer = ByteBuffer.allocate(bufferSize);
-        this.buffer.limit(0);
+        this.outBuffer = new byte[bufferSize];
+        this.inBuffer = new byte[bufferSize];
     }
 
     @Override
@@ -55,8 +60,9 @@ public int read(byte[] b, int off, int len) throws IOException {
 
         int readBytes = 0;
         do {
-            int remaining = Math.min(len - readBytes, buffer.remaining());
-            buffer.get(b, off + readBytes, remaining);
+            int remaining = Math.min(len - readBytes, limit - position);
+            System.arraycopy(outBuffer, position, b, off + readBytes, remaining);
+            position += remaining;
             readBytes += remaining;
         } while (readBytes < len && refill() != -1);
 
@@ -109,30 +115,33 @@ private int refill() throws IOException {
         int uncompressedSize = getInt32(headerBuff, 21);
 
         int offset = 9;
-        final byte[] block =  new byte[compressedSizeWithHeader];
-        block[0] = MAGIC;
-        setInt32(block, 1, compressedSizeWithHeader);
-        setInt32(block, 5, uncompressedSize);
+        if (inBuffer.length < compressedSizeWithHeader) {
+            LOG.debug("compressed buff too small, new size: {}", compressedSizeWithHeader);
+        }
+        inBuffer = inBuffer.length >= compressedSizeWithHeader ? inBuffer : new byte[(int) (compressedSizeWithHeader * 1.5)];
+        inBuffer[0] = MAGIC;
+        setInt32(inBuffer, 1, compressedSizeWithHeader);
+        setInt32(inBuffer, 5, uncompressedSize);
         // compressed data: compressed_size - 9 bytes
         int remaining = compressedSizeWithHeader - offset;
 
-        readFully = readFully(block, offset, remaining);
+        readFully = readFully(inBuffer, offset, remaining);
         if (!readFully) {
             throw new EOFException("Unexpected end of stream");
         }
 
-        long[] real = ClickHouseCityHash.cityHash128(block, 0, compressedSizeWithHeader);
+        long[] real = ClickHouseCityHash.cityHash128(inBuffer, 0, compressedSizeWithHeader);
         if (real[0] != getInt64(headerBuff, 0) || real[1] != ClickHouseByteUtils.getInt64(headerBuff, 8)) {
             throw new ClientException("Corrupted stream: checksum mismatch");
         }
 
-        if (buffer.capacity() < uncompressedSize) {
-            buffer = ByteBuffer.allocate(uncompressedSize);
-            LOG.warn("Buffer size is too small, reallocate buffer with size: " + uncompressedSize);
+        if (outBuffer.length < uncompressedSize) {
+            outBuffer = new byte[uncompressedSize];
+            LOG.debug("uncompressed buff too small, new size: {}", uncompressedSize);
         }
-        decompressor.decompress(ByteBuffer.wrap(block), offset,  buffer, 0, uncompressedSize);
-        buffer.position(0);
-        buffer.limit(uncompressedSize);
+        decompressor.decompress(inBuffer, offset, outBuffer, 0, uncompressedSize);
+        position = 0;
+        limit = uncompressedSize;
         return uncompressedSize;
     }
 

examples/demo-service/src/main/java/com/clickhouse/demo_service/DatasetController.java

@@ -61,18 +61,6 @@ public void setup() {
         datasetQuerySchema = chDirectClient.getTableSchemaFromQuery(DATASET_QUERY);
         chDirectClient.register(VirtualDatasetRecord.class, datasetQuerySchema);
         log.info("Dataset schema: " + datasetQuerySchema.getColumns());
-
-        pool = new BasicObjectsPool<>(new ConcurrentLinkedDeque<>(), 100) {
-            @Override
-            ObjectsPreparedCollection<VirtualDatasetRecord> create() {
-                return new ObjectsPreparedCollection<>(new LinkedList<>(), MAX_LIMIT) {
-                    @Override
-                    VirtualDatasetRecord create() {
-                        return new VirtualDatasetRecord();
-                    }
-                };
-            }
-        };
     }
 
     /**
@@ -124,7 +112,7 @@ public List<VirtualDatasetRecord> directDatasetFetch(@RequestParam(name = "limit
                     response.getMetrics().getMetric(ClientMetrics.OP_DURATION).getLong(),
                     TimeUnit.NANOSECONDS.toMillis(response.getServerTime())));
 
-            return result.stream().findFirst().stream().collect(Collectors.toCollection(ArrayList::new));
+            return result;
         } catch (Exception e) {
             throw new RuntimeException("Failed to fetch dataset", e);
         }
@@ -195,6 +183,25 @@ public CalculationResult directDatasetReadToPojo(@RequestParam(name = "limit", r
     private CalculationResult readToPOJO(int limit, boolean cache) {
         final String query = DATASET_QUERY + " LIMIT " + limit;
         List<VirtualDatasetRecord> result = null;
+
+        if (cache) {
+            synchronized (this) {
+                if (pool == null) {
+                    pool = new BasicObjectsPool<>(new ConcurrentLinkedDeque<>(), 100) {
+                        @Override
+                        ObjectsPreparedCollection<VirtualDatasetRecord> create() {
+                            return new ObjectsPreparedCollection<>(new LinkedList<>(), MAX_LIMIT) {
+                                @Override
+                                VirtualDatasetRecord create() {
+                                    return new VirtualDatasetRecord();
+                                }
+                            };
+                        }
+                    };
+                }
+            }
+        }
+
         Supplier<VirtualDatasetRecord> objectsPool = cache ? this.pool.lease()
                 : VirtualDatasetRecord::new;
         try  {

examples/demo-service/src/main/java/com/clickhouse/demo_service/DbConfiguration.java

@@ -11,7 +11,7 @@
 public class DbConfiguration {
 
     @Bean
-    public Client chDirectClient(LoggingMeterRegistry loggingMeterRegistry, @Value("${db.url}") String dbUrl, @Value("${db.user}") String dbUser,
+    public Client chDirectClient(@Value("${db.url}") String dbUrl, @Value("${db.user}") String dbUser,
                                  @Value("${db.pass}") String dbPassword) {
         return new Client.Builder()
                 .addEndpoint(dbUrl)
@@ -21,13 +21,10 @@ public Client chDirectClient(LoggingMeterRegistry loggingMeterRegistry, @Value("
                 // sets the maximum number of connections to the server at a time
                 // this is important for services handling many concurrent requests to ClickHouse
                 .setMaxConnections(100)
-                .setLZ4UncompressedBufferSize(1058576)
-                .setSocketRcvbuf(500_000)
                 .setSocketTcpNodelay(true)
-                .setSocketSndbuf(500_000)
-                .setClientNetworkBufferSize(500_000)
+                .setLZ4UncompressedBufferSize(1 * 1024 * 1024) // 1MB buffer size
+                .setClientNetworkBufferSize(1 * 1024 * 1024) // 1MB buffer size
                 .allowBinaryReaderToReuseBuffers(true) // using buffer pool for binary reader
-                .registerClientMetrics(loggingMeterRegistry, "clickhouse-client-metrics")
                 .build();
     }
 }