Disable skip metadata for CQL4 for broken cases

CQL <= 4 has multiple flaws in regard to prepared statement invalidation,
it is effectively completely broken.
As a result, the driver may read data that does not match the cached metadata,
leading to deserialization failures or incorrect deserialization.
To address it, this commit introduce:
`PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4(advanced.prepared-statements.disable-skip-metadata-for-cql4)`
Which allows to control skip metadata flag for older protocols.
It is defaulted to true, to make driver behave properly in all
scenarios.

More info at: scylladb/scylladb#20860

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,23 @@ public enum DefaultDriverOption implements DriverOption {
    * <p>Value-type: boolean
    */
   PREPARE_ON_ALL_NODES("advanced.prepared-statements.prepare-on-all-nodes"),
+
+  /**
+   * CQL <= 4 has multiple flaws in regard to prepared statement invalidation, it is effectively
+   * completely broken. As a result, the driver may read data that does not match the cached
+   * metadata, leading to deserialization failures or incorrect deserialization. To address it, by
+   * default, driver is configured to make server send metadata with every response, via setting
+   * skip metadata flag to `false` for older protocols. Unfortunately it leads to reduced
+   * performance both on driver and server side and to excessive traffic.
+   *
+   * <p>To avoid can set given flag to `false` and do the following: 1. Avoid altering table or
+   * types under use 2. Make sure that right after alter prepared statements are being re-prepared
+   *
+   * <p>Value-type: boolean
+   */
+  PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4(
+      "advanced.prepared-statements.disable-skip-metadata-for-cql4"),
+
   /**
    * 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_DISABLE_SKIP_METADATA_FOR_CQL4, true);
     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);
+  /** Disable skip metadata flag for CQL4. */
+  public static final TypedDriverOption<Boolean> PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4 =
+      new TypedDriverOption<>(
+          DefaultDriverOption.PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4, GenericType.BOOLEAN);
   /** 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,6 +26,7 @@
 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;
@@ -43,9 +44,12 @@
 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 +73,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 +101,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 +129,14 @@ public DefaultPreparedStatement(
     this.codecRegistry = codecRegistry;
     this.protocolVersion = protocolVersion;
     this.isLWT = isLWT;
+    this.defaultExecutionProfile = defaultExecutionProfile;
+    this.skipMetadata =
+        resolveSkipMetadata(
+            resultMetadataId,
+            resultSetDefinitions,
+            executionProfileForBoundStatements != null
+                ? executionProfileForBoundStatements
+                : this.defaultExecutionProfile);
   }
 
   @NonNull
@@ -147,6 +162,10 @@ public Partitioner getPartitioner() {
     return partitioner;
   }
 
+  public boolean isSkipMetadata() {
+    return skipMetadata;
+  }
+
   @NonNull
   @Override
   public List<Integer> getPartitionKeyIndices() {
@@ -172,6 +191,14 @@ public boolean isLWT() {
   @Override
   public void setResultMetadata(
       @NonNull ByteBuffer newResultMetadataId, @NonNull ColumnDefinitions newResultSetDefinitions) {
+    this.skipMetadata =
+        resolveSkipMetadata(
+            newResultMetadataId,
+            newResultSetDefinitions,
+            executionProfileForBoundStatements != null
+                ? executionProfileForBoundStatements
+                : this.defaultExecutionProfile);
+
     this.resultMetadata = new ResultMetadata(newResultMetadataId, newResultSetDefinitions);
   }
 
@@ -242,4 +269,43 @@ private ResultMetadata(ByteBuffer resultMetadataId, ColumnDefinitions resultSetD
       this.resultSetDefinitions = resultSetDefinitions;
     }
   }
+
+  private static boolean resolveSkipMetadata(
+      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) {
+      // 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.
+    boolean disableSkipMetadata = true;
+    if (executionProfileForBoundStatements != null) {
+      disableSkipMetadata =
+          executionProfileForBoundStatements.getBoolean(
+              DefaultDriverOption.PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4);
+    }
+
+    if (disableSkipMetadata) {
+      return false;
+    }
+
+    LOGGER.warn(
+        "Due to the flaws in CQL <=4 protocol using prepared statements with skip metadata flag can lead to "
+            + "prepared statement invalidation issues when table or returned type is altered."
+            + "These issues may lead to broken deserialization or data corruption. "
+            + "To avoid this you have following options:"
+            + "1. Avoid altering table or types under use, "
+            + "2. Make sure that right after alter prepared statements are being re-prepared, "
+            + "3. Set disable-skip-metadata-for-cql4 to true to make server return metadata with each response. It goes with negative performance impact.");
+    return true;
+  }
 }

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
+    disable-skip-metadata-for-cql4 = 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.getBoolean(DefaultDriverOption.PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4))
+        .thenReturn(true);
 
     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);
   }
 }

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/BoundStatementCcmIT.java

@@ -56,11 +56,15 @@
 import com.datastax.oss.driver.internal.core.DefaultProtocolFeature;
 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.DefaultPreparedStatement;
 import com.datastax.oss.driver.internal.core.util.RoutingKey;
 import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.protocol.internal.util.Bytes;
 import com.datastax.oss.protocol.internal.util.collection.NullAllowingImmutableMap;
+import com.tngtech.java.junit.dataprovider.DataProvider;
+import com.tngtech.java.junit.dataprovider.DataProviderRunner;
+import com.tngtech.java.junit.dataprovider.UseDataProvider;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.List;
@@ -75,8 +79,10 @@
 import org.junit.rules.RuleChain;
 import org.junit.rules.TestName;
 import org.junit.rules.TestRule;
+import org.junit.runner.RunWith;
 
 @Category(ParallelizableTests.class)
+@RunWith(DataProviderRunner.class)
 public class BoundStatementCcmIT {
 
   private CcmRule ccmRule = CcmRule.getInstance();
@@ -397,6 +403,59 @@ public void should_set_all_occurrences_of_variable() {
     should_set_all_occurrences_of_variable(ps.boundStatementBuilder().setInt(id, 12).build());
   }
 
+  @DataProvider
+  public static Object[][] disableSkipMetadataForCql4() {
+    return new Object[][] {
+      {
+        true,
+      },
+      {
+        false,
+      }
+    };
+  }
+
+  @Test
+  @UseDataProvider(value = "disableSkipMetadataForCql4")
+  public void prepared_stmt_metadata_update_loopholes_test(boolean disableSkipMetadataForCql4) {
+    // v0 is an int column, but we'll bind a String to it
+    try (CqlSession session =
+        sessionWithSkipCQL4MetadataResolveMethod(disableSkipMetadataForCql4)) {
+      String disableSkipMetadataForCql4Str = String.format("%b", disableSkipMetadataForCql4);
+
+      String table = String.format("skip_metadata_test_%s_table", disableSkipMetadataForCql4Str);
+
+      session.execute(String.format("CREATE TABLE %s (pk int, v int, PRIMARY KEY (pk))", table));
+
+      session.execute(String.format("INSERT INTO %s (pk, v) VALUES (1, 1)", table));
+
+      PreparedStatement stmt =
+          session.prepare(String.format("SELECT * FROM %s WHERE pk = ?", table));
+
+      boolean isCQL4orLower = stmt.getResultMetadataId() == null;
+      boolean isPreparedStatementInvalidationBroken = isCQL4orLower && !disableSkipMetadataForCql4;
+
+      if (isCQL4orLower) {
+        assertThat(((DefaultPreparedStatement) stmt).isSkipMetadata())
+            .isEqualTo(!disableSkipMetadataForCql4);
+      }
+
+      Row row = session.execute(stmt.bind(1)).one();
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+
+      session.execute(String.format("ALTER TABLE %s ADD z int;", table));
+
+      int expectedColumnCount = 3;
+      if (isPreparedStatementInvalidationBroken) {
+        // When case of CQL4 and skip metadata is set prepared statements will not be invalidated.
+        expectedColumnCount = 2;
+      }
+
+      row = session.execute(stmt.bind(1)).one();
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(expectedColumnCount);
+    }
+  }
+
   private void should_set_all_occurrences_of_variable(BoundStatement bs) {
     assertThat(bs.getInt(0)).isEqualTo(12);
     assertThat(bs.getInt(1)).isEqualTo(12);
@@ -448,6 +507,20 @@ private CqlSession sessionWithCustomCodec(CqlIntToStringCodec codec) {
             .build();
   }
 
+  private CqlSession sessionWithSkipCQL4MetadataResolveMethod(Boolean disableSkipMetadataForCql4) {
+    return (CqlSession)
+        SessionUtils.baseBuilder()
+            .addContactEndPoints(ccmRule.getContactPoints())
+            .withKeyspace(sessionRule.keyspace())
+            .withConfigLoader(
+                SessionUtils.configLoaderBuilder()
+                    .withBoolean(
+                        DefaultDriverOption.PREPARE_DISABLE_SKIP_METADATA_FOR_CQL4,
+                        disableSkipMetadataForCql4)
+                    .build())
+            .build();
+  }
+
   private boolean supportsPerRequestKeyspace(CqlSession session) {
     InternalDriverContext context = (InternalDriverContext) session.getContext();
     ProtocolVersionRegistry protocolVersionRegistry = context.getProtocolVersionRegistry();