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/CQL4SkipMetadataResolveMethod.java

@@ -0,0 +1,42 @@
+package com.datastax.oss.driver.api.core;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+public enum CQL4SkipMetadataResolveMethod {
+  // SMART (Default) - Disables the skip metadata flag only for wildcard selects (`SELECT * FROM`)
+  // and queries
+  //  that return UDTs (including UDT collections and maps containing UDTs).
+  SMART("smart"),
+  // ENABLED – Enables the `skip metadata` flag, preventing metadata from being sent
+  ENABLED("enabled"),
+  // DISABLED - Disables the `skip metadata` flag, ensuring metadata is included in every RESULT
+  // frame for bound statement execution.
+  DISABLED("disabled");
+
+  private final String value;
+
+  CQL4SkipMetadataResolveMethod(String value) {
+    this.value = value;
+  }
+
+  @Override
+  public String toString() {
+    return value;
+  }
+
+  // Case in-sensitive version of `valueOf`. To be used at all times instead of `valueOf`
+  @NonNull
+  public static CQL4SkipMetadataResolveMethod fromValue(@NonNull String value)
+      throws IllegalArgumentException {
+    switch (value.toLowerCase()) {
+      case "smart":
+        return SMART;
+      case "enabled":
+        return ENABLED;
+      case "disabled":
+        return DISABLED;
+      default:
+        throw new IllegalArgumentException("Unsupported value " + value);
+    }
+  }
+}

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

@@ -689,6 +689,34 @@ 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
+   *
+   * <p>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 bound statement RESULT query response.
+   *
+   * <p>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). -
+   * **"enabled"** – Enables the `skip metadata` flag, preventing metadata from being sent. -
+   * **"disabled"** – 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

@@ -17,6 +17,7 @@
  */
 package com.datastax.oss.driver.api.core.config;
 
