zstd

Author: bluestreak01

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "core/src/main/c/share/zstd"]
+	path = core/src/main/c/share/zstd
+	url = https://github.com/facebook/zstd.git

core/CMakeLists.txt

@@ -49,6 +49,42 @@ set(
         src/main/c/share/byte_sink.h
 )
 
+# libzstd is included via a git submodule at src/main/c/share/zstd (pinned to
+# upstream tag v1.5.7). Covers the client side of the QWP egress compression
+# feature; the server-side compressor lives in the Rust qdbr crate and isn't
+# linked into this library. Only the decompress-only subset of upstream is
+# compiled -- the compress/ directory is left out entirely. zstd_jni.c is our
+# JNI glue and lives alongside the submodule (not inside) so upstream resets
+# don't disturb it.
+if (NOT EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib/zstd.h)
+    message(FATAL_ERROR
+        "libzstd submodule not initialised. Run:\n"
+        "  git submodule update --init --recursive\n"
+        "from java-questdb-client/.")
+endif ()
+set(
+        ZSTD_FILES
+        src/main/c/share/zstd_jni.c
+        src/main/c/share/zstd/lib/common/debug.c
+        src/main/c/share/zstd/lib/common/entropy_common.c
+        src/main/c/share/zstd/lib/common/error_private.c
+        src/main/c/share/zstd/lib/common/fse_decompress.c
+        src/main/c/share/zstd/lib/common/pool.c
+        src/main/c/share/zstd/lib/common/threading.c
+        src/main/c/share/zstd/lib/common/xxhash.c
+        src/main/c/share/zstd/lib/common/zstd_common.c
+        src/main/c/share/zstd/lib/decompress/huf_decompress.c
+        src/main/c/share/zstd/lib/decompress/zstd_ddict.c
+        src/main/c/share/zstd/lib/decompress/zstd_decompress_block.c
+        src/main/c/share/zstd/lib/decompress/zstd_decompress.c
+)
+# x86_64-only hand-tuned Huffman decoder; C fallback kicks in when
+# ZSTD_DISABLE_ASM is set.
+if (ARCH_AMD64 AND NOT WIN32)
+    list(APPEND ZSTD_FILES src/main/c/share/zstd/lib/decompress/huf_decompress_amd64.S)
+endif ()
+list(APPEND SOURCE_FILES ${ZSTD_FILES})
+
 # JNI includes
 include_directories($ENV{JAVA_HOME}/include/)
 
@@ -111,6 +147,20 @@ set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${OUTPUT})
 
 add_library(questdb SHARED ${SOURCE_FILES})
 
+# libzstd public header is at zstd/lib/zstd.h; internal headers live under
+# zstd/lib/common/. Both directories go on the include path so zstd_jni.c can
+# use the short include "zstd.h" and the upstream .c files can find their own
+# siblings without patching.
+target_include_directories(questdb PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib
+        ${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib/common)
+
+# Drop the zstd-internal hand-written amd64 assembly on platforms that can't
+# assemble it; libzstd falls back to a C implementation when this is set.
+if (NOT ARCH_AMD64 OR WIN32)
+    target_compile_definitions(questdb PRIVATE ZSTD_DISABLE_ASM=1)
+endif ()
+
 set(COMMON_OPTIONS "-Wno-gnu-anonymous-struct;-Wno-nested-anon-types;-Wno-unused-parameter;-fPIC;-fno-rtti;-fno-exceptions")
 
 set(DEBUG_OPTIONS "-Wall;-pedantic;-Wextra;-g;-O0")

core/src/main/c/share/zstd

@@ -0,0 +1,70 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+/*
+ * JNI wrapper over the bundled libzstd (decompression only). The server ships
+ * compression support in the Rust qdbr crate; this file covers the client
+ * decompression path so RESULT_BATCH frames with FLAG_ZSTD can be decoded
+ * without any external native dependency.
+ *
+ * libzstd is vendored as a git submodule at share/zstd/ pinned to v1.5.7;
+ * this file lives alongside (not inside) the submodule so upstream resets
+ * don't nuke our JNI glue.
+ */
+
+#include <jni.h>
+#include <stdlib.h>
+#include "zstd.h"
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Zstd_createDCtx(
+        JNIEnv *env, jclass cls) {
+    return (jlong) (uintptr_t) ZSTD_createDCtx();
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_std_Zstd_freeDCtx(
+        JNIEnv *env, jclass cls, jlong ptr) {
+    if (ptr != 0) {
+        ZSTD_freeDCtx((ZSTD_DCtx *) (uintptr_t) ptr);
+    }
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Zstd_decompress(
+        JNIEnv *env, jclass cls,
+        jlong ctx,
+        jlong src_addr, jlong src_len,
+        jlong dst_addr, jlong dst_cap) {
+    if (ctx == 0) {
+        return -1;
+    }
+    ZSTD_DCtx *dctx = (ZSTD_DCtx *) (uintptr_t) ctx;
+    size_t n = ZSTD_decompressDCtx(
+            dctx,
+            (void *) (uintptr_t) dst_addr, (size_t) dst_cap,
+            (const void *) (uintptr_t) src_addr, (size_t) src_len);
+    if (ZSTD_isError(n)) {
+        unsigned code = ZSTD_getErrorCode(n);
+        return -(jlong) code;
+    }
+    return (jlong) n;
+}

core/src/main/c/share/zstd_jni.c

@@ -105,6 +105,12 @@ public abstract class WebSocketClient implements QuietCloseable {
     private CharSequence host;
     private int port;
     // QWP version negotiation
+    // Verbatim header value sent as X-QWP-Accept-Encoding during upgrade, e.g.
+    // "zstd;level=3,raw". When null, the header is omitted and the server ships
+    // batches uncompressed. The echoed X-QWP-Content-Encoding response header
+    // is intentionally not parsed: the RESULT_BATCH decoder branches on
+    // FLAG_ZSTD in every frame, which is the authoritative signal.
+    private String qwpAcceptEncoding;
     private String qwpClientId;
     private int qwpMaxVersion = 1;
     // Receive buffer (native memory)
@@ -376,6 +382,16 @@ public void sendPing(int timeout) {
         }
     }
 
+    /**
+     * Sets the value sent as the {@code X-QWP-Accept-Encoding} upgrade header,
+     * e.g. {@code "zstd;level=3,raw"}. Pass {@code null} to omit the header
+     * entirely (server ships uncompressed batches). Must be called before
+     * {@link #upgrade}.
+     */
+    public void setQwpAcceptEncoding(String acceptEncoding) {
+        this.qwpAcceptEncoding = acceptEncoding;
+    }
+
     /**
      * Sets the QWP client identifier sent in the X-QWP-Client-Id upgrade header.
      */
@@ -476,6 +492,11 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
             sendBuffer.putAscii(qwpClientId);
             sendBuffer.putAscii("\r\n");
         }
+        if (qwpAcceptEncoding != null) {
+            sendBuffer.putAscii("X-QWP-Accept-Encoding: ");
+            sendBuffer.putAscii(qwpAcceptEncoding);
+            sendBuffer.putAscii("\r\n");
+        }
         if (authorizationHeader != null) {
             sendBuffer.putAscii("Authorization: ");
             sendBuffer.putAscii(authorizationHeader);
@@ -958,7 +979,7 @@ private void validateUpgradeResponse(int headerEnd) {
             throw new HttpClientException("Invalid Sec-WebSocket-Accept header");
         }
 
-        // Extract X-QWP-Version (optional — defaults to 1 if absent)
+        // Extract X-QWP-Version (optional, defaults to 1 if absent)
         serverQwpVersion = extractQwpVersion(response);
     }