Skip to content

Commit f58f766

Browse files
bluestreak01claude
bluestreak01
and
claude
committed
test(ilp): JMH ingress latency benchmark for QWP Sender
Adds the user-facing counterpart to QwpEgressLatencyBenchmark in the OSS repo. Measures end-to-end wall time of a single row .at()+flush() against a locally running QuestDB. Default mode is SF on, which measures user-handover latency: flush() returns when the row is durable on the local SF segment. -Dsf=false switches to the no-SF path that blocks for the full server-ACK round-trip (apples-to-apples vs egress). Pulls in JMH 1.37 as a test-scope dependency, wires the annotation processor into maven-compiler-plugin, requires jmh.core + ch.qos.logback.classic in the test module-info. The benchmark's static initializer downgrades the logback root level to WARN before any other class loads -- DEBUG-level WS / SF logging would otherwise emit one log line per flush and inflate measured latency by ~70us. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 87320e5 commit f58f766

3 files changed

Lines changed: 245 additions & 0 deletions

File tree

‎core/pom.xml‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
<argLine>-ea -Dfile.encoding=UTF-8 -XX:+UseParallelGC -Dslf4j.provider=ch.qos.logback.classic.spi.LogbackServiceProvider</argLine>
3737
<test.exclude>None</test.exclude>
3838
<test.include>%regex[.*[^o].class]</test.include><!-- exclude module-info.class-->
39+
<jmh.version>1.37</jmh.version>
3940
</properties>
4041

4142
<version>1.2.1-SNAPSHOT</version>
@@ -88,6 +89,13 @@
8889
<excludes>
8990
<exclude>${excludeTestPattern1}</exclude>
9091
</excludes>
92+
<annotationProcessorPaths>
93+
<path>
94+
<groupId>org.openjdk.jmh</groupId>
95+
<artifactId>jmh-generator-annprocess</artifactId>
96+
<version>${jmh.version}</version>
97+
</path>
98+
</annotationProcessorPaths>
9199
</configuration>
92100
</plugin>
93101
<plugin>
@@ -434,5 +442,17 @@
434442
<version>1.5.25</version>
435443
<scope>test</scope>
436444
</dependency>
445+
<dependency>
446+
<groupId>org.openjdk.jmh</groupId>
447+
<artifactId>jmh-core</artifactId>
448+
<version>${jmh.version}</version>
449+
<scope>test</scope>
450+
</dependency>
451+
<dependency>
452+
<groupId>org.openjdk.jmh</groupId>
453+
<artifactId>jmh-generator-annprocess</artifactId>
454+
<version>${jmh.version}</version>
455+
<scope>test</scope>
456+
</dependency>
437457
</dependencies>
438458
</project>

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpIngressLatencyBenchmark.java‎

