self-review and metrics

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java

@@ -182,6 +182,15 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
         } else if (msgKind == QwpEgressMsgKind.QUERY_ERROR) {
             decodeAndEmitError(payloadPtr, payloadLen);
             currentQueryDone = true;
+        } else if (msgKind == QwpEgressMsgKind.CACHE_RESET) {
+            // Server reached a configured soft cap on the connection-scoped
+            // SYMBOL dict or schema-fingerprint cache. Drop the indicated
+            // caches on this side so the next RESULT_BATCH's deltaStart and
+            // schema-reference ids line up with the server's fresh counter.
+            // No user-visible event -- CACHE_RESET never arrives between the
+            // RESULT_BATCH / RESULT_END / EXEC_DONE / QUERY_ERROR of a query
+            // and the user callback, only after it.
+            handleCacheReset(payloadPtr, payloadLen);
         } else {
             emitTerminalError("unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
             currentQueryDone = true;
@@ -437,6 +446,22 @@ private void emitTerminalError(String message) {
         emitError(WebSocketResponse.STATUS_INTERNAL_ERROR, message);
     }
 
+    /**
+     * Decodes a {@code CACHE_RESET} frame body and clears the indicated
+     * connection-scoped caches on the client side. Body is a single byte mask:
+     * bit 0 = SYMBOL dict, bit 1 = schema-fingerprint cache.
+     */
+    private void handleCacheReset(long payloadPtr, int payloadLen) {
+        int bodyStart = QwpConstants.HEADER_SIZE + 1; // msg_kind byte consumed by caller
+        if (payloadLen < bodyStart + 1) {
+            emitTerminalError("CACHE_RESET frame truncated before reset_mask");
+            currentQueryDone = true;
+            return;
+        }
+        byte resetMask = Unsafe.getUnsafe().getByte(payloadPtr + bodyStart);
+        decoder.applyCacheReset(resetMask);
+    }
+
     private void handleResultBatch(long payloadPtr, int payloadLen) {
         QwpBatchBuffer buf;
         try {

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressMsgKind.java

@@ -30,6 +30,13 @@
  * egress payload identifies which of the egress message types it carries.
  */
 public final class QwpEgressMsgKind {
+    /**
+     * Server -> client. Connection-scoped cache reset. Body: {@code reset_mask:u8}
+     * with bit 0 = SYMBOL dict, bit 1 = schema-fingerprint cache. Sent between
+     * queries when a cache hits its server-side soft cap. Recipient clears the
+     * indicated caches; subsequent RESULT_BATCH delta sections start fresh.
+     */
+    public static final byte CACHE_RESET = 0x17;
     public static final byte CANCEL = 0x14;
     public static final byte CREDIT = 0x15;
     /**
@@ -39,21 +46,16 @@ public final class QwpEgressMsgKind {
     public static final byte EXEC_DONE = 0x16;
     public static final byte QUERY_ERROR = 0x13;
     public static final byte QUERY_REQUEST = 0x10;
-    public static final byte RESULT_BATCH = 0x11;
-    public static final byte RESULT_END = 0x12;
-
-    // Egress-specific QUERY_ERROR status codes. Extend the ingress
-    // QwpConstants.STATUS_* namespace (0x00-0x09).
     /**
-     * Status byte on a {@code QUERY_ERROR} frame: the query was cancelled,
-     * either by a client {@code CANCEL} frame or by explicit server-side cancel.
+     * Reset mask bit: clear the connection-scoped SYMBOL dict.
      */
-    public static final byte STATUS_CANCELLED = 0x0A;
+    public static final byte RESET_MASK_DICT = 0x01;
     /**
-     * Status byte on a {@code QUERY_ERROR} frame: a server-side limit was hit
-     * (query timeout, memory cap, circuit breaker).
+     * Reset mask bit: clear the connection-scoped schema-fingerprint cache.
      */
-    public static final byte STATUS_LIMIT_EXCEEDED = 0x0B;
+    public static final byte RESET_MASK_SCHEMAS = 0x02;
+    public static final byte RESULT_BATCH = 0x11;
+    public static final byte RESULT_END = 0x12;
 
     private QwpEgressMsgKind() {
     }

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpQueryClient.java

@@ -28,6 +28,7 @@
 import io.questdb.client.cutlass.http.client.HttpClientException;
 import io.questdb.client.cutlass.http.client.WebSocketClient;
 import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
+import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.impl.ConfStringParser;
 import io.questdb.client.std.Chars;
 import io.questdb.client.std.QuietCloseable;
@@ -65,7 +66,7 @@
  * Status byte convention on {@link QwpColumnBatchHandler#onError}: server-
  * emitted {@code QUERY_ERROR} frames surface with the server's status code
  * ({@link WebSocketResponse#STATUS_PARSE_ERROR}, {@code STATUS_INTERNAL_ERROR},
- * {@link QwpEgressMsgKind#STATUS_CANCELLED}, {@code STATUS_LIMIT_EXCEEDED},
+ * {@link QwpConstants#STATUS_CANCELLED}, {@link QwpConstants#STATUS_LIMIT_EXCEEDED},
  * etc.). Failures detected client-side (closed client, bind encoding error,
  * truncated / unknown frame, decoder out of sync, I/O thread interrupt) all
  * surface with {@link WebSocketResponse#STATUS_INTERNAL_ERROR} and the

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpResultBatchDecoder.java

@@ -120,6 +120,29 @@ public class QwpResultBatchDecoder implements QuietCloseable {
     private long varintPos;
     private long varintValue;
 
+    /**
+     * Clears the caches indicated by {@code resetMask} (bitwise OR of
+     * {@link QwpEgressMsgKind#RESET_MASK_DICT} and
+     * {@link QwpEgressMsgKind#RESET_MASK_SCHEMAS}). Drives the client-side
+     * state machine when the server emits a {@code CACHE_RESET} frame after
+     * hitting a connection-level soft cap: discards the SYMBOL dict and / or
+     * schema-fingerprint cache so the next batch's {@code deltaStart} and
+     * schema-reference ids line up with the server's fresh counter.
+     * <p>
+     * The native dict buffers are reused (positions reset to 0, capacity
+     * retained) so a workload that churns just above the cap does not reallocate
+     * on every reset.
+     */
+    public void applyCacheReset(byte resetMask) {
+        if ((resetMask & QwpEgressMsgKind.RESET_MASK_DICT) != 0) {
+            connDictSize = 0;
+            connDictHeapPos = 0;
+        }
+        if ((resetMask & QwpEgressMsgKind.RESET_MASK_SCHEMAS) != 0) {
+            schemaRegistry.clear();
+        }
+    }
+
     @Override
     public void close() {
         if (connDictHeapAddr != 0) {

core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpConstants.java

@@ -88,6 +88,19 @@ public final class QwpConstants {
      * Schema mode: Schema reference (ID lookup).
      */
     public static final byte SCHEMA_MODE_REFERENCE = 0x01;
+    /**
+     * Status byte on a {@code QUERY_ERROR} frame: the query was cancelled,
+     * either by a client {@code CANCEL} frame or by explicit server-side
+     * cancel. Egress extension of the ingress {@code STATUS_*} namespace
+     * (0x00-0x09).
+     */
+    public static final byte STATUS_CANCELLED = 0x0A;
+    /**
+     * Status byte on a {@code QUERY_ERROR} frame: a server-side limit was hit
+     * (query timeout, memory cap, circuit breaker, OOM). Egress extension of
+     * the ingress {@code STATUS_*} namespace (0x00-0x09).
+     */
+    public static final byte STATUS_LIMIT_EXCEEDED = 0x0B;
     /**
      * Column type: BINARY (length-prefixed opaque bytes).
      * Wire format: identical to VARCHAR — (N+1) x uint32 offsets + concatenated bytes.