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,18 @@ public enum DefaultDriverOption implements DriverOption {
    * <p>Value-type: boolean
    */
   PREPARE_ON_ALL_NODES("advanced.prepared-statements.prepare-on-all-nodes"),
+
+  /**
+   * 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
+   *
+   * <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,88 @@ 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. Such statements have prepared statement "
+              + "invalidation problem when run on CQL4, this problems can lead to broken deserialization or data corruption. "
+              + "Consider using targeted select instead, or alternatively, you can set `skip-cql4-metadata-resolve-method` "
+              + "to `always-on` in execution profile to make driver ignore this problem and proceed anyway.",
+          query);
+      return false;
+    }
+    // Disable skipping metadata if results contains udt and
+    for (ColumnDefinition columnDefinition : resultSet) {
+      if (contains_udt(columnDefinition.getType())) {
+        LOGGER.warn(
+            "Prepared statement `{}` contains UDTs in result set. Such statements have prepared statement "
+                + "invalidation problem when run on CQL4, this problems can lead to broken deserialization or data corruption. "
+                + "Consider using targeted select instead, or alternatively, you can set `skip-cql4-metadata-resolve-method` "
+                + "to `always-on` in execution profile to make driver ignore this problem and proceed anyway.",
+            query);
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static boolean contains_udt(DataType dataType) {
+    if (dataType instanceof ContainerType) {
+      return contains_udt(((ContainerType) dataType).getElementType());
+    } else if (dataType instanceof TupleType) {
+      for (DataType elementType : ((TupleType) dataType).getComponentTypes()) {
+        if (contains_udt(elementType)) {
+          return true;
+        }
+      }
+      return false;
+    } else if (dataType instanceof MapType) {
+      return contains_udt(((MapType) dataType).getKeyType())
+          || contains_udt(((MapType) dataType).getValueType());
+    }
+    return dataType instanceof UserDefinedType;
+  }
 }

core/src/main/resources/reference.conf

@@ -2182,6 +2182,22 @@ datastax-java-driver {
     # Overridable in a profile: yes
     prepare-on-all-nodes = true
 
+    # There is known problem in CQL 4.x when prepared statement invalidation could be voided:
+    # https://github.com/scylladb/scylladb/issues/20860
+    # 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
+    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/tracker/RequestLogFormatterTest.java

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