From 872c18231f87d7cef8c97f3d1a4d35b6ff699310 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 20 Jul 2026 13:29:04 +0200 Subject: [PATCH 1/7] 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/7] 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. From 478f297132ee5b935684a6341fde155d92a34b72 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 27 Jul 2026 10:27:33 +0200 Subject: [PATCH 3/7] Comment change --- .../test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 9ec82093..c8425776 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 @@ -38,7 +38,7 @@ /** * Tests for WebSocket transport support in the Sender.builder() API. * These tests verify the builder configuration and validation, - * not actual WebSocket connectivity (which requires a running server). + * not actual WebSocket connectivity (that requires a running server). */ public class LineSenderBuilderWebSocketTest extends AbstractTest { From a750e1a64dc590f487d11a87b2280721473f1d24 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 27 Jul 2026 11:22:31 +0200 Subject: [PATCH 4/7] Pin sf_max_bytes as a rejected key The rename of sf_max_bytes to sf_max_segment_bytes ships no alias, so the old spelling must reject as an unknown configuration key. Nothing asserted that. Every existing drift guard -- WsSenderConfigHonoredTest and QwpConfigKeysTest -- iterates ConfigSchema.all(), so by construction they can only catch a half-applied rename, never a whole-registry one: re-introducing the old key would make them pass just as happily. testWsConfigString_withSfMaxBytes_fails asserts the ws:: parse fails with "unknown configuration key: sf_max_bytes", using the existing assertBadConfig helper and joining the unknown-key rejection cluster next to the gorilla / init_buf_size / max_schemas_per_connection tests. Co-Authored-By: Claude Opus 5 (1M context) --- .../qwp/client/LineSenderBuilderWebSocketTest.java | 9 +++++++++ 1 file changed, 9 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 c8425776..0860750f 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 @@ -857,6 +857,15 @@ public void testWsConfigString_withPath_fails() { "unknown configuration key: path"); } + @Test + public void testWsConfigString_withSfMaxBytes_fails() { + // sf_max_bytes is the spelling this key carries in the 1.2.1-1.3.5 jars. + // sf_max_segment_bytes replaces it with no alias, so the old spelling + // rejects as an unknown key rather than silently configuring nothing. + assertBadConfig("ws::addr=localhost:9000;sf_max_bytes=4096;", + "unknown configuration key: sf_max_bytes"); + } + @Test public void testWsConfigString_withEgressOnlyKeysSilentlyAccepted() { // connect-string.md "Query client keys" and "Multi-host failover": these From 627327be4ee7c3ae68fee262c3eaeaa03facfca4 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 27 Jul 2026 11:22:55 +0200 Subject: [PATCH 5/7] Name the segment-size setter parameter for its unit storeAndForwardMaxSegmentBytes still took a parameter called maxBytes, which reads as the total cap and reintroduces exactly the ambiguity the sf_max_segment_bytes rename removes. Every sibling names its parameter after the method tail: storeAndForwardMaxTotalBytes takes maxTotalBytes, storeAndForwardDir takes dir, storeAndForwardDurability takes durability, storeAndForwardSyncIntervalMillis takes millis. The parameter is now maxSegmentBytes. Signature, the positivity guard, the exception text argument, and the field assignment all follow. No behavior changes. Co-Authored-By: Claude Opus 5 (1M context) --- core/src/main/java/io/questdb/client/Sender.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 64ee70e4..eda096db 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -2775,14 +2775,14 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) { * (4 MiB). Smaller segments mean faster trim of acked data; larger * segments mean fewer rotations. */ - public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxBytes) { + public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxSegmentBytes) { 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_segment_bytes must be positive: ").put(maxBytes); + if (maxSegmentBytes <= 0) { + throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxSegmentBytes); } - this.sfMaxSegmentBytes = maxBytes; + this.sfMaxSegmentBytes = maxSegmentBytes; return this; } From 558a70cb7402985c715660f85151b5f504f3d780 Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 27 Jul 2026 11:23:27 +0200 Subject: [PATCH 6/7] Document the WebSocket-only constraint on segment size The storeAndForwardMaxSegmentBytes javadoc omitted the "WebSocket transport only" sentence that storeAndForwardDir, storeAndForwardDurability and storeAndForwardMaxTotalBytes all carry, even though the setter throws for HTTP, TCP and UDP. The javadoc now states that constraint, names sf_max_segment_bytes so a reader can connect the builder method to the connect-string key, and documents the parameter. It gives the default as a plain 4 MiB rather than as DEFAULT_SEGMENT_BYTES, a private constant that resolves to nothing in published javadoc. The paragraph is re-flowed to the surrounding width. Co-Authored-By: Claude Opus 5 (1M context) --- core/src/main/java/io/questdb/client/Sender.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index eda096db..aa500d44 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -2770,10 +2770,12 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) { } /** - * Maximum bytes per segment file before rotation. Defaults to - * {@code DEFAULT_SEGMENT_BYTES} - * (4 MiB). Smaller segments mean faster trim of acked data; larger - * segments mean fewer rotations. + * Maximum bytes per segment file before rotation, the builder form of + * the {@code sf_max_segment_bytes} connect-string key. Smaller segments + * mean faster trim of acked data; larger segments mean fewer rotations. + * Default: {@code 4 MiB}. WebSocket transport only. + * + * @param maxSegmentBytes per-segment cap in bytes; must be positive */ public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxSegmentBytes) { if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { From 538dbead10dfb1d1c7514872c713bbc103972ced Mon Sep 17 00:00:00 2001 From: Marko Topolnik Date: Mon, 27 Jul 2026 11:25:16 +0200 Subject: [PATCH 7/7] Use the class rejection helper in the SF segment tests The two storeAndForwardMaxSegmentBytes rejection tests spelled out try/fail/catch inline, 19 lines for what the class already expresses in one call. assertThrows(String, Runnable) is the dominant idiom here, used by roughly twenty other tests in the file, and it produces a better failure message than a bare Assert.fail. The assertion moves from message equality to containment. The expected text is the whole message in both cases, so the check is as strong as it was; a mutation run with both production guards removed fails both tests. Co-Authored-By: Claude Opus 5 (1M context) --- .../LineSenderBuilderWebSocketTest.java | 23 ++++--------------- 1 file changed, 4 insertions(+), 19 deletions(-) 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 0860750f..fbbc5313 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 @@ -649,16 +649,8 @@ public void testStoreAndForwardMaxSegmentBytes() { 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() - ); - } + assertThrows("sf_max_segment_bytes must be positive: " + value, + () -> Sender.builder(Sender.Transport.WEBSOCKET).storeAndForwardMaxSegmentBytes(value)); } } @@ -670,15 +662,8 @@ public void testStoreAndForwardMaxSegmentBytesRejectedForNonWebSocketTransports( 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() - ); - } + assertThrows("store_and_forward is only supported for WebSocket transport", + () -> Sender.builder(transport).storeAndForwardMaxSegmentBytes(64 * 1024L)); } }