Skip to content

Commit 666e234

Browse files
authored
feat(ilp): add QwpQueryClient for QWP egress over WebSocket (#11)
1 parent d54f218 commit 666e234

78 files changed

Lines changed: 17901 additions & 31 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.gitmodules

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
[submodule "core/src/main/c/share/zstd"]
2+
path = core/src/main/c/share/zstd
3+
url = https://github.com/facebook/zstd.git

core/CMakeLists.txt

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,12 @@
11
cmake_minimum_required(VERSION 3.5)
22
project(questdb)
33

4+
# Required for zstd's huf_decompress_amd64.S to be assembled. Without this,
5+
# CMake silently drops the .S file from the build and the link fails at
6+
# _HUF_decompress4X1_usingDTable_internal_fast_asm_loop etc. (The asmlib
7+
# subdirectory enables ASM_NASM separately for Agner Fog's .asm files.)
8+
enable_language(ASM)
9+
410
include(ExternalProject)
511

612
set(CMAKE_CXX_STANDARD 17)
@@ -49,6 +55,42 @@ set(
4955
src/main/c/share/byte_sink.h
5056
)
5157

58+
# libzstd is included via a git submodule at src/main/c/share/zstd (pinned to
59+
# upstream tag v1.5.7). Covers the client side of the QWP egress compression
60+
# feature; the server-side compressor lives in the Rust qdbr crate and isn't
61+
# linked into this library. Only the decompress-only subset of upstream is
62+
# compiled -- the compress/ directory is left out entirely. zstd_jni.c is our
63+
# JNI glue and lives alongside the submodule (not inside) so upstream resets
64+
# don't disturb it.
65+
if (NOT EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib/zstd.h)
66+
message(FATAL_ERROR
67+
"libzstd submodule not initialised. Run:\n"
68+
" git submodule update --init --recursive\n"
69+
"from java-questdb-client/.")
70+
endif ()
71+
set(
72+
ZSTD_FILES
73+
src/main/c/share/zstd_jni.c
74+
src/main/c/share/zstd/lib/common/debug.c
75+
src/main/c/share/zstd/lib/common/entropy_common.c
76+
src/main/c/share/zstd/lib/common/error_private.c
77+
src/main/c/share/zstd/lib/common/fse_decompress.c
78+
src/main/c/share/zstd/lib/common/pool.c
79+
src/main/c/share/zstd/lib/common/threading.c
80+
src/main/c/share/zstd/lib/common/xxhash.c
81+
src/main/c/share/zstd/lib/common/zstd_common.c
82+
src/main/c/share/zstd/lib/decompress/huf_decompress.c
83+
src/main/c/share/zstd/lib/decompress/zstd_ddict.c
84+
src/main/c/share/zstd/lib/decompress/zstd_decompress_block.c
85+
src/main/c/share/zstd/lib/decompress/zstd_decompress.c
86+
)
87+
# x86_64-only hand-tuned Huffman decoder; C fallback kicks in when
88+
# ZSTD_DISABLE_ASM is set.
89+
if (ARCH_AMD64 AND NOT WIN32)
90+
list(APPEND ZSTD_FILES src/main/c/share/zstd/lib/decompress/huf_decompress_amd64.S)
91+
endif ()
92+
list(APPEND SOURCE_FILES ${ZSTD_FILES})
93+
5294
# JNI includes
5395
include_directories($ENV{JAVA_HOME}/include/)
5496

@@ -111,6 +153,20 @@ set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${OUTPUT})
111153

112154
add_library(questdb SHARED ${SOURCE_FILES})
113155

