Introduce skipCQL4MetadataResolveMethod to address CQL4 issues

Introduce skipCQL4MetadataResolveMethod to Mitigate CQL4 Issues
CQL4 has a flaw where, after a schema change, an affected prepared statement may not be invalidated on the client side.
As a result, the driver may read data that does not match the cached metadata, leading to deserialization failures or incorrect deserialization.

For more details, see: scylladb/scylladb#20860.

This commit introduces the skipCQL4MetadataResolveMethod, which determines how the driver resolves the skip metadata flag for CQL4 prepared statements.
It supports three modes:
SMART (default) – Disables the skip metadata flag only for wildcard selects and selects that return UDTs (including collections and maps).
ALWAYS_ON – Always enables the skip metadata flag.
ALWAYS_OFF – Always disables the skip metadata flag.

Author: dkropachev

driver-core/src/main/java/com/datastax/driver/core/DefaultPreparedStatement.java

@@ -24,13 +24,16 @@
 import static com.datastax.driver.core.ProtocolVersion.V4;
 
 import com.datastax.driver.core.policies.RetryPolicy;
+import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableMap;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class DefaultPreparedStatement implements PreparedStatement {
-
+  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultPreparedStatement.class);
   private static final String SCYLLA_CDC_LOG_SUFFIX = "_scylla_cdc_log";
 
   final PreparedId preparedId;
