CASSJAVA-130: Fix Result-Metadata Race Causing Corrupt SKIP_METADATA Row Decoding

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

@@ -120,7 +120,10 @@ public static DriverExecutionProfile resolveExecutionProfile(
   }
 
   public static Message toMessage(
-      Statement<?> statement, DriverExecutionProfile config, InternalDriverContext context) {
+      Statement<?> statement,
+      DriverExecutionProfile config,
+      InternalDriverContext context,
+      DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot) {
     ConsistencyLevelRegistry consistencyLevelRegistry = context.getConsistencyLevelRegistry();
     ConsistencyLevel consistency = statement.getConsistencyLevel();
     int consistencyCode =
@@ -185,8 +188,22 @@ public static Message toMessage(
           protocolVersion, DefaultProtocolFeature.UNSET_BOUND_VALUES)) {
         ensureAllSet(boundStatement);
       }
-      boolean skipMetadata =
-          boundStatement.getPreparedStatement().getResultSetDefinitions().size() > 0;
+      PreparedStatement preparedStatement = boundStatement.getPreparedStatement();
+      // Use the snapshot taken when sendRequest captured it (before this method was called), not
+      // the prepared statement's live cache: a concurrent execution's CASSANDRA-10786 schema-change
+      // update could otherwise swap the cache between that snapshot and this read, so the
+      // resultMetadataId sent on the wire would no longer match the definitions the response will
+      // later be decoded against (see resultMetadataSnapshot in CqlRequestHandler).
+      ByteBuffer resultMetadataId;
+      ColumnDefinitions resultSetDefinitions;
+      if (resultMetadataSnapshot != null) {
+        resultMetadataId = resultMetadataSnapshot.getResultMetadataId();
+        resultSetDefinitions = resultMetadataSnapshot.getResultSetDefinitions();
+      } else {
+        resultMetadataId = preparedStatement.getResultMetadataId();
+        resultSetDefinitions = preparedStatement.getResultSetDefinitions();
+      }
+      boolean skipMetadata = resultSetDefinitions.size() > 0;
       QueryOptions queryOptions =
           new QueryOptions(
               consistencyCode,
@@ -199,9 +216,7 @@ public static Message toMessage(
               timestamp,
               null,
               nowInSeconds);
-      PreparedStatement preparedStatement = boundStatement.getPreparedStatement();
       ByteBuffer id = preparedStatement.getId();
-      ByteBuffer resultMetadataId = preparedStatement.getResultMetadataId();
       return new Execute(
           Bytes.getArray(id),
           (resultMetadataId == null) ? null : Bytes.getArray(resultMetadataId),
@@ -320,11 +335,13 @@ public static AsyncResultSet toResultSet(
       Result result,
       ExecutionInfo executionInfo,
       CqlSession session,
-      InternalDriverContext context) {
+      InternalDriverContext context,
+      DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot) {
     if (result instanceof Rows) {
       Rows rows = (Rows) result;
       Statement<?> statement = (Statement<?>) executionInfo.getRequest();
-      ColumnDefinitions columnDefinitions = getResultDefinitions(rows, statement, context);
+      ColumnDefinitions columnDefinitions =
+          getResultDefinitions(rows, statement, context, resultMetadataSnapshot);
       return new DefaultAsyncResultSet(
           columnDefinitions, executionInfo, rows.getData(), session, context);
     } else if (result instanceof Prepared) {
@@ -336,12 +353,31 @@ public static AsyncResultSet toResultSet(
     }
   }
 
+  /**
+   * Returns {@code preparedStatement} narrowed to the internal implementation, or {@code null} if
+   * it isn't one (e.g. a test double implementing the public {@link PreparedStatement} interface
+   * directly).
+   */
+  static DefaultPreparedStatement asDefaultPreparedStatement(PreparedStatement preparedStatement) {
+    return (preparedStatement instanceof DefaultPreparedStatement)
+        ? (DefaultPreparedStatement) preparedStatement
+        : null;
+  }
+
   public static ColumnDefinitions getResultDefinitions(
-      Rows rows, Statement<?> statement, InternalDriverContext context) {
+      Rows rows,
+      Statement<?> statement,
+      InternalDriverContext context,
+      DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot) {
     RowsMetadata rowsMetadata = rows.getMetadata();
     if (rowsMetadata.columnSpecs.isEmpty()) {
       // If the response has no metadata, it means the request had SKIP_METADATA set, the driver
-      // only ever does that for bound statements.
+      // only ever does that for bound statements. Use the snapshot taken when the request was
+      // encoded, since the prepared statement's cached copy may have since been overwritten by a
+      // concurrent execution's CASSANDRA-10786 schema-change update (see getCurrentResultMetadata).
+      if (resultMetadataSnapshot != null) {
+        return resultMetadataSnapshot.getResultSetDefinitions();
+      }
       BoundStatement boundStatement = (BoundStatement) statement;
       return boundStatement.getPreparedStatement().getResultSetDefinitions();
     } else {

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

@@ -27,6 +27,7 @@
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.connection.FrameTooLongException;
 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.ExecutionInfo;
 import com.datastax.oss.driver.api.core.cql.Statement;
 import com.datastax.oss.driver.api.core.metadata.Node;
@@ -291,6 +292,11 @@ private void sendRequest(
               .map((g) -> g.getDecoratedStatement(finalStatement, nodeRequestId))
               .orElse(finalStatement);
 
+      // Captured once and reused for both the callback and the outgoing message, so a concurrent
+      // schema-change update racing between the two can't make the wire message's
+      // resultMetadataId disagree with the definitions the response will later be decoded against.
+      DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot =
+          resultMetadataSnapshot(statement);
       NodeResponseCallback nodeResponseCallback =
           new NodeResponseCallback(
               statement,
@@ -300,14 +306,31 @@ private void sendRequest(
               currentExecutionIndex,
               retryCount,
               scheduleNextExecution,
-              logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex));
-      Message message = Conversions.toMessage(statement, executionProfile, context);
+              logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex),
+              resultMetadataSnapshot);
+      Message message =
+          Conversions.toMessage(statement, executionProfile, context, resultMetadataSnapshot);
       channel
           .write(message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback)
           .addListener(nodeResponseCallback);
     }
   }
 
+  /**
+   * Captures the prepared statement's result metadata as it stands right now, so that decoding this
+   * request's response later (potentially after a concurrent execution has swapped that metadata,
+   * see CASSANDRA-10786) uses what was true when this request was actually encoded.
+   */
+  private static DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot(
+      Statement<?> statement) {
+    if (!(statement instanceof BoundStatement)) {
+      return null;
+    }
+    DefaultPreparedStatement preparedStatement =
+        Conversions.asDefaultPreparedStatement(((BoundStatement) statement).getPreparedStatement());
+    return (preparedStatement == null) ? null : preparedStatement.getCurrentResultMetadata();
+  }
+
   private void recordError(Node node, Throwable error) {
     // Use a local variable to do only a single single volatile read in the nominal case
     List<Map.Entry<Node, Throwable>> errorsSnapshot = this.errors;
@@ -345,7 +368,8 @@ private void setFinalResult(
       ExecutionInfo executionInfo =
           buildExecutionInfo(callback, resultMessage, responseFrame, schemaInAgreement);
       AsyncResultSet resultSet =
-          Conversions.toResultSet(resultMessage, executionInfo, session, context);
+          Conversions.toResultSet(
+              resultMessage, executionInfo, session, context, callback.resultMetadataSnapshot);
       if (result.complete(resultSet)) {
         cancelScheduledTasks();
         throttler.signalSuccess(this);
@@ -505,6 +529,9 @@ private class NodeResponseCallback
     private final int retryCount;
     private final boolean scheduleNextExecution;
     private final String logPrefix;
+    // Snapshot of the prepared statement's result metadata taken when this execution's request
+    // was encoded (null for non-bound statements). See resultMetadataSnapshot(Statement).
+    private final DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot;
 
     private NodeResponseCallback(
         Statement<?> statement,
@@ -514,7 +541,8 @@ private NodeResponseCallback(
         int execution,
         int retryCount,
         boolean scheduleNextExecution,
-        String logPrefix) {
+        String logPrefix,
+        DefaultPreparedStatement.ResultMetadata resultMetadataSnapshot) {
       this.statement = statement;
       this.node = node;
       this.queryPlan = queryPlan;
@@ -523,6 +551,7 @@ private NodeResponseCallback(
       this.retryCount = retryCount;
       this.scheduleNextExecution = scheduleNextExecution;
       this.logPrefix = logPrefix;
+      this.resultMetadataSnapshot = resultMetadataSnapshot;
     }
 
     // this gets invoked once the write completes.

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

@@ -152,6 +152,16 @@ public void setResultMetadata(
     this.resultMetadata = new ResultMetadata(newResultMetadataId, newResultSetDefinitions);
   }
 
+  /**
+   * Returns an atomic snapshot of the id+definitions pair currently in effect, in a single volatile
+   * read. Used to capture the metadata that was in effect when a request was encoded, so that
+   * decoding its response later can use that snapshot instead of racing against a concurrent {@link
+   * #setResultMetadata} call (see CASSANDRA-10786 and the SKIP_METADATA optimization).
+   */
+  ResultMetadata getCurrentResultMetadata() {
+    return this.resultMetadata;
+  }
+
   @NonNull
   @Override
   public BoundStatement bind(@NonNull Object... values) {
@@ -210,13 +220,21 @@ public RepreparePayload getRepreparePayload() {
     return this.repreparePayload;
   }
 
-  private static class ResultMetadata {
-    private ByteBuffer resultMetadataId;
-    private ColumnDefinitions resultSetDefinitions;
+  static class ResultMetadata {
+    private final ByteBuffer resultMetadataId;
+    private final ColumnDefinitions resultSetDefinitions;
 
     private ResultMetadata(ByteBuffer resultMetadataId, ColumnDefinitions resultSetDefinitions) {
       this.resultMetadataId = resultMetadataId;
       this.resultSetDefinitions = resultSetDefinitions;
     }
+
+    ByteBuffer getResultMetadataId() {
+      return resultMetadataId;
+    }
+
+    ColumnDefinitions getResultSetDefinitions() {
+      return resultSetDefinitions;
+    }
   }
 }

core/src/test/java/com/datastax/dse/driver/internal/core/cql/reactive/CqlRequestReactiveProcessorTest.java

@@ -152,7 +152,8 @@ public void should_complete_multi_page_result(ProtocolVersion version) {
               DseTestFixtures.tenDseRows(2, true),
               mockInfo,
               harness.getSession(),
-              harness.getContext()));
+              harness.getContext(),
+              null));
 
       List<ReactiveRow> rows = rowsPublisher.toList().blockingGet();
       assertThat(rows).hasSize(20);