Intruduce flag to disable skip metadata

CQL 4.x 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 REQUEST_DISABLE_SKIP_METADATA(basic.request.disable-skip-metadata)
which is set to true by default.
Which will make server reply every time with metadata in response.

Author: dkropachev

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

@@ -116,7 +116,10 @@ public static Message toContinuousPagingMessage(
         Conversions.ensureAllSet(boundStatement);
       }
       boolean skipMetadata =
-          boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+          Conversions.resolveSkipMetadata(
+              boundStatement,
+              context,
+              boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0);
       DseQueryOptions queryOptions =
           new DseQueryOptions(
               consistencyCode,

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

@@ -68,6 +68,12 @@ public enum DefaultDriverOption implements DriverOption {
    * <p>Value-Type: int
    */
   REQUEST_PAGE_SIZE("basic.request.page-size"),
+  /**
+   * The page size.
+   *
+   * <p>Value-Type: int
+   */
+  REQUEST_DISABLE_SKIP_METADATA("basic.request.disable-skip-metadata"),
   /**
    * The serial consistency level.
    *

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.REQUEST_DISABLE_SKIP_METADATA, 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

@@ -126,6 +126,10 @@ public String toString() {
   /** The page size. */
   public static final TypedDriverOption<Integer> REQUEST_PAGE_SIZE =
       new TypedDriverOption<>(DefaultDriverOption.REQUEST_PAGE_SIZE, GenericType.INTEGER);
+  /** The disable skip metadata. */
+  public static final TypedDriverOption<Boolean> REQUEST_DISABLE_SKIP_METADATA =
+      new TypedDriverOption<>(
+          DefaultDriverOption.REQUEST_DISABLE_SKIP_METADATA, GenericType.BOOLEAN);
   /** The serial consistency level. */
   public static final TypedDriverOption<String> REQUEST_SERIAL_CONSISTENCY =
       new TypedDriverOption<>(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY, GenericType.STRING);

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

@@ -194,8 +194,12 @@ public static Message toMessage(
           protocolVersion, DefaultProtocolFeature.UNSET_BOUND_VALUES)) {
         ensureAllSet(boundStatement);
       }
+
       boolean skipMetadata =
-          boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+          resolveSkipMetadata(
+              boundStatement,
+              context,
+              boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0);
       QueryOptions queryOptions =
           new QueryOptions(
               consistencyCode,
@@ -556,6 +560,16 @@ public static boolean resolveIdempotence(Request request, InternalDriverContext
         : requestIsIdempotent;
   }
 
+  public static boolean resolveSkipMetadata(
+      Request request, InternalDriverContext context, boolean responseHasMetadata) {
+    if (!responseHasMetadata) {
+      // If no reason to send this flag, there will be no metadata anyway.
+      return false;
+    }
+    DriverExecutionProfile executionProfile = resolveExecutionProfile(request, context);
+    return !executionProfile.getBoolean(DefaultDriverOption.REQUEST_DISABLE_SKIP_METADATA);
+  }
+
   public static Duration resolveRequestTimeout(Request request, InternalDriverContext context) {
     DriverExecutionProfile executionProfile = resolveExecutionProfile(request, context);
     return request.getTimeout() != null

core/src/main/resources/reference.conf

@@ -134,6 +134,18 @@ datastax-java-driver {
     # Overridable in a profile: yes
     page-size = 5000
 
+    # Disable skipping metadata for every query
+    # 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 that turn this option on, it will make server respond with metadata on every query.
+    # Unfortunately it comes with prise of having more traffic and more deserialization CPU footprint on client side.
+    #
+    # 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 = true
+
     # The serial consistency level.
     # The allowed values are SERIAL and LOCAL_SERIAL.
     #

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

@@ -36,6 +36,7 @@
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
+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.ResultSet;
@@ -56,6 +57,7 @@
 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.type.DefaultUserDefinedType;
 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;
@@ -397,6 +399,53 @@ public void should_set_all_occurrences_of_variable() {
     should_set_all_occurrences_of_variable(ps.boundStatementBuilder().setInt(id, 12).build());
   }
 
+  @Test
+  public void disable_skip_metadata() {
+    // v0 is an int column, but we'll bind a String to it
+    try (CqlSession session = sessionWithNoSkipMetadata()) {
+      session.execute("CREATE TYPE IF NOT EXISTS udttype(x int, y int)");
+
+      session.execute("CREATE TABLE udt_test (pk int, v udttype, PRIMARY KEY (pk))");
+      session.execute("CREATE TABLE regular_table (pk int, v int, PRIMARY KEY (pk))");
+
+      session.execute("INSERT INTO regular_table (pk, v) VALUES (1, 1)");
+      session.execute("INSERT INTO udt_test (pk, v) VALUES (1, {x: 1, y: 1})");
+
+      PreparedStatement prepared_regular_table =
+          session.prepare("SELECT * FROM regular_table WHERE pk = ?");
+      PreparedStatement prepared_udt_all = session.prepare("SELECT * FROM udt_test WHERE pk = ?");
+      PreparedStatement prepared_udt_concrete =
+          session.prepare("SELECT pk, v FROM udt_test WHERE pk = ?");
+
+      Row row = session.execute(prepared_udt_all.bind(1)).one();
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+      assertThat(get_udt_column_count(row.getColumnDefinitions().get(1))).isEqualTo(2);
+      row = session.execute(prepared_udt_concrete.bind(1)).one();
+      assertThat(get_udt_column_count(row.getColumnDefinitions().get(1))).isEqualTo(2);
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+      row = session.execute(prepared_regular_table.bind(1)).one();
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+
+      session.execute("ALTER TABLE regular_table ADD z int;");
+      session.execute("ALTER TYPE udttype ADD z int;");
+
+      row = session.execute(prepared_udt_all.bind(1)).one();
+      row.getUdtValue(1);
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+      assertThat(get_udt_column_count(row.getColumnDefinitions().get(1))).isEqualTo(3);
+      row = session.execute(prepared_udt_concrete.bind(1)).one();
+      row.getUdtValue(1);
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(2);
+      assertThat(get_udt_column_count(row.getColumnDefinitions().get(1))).isEqualTo(3);
+      row = session.execute(prepared_regular_table.bind(1)).one();
+      assertThat(row.getColumnDefinitions().size()).isEqualTo(3);
+    }
+  }
+
+  private int get_udt_column_count(ColumnDefinition cd) {
+    return ((DefaultUserDefinedType) cd.getType()).getFieldTypes().size();
+  }
+
   private void should_set_all_occurrences_of_variable(BoundStatement bs) {
     assertThat(bs.getInt(0)).isEqualTo(12);
     assertThat(bs.getInt(1)).isEqualTo(12);
@@ -448,6 +497,18 @@ private CqlSession sessionWithCustomCodec(CqlIntToStringCodec codec) {
             .build();
   }
 
+  private CqlSession sessionWithNoSkipMetadata() {
+    return (CqlSession)
+        SessionUtils.baseBuilder()
+            .addContactEndPoints(ccmRule.getContactPoints())
+            .withKeyspace(sessionRule.keyspace())
+            .withConfigLoader(
+                SessionUtils.configLoaderBuilder()
+                    .withBoolean(DefaultDriverOption.REQUEST_DISABLE_SKIP_METADATA, true)
+                    .build())
+            .build();
+  }
+
   private boolean supportsPerRequestKeyspace(CqlSession session) {
     InternalDriverContext context = (InternalDriverContext) session.getContext();
     ProtocolVersionRegistry protocolVersionRegistry = context.getProtocolVersionRegistry();