Lines changed: 223 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,223 @@
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+
package io.questdb.client.test.cutlass.qwp.client;
26+
27+
import io.questdb.client.Sender;
28+
import org.openjdk.jmh.annotations.Benchmark;
29+
import org.openjdk.jmh.annotations.BenchmarkMode;
30+
import org.openjdk.jmh.annotations.Level;
31+
import org.openjdk.jmh.annotations.Mode;
32+
import org.openjdk.jmh.annotations.OutputTimeUnit;
33+
import org.openjdk.jmh.annotations.Scope;
34+
import org.openjdk.jmh.annotations.Setup;
35+
import org.openjdk.jmh.annotations.State;
36+
import org.openjdk.jmh.annotations.TearDown;
37+
import org.openjdk.jmh.runner.Runner;
38+
import org.openjdk.jmh.runner.RunnerException;
39+
import org.openjdk.jmh.runner.options.Options;
40+
import org.openjdk.jmh.runner.options.OptionsBuilder;
41+
import org.openjdk.jmh.runner.options.TimeValue;
42+
43+
import java.nio.file.Paths;
44+
import java.sql.Connection;
45+
import java.sql.DriverManager;
46+
import java.sql.Statement;
47+
import java.time.temporal.ChronoUnit;
48+
import java.util.Properties;
49+
import java.util.TimeZone;
50+
import java.util.concurrent.TimeUnit;
51+
52+
/**
53+
* JMH latency benchmark for QWP ingress -- the user-facing counterpart to
54+
* {@code QwpEgressLatencyBenchmark} in the QuestDB OSS repo. Measures the
55+
* end-to-end wall time of a single row {@code .at(...) + flush()} against a
56+
* locally running QuestDB, excluding connection setup (the {@link Sender} is
57+
* opened once per trial and reused across every benchmarked invocation).
58+
* <p>
59+
* Default mode (SF on) measures user-handover latency: {@code flush()} blocks
60+
* only until the row is durable on the local SF segment (CRC + two pwrites);
61+
* the wire send and server ACK are processed asynchronously by the I/O thread
62+
* and are NOT included in the measurement window. This is the number to quote
63+
* when the user app's contract is "the row is recoverable if I crash now",
64+
* not "the server has confirmed the row".
65+
* <p>
66+
* With {@code -Dsf=false}, store-and-forward is disabled. {@code flush()} then
67+
* blocks for the full row encode &rarr; WS send &rarr; server ACK round-trip.
68+
* This is the symmetric counterpart of the egress benchmark's {@code SELECT 1}
69+
* round-trip -- useful when comparing the ingress and egress wire paths head
70+
* to head, but it is NOT what a real SF-enabled user app experiences.
71+
* <p>
72+
* Runs two modes on each invocation:
73+
* <ul>
74+
* <li>{@code SampleTime} -- reports p50/p90/p99/p99.9 percentiles per
75+
* iteration. This is the main signal; ingest UX is gated by the tail,
76+
* not the mean.</li>
77+
* <li>{@code AverageTime} -- arithmetic mean. Useful when comparing two
78+
* builds: a smaller mean with an unchanged tail is usually the honest
79+
* win (no outlier distortion).</li>
80+
* </ul>
81+
* <p>
82+
* Prerequisites:
83+
* <ul>
84+
* <li>A QuestDB server listening on 9000 (HTTP/WS) and 8812 (PG wire).</li>
85+
* </ul>
86+
* <p>
87+
* Tune via system properties:
88+
* <ul>
89+
* <li>{@code -Dskip.populate=true} to re-use an existing
90+
* {@code latency_bench_ingress} table instead of dropping and recreating
91+
* it in {@code @Setup}.</li>
92+
* <li>{@code -Dsf=true} to enable store-and-forward. {@code -Dsf.dir=<path>}
93+
* overrides the SF directory (default: a fresh tmp dir per trial).</li>
94+
* <li>{@code -Dfsync.on.flush=true} to also fsync the SF segment on every
95+
* flush ({@code sf_fsync_on_flush=on}; only meaningful with
96+
* {@code -Dsf=true}).</li>
97+
* </ul>
98+
* <p>
99+
* Run via Maven exec:
100+
* <pre>
101+
* mvn -pl core test-compile
102+
* mvn -pl core exec:java \
103+
* -Dexec.classpathScope=test \
104+
* -Dexec.mainClass=io.questdb.client.test.cutlass.qwp.client.QwpIngressLatencyBenchmark
105+
* </pre>
106+
*/
107+
@State(Scope.Benchmark)
108+
@OutputTimeUnit(TimeUnit.MICROSECONDS)
109+
@BenchmarkMode({Mode.SampleTime, Mode.AverageTime})
110+
public class QwpIngressLatencyBenchmark {
111+
112+
static {
113+
// The WS / SF code paths emit a handful of DEBUG lines per flush.
114+
// At 7-8k flushes/sec that's enough I/O to inflate measured latency
115+
// by ~70 us (verified: same harness, root=DEBUG vs root=WARN, p50 went
116+
// 200 us -> 38 us). Force WARN before any other class loads so the
117+
// first log line we'd otherwise emit is also gone. If SLF4J is bound
118+
// to something other than logback, leave the level alone -- the
119+
// benchmark still runs, just with whatever the binding's default is.
120+
org.slf4j.ILoggerFactory factory = org.slf4j.LoggerFactory.getILoggerFactory();
121+
if (factory instanceof ch.qos.logback.classic.LoggerContext) {
122+
((ch.qos.logback.classic.LoggerContext) factory)
123+
.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME)
124+
.setLevel(ch.qos.logback.classic.Level.WARN);
125+
}
126+
}
127+
128+
private static final boolean FSYNC_ON_FLUSH = Boolean.parseBoolean(System.getProperty("fsync.on.flush", "false"));
129+
private static final String HOST = "localhost";
130+
private static final int HTTP_PORT = 9000;
131+
private static final int PG_PORT = 8812;
132+
private static final boolean SF_ENABLED = Boolean.parseBoolean(System.getProperty("sf", "true"));
133+
private static final String SF_DIR_OVERRIDE = System.getProperty("sf.dir");
134+
private static final boolean SKIP_POPULATE = Boolean.parseBoolean(System.getProperty("skip.populate", "false"));
135+
private static final String TABLE = "latency_bench_ingress";
136+
137+
private long rowCounter;
138+
private Sender sender;
139+
140+
public static void main(String[] args) throws RunnerException {
141+
Options opt = new OptionsBuilder()
142+
.include(QwpIngressLatencyBenchmark.class.getSimpleName())
143+
// Five warmup iterations at two seconds each so the JIT gets
144+
// past C2 tiering and the WAL writer / WS encoder are hot
145+
// before we record samples.
146+
.warmupIterations(5)
147+
.warmupTime(TimeValue.seconds(2))
148+
.measurementIterations(10)
149+
.measurementTime(TimeValue.seconds(2))
150+
.threads(1)
151+
.forks(2)
152+
.build();
153+
new Runner(opt).run();
154+
}
155+
156+
@Benchmark
157+
public void ingestSingleRow() {
158+
// Monotonic id and ts so rows are unique and the WAL writer is
159+
// exercised in append-mostly mode (no out-of-order rewrites).
160+
long n = ++rowCounter;
161+
sender.table(TABLE)
162+
.longColumn("id", n)
163+
.at(n, ChronoUnit.MICROS);
164+
sender.flush();
165+
}
166+
167+
@Setup(Level.Trial)
168+
public void setUp() throws Exception {
169+
if (!SKIP_POPULATE) {
170+
recreateTable();
171+
} else {
172+
System.out.println("skip.populate=true, re-using existing " + TABLE);
173+
}
174+
175+
String cfg = "ws::addr=" + HOST + ":" + HTTP_PORT + ";";
176+
if (SF_ENABLED) {
177+
String sfDir = SF_DIR_OVERRIDE != null
178+
? SF_DIR_OVERRIDE
179+
: Paths.get(System.getProperty("java.io.tmpdir"),
180+
"qdb-sf-ingress-bench-" + System.nanoTime()).toString();
181+
cfg += "store_and_forward=on;sf_dir=" + sfDir + ";";
182+
if (FSYNC_ON_FLUSH) {
183+
cfg += "sf_fsync_on_flush=on;";
184+
}
185+
System.out.println("SF enabled, dir=" + sfDir + ", sf_fsync_on_flush=" + FSYNC_ON_FLUSH);
186+
}
187+
sender = Sender.fromConfig(cfg);
188+
189+
// Prime: first flush registers the table schema with the server and
190+
// warms WS encoder / async pipeline state. Keeps those one-time
191+
// costs out of the measurement window.
192+
rowCounter = 0;
193+
sender.table(TABLE)
194+
.longColumn("id", 0L)
195+
.at(0L, ChronoUnit.MICROS);
196+
sender.flush();
197+
}
198+
199+
@TearDown(Level.Trial)
200+
public void tearDown() {
201+
if (sender != null) {
202+
sender.close();
203+
}
204+
}
205+
206+
private static Connection createPgConnection() throws Exception {
207+
Properties p = new Properties();
208+
p.setProperty("user", "admin");
209+
p.setProperty("password", "quest");
210+
p.setProperty("sslmode", "disable");
211+
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
212+
return DriverManager.getConnection(
213+
String.format("jdbc:postgresql://%s:%d/qdb", HOST, PG_PORT), p);
214+
}
215+
216+
private static void recreateTable() throws Exception {
217+
try (Connection c = createPgConnection(); Statement st = c.createStatement()) {
218+
st.execute("DROP TABLE IF EXISTS " + TABLE);
219+
st.execute("CREATE TABLE " + TABLE + " (id LONG, ts TIMESTAMP) "
220+
+ "TIMESTAMP(ts) PARTITION BY DAY WAL");
221+
}
222+
}
223+
}

‎core/src/test/java/module-info.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
requires org.slf4j;
3333
requires java.sql;
3434
requires org.postgresql.jdbc;
35+
requires jmh.core;
36+
requires ch.qos.logback.classic;
3537

3638
exports io.questdb.client.test;
3739
exports io.questdb.client.test.cairo;

0 commit comments

Comments
 (0)