[client-v2] Add property for network buffer size

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

@@ -714,6 +714,19 @@ public Builder setSharedOperationExecutor(ExecutorService executorService) {
             return this;
         }
 
+        /**
+         * Set size of a buffers that are used to read/write data from the server. It is mainly used to copy data from
+         * a socket to application memory and visa-versa. Setting is applied for both read and write operations.
+         * Default is 8192 bytes.
+         *
+         * @param size - size in bytes
+         * @return
+         */
+        public Builder setClientNetworkBufferSize(int size) {
+            this.configuration.put("client_network_buffer_size", String.valueOf(size));
+            return this;
+        }
+
         public Client build() {
             this.configuration = setDefaults(this.configuration);
 
@@ -816,6 +829,10 @@ private Map<String, String> setDefaults(Map<String, String> userConfig) {
                 userConfig.put("connection_ttl", "-1");
             }
 
+            if (!userConfig.containsKey("client_network_buffer_size")) {
+                setClientNetworkBufferSize(8192);
+            }
+
             return userConfig;
         }
     }

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

@@ -18,6 +18,7 @@
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
 import org.apache.hc.client5.http.impl.io.BasicHttpClientConnectionManager;
+import org.apache.hc.client5.http.impl.io.ManagedHttpClientConnectionFactory;
 import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
 import org.apache.hc.client5.http.io.HttpClientConnectionManager;
 import org.apache.hc.client5.http.protocol.HttpClientContext;
@@ -33,7 +34,10 @@
 import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpStatus;
 import org.apache.hc.core5.http.NoHttpResponseException;
+import org.apache.hc.core5.http.config.CharCodingConfig;
+import org.apache.hc.core5.http.config.Http1Config;
 import org.apache.hc.core5.http.config.RegistryBuilder;
+import org.apache.hc.core5.http.impl.io.DefaultHttpResponseParserFactory;
 import org.apache.hc.core5.http.io.SocketConfig;
 import org.apache.hc.core5.http.io.entity.EntityTemplate;
 import org.apache.hc.core5.io.IOCallback;
@@ -161,6 +165,7 @@ private HttpClientConnectionManager poolConnectionManager(SSLContext sslContext,
         PoolingHttpClientConnectionManagerBuilder connMgrBuilder = PoolingHttpClientConnectionManagerBuilder.create()
                 .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.LAX);
 