@@ -50,6 +53,7 @@ public class DefaultPreparedStatement implements PreparedStatement {
   volatile RetryPolicy retryPolicy;
   volatile ImmutableMap<String, ByteBuffer> outgoingPayload;
   volatile Boolean idempotent;
+  volatile boolean skipMetadata;
 
   private DefaultPreparedStatement(
       PreparedId id,
@@ -66,6 +70,7 @@ private DefaultPreparedStatement(
     this.cluster = cluster;
     this.isLWT = isLWT;
     this.partitioner = partitioner;
+    this.skipMetadata = this.calculateSkipMetadata();
   }
 
   static DefaultPreparedStatement fromMessage(
@@ -172,6 +177,71 @@ private static Token.Factory partitioner(ColumnDefinitions defs, Cluster cluster
     return null;
   }
 
+  private boolean calculateSkipMetadata() {
+    if (cluster.manager.protocolVersion() == ProtocolVersion.V1
+        || preparedId.resultSetMetadata.variables == null) {
+      // CQL1 does not support it.
+      // If no rows returned there is no reason to send this flag, consequently, no metadata.
+      return false;
+    }
+
+    if (preparedId.resultSetMetadata.id != null
+        && preparedId.resultSetMetadata.id.bytes.length > 0) {
+      // It is CQL 5 or higher.
+      // Prepared statement invalidation works perfectly no need to disable skip metadata
+      return true;
+    }
+
+    switch (cluster.getConfiguration().getQueryOptions().getSkipCQL4MetadataResolveMethod()) {
+      case ALWAYS_ON:
+        return true;
+      case ALWAYS_OFF:
+        return false;
+    }
+
+    List<String> chunks = Splitter.onPattern("\\s+").splitToList(query);
+    if (chunks.size() < 2) {
+      // Weird query, assuming no result expected
+      return false;
+    }
+    if (!chunks.get(0).toLowerCase().startsWith("select")) {
+      // In case if non-select sneaks in, disable skip metadata for it no result expected.
+      return false;
+    }
+    if (chunks.get(1).equals("*")) {
+      LOGGER.warn(
+          "Prepared statement {} is a wildcard select, which can cause prepared statement invalidation issues when executed on CQL4. "
+              + "These issues may lead to broken deserialization or data corruption. "
+              + "To mitigate this, the driver ensures that the server returns metadata with each query for such statements, "
+              + "though this negatively impacts performance. "
+              + "To avoid this, consider using a targeted select instead. "
+              + "Alternatively, you can enable the skip-cql4-metadata-resolve-method option in the execution profile by setting it to `always-on`, "
+              + "allowing the driver to ignore this issue and proceed regardless, risking broken deserialization or data corruption.",
+          query);
+      return false;
+    }
+    // Disable skipping metadata if results contains udt and
+    for (ColumnDefinitions.Definition columnDefinition : preparedId.resultSetMetadata.variables) {
+      if (containsUDT(columnDefinition.getType())) {
+        LOGGER.warn(
+            "Prepared statement {} contains UDT in result, which can cause prepared statement invalidation issues when executed on CQL4. "
+                + "These issues may lead to broken deserialization or data corruption. "
+                + "To mitigate this, the driver ensures that the server returns metadata with each query for such statements, "
+                + "though this negatively impacts performance. "
+                + "To avoid this, consider using a targeted select instead. "
+                + "Alternatively, you can enable the skip-cql4-metadata-resolve-method option in the execution profile by setting it to `always-on`, "
+                + "allowing the driver to ignore this issue and proceed regardless, risking broken deserialization or data corruption.",
+            query);
+        return false;
+      }
+    }
+    return true;
+  }
+
+  public boolean isSkipMetadata() {
+    return skipMetadata;
+  }
+
   @Override
   public ColumnDefinitions getVariables() {
     return preparedId.boundValuesMetadata.variables;
@@ -315,4 +385,16 @@ public Boolean isIdempotent() {
   public boolean isLWT() {
     return isLWT;
   }
+
+  private static boolean containsUDT(DataType dataType) {
+    if (dataType.isCollection()) {
+      for (DataType elementType : dataType.getTypeArguments()) {
+        if (containsUDT(elementType)) {
+          return true;
+        }
+      }
+      return false;
+    }
+    return dataType instanceof UserType;
+  }
 }

driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java

@@ -69,6 +69,8 @@ public class QueryOptions {
   private volatile boolean reprepareOnUp = true;
   private volatile Cluster.Manager manager;
   private volatile boolean prepareOnAllHosts = true;
+  private volatile CQL4SkipMetadataResolveMethod skipCQL4MetadataResolveMethod =
+      CQL4SkipMetadataResolveMethod.SMART;
 
   private volatile boolean schemaQueriesPaged = true;
 
@@ -193,6 +195,37 @@ public boolean getDefaultIdempotence() {
     return defaultIdempotence;
   }
 
+  /**
+   * There is known problem in CQL 4.x when prepared statement invalidation could be voided: <a
+   * href="https://github.com/scylladb/scylladb/issues/20860">more info</a> When it happens metadata
+   * on client side does not match data and deserialization can go wrong in many ways To avoid
+   * driver can disable skip metadata flag to make server respond with metadata on every query.
+   * Unfortunately it causes excessive network traffic and CPU overhead on both server and driver
+   * side. This option controls how driver resolves skip metadata flag for CQL4 prepared statements.
+   * SMART - disable flag only for wildcard selects (select * from) and selects that return UDTs,
+   * including collections of UDTs and maps that contain UDTs ALWAYS_ON - flag is always set
+   * ALWAYS_OFF - flag is always disabled Default is SMART Required: yes Modifiable at runtime: yes,
+   * the new value will be used for requests issued after the change. Overridable in a profile: yes
+   *
+   * @param method the new value to set as skip metadata resolve method.
+   * @return this {@code QueryOptions} instance.
+   */
+  public QueryOptions setSkipCQL4MetadataResolveMethod(CQL4SkipMetadataResolveMethod method) {
+    this.skipCQL4MetadataResolveMethod = method;
+    return this;
+  }
+
+  /**
+   * Skip metadata resolve method .
+   *
+   * <p>It defaults to {@link #skipCQL4MetadataResolveMethod.SMART}.
+   *
+   * @return the default idempotence for queries.
+   */
+  public CQL4SkipMetadataResolveMethod getSkipCQL4MetadataResolveMethod() {
+    return this.skipCQL4MetadataResolveMethod;
+  }
+
   /**
    * Set whether the driver should prepare statements on all hosts in the cluster.
    *
@@ -583,4 +616,10 @@ public int hashCode() {
   public boolean isConsistencySet() {
     return consistencySet;
   }
+
+  public enum CQL4SkipMetadataResolveMethod {
+    ALWAYS_ON,
+    ALWAYS_OFF,
+    SMART
+  }
 }

driver-core/src/main/java/com/datastax/driver/core/SessionManager.java

@@ -654,11 +654,17 @@ else if (fetchSize != Integer.MAX_VALUE)
       }
       if (protocolVersion.compareTo(ProtocolVersion.V4) < 0) bs.ensureAllSet();
 
-      // skip resultset metadata if version > 1 (otherwise this feature is not supported)
-      // and if we already have metadata for the prepared statement being executed.
-      boolean skipMetadata =
-          protocolVersion != ProtocolVersion.V1
-              && bs.statement.getPreparedId().resultSetMetadata.variables != null;
+      boolean skipMetadata;
+      if (bs.statement instanceof DefaultPreparedStatement) {
+        skipMetadata = ((DefaultPreparedStatement) bs.statement).isSkipMetadata();
+      } else {
+        skipMetadata =
+            protocolVersion != ProtocolVersion.V1
+                && bs.statement.getPreparedId().resultSetMetadata.variables != null;
+        // skip resultset metadata if version > 1 (otherwise this feature is not supported)
+        // and if we already have metadata for the prepared statement being executed.
+      }
+
       Requests.QueryProtocolOptions options =
           new Requests.QueryProtocolOptions(
               Message.Request.Type.EXECUTE,