Patch v0.11.1

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

@@ -21,12 +21,12 @@ public abstract class CompressedBlockOutputStream extends OutputStream {
 
     private final ByteBuffer compressedBuffer;
 
-    public CompressedBlockOutputStream(OutputStream out, int bufferSize) {
+    public CompressedBlockOutputStream(OutputStream out, int bufferSize, int maxCompressedBufferSize) {
         super();
         LOG.debug("Using compressor with buffer size {}", bufferSize);
         this.inBuffer = ByteBuffer.allocate(bufferSize);
         this.out = out;
-        this.compressedBuffer = ByteBuffer.allocate(bufferSize + CompressedBlockInputStream.HEADER_LENGTH);
+        this.compressedBuffer = ByteBuffer.allocate(maxCompressedBufferSize + CompressedBlockInputStream.HEADER_LENGTH);
     }
 
     @Override
@@ -116,7 +116,7 @@ public static class LZ4OutputStream extends CompressedBlockOutputStream {
         private final LZ4Compressor compressor;
 
         public LZ4OutputStream(OutputStream out, LZ4Compressor compressor, int bufferSize) {
-            super(out, bufferSize);
+            super(out, bufferSize, compressor.maxCompressedLength(bufferSize));
             this.compressor = compressor;
         }
 
@@ -137,7 +137,7 @@ public static class ZSTDOutputStream extends CompressedBlockOutputStream {
         private static final int COMPRESSION_LEVEL = 3;
 
         public ZSTDOutputStream(OutputStream out, int bufferSize) {
-            super(out, bufferSize);
+            super(out, bufferSize, (int) Zstd.compressBound(bufferSize));
         }
 
         @Override

client-v2/src/test/java/com/clickhouse/client/api/internal/CompressedBlockOutputStreamTest.java

@@ -0,0 +1,72 @@
+package com.clickhouse.client.api.internal;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import java.util.function.Function;
+
+import net.jpountz.lz4.LZ4Factory;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class CompressedBlockOutputStreamTest {
+
+    private static final LZ4Factory LZ4_FACTORY = LZ4Factory.fastestInstance();
+
+    private static final int BUFFER_SIZE = CompressedBlockOutputStream.UNCOMPRESSED_BUFF_SIZE;
+
+    @DataProvider(name = "streamsAndPayloads")
+    public static Object[][] streamsAndPayloads() {
+        Function<OutputStream, CompressedBlockOutputStream> lz4 =
+                out -> new CompressedBlockOutputStream.LZ4OutputStream(out, LZ4_FACTORY.fastCompressor(), BUFFER_SIZE);
+        Function<OutputStream, CompressedBlockOutputStream> zstd =
+                out -> new CompressedBlockOutputStream.ZSTDOutputStream(out, BUFFER_SIZE);
+
+        return new Object[][] {
+                { "LZ4", "incompressible", lz4, incompressibleData(BUFFER_SIZE) },
+                { "ZSTD", "incompressible", zstd, incompressibleData(BUFFER_SIZE) },
+                { "LZ4", "compressible", lz4, compressibleData(BUFFER_SIZE) },
+                { "ZSTD", "compressible", zstd, compressibleData(BUFFER_SIZE) },
+        };
+    }
+
+    @Test(dataProvider = "streamsAndPayloads")
+    public void testFullBlockRoundTrip(String method, String dataKind,
+                                       Function<OutputStream, CompressedBlockOutputStream> streamFactory,
+                                       byte[] payload) throws IOException {
+        ByteArrayOutputStream compressed = new ByteArrayOutputStream();
+        try (CompressedBlockOutputStream out = streamFactory.apply(compressed)) {
+            out.write(payload);
+        }
+
+        try (CompressedBlockInputStream in = new CompressedBlockInputStream(
+                new ByteArrayInputStream(compressed.toByteArray()), LZ4_FACTORY.fastDecompressor(), BUFFER_SIZE)) {
+            ByteArrayOutputStream decompressed = new ByteArrayOutputStream();
+            byte[] buf = new byte[8192];
+            int n;
+            while ((n = in.read(buf)) != -1) {
+                decompressed.write(buf, 0, n);
+            }
+            Assert.assertEquals(decompressed.toByteArray(), payload, method + " " + dataKind + " round trip");
+        }
+    }
+
+    private static byte[] incompressibleData(int length) {
+        byte[] data = new byte[length];
+        new Random(42).nextBytes(data);
+        return data;
+    }
+
+    private static byte[] compressibleData(int length) {
+        byte[] pattern = "ClickHouse compressed block test line. ".getBytes(StandardCharsets.US_ASCII);
+        byte[] data = new byte[length];
+        for (int i = 0; i < length; i++) {
+            data[i] = pattern[i % pattern.length];
+        }
+        return data;
+    }
+}

history/latest/3302.md

@@ -0,0 +1,6 @@
+- **[client-v2]** Fixed `INSERT` failing with `LZ4Exception: maxDestLen is too small` (LZ4) or
+  `ZstdException: Destination buffer is too small` (ZSTD) when request compression is enabled
+  (`compressClientRequest(true)`) and a 64 KiB block of the data does not compress, for example random bytes. The
+  compressed output buffer was as large as the uncompressed block, but such a block grows when it is compressed. The
+  buffer is now sized with the worst-case bound of the codec (`LZ4Compressor.maxCompressedLength`,
+  `Zstd.compressBound`). The bug was introduced in 0.11.0. (https://github.com/ClickHouse/clickhouse-java/issues/3302)