Add schema ID validation and configurable limit

Add server-side validation that schema IDs arrive in strict
monotonic sequence and do not exceed a configurable per-connection
limit. The client-side sender exposes the limit through the
builder API and configuration string.

Server side:
- QwpSchemaCache.put() validates monotonic sequence and rejects
  IDs that exceed maxSchemasPerConnection
- Add INVALID_SCHEMA_ID error code to QwpParseException
- Add DEFAULT_MAX_SCHEMAS_PER_CONNECTION (65,535) to QwpConstants
- Wire the limit through LineHttpProcessorConfiguration,
  PropServerConfiguration, and QwpProcessorState
- QwpStreamingDecoder.resetConnectionState() clears schema cache
  on disconnect

Client side:
- Sender.LineSenderBuilder.maxSchemasPerConnection() builder
  method and max_schemas_per_connection config string parameter
- QwpWebSocketSender accepts and forwards the limit
- Replace static import of QwpConstants.* with explicit prefix

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mtopolnik

core/src/main/java/io/questdb/client/Sender.java

@@ -585,6 +585,7 @@ final class LineSenderBuilder {
         private int maxBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private int maxDatagramSize = PARAMETER_NOT_SET_EXPLICITLY;
         private int maxNameLength = PARAMETER_NOT_SET_EXPLICITLY;
+        private int maxSchemasPerConnection = PARAMETER_NOT_SET_EXPLICITLY;
         private int maximumBufferCapacity = PARAMETER_NOT_SET_EXPLICITLY;
         private final HttpClientConfiguration httpClientConfiguration = new DefaultHttpClientConfiguration() {
             @Override
@@ -711,15 +712,14 @@ public AdvancedTlsSettings advancedTls() {
         }
 
         /**
+         * @param enabled ignored
+         * @return this instance for method chaining
          * @deprecated Async mode is now derived from {@link #inFlightWindowSize(int)}.
          * Window size 1 implies synchronous mode, greater than 1 implies asynchronous mode.
          * The default window size is 128 (asynchronous). Call {@code inFlightWindowSize(1)}
          * for synchronous behavior.
          * <br>
          * This method is a no-op and will be removed in a future release.
-         *
-         * @param enabled ignored
-         * @return this instance for method chaining
          */
         @Deprecated
         public LineSenderBuilder asyncMode(boolean enabled) {
@@ -901,6 +901,8 @@ public Sender build() {
                         ? DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS
                         : TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
                 int actualInFlightWindowSize = inFlightWindowSize == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_IN_FLIGHT_WINDOW_SIZE : inFlightWindowSize;
+                int actualMaxSchemasPerConnection = maxSchemasPerConnection == PARAMETER_NOT_SET_EXPLICITLY
+                        ? QwpWebSocketSender.DEFAULT_MAX_SCHEMAS_PER_CONNECTION : maxSchemasPerConnection;
 
                 String wsAuthHeader = buildWebSocketAuthHeader();
 
@@ -912,7 +914,8 @@ public Sender build() {
                         actualAutoFlushBytes,
                         actualAutoFlushIntervalNanos,
                         actualInFlightWindowSize,
-                        wsAuthHeader
+                        wsAuthHeader,
+                        actualMaxSchemasPerConnection
                 );
             }
 
@@ -1093,7 +1096,7 @@ public LineSenderBuilder httpPath(String path) {
          *
          * <b>Example:</b> If the server configures {@code http.context.settings=/custom/settings},
          * call {@code httpSettingPath("/custom/settings")}.
-         *
+         * <p>
          * This is only used when communicating over HTTP transport.
          *
          * @param path The HTTP path to query for server protocol settings. Must:
@@ -1345,6 +1348,25 @@ public LineSenderBuilder maxNameLength(int maxNameLength) {
             return this;
         }
 
+        /**
+         * Sets the maximum number of distinct schemas the WebSocket sender may assign on one connection.
+         */
+        public LineSenderBuilder maxSchemasPerConnection(int maxSchemasPerConnection) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("max schemas per connection is only supported for WebSocket transport");
+            }
+            if (this.maxSchemasPerConnection != PARAMETER_NOT_SET_EXPLICITLY) {
+                throw new LineSenderException("max schemas per connection was already configured")
+                        .put("[maxSchemasPerConnection=").put(this.maxSchemasPerConnection).put("]");
+            }
+            if (maxSchemasPerConnection < 1) {
+                throw new LineSenderException("max schemas per connection must be positive")
+                        .put("[maxSchemasPerConnection=").put(maxSchemasPerConnection).put("]");
+            }
+            this.maxSchemasPerConnection = maxSchemasPerConnection;
+            return this;
+        }
+
         /**
          * Minimum expected throughput in bytes per second for HTTP requests.
          * <br>
@@ -1829,6 +1851,13 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     pos = getValue(configurationString, pos, sink, "in_flight_window");
                     int windowSize = parseIntValue(sink, "in_flight_window");
                     inFlightWindowSize(windowSize);
+                } else if (Chars.equals("max_schemas_per_connection", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("max_schemas_per_connection is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "max_schemas_per_connection");
+                    int maxSchemas = parseIntValue(sink, "max_schemas_per_connection");
+                    maxSchemasPerConnection(maxSchemas);
                 } else if (Chars.equals("max_datagram_size", sink)) {
                     pos = getValue(configurationString, pos, sink, "max_datagram_size");
                     int mds = parseIntValue(sink, "max_datagram_size");