156+
# libzstd public header is at zstd/lib/zstd.h; internal headers live under
157+
# zstd/lib/common/. Both directories go on the include path so zstd_jni.c can
158+
# use the short include "zstd.h" and the upstream .c files can find their own
159+
# siblings without patching.
160+
target_include_directories(questdb PRIVATE
161+
${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib
162+
${CMAKE_CURRENT_SOURCE_DIR}/src/main/c/share/zstd/lib/common)
163+
164+
# Drop the zstd-internal hand-written amd64 assembly on platforms that can't
165+
# assemble it; libzstd falls back to a C implementation when this is set.
166+
if (NOT ARCH_AMD64 OR WIN32)
167+
target_compile_definitions(questdb PRIVATE ZSTD_DISABLE_ASM=1)
168+
endif ()
169+
114170
set(COMMON_OPTIONS "-Wno-gnu-anonymous-struct;-Wno-nested-anon-types;-Wno-unused-parameter;-fPIC;-fno-rtti;-fno-exceptions")
115171

116172
set(DEBUG_OPTIONS "-Wall;-pedantic;-Wextra;-g;-O0")

core/src/main/c/share/zstd

Submodule zstd added at f8745da

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

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
/*+*****************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
/*
26+
* JNI wrapper over the bundled libzstd (decompression only). The server ships
27+
* compression support in the Rust qdbr crate; this file covers the client
28+
* decompression path so RESULT_BATCH frames with FLAG_ZSTD can be decoded
29+
* without any external native dependency.
30+
*
31+
* libzstd is vendored as a git submodule at share/zstd/ pinned to v1.5.7;
32+
* this file lives alongside (not inside) the submodule so upstream resets
33+
* don't nuke our JNI glue.
34+
*/
35+
36+
#include <jni.h>
37+
#include <stdint.h>
38+
#include <stdlib.h>
39+
#include "zstd.h"
40+
41+
JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Zstd_createDCtx(
42+
JNIEnv *env, jclass cls) {
43+
return (jlong) (uintptr_t) ZSTD_createDCtx();
44+
}
45+
46+
JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Zstd_getFrameContentSize(
47+
JNIEnv *env, jclass cls,
48+
jlong src_addr, jlong src_len) {
49+
/*
50+
* Peeks the zstd frame header at src_addr to recover the declared
51+
* uncompressed size. Returns:
52+
* positive -- declared content size in bytes
53+
* -1 -- frame valid, content size not stored (ZSTD_CONTENTSIZE_UNKNOWN)
54+
* -2 -- invalid frame, truncated header, or size > INT64_MAX
55+
*
56+
* Lets the Java caller size the destination buffer in a single allocation
57+
* instead of retrying decompress on dst-too-small. Crucially, it also lets
58+
* a corrupt frame fail BEFORE any output buffer growth, eliminating a
59+
* memory-amplification vector where one bad frame would have driven
60+
* scratch growth all the way to the 64 MiB cap.
61+
*/
62+
if (src_len < 0 || (src_len > 0 && src_addr == 0)) {
63+
return -2;
64+
}
65+
unsigned long long size = ZSTD_getFrameContentSize(
66+
(const void *) (uintptr_t) src_addr, (size_t) src_len);
67+
if (size == ZSTD_CONTENTSIZE_UNKNOWN) {
68+
return -1;
69+
}
70+
if (size == ZSTD_CONTENTSIZE_ERROR) {
71+
return -2;
72+
}
73+
if (size > (unsigned long long) INT64_MAX) {
74+
/* Cast to jlong would wrap to negative and look like an error code;
75+
* reject upfront so the caller doesn't double-interpret. */
76+
return -2;
77+
}
78+
return (jlong) size;
79+
}
80+
81+
JNIEXPORT void JNICALL Java_io_questdb_client_std_Zstd_freeDCtx(
82+
JNIEnv *env, jclass cls, jlong ptr) {
83+
if (ptr != 0) {
84+
ZSTD_freeDCtx((ZSTD_DCtx *) (uintptr_t) ptr);
85+
}
86+
}
87+
88+
JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Zstd_decompress(
89+
JNIEnv *env, jclass cls,
90+
jlong ctx,
91+
jlong src_addr, jlong src_len,
92+
jlong dst_addr, jlong dst_cap) {
93+
if (ctx == 0) {
94+
return -1;
95+
}
96+
ZSTD_DCtx *dctx = (ZSTD_DCtx *) (uintptr_t) ctx;
97+
size_t n = ZSTD_decompressDCtx(
98+
dctx,
99+
(void *) (uintptr_t) dst_addr, (size_t) dst_cap,
100+
(const void *) (uintptr_t) src_addr, (size_t) src_len);
101+
if (ZSTD_isError(n)) {
102+
unsigned code = ZSTD_getErrorCode(n);
103+
return -(jlong) code;
104+
}
105+
return (jlong) n;
106+
}

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

Lines changed: 43 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,17 @@ public abstract class WebSocketClient implements QuietCloseable {
105105
private CharSequence host;
106106
private int port;
107107
// QWP version negotiation
108+
// Verbatim header value sent as X-QWP-Accept-Encoding during upgrade, e.g.
109+
// "zstd;level=3,raw". When null, the header is omitted and the server ships
110+
// batches uncompressed. The echoed X-QWP-Content-Encoding response header
111+
// is intentionally not parsed: the RESULT_BATCH decoder branches on
112+
// FLAG_ZSTD in every frame, which is the authoritative signal.
113+
private String qwpAcceptEncoding;
108114
private String qwpClientId;
115+
// Client-requested per-batch row cap advertised via X-QWP-Max-Batch-Rows.
116+
// 0 means "omit the header" (server uses its default cap). Server may clamp
117+
// down to its own hard limit.
118+
private int qwpMaxBatchRows;
109119
private int qwpMaxVersion = 1;
110120
// Opt-in for STATUS_DURABLE_ACK frames; sent as X-QWP-Request-Durable-Ack: true
111121
private boolean qwpRequestDurableAck;
@@ -378,13 +388,35 @@ public void sendPing(int timeout) {
378388
}
379389
}
380390

391+
/**
392+
* Sets the value sent as the {@code X-QWP-Accept-Encoding} upgrade header,
393+
* e.g. {@code "zstd;level=3,raw"}. Pass {@code null} to omit the header
394+
* entirely (server ships uncompressed batches). Must be called before
395+
* {@link #upgrade}.
396+
*/
397+
public void setQwpAcceptEncoding(String acceptEncoding) {
398+
this.qwpAcceptEncoding = acceptEncoding;
399+
}
400+
381401
/**
382402
* Sets the QWP client identifier sent in the X-QWP-Client-Id upgrade header.
383403
*/
384404
public void setQwpClientId(String clientId) {
385405
this.qwpClientId = clientId;
386406
}
387407

408+
/**
409+
* Sets the client's preferred per-batch row cap, sent in the
410+
* {@code X-QWP-Max-Batch-Rows} upgrade header. {@code 0} (the default)
411+
* omits the header entirely and the server uses its own cap. Positive
412+
* values ask the server to flush batches sooner (lower time-to-first-row
413+
* for streaming consumers, at the cost of more per-batch overhead); the
414+
* server clamps down to its own maximum.
415+
*/
416+
public void setQwpMaxBatchRows(int maxBatchRows) {
417+
this.qwpMaxBatchRows = maxBatchRows;
418+
}
419+
388420
/**
389421
* Sets the maximum QWP version this client supports, sent in the X-QWP-Max-Version upgrade header.
390422
*/
@@ -488,6 +520,16 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
488520
sendBuffer.putAscii(qwpClientId);
489521
sendBuffer.putAscii("\r\n");
490522
}
523+
if (qwpAcceptEncoding != null) {
524+
sendBuffer.putAscii("X-QWP-Accept-Encoding: ");
525+
sendBuffer.putAscii(qwpAcceptEncoding);
526+
sendBuffer.putAscii("\r\n");
527+
}
528+
if (qwpMaxBatchRows > 0) {
529+
sendBuffer.putAscii("X-QWP-Max-Batch-Rows: ");
530+
sendBuffer.putAscii(Integer.toString(qwpMaxBatchRows));
531+
sendBuffer.putAscii("\r\n");
532+
}
491533
if (qwpRequestDurableAck) {
492534
sendBuffer.putAscii("X-QWP-Request-Durable-Ack: true\r\n");
493535
}
@@ -973,7 +1015,7 @@ private void validateUpgradeResponse(int headerEnd) {
9731015
throw new HttpClientException("Invalid Sec-WebSocket-Accept header");
9741016
}
9751017

976-
// Extract X-QWP-Version (optional defaults to 1 if absent)
1018+
// Extract X-QWP-Version (optional, defaults to 1 if absent)
9771019
serverQwpVersion = extractQwpVersion(response);
9781020
}
9791021

0 commit comments

Comments
 (0)