+import com.datastax.oss.driver.api.core.CQL4SkipMetadataResolveMethod;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -258,6 +259,9 @@ 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,
+        CQL4SkipMetadataResolveMethod.SMART.toString());
     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,17 @@ public static DefaultPreparedStatement toPreparedStatement(
 
     Partitioner partitioner = PartitionerFactory.partitioner(variableDefinitions, context);
 
+    DriverExecutionProfile executionProfile = request.getExecutionProfileForBoundStatements();
+    if (executionProfile == null) {
+      if (request.getExecutionProfileNameForBoundStatements() != null
+          && !request.getExecutionProfileNameForBoundStatements().isEmpty()) {
+        executionProfile =
+            context.getConfig().getProfile(request.getExecutionProfileNameForBoundStatements());
+      } else {
+        executionProfile = context.getConfig().getDefaultProfile();
+      }
+    }
+
     return new DefaultPreparedStatement(
         ByteBuffer.wrap(response.preparedQueryId).asReadOnlyBuffer(),
         request.getQuery(),
@@ -395,7 +413,7 @@ public static DefaultPreparedStatement toPreparedStatement(
         partitioner,
         NullAllowingImmutableMap.copyOf(request.getCustomPayload()),
         request.getExecutionProfileNameForBoundStatements(),
-        request.getExecutionProfileForBoundStatements(),
+        executionProfile,
         request.getRoutingKeyspaceForBoundStatements(),
         request.getRoutingKeyForBoundStatements(),
         request.getRoutingTokenForBoundStatements(),

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

@@ -23,29 +23,43 @@
  */
 package com.datastax.oss.driver.internal.core.cql;
 
+import com.datastax.oss.driver.api.core.CQL4SkipMetadataResolveMethod;
 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 static final Splitter SPACE_SPLITTER = Splitter.onPattern("\\s+");
+  private static final Splitter COMMA_SPLITTER = Splitter.onPattern(",");
 
   private final ByteBuffer id;
   private final RepreparePayload repreparePayload;
@@ -69,6 +83,7 @@ public class DefaultPreparedStatement implements PreparedStatement {
   private final Duration timeoutForBoundStatements;
   private final Partitioner partitioner;
   private final boolean isLWT;
+  private volatile boolean skipMetadata;
 
   public DefaultPreparedStatement(
       ByteBuffer id,
@@ -122,6 +137,9 @@ public DefaultPreparedStatement(
     this.codecRegistry = codecRegistry;
     this.protocolVersion = protocolVersion;
     this.isLWT = isLWT;
+    this.skipMetadata =
+        resolveSkipMetadata(
+            query, resultMetadataId, resultSetDefinitions, this.executionProfileForBoundStatements);
   }
 
   @NonNull
@@ -147,6 +165,10 @@ public Partitioner getPartitioner() {
     return partitioner;
   }
 
+  public boolean isSkipMetadata() {
+    return skipMetadata;
+  }
+
   @NonNull
   @Override
   public List<Integer> getPartitionKeyIndices() {
@@ -172,6 +194,13 @@ public boolean isLWT() {
   @Override
   public void setResultMetadata(
       @NonNull ByteBuffer newResultMetadataId, @NonNull ColumnDefinitions newResultSetDefinitions) {
+    this.skipMetadata =
+        resolveSkipMetadata(
+            this.getQuery(),
+            newResultMetadataId,
+            newResultSetDefinitions,
+            executionProfileForBoundStatements);
+
     this.resultMetadata = new ResultMetadata(newResultMetadataId, newResultSetDefinitions);
   }
 
@@ -242,4 +271,132 @@ private ResultMetadata(ByteBuffer resultMetadataId, ColumnDefinitions resultSetD
       this.resultSetDefinitions = resultSetDefinitions;
     }
   }
+
+  private static boolean resolveSkipMetadata(
+      String query,
+      ByteBuffer resultMetadataId,
+      ColumnDefinitions resultSet,
+      DriverExecutionProfile executionProfileForBoundStatements) {
+    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) {
+      // Result metadata ID feature is supported, it makes prepared statement invalidation work
+      // properly.
+      // Skip Metadata should be enabled.
+      // Prepared statement invalidation works perfectly no need to disable skip metadata
+      return true;
+    }
+
+    CQL4SkipMetadataResolveMethod resolveMethod = CQL4SkipMetadataResolveMethod.SMART;
+
+    if (executionProfileForBoundStatements != null) {
+      String resolveMethodName =
+          executionProfileForBoundStatements.getString(
+              DefaultDriverOption.PREPARE_SKIP_CQL4_METADATA_RESOLVE_METHOD);
+      try {
+        resolveMethod = CQL4SkipMetadataResolveMethod.fromValue(resolveMethodName);
+      } catch (IllegalArgumentException e) {
+        LOGGER.warn(
+            "Property advanced.prepared-statements.skip-cql4-metadata-resolve-method is incorrectly set to `{}`, "
+                + "available options: smart, enabled, disabled. Defaulting to `SMART`",
+            resolveMethodName);
+        resolveMethod = CQL4SkipMetadataResolveMethod.SMART;
+      }
+    }
+
+    switch (resolveMethod) {
+      case ENABLED:
+        return true;
+      case DISABLED:
+        return false;
+      case SMART:
+        break;
+    }
+
+    List<String> chunks = SPACE_SPLITTER.splitToList(query.trim().toLowerCase());
+    if (chunks.size() < 2) {
+      // Weird query, assuming no result expected
+      return false;
+    }
+
+    if (!chunks.get(0).equals("select")) {
+      // In case if non-select sneaks in, disable skip metadata for it no result expected.
+      return false;
+    }
+
+    if (isStartPresent(query)) {
+      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;
+  }
+
+  private static boolean isStartPresent(String query) {
+    List<String> chunks = SPACE_SPLITTER.splitToList(query.trim().toLowerCase());
+    if (chunks.size() < 2) {
+      // Weird query, assuming no result expected
+      return false;
+    }
+
+    if (!chunks.get(0).equals("select")) {
+      // In case if non-select sneaks in, disable skip metadata for it no result expected.
+      return false;
+    }
+
+    for (String chunk : chunks) {
+      if (chunk.equals("from")) {
+        return false;
+      }
+      if (chunk.equals("*")) {
+        return true;
+      }
+      for (String part : COMMA_SPLITTER.split(chunk)) {
+        if (part.equals("*")) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
 }