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 <noreply@anthropic.com>

Author: mtopolnik

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,17 +1446,17 @@ 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.
                 long defaultMaxTotal = sfDir == null
                         ? 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<String, Object> 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);

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<PendingDurableEntry> pendingDurable = new ArrayDeque<>();
     private final ArrayDeque<PendingDurableEntry> pendingDurablePool = new ArrayDeque<>();

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.

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

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);
 

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",
         };

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++) {

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"));
             }
         });
     }

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

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);