fixed LZ4OutputStream to avoid compressing extra bytes when income buffer is empty

Author: chernser

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

@@ -1501,7 +1501,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
 
                                              byte[] buffer = new byte[writeBufferSize];
                                              int bytesRead;
-                                             while ((bytesRead = data.read(buffer)) != -1) {
+                                             while ((bytesRead = data.read(buffer)) > 0) {
                                                  out.write(buffer, 0, bytesRead);
                                              }
                                              out.close();

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

@@ -65,21 +65,23 @@ public void write( byte[] b, int off, int len) throws IOException {
 
     @Override
     public void flush() throws IOException {
-        compressedBuffer.clear();
-        compressedBuffer.put(16, ClickHouseLZ4InputStream.MAGIC);
-        int uncompressedLen = buffer.position();
-        buffer.flip();
-        int compressed = compressor.compress(buffer, 0, uncompressedLen, compressedBuffer, 25,
-                compressedBuffer.remaining() - 25);
-        int compressedSizeWithHeader = compressed + 9;
-        ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 17, compressedSizeWithHeader); // compressed size with header
-        ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 21, uncompressedLen); // uncompressed size
-        long[] hash = ClickHouseCityHash.cityHash128(compressedBuffer.array(), 16, compressedSizeWithHeader);
-        setInt64(compressedBuffer.array(), 0, hash[0]);
-        setInt64(compressedBuffer.array(), 8, hash[1]);
-        compressedBuffer.flip();
-        out.write(compressedBuffer.array(), 0, compressed + 25);
-        buffer.clear();
+        if (buffer.position() > 0) {
+            compressedBuffer.clear();
+            compressedBuffer.put(16, ClickHouseLZ4InputStream.MAGIC);
+            int uncompressedLen = buffer.position();
+            buffer.flip();
+            int compressed = compressor.compress(buffer, 0, uncompressedLen, compressedBuffer, 25,
+                    compressedBuffer.remaining() - 25);
+            int compressedSizeWithHeader = compressed + 9;
+            ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 17, compressedSizeWithHeader); // compressed size with header
+            ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 21, uncompressedLen); // uncompressed size
+            long[] hash = ClickHouseCityHash.cityHash128(compressedBuffer.array(), 16, compressedSizeWithHeader);
+            setInt64(compressedBuffer.array(), 0, hash[0]);
+            setInt64(compressedBuffer.array(), 8, hash[1]);
+            compressedBuffer.flip();
+            out.write(compressedBuffer.array(), 0, compressed + 25);
+            buffer.clear();
+        }
     }
 
 

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

@@ -73,6 +73,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import static com.clickhouse.client.api.ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT;
+
 public class HttpAPIClientHelper {
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
 
@@ -235,6 +237,9 @@ public CloseableHttpClient createHttpClient() {
                 soCfgBuilder::setSndBufSize);
         MapUtils.applyInt(chConfiguration, ClientConfigProperties.SOCKET_LINGER_OPT.getKey(),
                     (v) -> soCfgBuilder.setSoLinger(v, TimeUnit.SECONDS));
+        if (MapUtils.getFlag(chConfiguration, ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT.getKey(), false)) {
+            soCfgBuilder.setTcpNoDelay(true);
+        }
 
         // Proxy
         String proxyHost = chConfiguration.get(ClientConfigProperties.PROXY_HOST.getKey());

client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java

@@ -20,7 +20,6 @@
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.data.ClickHouseFormat;
 import com.clickhouse.data.ClickHouseVersion;
-import com.clickhouse.data.stream.ByteArrayQueueInputStream;
 import org.apache.commons.lang3.StringEscapeUtils;
 import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
 import org.testng.Assert;
@@ -38,9 +37,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.LinkedList;
 import java.util.List;
-import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
@@ -70,7 +67,6 @@ public InsertTests(boolean useClientCompression, boolean useHttpCompression) {
     public void setUp() throws IOException {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
         int bufferSize = (7 * 65500);
-
         client = new Client.Builder()
                 .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), false)
                 .setUsername("default")
@@ -81,6 +77,7 @@ public void setUp() throws IOException {
                 .setSocketRcvbuf(bufferSize)
                 .setClientNetworkBufferSize(bufferSize)
                 .build();
+
         settings = new InsertSettings()
                 .setDeduplicationToken(RandomStringUtils.randomAlphabetic(36))
                 .setQueryId(String.valueOf(UUID.randomUUID()));
@@ -233,38 +230,6 @@ public void insertRawData() throws Exception {
         assertEquals(records.size(), 1000);
     }
 
-
-    @Test(groups = { "integration" }, enabled = true)
-    public void insertRawDataQueued() throws Exception {
-        final String tableName = "raw_data_table";
-        final String createSQL = "CREATE TABLE " + tableName +
-                " (Id UInt32, event_ts Timestamp, name String, p1 Int64, p2 String) ENGINE = MergeTree() ORDER BY ()";
-
-        initTable(tableName, createSQL);
-        settings.setInputStreamCopyBufferSize(8198 * 2);
-        settings.compressClientRequest(true);
-        Queue<byte[]> queue = new LinkedList<>();
-        ByteArrayQueueInputStream qIn = new ByteArrayQueueInputStream(queue, null);
-        for (int i = 0; i < 10; i++) {
-            if (i > 2 && i < 5) {
-                queue.add(new byte[0]);
-            } else {
-                queue.add(String.format("{ \"Id\": %d, \"events_ts\": \"%s\", \"name\": \"%s\", \"p1\": \"%d\", \"p2\": \"%s\"}\n", i, "2021-01-01 00:00:00", "name" + i, i, "p2").getBytes());
-            }
-        }
-        InsertResponse response = client.insert(tableName, qIn,
-                ClickHouseFormat.JSONEachRow, settings).get(30, TimeUnit.SECONDS);
-
-        assertEquals((int)response.getWrittenRows(), 10 );
-
-        List<GenericRecord> records = client.queryAll("SELECT * FROM " + tableName);
-        assertEquals(records.size(), 10);
-        for (GenericRecord record : records) {
-            System.out.println(record.getString(1) + " " +record.getString(2) + " " +record.getString(3) + " " +record.getString(4) + " " +record.getString(5) + " ");
-        }
-    }
-
-
     @Test(groups = { "integration" }, enabled = true)
     public void insertRawDataSimple() throws Exception {
         insertRawDataSimple(1000);