Merge origin/vi_egress: adopt CACHE_RESET, move SERVER_INFO to 0x18

# Conflicts:
#	core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressMsgKind.java
#	core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpQueryClient.java

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 {
             emitTerminalTransportError("unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
             currentQueryDone = true;
@@ -451,6 +460,22 @@ private void emitTerminalTransportError(String message) {
         events.offer(new QueryEvent().asTransportError(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) {
+            emitTerminalTransportError("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,6 +46,14 @@ 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;
+    /**
+     * Reset mask bit: clear the connection-scoped SYMBOL dict.
+     */
+    public static final byte RESET_MASK_DICT = 0x01;
+    /**
+     * Reset mask bit: clear the connection-scoped schema-fingerprint cache.
+     */
+    public static final byte RESET_MASK_SCHEMAS = 0x02;
     public static final byte RESULT_BATCH = 0x11;
     public static final byte RESULT_END = 0x12;
     /**
@@ -66,19 +81,10 @@ public final class QwpEgressMsgKind {
      * Server -> client. Unsolicited frame delivered as the first QWP message
      * on every v2 WebSocket connection. Body (little-endian): {@code
      * msg_kind:u8, role:u8, epoch:u64, capabilities:u32, server_wall_ns:i64,
-     * cluster_id:u16_len+utf8, node_id:u16_len+utf8}.
-     */
-    public static final byte SERVER_INFO = 0x17;
-    /**
-     * 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.
-     */
-    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).
+     * cluster_id:u16_len+utf8, node_id:u16_len+utf8}. The byte-value 0x17 is
+     * claimed by {@link #CACHE_RESET}; SERVER_INFO lives at 0x18.
      */
-    public static final byte STATUS_LIMIT_EXCEEDED = 0x0B;
+    public static final byte SERVER_INFO = 0x18;
 
     private QwpEgressMsgKind() {
     }

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

@@ -84,6 +84,15 @@
  * {@link QwpColumnBatchHandler#onError} with the stored status/message. Per-
  * query {@code QUERY_ERROR} responses are NOT terminal -- the connection
  * remains usable for the next query.
+ * <p>
+ * 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 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
+ * specific cause in the message.
  */
 public class QwpQueryClient implements QuietCloseable {
 

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.