Disable skip metadata for CQL4 for broken cases

CQL4 has flaws that allows scenario when after schema change affected
prepare statement is not invalidated on client side.
As result, data that is read by driver does no match cached metadata and
client will fail to deserialize or deserialization will go wrong.

More info at: scylladb/scylladb#20860

This commit introduces PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD(advanced.prepared-statements.skip-cql4-metadata-resolve-method)
that controls how driver resolves skip metadata flag for CQL4 prepared
statements, it can be "smart", "always-on", "always-off". Default is
"smart".
It makes driver disable skip metadata flag only for wildcard selects and
selects that returns udts (including collections and maps)

Author: dkropachev

core/src/main/java/com/datastax/dse/driver/internal/core/cql/DseConversions.java

@@ -39,6 +39,7 @@
 import com.datastax.oss.driver.internal.core.ProtocolVersionRegistry;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
 import com.datastax.oss.driver.internal.core.cql.Conversions;
+import com.datastax.oss.driver.internal.core.cql.DefaultPreparedStatement;
 import com.datastax.oss.protocol.internal.Message;
 import com.datastax.oss.protocol.internal.request.Execute;
 import com.datastax.oss.protocol.internal.request.Query;
@@ -115,8 +116,15 @@ public static Message toContinuousPagingMessage(
           protocolVersion, DefaultProtocolFeature.UNSET_BOUND_VALUES)) {
         Conversions.ensureAllSet(boundStatement);
       }
