Merge pull request #1795 from ClickHouse/clientv2_optimize_reading_record

[client-v2] Fixed reusing map without creating new nodes

Author: chernser

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

@@ -39,6 +39,7 @@
 import java.util.TimeZone;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
 public abstract class AbstractBinaryFormatReader implements ClickHouseBinaryFormatReader {
@@ -73,6 +74,7 @@ protected AbstractBinaryFormatReader(InputStream inputStream, QuerySettings quer
     protected Map<String, Object> currentRecord = new ConcurrentHashMap<>();
     protected Map<String, Object> nextRecord = new ConcurrentHashMap<>();
 
+    protected AtomicBoolean nextRecordEmpty = new AtomicBoolean(true);
 
     /**
      * It is still internal method and should be used with care.
@@ -91,7 +93,9 @@ public boolean readRecord(Map<String, Object> record) throws IOException {
             try {
                 Object val = binaryStreamReader.readValue(column);
                 if (val != null) {
-                    record.put(column.getColumnName(),val);
+                    record.put(column.getColumnName(), val);
+                } else {
+                    record.remove(column.getColumnName());
                 }
                 firstColumn = false;
             } catch (EOFException e) {
@@ -121,15 +125,17 @@ public <T> T readValue(String colName) {
 
     @Override
     public boolean hasNext() {
-         return hasNext;
+        return hasNext;
     }
 
 
     protected void readNextRecord() {
         try {
-            nextRecord.clear();
+            nextRecordEmpty.set(true);
             if (!readRecord(nextRecord)) {
                 hasNext = false;
+            } else {
+                nextRecordEmpty.compareAndSet(true, false);
             }
         } catch (IOException e) {
             hasNext = false;
@@ -143,15 +149,14 @@ public Map<String, Object> next() {
             return null;
         }
 
-        if (!nextRecord.isEmpty()) {
+        if (!nextRecordEmpty.get()) {
             Map<String, Object> tmp = currentRecord;
             currentRecord = nextRecord;
             nextRecord = tmp;
             readNextRecord();
             return currentRecord;
         } else {
             try {
-                currentRecord.clear();
                 if (readRecord(currentRecord)) {
                     readNextRecord();
                     return currentRecord;

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

@@ -18,10 +18,8 @@ public Client chDirectClient(@Value("${db.url}") String dbUrl, @Value("${db.user
                 .setUsername(dbUser)
                 .setPassword(dbPassword)
                 .useNewImplementation(true) // using new transport layer implementation
-
-                // sets the maximum number of connections to the server at a time
-                // this is important for services handling many concurrent requests to ClickHouse
-                .setOption(ClickHouseHttpOption.MAX_OPEN_CONNECTIONS.getKey(), "2000")
+                .setLZ4UncompressedBufferSize(1050000) // increase a LZ4 buffer size
+                .setMaxConnections(50)
                 .build();
     }
 }