From 872c18231f87d7cef8c97f3d1a4d35b6ff699310 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 20 Jul 2026 13:29:04 +0200 Subject: [PATCH 1/2] refactor(qwp): rename sf_max_bytes to sf_max_segment_bytes The connect-string key caps a single SF segment, and the new name says so; it also pairs cleanly with sf_max_total_bytes. The builder method follows: storeAndForwardMaxBytes() is now storeAndForwardMaxSegmentBytes(). Internal identifiers and error messages rename in step so the old spelling is gone entirely. Co-Authored-By: Claude Fable 5 --- .../main/java/io/questdb/client/Sender.java | 32 +++++++++---------- .../sf/cursor/CursorWebSocketSendLoop.java | 2 +- .../qwp/client/sf/cursor/MmapSegment.java | 2 +- .../qwp/client/sf/cursor/SegmentRing.java | 2 +- .../io/questdb/client/impl/ConfigSchema.java | 2 +- .../client/QwpQueryClientFromConfigTest.java | 2 +- .../qwp/client/RecoveryReplayTest.java | 2 +- .../qwp/client/sf/SfFromConfigTest.java | 16 +++++----- .../qwp/client/sf/cursor/SegmentRingTest.java | 2 +- .../test/impl/WsSenderConfigHonoredTest.java | 2 +- 10 files changed, 32 insertions(+), 32 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 4ec70dbd..6cc1b873 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -1123,7 +1123,7 @@ public int getConnectTimeout() { // implemented; FLUSH and APPEND are deferred follow-ups (cursor needs // to learn fsync first). private SfDurability sfDurability = SfDurability.MEMORY; - private long sfMaxBytes = PARAMETER_NOT_SET_EXPLICITLY; + private long sfMaxSegmentBytes = PARAMETER_NOT_SET_EXPLICITLY; private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY; private boolean shouldDestroyPrivKey; private boolean tlsEnabled; @@ -1446,9 +1446,9 @@ public Sender build() { // (same lock-free architecture, no disk involvement). The // sf_durability != memory rejection lives in validateParameters // so it is reached by build() and by no-connect validation alike. - long actualSfMaxBytes = sfMaxBytes == PARAMETER_NOT_SET_EXPLICITLY + long actualSfMaxSegmentBytes = sfMaxSegmentBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_SEGMENT_BYTES - : sfMaxBytes; + : sfMaxSegmentBytes; // Default cap depends on backing: RAM (memory mode) is tight // by default; disk (SF mode) is cheap so the default is // generous enough that normal traffic never hits it. @@ -1456,7 +1456,7 @@ public Sender build() { ? DEFAULT_MAX_BYTES_MEMORY : DEFAULT_MAX_BYTES_SF; long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY - ? Math.max(defaultMaxTotal, actualSfMaxBytes * 2) + ? Math.max(defaultMaxTotal, actualSfMaxSegmentBytes * 2) : sfMaxTotalBytes; long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == CLOSE_FLUSH_TIMEOUT_NOT_SET ? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS @@ -1541,7 +1541,7 @@ public Sender build() { ? CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS : sfAppendDeadlineMillis * 1_000_000L; CursorSendEngine cursorEngine = new CursorSendEngine( - slotPath, actualSfMaxBytes, + slotPath, actualSfMaxSegmentBytes, actualSfMaxTotalBytes, actualSfAppendDeadlineNanos); int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY ? errorInboxCapacity @@ -1622,7 +1622,7 @@ public Sender build() { connected.startOrphanDrainers( orphans, maxBackgroundDrainers, - actualSfMaxBytes, + actualSfMaxSegmentBytes, actualSfMaxTotalBytes); } } @@ -2757,14 +2757,14 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) { * (4 MiB). Smaller segments mean faster trim of acked data; larger * segments mean fewer rotations. */ - public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) { + public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxBytes) { if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { throw new LineSenderException("store_and_forward is only supported for WebSocket transport"); } if (maxBytes <= 0) { - throw new LineSenderException("sf_max_bytes must be positive: ").put(maxBytes); + throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxBytes); } - this.sfMaxBytes = maxBytes; + this.sfMaxSegmentBytes = maxBytes; return this; } @@ -3352,12 +3352,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) { } pos = getValue(configurationString, pos, sink, "sender_id"); senderId(sink.toString()); - } else if (Chars.equals("sf_max_bytes", sink)) { + } else if (Chars.equals("sf_max_segment_bytes", sink)) { if (protocol != PROTOCOL_WEBSOCKET) { - throw new LineSenderException("sf_max_bytes is only supported for WebSocket transport"); + throw new LineSenderException("sf_max_segment_bytes is only supported for WebSocket transport"); } - pos = getValue(configurationString, pos, sink, "sf_max_bytes"); - storeAndForwardMaxBytes(parseSizeValue(sink, "sf_max_bytes")); + pos = getValue(configurationString, pos, sink, "sf_max_segment_bytes"); + storeAndForwardMaxSegmentBytes(parseSizeValue(sink, "sf_max_segment_bytes")); } else if (Chars.equals("sf_max_total_bytes", sink)) { if (protocol != PROTOCOL_WEBSOCKET) { throw new LineSenderException("sf_max_total_bytes is only supported for WebSocket transport"); @@ -3689,8 +3689,8 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString) if (view.has("sf_append_deadline_millis")) { sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis")); } - if (view.has("sf_max_bytes")) { - storeAndForwardMaxBytes(wsSize(view, v, "sf_max_bytes")); + if (view.has("sf_max_segment_bytes")) { + storeAndForwardMaxSegmentBytes(wsSize(view, v, "sf_max_segment_bytes")); } if (view.has("sf_max_total_bytes")) { storeAndForwardMaxTotalBytes(wsSize(view, v, "sf_max_total_bytes")); @@ -3848,7 +3848,7 @@ public java.util.Map wsConfigSnapshotForTest() { m.put("request_durable_ack", requestDurableAck); m.put("sender_id", senderId); m.put("sf_dir", sfDir); - m.put("sf_max_bytes", sfMaxBytes); + m.put("sf_max_segment_bytes", sfMaxSegmentBytes); m.put("sf_max_total_bytes", sfMaxTotalBytes); m.put("sf_durability", sfDurability == null ? null : sfDurability.name()); m.put("sf_append_deadline_millis", sfAppendDeadlineMillis); diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java index 13a69f77..6d59f344 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java @@ -159,7 +159,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // drained from the head every time a STATUS_DURABLE_ACK frame advances // any watermark; an entry pops when every (name, seqTxn) it carries is // covered by durableTableWatermarks. Bounded in practice by the SF on-disk - // cap: once the producer hits sf_max_bytes it blocks, which caps how far + // cap: once the producer hits sf_max_segment_bytes it blocks, which caps how far // the durable watermark can lag behind the OK watermark. private final ArrayDeque pendingDurable = new ArrayDeque<>(); private final ArrayDeque pendingDurablePool = new ArrayDeque<>(); diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java index 78e5db9f..bf44d3c3 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java @@ -177,7 +177,7 @@ public static MmapSegment create(FilesFacade ff, long pathPtr, String displayPat // the JVM). if (!ff.allocate(fd, sizeBytes)) { ff.close(fd); - // Unlink the partially-created file so a sf_max_bytes-sized + // Unlink the partially-created file so a sf_max_segment_bytes-sized // empty file does not survive the failure. Under sustained // disk-full pressure with the manager polling, hundreds would // otherwise accumulate. diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java index c8516c4c..2b5ca720 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java @@ -255,7 +255,7 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) { return null; } // Sort by baseSeq ascending. Worst-case segment count is - // sf_max_total_bytes / sf_max_bytes -- at the documented ceiling + // sf_max_total_bytes / sf_max_segment_bytes -- at the documented ceiling // (1 TiB / 64 MiB) that is ~16K entries, where an O(N²) sort spends // multiple seconds in compares + shifts before the I/O thread can // start. In-place quicksort with median-of-three pivot keeps the diff --git a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java index a255e638..5c76fe55 100644 --- a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java +++ b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java @@ -86,7 +86,7 @@ public final class ConfigSchema { str("sf_append_deadline_millis", Side.INGRESS); str("sf_dir", Side.INGRESS); str("sf_durability", Side.INGRESS); - str("sf_max_bytes", Side.INGRESS); + str("sf_max_segment_bytes", Side.INGRESS); str("sf_max_total_bytes", Side.INGRESS); str("transaction", Side.INGRESS); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java index 45f34fef..42d668fb 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java @@ -720,7 +720,7 @@ public void testIngressOnlyKeysSilentlyAcceptedOnEgress() { "sf_append_deadline_millis=30000", "sf_dir=/var/lib/qdb-sf", "sf_durability=memory", - "sf_max_bytes=4m", + "sf_max_segment_bytes=4m", "sf_max_total_bytes=10g", "transaction=on", }; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java index e868dafc..98a59e6f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java @@ -86,7 +86,7 @@ public void testRestartReplaysSealedSegmentsAgainstFreshServer() throws Exceptio String pad = repeat("x", 64); String cfg1 = "ws::addr=localhost:" + port1 + ";sf_dir=" + sfDir - + ";sf_max_bytes=4096" + + ";sf_max_segment_bytes=4096" + ";close_flush_timeout_millis=0;"; try (Sender s1 = Sender.fromConfig(cfg1)) { for (int i = 0; i < 50; i++) { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java index 0c27d698..63b1fcdb 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java @@ -92,7 +92,7 @@ public void testSfDirOnTcpRejected() throws Exception { } @Test - public void testSfMaxBytesParsing() throws Exception { + public void testSfMaxSegmentBytesParsing() throws Exception { TestUtils.assertMemoryLeak(() -> { AckHandler handler = new AckHandler(); try (TestWebSocketServer server = new TestWebSocketServer(handler)) { @@ -101,7 +101,7 @@ public void testSfMaxBytesParsing() throws Exception { int port = server.getPort(); String config = "ws::addr=localhost:" + port - + ";sf_dir=" + sfDir + ";sf_max_bytes=131072;"; + + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=131072;"; try (Sender sender = Sender.fromConfig(config)) { // Write enough data that segments rotate at ~128 KiB boundary. for (int i = 0; i < 50; i++) { @@ -139,7 +139,7 @@ public void testNoSfDirMeansNoSf() throws Exception { } /** - * Regression test for the connect-string {@code sf_max_bytes} / + * Regression test for the connect-string {@code sf_max_segment_bytes} / * {@code sf_max_total_bytes} parser accepting values larger than * {@code Integer.MAX_VALUE}. The pre-cursor parser used parseInt which * artificially capped the SF size from the connect string at ~2 GiB. @@ -222,7 +222,7 @@ public void testSfDurabilityOnTcpRejected() throws Exception { } @Test - public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception { + public void testSfMaxSegmentBytesAcceptsSizeSuffixes() throws Exception { TestUtils.assertMemoryLeak(() -> { AckHandler handler = new AckHandler(); try (TestWebSocketServer server = new TestWebSocketServer(handler)) { @@ -233,7 +233,7 @@ public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception { // 64m / 4g should parse identically to their byte-count equivalents. String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir - + ";sf_max_bytes=64m" + + ";sf_max_segment_bytes=64m" + ";sf_max_total_bytes=4g;"; try (Sender sender = Sender.fromConfig(config)) { sender.table("foo").longColumn("v", 1L).atNow(); @@ -346,14 +346,14 @@ public void testSenderIdInvalidCharRejected() throws Exception { } @Test - public void testSfMaxBytesInvalidSizeSuffixRejected() throws Exception { + public void testSfMaxSegmentBytesInvalidSizeSuffixRejected() throws Exception { TestUtils.assertMemoryLeak(() -> { - String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_bytes=64x;"; + String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_segment_bytes=64x;"; try (Sender ignored = Sender.fromConfig(config)) { Assert.fail("expected rejection of unknown unit suffix"); } catch (LineSenderException expected) { Assert.assertTrue(expected.getMessage(), - expected.getMessage().contains("invalid sf_max_bytes")); + expected.getMessage().contains("invalid sf_max_segment_bytes")); } }); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java index f1a0fcde..e0824035 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java @@ -530,7 +530,7 @@ public void testNextSealedAfterStillReturnsCorrectlyWhenCursorWasTrimmed() throw /** * Open-time sort regression: at the documented {@code sf_max_total_bytes - * / sf_max_bytes} ceiling (~16K segments) an O(N²) sort over the + * / sf_max_segment_bytes} ceiling (~16K segments) an O(N²) sort over the * recovered segments burns multi-second wall time before the I/O thread * can start. The previous selection-sort implementation regressed an * earlier perf fix on the legacy {@code SegmentLog} path; this test diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java index dba5765f..f59b276d 100644 --- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java @@ -61,7 +61,7 @@ public void testEveryIngressKeyIsHonored() { assertHonored("request_durable_ack=on", "request_durable_ack", true); assertHonored("sender_id=probe-1", "sender_id", "probe-1"); assertHonored("sf_dir=/var/probe", "sf_dir", "/var/probe"); - assertHonored("sf_max_bytes=4096", "sf_max_bytes", 4096L); + assertHonored("sf_max_segment_bytes=4096", "sf_max_segment_bytes", 4096L); assertHonored("sf_max_total_bytes=8192", "sf_max_total_bytes", 8192L); assertHonored("sf_durability=flush", "sf_durability", "FLUSH"); assertHonored("sf_append_deadline_millis=1500", "sf_append_deadline_millis", 1500L); From 93d3eb986e96251851ee8bcbe35d52be60c60151 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Fri, 24 Jul 2026 17:35:09 +0200 Subject: [PATCH 2/2] test(qwp): cover SF max segment bytes setter --- .../LineSenderBuilderWebSocketTest.java | 48 +++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java index caec264b..9ec82093 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java @@ -634,6 +634,54 @@ public void testRetryTimeout_notSupportedForWebSocket() { "not supported for WebSocket"); } + @Test + public void testStoreAndForwardMaxSegmentBytes() { + Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET); + + Assert.assertSame(builder, builder.storeAndForwardMaxSegmentBytes(64 * 1024L)); + Assert.assertEquals( + 64 * 1024L, + ((Number) builder.wsConfigSnapshotForTest().get("sf_max_segment_bytes")).longValue() + ); + } + + @Test + public void testStoreAndForwardMaxSegmentBytesRejectsNonPositiveValues() { + long[] rejected = {0L, -1L}; + for (long value : rejected) { + try { + Sender.builder(Sender.Transport.WEBSOCKET) + .storeAndForwardMaxSegmentBytes(value); + Assert.fail("expected storeAndForwardMaxSegmentBytes(" + value + ") to fail"); + } catch (LineSenderException expected) { + Assert.assertEquals( + "sf_max_segment_bytes must be positive: " + value, + expected.getMessage() + ); + } + } + } + + @Test + public void testStoreAndForwardMaxSegmentBytesRejectedForNonWebSocketTransports() { + Sender.Transport[] rejected = { + Sender.Transport.HTTP, + Sender.Transport.TCP, + Sender.Transport.UDP + }; + for (Sender.Transport transport : rejected) { + try { + Sender.builder(transport).storeAndForwardMaxSegmentBytes(64 * 1024L); + Assert.fail("expected " + transport + " to reject storeAndForwardMaxSegmentBytes"); + } catch (LineSenderException expected) { + Assert.assertEquals( + "store_and_forward is only supported for WebSocket transport", + expected.getMessage() + ); + } + } + } + @Test public void testSyncModeAutoFlushDefaults() throws Exception { // Regression test: connect() must not hardcode autoFlush to 0.