-      boolean skipMetadata =
-          boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+
+      boolean skipMetadata;
+      if (boundStatement.getPreparedStatement() instanceof DefaultPreparedStatement) {
+        skipMetadata =
+            ((DefaultPreparedStatement) boundStatement.getPreparedStatement()).isSkipMetadata();
+      } else {
+        skipMetadata = boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+        ;
+      }
       DseQueryOptions queryOptions =
           new DseQueryOptions(
               consistencyCode,

core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java

@@ -689,6 +689,30 @@ public enum DefaultDriverOption implements DriverOption {
    * <p>Value-type: boolean
    */
   PREPARE_ON_ALL_NODES("advanced.prepared-statements.prepare-on-all-nodes"),
+
+  /**
+   * CQL 4.x has a known issue where prepared statement invalidation may be bypassed on the client
+   * side. Reference: https://github.com/scylladb/scylladb/issues/20860 When this occurs, the
+   * client's metadata can become outdated, leading to various deserialization errors.
+   *
+   * <p>To mitigate this, the driver can disable the `skip metadata` flag, ensuring the server
+   * includes metadata with every query response. This setting determines how the driver handles the
+   * `skip metadata` flag for CQL 4 prepared statements: - **"smart"** (default) – Disables the flag
+   * only for wildcard selects (`SELECT * FROM`) and queries that return UDTs (including UDT
+   * collections and maps containing UDTs). - **"always-on"** – Enables the `skip metadata` flag,
+   * preventing metadata from being sent. - **"always-off"** – Disables the `skip metadata` flag,
+   * ensuring metadata is included in every RESULT frame.
+   *
+   * <p>Sending metadata reduces performance on both the driver and server while increasing traffic.
+   * If you need to use UDTs or wildcard selects, you must either accept the performance impact or
+   * ensure: 1. No schema alterations are performed on tables or UDTs in use. 2. After any schema
+   * change, all relevant prepared statements are re-prepared.
+   *
+   * <p>Value-type: string
+   */
+  PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD(
+      "advanced.prepared-statements.skip-cql4-metadata-resolve-method"),
+
   /**
    * Whether the driver tries to prepare on new nodes at all.
    *

core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java

@@ -258,6 +258,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
     map.put(TypedDriverOption.REQUEST_TIMEOUT, requestTimeout);
     map.put(TypedDriverOption.REQUEST_CONSISTENCY, "LOCAL_ONE");
     map.put(TypedDriverOption.REQUEST_PAGE_SIZE, requestPageSize);
+    map.put(TypedDriverOption.PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD, "smart");
     map.put(TypedDriverOption.REQUEST_SERIAL_CONSISTENCY, "SERIAL");
     map.put(TypedDriverOption.REQUEST_DEFAULT_IDEMPOTENCE, false);
     map.put(TypedDriverOption.GRAPH_TRAVERSAL_SOURCE, "g");

core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java

@@ -583,6 +583,10 @@ public String toString() {
   /** Whether `Session.prepare` calls should be sent to all nodes in the cluster. */
   public static final TypedDriverOption<Boolean> PREPARE_ON_ALL_NODES =
       new TypedDriverOption<>(DefaultDriverOption.PREPARE_ON_ALL_NODES, GenericType.BOOLEAN);
+  /** Method to resolve skip metadata flag for CQL4. */
+  public static final TypedDriverOption<String> PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD =
+      new TypedDriverOption<>(
+          DefaultDriverOption.PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD, GenericType.STRING);
   /** Whether the driver tries to prepare on new nodes at all. */
   public static final TypedDriverOption<Boolean> REPREPARE_ENABLED =
       new TypedDriverOption<>(DefaultDriverOption.REPREPARE_ENABLED, GenericType.BOOLEAN);

core/src/main/java/com/datastax/oss/driver/internal/core/cql/Conversions.java

@@ -194,8 +194,15 @@ public static Message toMessage(
           protocolVersion, DefaultProtocolFeature.UNSET_BOUND_VALUES)) {
         ensureAllSet(boundStatement);
       }
-      boolean skipMetadata =
-          boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+
+      boolean skipMetadata;
+      if (boundStatement.getPreparedStatement() instanceof DefaultPreparedStatement) {
+        skipMetadata =
+            ((DefaultPreparedStatement) boundStatement.getPreparedStatement()).isSkipMetadata();
+      } else {
+        skipMetadata = boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+      }
+
       QueryOptions queryOptions =
           new QueryOptions(
               consistencyCode,
@@ -382,6 +389,12 @@ public static DefaultPreparedStatement toPreparedStatement(
 
     Partitioner partitioner = PartitionerFactory.partitioner(variableDefinitions, context);
 
+    DriverExecutionProfile defaultExecutionProfile =
+        (request.getExecutionProfileNameForBoundStatements() == null
+                || request.getExecutionProfileNameForBoundStatements().isEmpty())
+            ? context.getConfig().getDefaultProfile()
+            : context.getConfig().getProfile(request.getExecutionProfileNameForBoundStatements());
+
     return new DefaultPreparedStatement(
         ByteBuffer.wrap(response.preparedQueryId).asReadOnlyBuffer(),
         request.getQuery(),
@@ -409,7 +422,8 @@ public static DefaultPreparedStatement toPreparedStatement(
         request.areBoundStatementsTracing(),
         context.getCodecRegistry(),
         context.getProtocolVersion(),
-        lwtInfo != null && lwtInfo.isLwt(response.variablesMetadata.flags));
+        lwtInfo != null && lwtInfo.isLwt(response.variablesMetadata.flags),
+        defaultExecutionProfile);
   }
 
   public static ColumnDefinitions toColumnDefinitions(

core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java

@@ -26,26 +26,37 @@
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.BoundStatementBuilder;
+import com.datastax.oss.driver.api.core.cql.ColumnDefinition;
 import com.datastax.oss.driver.api.core.cql.ColumnDefinitions;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
 import com.datastax.oss.driver.api.core.metadata.token.Partitioner;
 import com.datastax.oss.driver.api.core.metadata.token.Token;
+import com.datastax.oss.driver.api.core.type.ContainerType;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.MapType;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
 import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
 import com.datastax.oss.driver.internal.core.data.ValuesHelper;
 import com.datastax.oss.driver.internal.core.session.RepreparePayload;
+import com.datastax.oss.driver.shaded.guava.common.base.Splitter;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import net.jcip.annotations.ThreadSafe;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @ThreadSafe
 public class DefaultPreparedStatement implements PreparedStatement {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultPreparedStatement.class);
 
   private final ByteBuffer id;
   private final RepreparePayload repreparePayload;
@@ -69,6 +80,8 @@ public class DefaultPreparedStatement implements PreparedStatement {
   private final Duration timeoutForBoundStatements;
   private final Partitioner partitioner;
   private final boolean isLWT;
+  private volatile boolean skipMetadata;
+  private final DriverExecutionProfile defaultExecutionProfile;
 
   public DefaultPreparedStatement(
       ByteBuffer id,
@@ -95,7 +108,8 @@ public DefaultPreparedStatement(
       boolean areBoundStatementsTracing,
       CodecRegistry codecRegistry,
       ProtocolVersion protocolVersion,
-      boolean isLWT) {
+      boolean isLWT,
+      DriverExecutionProfile defaultExecutionProfile) {
     this.id = id;
     this.partitionKeyIndices = partitionKeyIndices;
     // It's important that we keep a reference to this object, so that it only gets evicted from
@@ -122,6 +136,15 @@ public DefaultPreparedStatement(
     this.codecRegistry = codecRegistry;
     this.protocolVersion = protocolVersion;
     this.isLWT = isLWT;
+    this.defaultExecutionProfile = defaultExecutionProfile;
+    this.skipMetadata =
+        resolveSkipMetadata(
+            query,
+            resultMetadataId,
+            resultSetDefinitions,
+            executionProfileForBoundStatements != null
+                ? executionProfileForBoundStatements
+                : this.defaultExecutionProfile);
   }
 
   @NonNull
@@ -147,6 +170,10 @@ public Partitioner getPartitioner() {
     return partitioner;
   }
 
+  public boolean isSkipMetadata() {
+    return skipMetadata;
+  }
+
   @NonNull
   @Override
   public List<Integer> getPartitionKeyIndices() {
@@ -172,6 +199,15 @@ public boolean isLWT() {
   @Override
   public void setResultMetadata(
       @NonNull ByteBuffer newResultMetadataId, @NonNull ColumnDefinitions newResultSetDefinitions) {
+    this.skipMetadata =
+        resolveSkipMetadata(
+            this.getQuery(),
+            newResultMetadataId,
+            newResultSetDefinitions,
+            executionProfileForBoundStatements != null
+                ? executionProfileForBoundStatements
+                : this.defaultExecutionProfile);
+
     this.resultMetadata = new ResultMetadata(newResultMetadataId, newResultSetDefinitions);
   }
 
@@ -242,4 +278,93 @@ private ResultMetadata(ByteBuffer resultMetadataId, ColumnDefinitions resultSetD
       this.resultSetDefinitions = resultSetDefinitions;
     }
   }
+
+  private static boolean resolveSkipMetadata(
+      String query,
+      ByteBuffer resultMetadataId,
+      ColumnDefinitions resultSet,
+      DriverExecutionProfile executionProfileForBoundStatements) {
+    String resolveMethod =
+        (executionProfileForBoundStatements != null)
+            ? executionProfileForBoundStatements.getString(
+                DefaultDriverOption.PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD)
+            : "smart";
+
+    if (resultSet == null || resultSet.size() == 0) {
+      // there is no reason to send this flag, there will be no rows in the response and,
+      // consequently, no metadata.
+      return false;
+    }
+    if (resultMetadataId != null && resultMetadataId.capacity() > 0) {
+      // It is CQL 5 or higher.
+      // Prepared statement invalidation works perfectly no need to disable skip metadata
+      return true;
+    }
+    // It is CQL 4 or lower.
+    switch (resolveMethod) {
+      case "always-on":
+        return true;
+      case "always-off":
+        return false;
+    }
+
+    if (!resolveMethod.equals("smart")) {
+      LOGGER.warn(
+          "Property advanced.prepared-statements.skip-cql4-metadata-resolve-method is incorrectly set to `{}`, "
+              + "available options: smart, always-on, always-off. Defaulting to `smart`",
+          resolveMethod);
+    }
+    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. "
+              + "Find more mitigation options in description of `advanced.prepared-statements.skip-cql4-metadata-resolve-method` flag",
+          query);
+      return false;
+    }
+    // Disable skipping metadata if results contains udt and
+    for (ColumnDefinition columnDefinition : resultSet) {
+      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 regular columns instead of UDT. "
+                + "Find more mitigation options in description of `advanced.prepared-statements.skip-cql4-metadata-resolve-method` flag",
+            query);
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static boolean containsUDT(DataType dataType) {
+    if (dataType instanceof ContainerType) {
+      return containsUDT(((ContainerType) dataType).getElementType());
+    } else if (dataType instanceof TupleType) {
+      for (DataType elementType : ((TupleType) dataType).getComponentTypes()) {
+        if (containsUDT(elementType)) {
+          return true;
+        }
+      }
+      return false;
+    } else if (dataType instanceof MapType) {
+      return containsUDT(((MapType) dataType).getKeyType())
+          || containsUDT(((MapType) dataType).getValueType());
+    }
+    return dataType instanceof UserDefinedType;
+  }
 }

core/src/main/resources/reference.conf

@@ -2182,6 +2182,27 @@ datastax-java-driver {
     # Overridable in a profile: yes
     prepare-on-all-nodes = true
 
+    # CQL 4.x has a known issue where prepared statement invalidation may be bypassed on the client side.
+    # Reference: https://github.com/scylladb/scylladb/issues/20860
+    # When this occurs, the client's metadata can become outdated, leading to various deserialization errors.
+    #
+    # To mitigate this, the driver can disable the `skip metadata` flag, ensuring the server includes metadata with every query response.
+    # This setting determines how the driver handles the `skip metadata` flag for CQL 4 prepared statements:
+    # - **"smart"** (default) – Disables the flag only for wildcard selects (`SELECT * FROM`) and queries
+    #   that return UDTs (including UDT collections and maps containing UDTs).
+    # - **"always-on"** – Enables the `skip metadata` flag, preventing metadata from being sent.
+    # - **"always-off"** – Disables the `skip metadata` flag, ensuring metadata is included in every RESULT frame.
+    #
+    # Sending metadata reduces performance on both the driver and server while increasing traffic.
+    # If you need to use UDTs or wildcard selects, you must either accept the performance impact or ensure:
+    # 1. No schema alterations are performed on tables or UDTs in use.
+    # 2. After any schema change, all relevant prepared statements are re-prepared.
+    #
+    # Required: yes
+    # Modifiable at runtime: yes, the new value will be used for requests issued after the change.
+    # Overridable in a profile: yes
+    skip-cql4-metadata-resolve-method = smart
+
     # How the driver replicates prepared statements on a node that just came back up or joined the
     # cluster.
     reprepare-on-up {

core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java

@@ -122,6 +122,8 @@ protected RequestHandlerTestHarness(Builder builder) {
     when(defaultProfile.getBoolean(DefaultDriverOption.REQUEST_DEFAULT_IDEMPOTENCE))
         .thenReturn(builder.defaultIdempotence);
     when(defaultProfile.getBoolean(DefaultDriverOption.PREPARE_ON_ALL_NODES)).thenReturn(true);
+    when(defaultProfile.getString(DefaultDriverOption.PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD))
+        .thenReturn("smart");
 
     when(config.getDefaultProfile()).thenReturn(defaultProfile);
     when(context.getConfig()).thenReturn(config);

core/src/test/java/com/datastax/oss/driver/internal/core/tracker/RequestLogFormatterTest.java

@@ -294,6 +294,7 @@ private PreparedStatement mockPreparedStatement(String query, Map<String, DataTy
         false,
         context.getCodecRegistry(),
         context.getProtocolVersion(),
-        false);
+        false,
+        null);
   }
 }