+
         ConnectionReuseStrategy connectionReuseStrategy =
                 ConnectionReuseStrategy.valueOf(chConfiguration.get("connection_reuse_strategy"));
         switch (connectionReuseStrategy) {
@@ -181,6 +186,15 @@ private HttpClientConnectionManager poolConnectionManager(SSLContext sslContext,
                 connMgrBuilder::setMaxConnPerRoute);
 
 
+        int networkBufferSize = MapUtils.getInt(chConfiguration, "client_network_buffer_size");
+        ManagedHttpClientConnectionFactory connectionFactory = new ManagedHttpClientConnectionFactory(
+                Http1Config.custom()
+                        .setBufferSize(networkBufferSize)
+                        .build(),
+                CharCodingConfig.DEFAULT,
+                DefaultHttpResponseParserFactory.INSTANCE);
+
+        connMgrBuilder.setConnectionFactory(connectionFactory);
         connMgrBuilder.setSSLSocketFactory(new SSLConnectionSocketFactory(sslContext));
         connMgrBuilder.setDefaultSocketConfig(socketConfig);
         return connMgrBuilder.build();
@@ -272,6 +286,7 @@ public ClassicHttpResponse executeRequest(ClickHouseNode server, Map<String, Obj
             throw new RuntimeException(e);
         }
         HttpPost req = new HttpPost(uri);
+//        req.setVersion(new ProtocolVersion("HTTP", 1, 0)); // to disable chunk transfer encoding
         addHeaders(req, chConfiguration, requestConfig);
 
         RequestConfig httpReqConfig = RequestConfig.copy(baseRequestConfig)

examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/BigDatasetExamples.java

@@ -0,0 +1,157 @@
+package com.clickhouse.examples.client_v2;
+
+import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
+import com.clickhouse.client.api.metrics.ClientMetrics;
+import com.clickhouse.client.api.query.QueryResponse;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class BigDatasetExamples {
+
+    private final String endpoint;
+    private final String user;
+    private final String password;
+    private final String database;
+
+    public BigDatasetExamples(String endpoint, String user, String password, String database) {
+        this.endpoint = endpoint;
+        this.user = user;
+        this.password = password;
+        this.database = database;
+    }
+
+    /**
+     * Reads {@code system.numbers} table into a result set of numbers of different types.
+     *
+     */
+    void readBigSetOfNumbers(int limit, int iterations, int concurrency) {
+        Client client = new Client.Builder()
+                .addEndpoint(endpoint)
+                .setUsername(user)
+                .setPassword(password)
+                .setDefaultDatabase(database)
+                .compressServerResponse(false)
+                .compressClientRequest(false)
+                .setLZ4UncompressedBufferSize(1048576)
+                .useNewImplementation(true)
+                // when network buffer and socket buffer are the same size - it is less IO calls and more efficient
+                .setSocketRcvbuf(1_000_000)
+                .setClientNetworkBufferSize(1_000_000)
+                .setMaxConnections(20)
+                .build();
+        try {
+            client.ping(10); // warmup connections pool. required once per client.
+
+            Runnable task = () -> {
+                StringBuilder sb = new StringBuilder();
+
+                for (int i = 0; i < iterations; i++) {
+                    try {
+                        long[] stats = doReadNumbersSet(client, limit);
+                        for (long stat : stats) {
+                            sb.append(stat).append(", ");
+                        }
+                        sb.append("\n");
+                    } catch (Exception e) {
+                        log.error("Failed to read dataset", e);
+                    }
+                }
+
+                System.out.print(sb.toString());
+            };
+
+            System.out.println("Records, Read Time, Request Time, Server Time");
+            if (concurrency == 1) {
+                task.run();
+            } else {
+                ExecutorService executor = new ThreadPoolExecutor(concurrency, Integer.MAX_VALUE,
+                        60L, TimeUnit.SECONDS,
+                        new SynchronousQueue<Runnable>());
+
+                for (int i = 0; i < concurrency; i++) {
+                    executor.submit(task);
+                }
+
+                executor.shutdown();
+                executor.awaitTermination(3, TimeUnit.MINUTES);
+            }
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } finally {
+            client.close();
+        }
+    }
+
+    /**
+     * Does actual request and returns time stats in format:
+     *  [number of records, read time in ms, request initiation time in ms, server time in ms]
+     * @param client
+     * @param limit
+     * @return
+     */
+    private long[] doReadNumbersSet(Client client, int limit) {
+        final String query = DATASET_QUERY + " LIMIT " + limit;
+        try (QueryResponse response = client.query(query).get(3000, TimeUnit.MILLISECONDS)) {
+            ArrayList<com.clickhouse.demo_service.data.NumbersRecord> result = new ArrayList<>();
+
+            // iterable approach is more efficient for large datasets because it doesn't load all records into memory
+            ClickHouseBinaryFormatReader reader = Client.newBinaryFormatReader(response);
+
+            long start = System.nanoTime();
+            while (reader.next() != null) {
+                result.add(new com.clickhouse.demo_service.data.NumbersRecord(
+                        reader.getUUID("id"),
+                        reader.getLong("p1"),
+                        reader.getBigInteger("number"),
+                        reader.getFloat("p2"),
+                        reader.getDouble("p3")
+                ));
+            }
+            long duration = System.nanoTime() - start;
+
+            return new long[] { result.size(), TimeUnit.NANOSECONDS.toMillis(duration), response.getMetrics().getMetric(ClientMetrics.OP_DURATION).getLong(),
+                    TimeUnit.NANOSECONDS.toMillis(response.getServerTime()) };
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to fetch dataset", e);
+        }
+    }
+
+    private static final String DATASET_QUERY =
+            "SELECT generateUUIDv4() as id, " +
+                    "toUInt32(number) as p1, " +
+                    "number,  " +
+                    "toFloat32(number/100000) as p2, " +
+                    "toFloat64(number/100000) as p3" +
+                    " FROM system.numbers";
+
+    public static void main(String[] args) {
+        final String endpoint = System.getProperty("chEndpoint", "http://localhost:8123");
+        final String user = System.getProperty("chUser", "default");
+        final String password = System.getProperty("chPassword", "");
+        final String database = System.getProperty("chDatabase", "default");
+
+//        profilerDelay();
+
+        BigDatasetExamples examples = new BigDatasetExamples(endpoint, user, password, database);
+
+        examples.readBigSetOfNumbers(100_000, 100, 10);
+
+//        profilerDelay();
+    }
+
+    private static void profilerDelay() {
+        // Delay for a profiler
+        try {
+            Thread.sleep(30000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}

examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/data/NumbersRecord.java

@@ -0,0 +1,24 @@
+package com.clickhouse.demo_service.data;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.math.BigInteger;
+import java.util.UUID;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class NumbersRecord {
+
+    private UUID id;
+
+    private long p1;
+
+    private BigInteger number;
+
+    private float p2;
+
+    private double p3;
+}