diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/Conversions.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/Conversions.java index ff9384b3e24..4326e820c0c 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/Conversions.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/Conversions.java @@ -320,11 +320,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 +338,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 { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java index 6842547b11a..b7a9e003d35 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java +++ b/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; @@ -300,7 +301,8 @@ private void sendRequest( currentExecutionIndex, retryCount, scheduleNextExecution, - logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex)); + logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex), + resultMetadataSnapshot(statement)); Message message = Conversions.toMessage(statement, executionProfile, context); channel .write(message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback) @@ -308,6 +310,21 @@ private void sendRequest( } } + /** + * 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> errorsSnapshot = this.errors; @@ -345,7 +362,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 +523,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 +535,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 +545,7 @@ private NodeResponseCallback( this.retryCount = retryCount; this.scheduleNextExecution = scheduleNextExecution; this.logPrefix = logPrefix; + this.resultMetadataSnapshot = resultMetadataSnapshot; } // this gets invoked once the write completes. diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java index e45e1e5add0..d5d4a8278a1 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java +++ b/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; + } } } diff --git a/core/src/test/java/com/datastax/dse/driver/internal/core/cql/reactive/CqlRequestReactiveProcessorTest.java b/core/src/test/java/com/datastax/dse/driver/internal/core/cql/reactive/CqlRequestReactiveProcessorTest.java index a7a6bced9e8..c5bc2305027 100644 --- a/core/src/test/java/com/datastax/dse/driver/internal/core/cql/reactive/CqlRequestReactiveProcessorTest.java +++ b/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 rows = rowsPublisher.toList().blockingGet(); assertThat(rows).hasSize(20); diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/ConversionsMetadataRaceTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/ConversionsMetadataRaceTest.java new file mode 100644 index 00000000000..dce2e319a5d --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/ConversionsMetadataRaceTest.java @@ -0,0 +1,191 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.cql; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +import com.datastax.oss.driver.api.core.DefaultProtocolVersion; +import com.datastax.oss.driver.api.core.cql.BoundStatement; +import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; +import com.datastax.oss.driver.api.core.cql.Row; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry; +import com.datastax.oss.driver.internal.core.context.InternalDriverContext; +import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; +import com.datastax.oss.protocol.internal.ProtocolConstants; +import com.datastax.oss.protocol.internal.response.result.ColumnSpec; +import com.datastax.oss.protocol.internal.response.result.DefaultRows; +import com.datastax.oss.protocol.internal.response.result.RawType; +import com.datastax.oss.protocol.internal.response.result.Rows; +import com.datastax.oss.protocol.internal.response.result.RowsMetadata; +import com.datastax.oss.protocol.internal.util.Bytes; +import java.nio.ByteBuffer; +import java.util.ArrayDeque; +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import org.junit.Test; +import org.mockito.Mockito; + +/** + * Deterministic (non-timing-based) regression test for the SKIP_METADATA / CASSANDRA-10786 race: + * execution A's request is encoded while the prepared statement's cached result metadata is + * OLD_DEFS, then a concurrent execution B's response arrives first and swaps the shared {@link + * DefaultPreparedStatement} metadata to NEW_DEFS before A's own SKIP_METADATA response is decoded. + * + *

NEW_DEFS here models a real {@code ALTER TABLE ... ADD extra_0 bigint} (as issued by the live + * harness in {@code MetadataRaceHarness}): Cassandra returns non-primary-key columns in + * alphabetical order, so the added column ("extra_0") sorts before the existing ones and is + * inserted at index 0, shifting "group_id", "is_deleted" and "last_modified" each one slot to the + * right. Their relative order to each other is unchanged -- ALTER TABLE can only insert a column, + * it cannot reorder existing ones -- but their absolute indices shift, which is exactly what used + * to corrupt decoding of A's already-encoded, old-layout row bytes before the fix. + * + *

The fix snapshots the metadata at encode time and threads it through to decode time (see + * {@link Conversions#getResultDefinitions}), so A correctly decodes against OLD_DEFS regardless of + * B's concurrent swap. Response arrival order here is controlled by direct, sequential calls into + * {@code getResultDefinitions}, so the scenario triggers identically on every run. + */ +public class ConversionsMetadataRaceTest { + + private static final ByteBuffer OLD_RESULT_METADATA_ID = Bytes.fromHexString("0x01"); + private static final ByteBuffer NEW_RESULT_METADATA_ID = Bytes.fromHexString("0x02"); + + @Test + public void should_decode_skip_metadata_response_using_encode_time_snapshot() { + InternalDriverContext context = Mockito.mock(InternalDriverContext.class); + Mockito.when(context.getCodecRegistry()).thenReturn(CodecRegistry.DEFAULT); + Mockito.when(context.getProtocolVersion()).thenReturn(DefaultProtocolVersion.V4); + + // OLD_DEFS: the layout in effect when A's and B's requests were encoded (SKIP_METADATA + // decided here). Regular (non-PK) columns come back from Cassandra in alphabetical order, so + // for the schema "group_id, is_deleted, last_modified" that's also declaration order here. + ColumnDefinitions oldDefs = + Conversions.toColumnDefinitions( + new RowsMetadata( + ImmutableList.of( + columnSpec("group_id", 0, ProtocolConstants.DataType.VARCHAR), + columnSpec("is_deleted", 1, ProtocolConstants.DataType.BOOLEAN), + columnSpec("last_modified", 2, ProtocolConstants.DataType.BIGINT)), + null, + new int[0], + null), + context); + + DefaultPreparedStatement preparedStatement = + new DefaultPreparedStatement( + Bytes.fromHexString("0xAAAA"), + "SELECT * FROM t2 WHERE k = ?", + DefaultColumnDefinitions.valueOf(Collections.emptyList()), + Collections.emptyList(), + OLD_RESULT_METADATA_ID, + oldDefs, + null, + Collections.emptyMap(), + null, + null, + null, + null, + null, + Collections.emptyMap(), + null, + null, + null, + 5000, + null, + null, + false, + CodecRegistry.DEFAULT, + DefaultProtocolVersion.V4); + + // Two concurrent executions of the same cached PreparedStatement, both encoded while OLD_DEFS + // is current -- so both capture the same encode-time snapshot, exactly as + // CqlRequestHandler.sendRequest does for each execution before calling Conversions.toMessage. + BoundStatement stmtA = preparedStatement.bind(); + BoundStatement stmtB = preparedStatement.bind(); + DefaultPreparedStatement.ResultMetadata encodeTimeSnapshot = + preparedStatement.getCurrentResultMetadata(); + + // B's response lands first: the server signals a schema change (CASSANDRA-10786) triggered by + // a real "ALTER TABLE t2 ADD extra_0 bigint". Non-PK columns are returned alphabetically, so + // "extra_0" sorts before "group_id" and is inserted at index 0 -- shifting the three existing + // columns one slot to the right, without reordering them relative to each other. Decoding this + // response swaps the prepared statement's shared cached metadata to NEW_DEFS. + Rows rowsB = + new DefaultRows( + new RowsMetadata( + ImmutableList.of( + columnSpec("extra_0", 0, ProtocolConstants.DataType.BIGINT), + columnSpec("group_id", 1, ProtocolConstants.DataType.VARCHAR), + columnSpec("is_deleted", 2, ProtocolConstants.DataType.BOOLEAN), + columnSpec("last_modified", 3, ProtocolConstants.DataType.BIGINT)), + null, + new int[0], + Bytes.getArray(NEW_RESULT_METADATA_ID)), + oneRow( + TypeCodecs.BIGINT.encode(0L, DefaultProtocolVersion.V4), + TypeCodecs.TEXT.encode("g1", DefaultProtocolVersion.V4), + TypeCodecs.BOOLEAN.encode(true, DefaultProtocolVersion.V4), + TypeCodecs.BIGINT.encode(1_700_000_000_000L, DefaultProtocolVersion.V4))); + + ColumnDefinitions defsForB = + Conversions.getResultDefinitions(rowsB, stmtB, context, encodeTimeSnapshot); + assertThat(defsForB.get(2).getName().asInternal()).isEqualTo("is_deleted"); + + // The swap happened as a side effect of decoding B: the shared cache now holds NEW_DEFS. + assertThat(preparedStatement.getResultMetadataId()).isEqualTo(NEW_RESULT_METADATA_ID); + + // A's response arrives and is decoded now: SKIP_METADATA, row bytes laid out per OLD_DEFS + // (group_id text, is_deleted boolean, last_modified bigint) -- only 3 values, since A's + // request was encoded before the ALTER TABLE added "extra_0". + Rows rowsA = + new DefaultRows( + new RowsMetadata(oldDefs.size(), null, new int[0], null), + oneRow( + TypeCodecs.TEXT.encode("g2", DefaultProtocolVersion.V4), + TypeCodecs.BOOLEAN.encode(false, DefaultProtocolVersion.V4), + TypeCodecs.BIGINT.encode(1_700_000_000_001L, DefaultProtocolVersion.V4))); + + ColumnDefinitions defsForA = + Conversions.getResultDefinitions(rowsA, stmtA, context, encodeTimeSnapshot); + + // Fixed behaviour: A decodes against its own encode-time snapshot (OLD_DEFS, 3 columns), not + // the shared cache's current state (NEW_DEFS, 4 columns), even though B already swapped it. + assertThat(defsForA).isEqualTo(oldDefs); + assertThat(defsForA.get(1).getName().asInternal()).isEqualTo("is_deleted"); + + Row row = new DefaultRow(defsForA, rowsA.getData().poll(), context); + + // No column-offset shift: OLD_DEFS says index 1 ("is_deleted") is a 1-byte boolean, and A's + // bytes at that slot really are the 1-byte boolean OLD_DEFS put there. + assertThatCode(() -> row.getBoolean("is_deleted")).doesNotThrowAnyException(); + assertThat(row.getBoolean("is_deleted")).isFalse(); + assertThat(row.getLong("last_modified")).isEqualTo(1_700_000_000_001L); + } + + private static ColumnSpec columnSpec(String name, int index, int protocolDataType) { + return new ColumnSpec("race_test", "t2", name, index, RawType.PRIMITIVES.get(protocolDataType)); + } + + private static Queue> oneRow(ByteBuffer... values) { + Queue> data = new ArrayDeque<>(); + data.add(ImmutableList.copyOf(values)); + return data; + } +}