Fix client-v2: detect the codec of a compressed response instead of assuming LZ4

CHANGELOG.md

@@ -5,6 +5,20 @@
 
 ### Breaking Changes
 
+- **[client-v2] The algorithm of a compressed response is now requested with `Accept-Encoding` and is no longer the
+  one the server picks.** A response was requested with the `compress=1` framing, whose codec the server chooses on
+  its own: ClickHouse `26.9` changed that codec from `LZ4` to `ZSTD(3)`, the framed output follows the built-in
+  default and no setting overrides it, so the client could not keep reading a response it asked for. A response is
+  now requested with the content coding of the new `client.compression_algorithm` property
+  (`Client.Builder#compressionAlgorithm`), which defaults to `LZ4` and keeps the algorithm of a compressed body the
+  same on every server version. Set the property to `ZSTD` to select the other algorithm; the client now brings
+  `com.github.luben:zstd-jni` itself, and the dependency of `clickhouse-jdbc` is no longer `provided`, so the shaded
+  jars carry it and an application needs no additional dependency. The property selects only *how* a body is
+  compressed - *whether* it is compressed stays with `compressServerResponse` and `compressClientRequest`. A client
+  that reads a compressed response now also sends `enable_http_compression=1`, which a user profile that forbids
+  setting changes (`readonly = 1`) rejects - such a profile has to use `readonly = 2` or read an uncompressed
+  response (`compressServerResponse(false)`).
+  (https://github.com/ClickHouse/clickhouse-java/issues/3105)
 - **[client-v2]** `com.clickhouse.client.api.observability.SpanSupport` now uses `QUERY` and `INSERT` operation
   constants (`QUERY <database>` and `INSERT <database>.<table>` span names). `db.operation.name` attribute is set to
   `INSERT` for insert operations and left unset for queries because SQL statements are not parsed on the client.
@@ -160,6 +174,13 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed every compressed read failing with `Invalid LZ4 magic byte: '-112'` against ClickHouse `26.9`
+  and later. The server chooses the codec of the `compress=1` framing the client requested and switched that codec to
+  `ZSTD(3)`, while the response reader asserted the LZ4 method byte of every block, so any query answered with a
+  compressed body died before the first row was parsed. The algorithm of a response is now requested with
+  `Accept-Encoding`, so the client reads the algorithm it asked for; see the breaking-changes entry above.
+  (https://github.com/ClickHouse/clickhouse-java/issues/3105)
+
 - **[jdbc-v2]** Added the non-reserved keywords `AGGREGATE`, `BOUNDED`, `EXTEND`, `HANDLER`, `IDLE`, `PROTOCOL`,
   `RECENT`, `TIMEOUT` and `UNORDERED` (ClickHouse `26.8+`; `IDLE`, `TIMEOUT` and `RECENT` come from the multi-word
   keywords `IDLE TIMEOUT` and `RECENT SAMPLES`) to the list of keywords allowed in identifier positions. The server

clickhouse-jdbc/pom.xml

@@ -61,7 +61,6 @@
         <dependency>
             <groupId>com.github.luben</groupId>
             <artifactId>zstd-jni</artifactId>
-            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>

client-v2/pom.xml

@@ -58,6 +58,11 @@
             <version>${lz4.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.luben</groupId>
+            <artifactId>zstd-jni</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
@@ -188,12 +193,6 @@
             <version>5.19.0</version>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>com.github.luben</groupId>
-            <artifactId>zstd-jni</artifactId>
-            <version>1.5.7-6</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.bouncycastle</groupId>
             <artifactId>bcprov-jdk18on</artifactId>

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

@@ -11,6 +11,7 @@
 import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader;
 import com.clickhouse.client.api.data_formats.internal.MapBackedRecord;
 import com.clickhouse.client.api.data_formats.internal.ProcessParser;
+import com.clickhouse.client.api.enums.CompressionAlgorithm;
 import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.enums.ProxyType;
 import com.clickhouse.client.api.enums.SSLMode;
@@ -659,10 +660,31 @@ public Builder compressClientRequest(boolean enabled) {
             return this;
         }
 
+        /**
+         * Algorithm of a compressed request or response body. The algorithm is requested with the HTTP
+         * content-coding of the operation, so a compressed body always uses the algorithm set here and
+         * never one the server picks on its own. Default is {@link CompressionAlgorithm#LZ4}.
+         * <p>
+         * The algorithm selects only how a body is compressed. Whether a body is compressed is controlled by
+         * {@link #compressServerResponse(boolean)} and {@link #compressClientRequest(boolean)}.
+         * <p>
+         * A request body follows this algorithm only together with {@link #useHttpCompression(boolean)};
+         * the ClickHouse framing of a request compressed without it is always LZ4.
+         *
+         * @param algorithm - algorithm of a compressed body
+         * @return same instance of the builder
+         */
+        public Builder compressionAlgorithm(CompressionAlgorithm algorithm) {
+            ValidationUtils.checkNotNull(algorithm, "algorithm");
+            this.configuration.put(ClientConfigProperties.COMPRESSION_ALGORITHM.getKey(), algorithm.name());
+            return this;
+        }
+
         /**
          * Configures the client to use HTTP compression. In this case compression is controlled by
-         * http headers. Client compression will set {@code Content-Encoding: lz4} header and server
-         * compression will set {@code Accept-Encoding: lz4} header. Default is false.
+         * http headers. Client compression will set the {@code Content-Encoding} header and server
+         * compression will set the {@code Accept-Encoding} header, both to the content coding of
+         * {@link #compressionAlgorithm(CompressionAlgorithm)}. Default is false.
          *
          * @param enabled - indicates if http compression is enabled
          * @return

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

@@ -2,6 +2,7 @@
 
 import com.clickhouse.client.api.data_formats.ClickHouseFormatReader;
 import com.clickhouse.client.api.data_formats.internal.AbstractBinaryFormatReader;
+import com.clickhouse.client.api.enums.CompressionAlgorithm;
 import com.clickhouse.client.api.enums.SSLMode;
 import com.clickhouse.client.api.internal.ClickHouseLZ4OutputStream;
 import com.clickhouse.data.ClickHouseDataType;
@@ -245,6 +246,26 @@ public Object parseValue(String value) {
                     .collect(Collectors.toList());
         }
     },
+
+    /**
+     * Algorithm of a compressed request or response body. The algorithm is requested with the HTTP
+     * content-coding of the operation ({@code Accept-Encoding} for a response, {@code Content-Encoding}
+     * for a request), so a compressed body always uses the algorithm the client asked for and never one
+     * the server picks on its own. The algorithm selects only how a body is compressed; whether a body is
+     * compressed is controlled by {@link #COMPRESS_SERVER_RESPONSE} and {@link #COMPRESS_CLIENT_REQUEST}.
+     * <p>
+     * The name of an algorithm and its content-coding token are both accepted, in any case.
+     * <p>
+     * Appended at the end of the enum on purpose: adding a constant in the middle would shift the ordinal
+     * of every following constant (see {@code docs/changes_checklist.md}).
+     */
+    COMPRESSION_ALGORITHM("client.compression_algorithm", CompressionAlgorithm.class,
+            CompressionAlgorithm.LZ4.name()) {
+        @Override
+        public Object parseValue(String value) {
+            return value == null ? null : CompressionAlgorithm.fromValue(value);
+        }
+    },
     ;
 
     private static final Logger LOG = LoggerFactory.getLogger(ClientConfigProperties.class);

client-v2/src/main/java/com/clickhouse/client/api/enums/CompressionAlgorithm.java

@@ -0,0 +1,62 @@
+package com.clickhouse.client.api.enums;
+
+/**
+ * Enumerates the compression algorithms the client can ask the server for and can apply itself.
+ *
+ * <p>The algorithm is requested with the HTTP content-coding of the operation - {@code Accept-Encoding}
+ * for a response and {@code Content-Encoding} for a request - so a compressed body always uses the
+ * algorithm of the request and never one the server picks on its own.</p>
+ *
+ * <p>The algorithm selects only <i>how</i> a body is compressed. <i>Whether</i> a body is compressed is
+ * controlled by the compression flags - {@code compressServerResponse} for a response and
+ * {@code compressClientRequest} for a request.</p>
+ *
+ * <ul>
+ *     <li>{@link #LZ4} - default. Needs {@code org.lz4:lz4-java}, which the client depends on.</li>
+ *     <li>{@link #ZSTD} - needs {@code com.github.luben:zstd-jni}, which the client depends on.</li>
+ * </ul>
+ */
+public enum CompressionAlgorithm {
+
+    /**
+     * ClickHouse LZ4. Default algorithm.
+     */
+    LZ4("lz4"),
+
+    /**
+     * Zstandard.
+     */
+    ZSTD("zstd");
+
+    private final String httpContentCoding;
+
+    CompressionAlgorithm(String httpContentCoding) {
+        this.httpContentCoding = httpContentCoding;
+    }
+
+    /**
+     * Returns the HTTP content-coding token of the algorithm, as used in the {@code Accept-Encoding}
+     * and {@code Content-Encoding} headers.
+     *
+     * @return content-coding token
+     */
+    public String getHttpContentCoding() {
+        return httpContentCoding;
+    }
+
+    /**
+     * Case-insensitive variant of {@link #valueOf(String)} that also accepts the content-coding token.
+     *
+     * @param value algorithm name or content-coding token in any case
+     * @return matching algorithm
+     * @throws IllegalArgumentException when the value does not match any algorithm
+     */
+    public static CompressionAlgorithm fromValue(String value) {
+        for (CompressionAlgorithm algorithm : values()) {
+            if (algorithm.name().equalsIgnoreCase(value) || algorithm.httpContentCoding.equalsIgnoreCase(value)) {
+                return algorithm;
+            }
+        }
+        throw new IllegalArgumentException("Unknown compression algorithm '" + value + "'");
+    }
+}

client-v2/src/main/java/com/clickhouse/client/api/insert/InsertSettings.java

@@ -2,6 +2,7 @@
 
 import com.clickhouse.client.api.Client;
 import com.clickhouse.client.api.ClientConfigProperties;
+import com.clickhouse.client.api.enums.CompressionAlgorithm;
 import com.clickhouse.client.api.Session;
 import com.clickhouse.client.api.internal.CommonSettings;
 import org.apache.hc.core5.http.HttpHeaders;
@@ -213,6 +214,20 @@ public InsertSettings compressClientRequest(boolean enabled) {
         return this;
     }
 
+    /**
+     * Algorithm of a compressed request or response body of this operation. The algorithm is requested with
+     * the HTTP content coding of the operation, so a compressed body always uses the algorithm set here.
+     * Whether the body is compressed is controlled by the compression flags of the client. Defaults to
+     * the algorithm of the client.
+     *
+     * @param algorithm - algorithm of a compressed body
+     * @return same instance of the settings
+     */
+    public InsertSettings compressionAlgorithm(CompressionAlgorithm algorithm) {
+        settings.setOption(ClientConfigProperties.COMPRESSION_ALGORITHM.getKey(), algorithm);
+        return this;
+    }
+
     public InsertSettings useHttpCompression(boolean enabled) {
         settings.setOption(ClientConfigProperties.USE_HTTP_COMPRESSION.getKey(), enabled);
         return this;

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

@@ -11,6 +11,7 @@
 import com.clickhouse.client.api.DataTransferException;
 import com.clickhouse.client.api.ServerException;
 import com.clickhouse.client.api.TransportException;
+import com.clickhouse.client.api.enums.CompressionAlgorithm;
 import com.clickhouse.client.api.enums.ProxyType;
 import com.clickhouse.client.api.enums.SSLMode;
 import com.clickhouse.client.api.http.ClickHouseHttpProto;
@@ -123,7 +124,6 @@ public class HttpAPIClientHelper {
 
     private static final int ERROR_BODY_BUFFER_SIZE = 1024; // Error messages are usually small
 
-    private final String DEFAULT_HTTP_COMPRESSION_ALGO = "lz4";
 
     private static final Pattern PATTERN_HEADER_VALUE_ASCII = Pattern.compile(
         "\\p{Graph}+(?:[ ]\\p{Graph}+)*");
@@ -163,7 +163,13 @@ public HttpAPIClientHelper(Map<String, Object> configuration, Object metricsRegi
         boolean usingServerCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(configuration);
         boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(configuration);
 
-        LOG.debug("client compression: {}, server compression: {}, http compression: {}", usingClientCompression, usingServerCompression, useHttpCompression);
+        CompressionAlgorithm algorithm = compressionAlgorithm(configuration);
+        LOG.debug("client compression: {}, server compression: {}, http compression: {}, algorithm: {}",
+                usingClientCompression, usingServerCompression, useHttpCompression, algorithm);
+        if (usingClientCompression && !useHttpCompression && algorithm != CompressionAlgorithm.LZ4) {
+            LOG.warn("Request compression uses LZ4 instead of {}: the ClickHouse framing of a request is LZ4 " +
+                    "unless http compression is used", algorithm);
+        }
 
         defaultRetryCauses = new HashSet<>(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(configuration));
         if (defaultRetryCauses.contains(ClientFaultCause.None)) {
@@ -779,9 +785,7 @@ private TransportResponse doExecuteRequest(TransportRequest transportRequest, Sp
                 spanRecorder.recordHttpStatus(requestSpan, httpResponse.getCode());
             }
 
-            httpResponse.setEntity(wrapResponseEntity(httpResponse.getEntity(),
-                    httpResponse.getCode(),
-                    requestConfig));
+            httpResponse.setEntity(wrapResponseEntity(httpResponse.getEntity()));
 
             if (httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) {
                 throw readError(req, httpResponse);
@@ -936,15 +940,16 @@ private void addHeaders(HttpPost req, Map<String, Object> requestConfig) {
         boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig);
         boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig);
         boolean appCompressedData = ClientConfigProperties.APP_COMPRESSED_DATA.getOrDefault(requestConfig);
+        CompressionAlgorithm algorithm = compressionAlgorithm(requestConfig);
 
-        if (useHttpCompression) {
-            if (serverCompression) {
-                setHeader(req, HttpHeaders.ACCEPT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO);
-            }
+        if (serverCompression) {
+            // the codec of a compressed response is the one requested here: the server picks its own
+            // default codec for the compress=1 framing and does not let a client select it
+            setHeader(req, HttpHeaders.ACCEPT_ENCODING, algorithm.getHttpContentCoding());
+        }
 
-            if (clientCompression && !appCompressedData) {
-                setHeader(req, HttpHeaders.CONTENT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO);
-            }
+        if (useHttpCompression && clientCompression && !appCompressedData) {
+            setHeader(req, HttpHeaders.CONTENT_ENCODING, algorithm.getHttpContentCoding());
         }
 
         for (String key : requestConfig.keySet()) {
@@ -983,13 +988,16 @@ private void addRequestParams(Map<String, Object> requestConfig, BiConsumer<Stri
         boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig);
 
         if (useHttpCompression) {
-            // enable_http_compression make server react on http header
+            // enable_http_compression makes the server react on the http header
             // for client side compression Content-Encoding should be set
             // for server side compression Accept-Encoding should be set
             consumer.accept(ClickHouseHttpProto.QPARAM_ENABLE_HTTP_COMPRESSION, "1");
         } else {
             if (serverCompression) {
-                consumer.accept(ClickHouseHttpProto.QPARAM_COMPRESS, "1");
+                // the response is requested with Accept-Encoding, which the server honours only for
+                // an http-compressed response; the compress=1 framing would be compressed with the
+                // codec of the server instead of the requested one
+                consumer.accept(ClickHouseHttpProto.QPARAM_ENABLE_HTTP_COMPRESSION, "1");
             }
             if (clientCompression) {
                 consumer.accept(ClickHouseHttpProto.QPARAM_DECOMPRESS, "1");
@@ -1018,6 +1026,20 @@ private void addStatementParams(Map<String, Object> requestConfig, BiConsumer<St
         }
     }
 
+    /**
+     * Algorithm of a compressed body of the operation. An operation may set the option to the name of an
+     * algorithm instead of a constant, because a per-operation option is not parsed when it is set.
+     */
+    private static CompressionAlgorithm compressionAlgorithm(Map<String, Object> requestConfig) {
+        Object value = requestConfig.get(ClientConfigProperties.COMPRESSION_ALGORITHM.getKey());
+        if (value == null) {
+            return ClientConfigProperties.COMPRESSION_ALGORITHM.getDefObjVal();
+        }
+        return value instanceof CompressionAlgorithm
+                ? (CompressionAlgorithm) value
+                : CompressionAlgorithm.fromValue(String.valueOf(value));
+    }
+
     private HttpEntity wrapRequestEntity(HttpEntity httpEntity, Map<String, Object> requestConfig) {
 
         boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig);
@@ -1036,21 +1058,14 @@ private HttpEntity wrapRequestEntity(HttpEntity httpEntity, Map<String, Object>
         }
     }
 
-    private HttpEntity wrapResponseEntity(HttpEntity httpEntity, int httpStatus, Map<String, Object> requestConfig) {
-        boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig);
-        boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig);
-
+    private HttpEntity wrapResponseEntity(HttpEntity httpEntity) {
         if (httpEntity.getContentEncoding() != null) {
-            // http compressed response
+            // the algorithm of a compressed response is the one the request asked for
             return new CompressedEntity(httpEntity, true, CompressorStreamFactory.getSingleton());
         }
 
-        // data compression
-        if (serverCompression && !(httpStatus == HttpStatus.SC_FORBIDDEN || httpStatus == HttpStatus.SC_UNAUTHORIZED)) {
-            int buffSize = ClientConfigProperties.COMPRESSION_LZ4_UNCOMPRESSED_BUF_SIZE.getOrDefault(requestConfig);
-            return new LZ4Entity(httpEntity, useHttpCompression, true, false, buffSize, true, lz4Factory);
-        }
-
+        // a response without a content coding is not compressed: the server answers an unsupported
+        // Accept-Encoding, and a request that asks for no compression, with a plain body
         return httpEntity;
     }