diff --git a/src/java/com/datastax/driver/core/PreparedStatementHelper.java b/src/java/com/datastax/driver/core/PreparedStatementHelper.java new file mode 100644 index 0000000000..020bf7e98c --- /dev/null +++ b/src/java/com/datastax/driver/core/PreparedStatementHelper.java @@ -0,0 +1,119 @@ +/* + * 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.driver.core; + +import java.io.UnsupportedEncodingException; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; + +// Unfortunately, MD5Digest and fields in PreparedStatement are package-private, so the easiest way to test these +// things while still using the driver was to create a class in DS package. +public class PreparedStatementHelper +{ + private static final MessageDigest cachedDigest; + + static { + try { + cachedDigest = MessageDigest.getInstance("MD5"); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException(e); + } + } + + private static MD5Digest id(PreparedStatement statement) + { + return statement.getPreparedId().id; + } + + public static void assertStable(PreparedStatement first, PreparedStatement subsequent) + { + if (!id(first).equals(id(subsequent))) + { + throw new AssertionError(String.format("Subsequent id (%s) is different from the first one (%s)", + id(first), + id(subsequent))); + } + } + + public static void assertHashWithoutKeyspace(PreparedStatement statement, String queryString, String ks) + { + MD5Digest returned = id(statement); + if (!returned.equals(hashWithoutKeyspace(queryString, ks))) + { + if (returned.equals(hashWithKeyspace(queryString, ks))) + throw new AssertionError(String.format("Got hash with keyspace from the cluster: %s, should have gotten %s", + returned, hashWithoutKeyspace(queryString, ks))); + else + throw new AssertionError(String.format("Got unrecognized hash: %s", + returned)); + } + } + + public static void assertHashWithKeyspace(PreparedStatement statement, String queryString, String ks) + { + MD5Digest returned = id(statement); + if (!returned.equals(hashWithKeyspace(queryString, ks))) + { + if (returned.equals(hashWithoutKeyspace(queryString, ks))) + throw new AssertionError(String.format("Got hash without keyspace from the cluster: %s, should have gotten %s", + returned, hashWithKeyspace(queryString, ks))); + else + throw new AssertionError(String.format("Got unrecognized hash: %s", + returned)); + } + + } + + public static boolean equalsToHashWithKeyspace(byte[] digest, String queryString, String ks) + { + return MD5Digest.wrap(digest).equals(hashWithKeyspace(queryString, ks)); + } + + public static MD5Digest hashWithKeyspace(String queryString, String ks) + { + return computeId(queryString, ks); + } + + public static boolean equalsToHashWithoutKeyspace(byte[] digest, String queryString, String ks) + { + return MD5Digest.wrap(digest).equals(hashWithoutKeyspace(queryString, ks)); + } + + public static MD5Digest hashWithoutKeyspace(String queryString, String ks) + { + return computeId(queryString, null); + } + + private static MD5Digest computeId(String queryString, String keyspace) { + return compute(keyspace == null ? queryString : keyspace + queryString); + } + + public static MD5Digest compute(String toHash) { + try { + return compute(toHash.getBytes("UTF-8")); + } catch (UnsupportedEncodingException e) { + throw new RuntimeException(e.getMessage()); + } + } + + public static synchronized MD5Digest compute(byte[] toHash) { + cachedDigest.reset(); + return MD5Digest.wrap(cachedDigest.digest(toHash)); + } +} diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 4680c8345f..f6032dd2e3 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -420,6 +420,7 @@ public class Config public volatile boolean check_for_duplicate_rows_during_reads = true; public volatile boolean check_for_duplicate_rows_during_compaction = true; + public volatile boolean force_new_prepared_statement_behaviour = false; /** * Client mode means that the process is a pure client, that uses C* code base but does * not read or write local C* database files. diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 7a4d116418..a5a02ced98 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2682,4 +2682,17 @@ public class DatabaseDescriptor conf.check_for_duplicate_rows_during_compaction = enabled; } + public static boolean getForceNewPreparedStatementBehaviour() + { + return conf.force_new_prepared_statement_behaviour; + } + + public static void setForceNewPreparedStatementBehaviour(boolean value) + { + if (value != conf.force_new_prepared_statement_behaviour) + { + logger.info("Setting force_new_prepared_statement_behaviour to {}", value); + conf.force_new_prepared_statement_behaviour = value; + } + } } diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index 589a25a8c9..7de33a78b9 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -63,16 +63,9 @@ public class QueryProcessor implements QueryHandler { public static final CassandraVersion CQL_VERSION = new CassandraVersion("3.4.4"); - /** - * If a query is prepared with a fully qualified name, but the user also uses USE (specifically when USE keyspace - * is different) then the IDs generated could change over time; invalidating the assumption that IDs won't ever - * change. In the version defined below, the USE keyspace is ignored when a fully-qualified name is used as an - * attempt to make IDs stable. - */ - private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_30 = new CassandraVersion("3.0.26"); - private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_3X = new CassandraVersion("3.11.12"); - private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_40 = new CassandraVersion("4.0.1"); - + // See comments on QueryProcessor #prepare + public static final CassandraVersion NEW_PREPARED_STATEMENT_BEHAVIOUR_SINCE_30 = new CassandraVersion("3.0.26"); + public static final CassandraVersion NEW_PREPARED_STATEMENT_BEHAVIOUR_SINCE_3X = new CassandraVersion("3.11.12"); public static final QueryProcessor instance = new QueryProcessor(); @@ -155,26 +148,33 @@ public class QueryProcessor implements QueryHandler } } - public static void preloadPreparedStatement() + public void preloadPreparedStatements() { - ClientState clientState = ClientState.forInternalCalls(); - int count = 0; - for (Pair useKeyspaceAndCQL : SystemKeyspace.loadPreparedStatements()) - { + int count = SystemKeyspace.loadPreparedStatements((id, query, keyspace) -> { try { - clientState.setKeyspace(useKeyspaceAndCQL.left); - prepare(useKeyspaceAndCQL.right, clientState, false); - count++; + ClientState clientState = ClientState.forInternalCalls(); + if (keyspace != null) + clientState.setKeyspace(keyspace); + ParsedStatement.Prepared prepared = getStatement(query, clientState); + preparedStatements.putIfAbsent(id, prepared); + // Preload `null` statement for non-fully qualified statements, since it can't be parsed if loaded from cache and will be dropped + if (!prepared.fullyQualified) + preparedStatements.putIfAbsent(computeId(query, null), getStatement(query, clientState)); + return true; } - catch (RequestValidationException e) + catch (Throwable e) { - logger.warn("prepared statement recreation error: {}", useKeyspaceAndCQL.right, e); + JVMStabilityInspector.inspectThrowable(e); + logger.warn(String.format("Prepared statement recreation error, removing statement: %s %s %s", id, query, keyspace)); + SystemKeyspace.removePreparedStatement(id); + return false; } - } + }); logger.info("Preloaded {} prepared statements", count); } + /** * Clears the prepared statement cache. * @param memoryOnly {@code true} if only the in memory caches must be cleared, {@code false} otherwise. @@ -199,6 +199,12 @@ public class QueryProcessor implements QueryHandler MigrationManager.instance.register(new MigrationSubscriber()); } + @VisibleForTesting + public void evictPrepared(MD5Digest id) + { + preparedStatements.remove(id); + } + public ParsedStatement.Prepared getPrepared(MD5Digest id) { return preparedStatements.get(id); @@ -424,41 +430,113 @@ public class QueryProcessor implements QueryHandler return prepare(queryString, cState, cState instanceof ThriftClientState); } - public static ResultMessage.Prepared prepare(String queryString, ClientState clientState, boolean forThrift) + private volatile boolean newPreparedStatementBehaviour = false; + public boolean useNewPreparedStatementBehaviour() { - ResultMessage.Prepared existing = getStoredPreparedStatement(queryString, clientState.getRawKeyspace(), forThrift); - if (existing != null) - return existing; + if (newPreparedStatementBehaviour || DatabaseDescriptor.getForceNewPreparedStatementBehaviour()) + return true; + + synchronized (this) + { + CassandraVersion minVersion = Gossiper.instance.getMinVersion(DatabaseDescriptor.getWriteRpcTimeout(), TimeUnit.MILLISECONDS); + if (minVersion != null && + ((minVersion.is30() && minVersion.compareTo(NEW_PREPARED_STATEMENT_BEHAVIOUR_SINCE_30) >= 0) || + (minVersion.compareTo(NEW_PREPARED_STATEMENT_BEHAVIOUR_SINCE_3X) >= 0))) + { + logger.info("Fully upgraded to at least {}", minVersion); + newPreparedStatementBehaviour = true; + } + + return newPreparedStatementBehaviour; + } + } + + /** + * This method got slightly out of hand, but this is with best intentions: to allow users to be upgraded from any + * prior version, and help implementers avoid previous mistakes by clearly separating fully qualified and non-fully + * qualified statement behaviour. + * + * Basically we need to handle 4 different hashes here; + * 1. fully qualified query with keyspace + * 2. fully qualified query without keyspace + * 3. unqualified query with keyspace + * 4. unqualified query without keyspace + * + * The correct combination to return is 2/3 - the problem is during upgrades (assuming upgrading from < 3.0.26) + * - Existing clients have hash 1 or 3 + * - Query prepared on a 3.0.26/3.11.12 instance needs to return hash 1/3 to be able to execute it on a 3.0.25 instance + * - This is handled by the useNewPreparedStatementBehaviour flag - while there still are 3.0.25 instances in + * the cluster we always return hash 1/3 + * - Once fully upgraded we start returning hash 2/3, this will cause a prepared statement id mismatch for existing + * clients, but they will be able to continue using the old prepared statement id after that exception since we + * store the query both with and without keyspace. + */ + public ResultMessage.Prepared prepare(String queryString, ClientState clientState, boolean forThrift) + { + boolean useNewPreparedStatementBehaviour = useNewPreparedStatementBehaviour(); + + MD5Digest hashWithoutKeyspace = computeId(queryString, null); + MD5Digest hashWithKeyspace = computeId(queryString, clientState.getRawKeyspace()); + ParsedStatement.Prepared cachedWithoutKeyspace = preparedStatements.get(hashWithoutKeyspace); + ParsedStatement.Prepared cachedWithKeyspace = preparedStatements.get(hashWithKeyspace); + // We assume it is only safe to return cached prepare if we have both instances + boolean safeToReturnCached = cachedWithoutKeyspace != null && cachedWithKeyspace != null; + + if (!forThrift) + { + if (safeToReturnCached) + { + if (useNewPreparedStatementBehaviour) + { + if (cachedWithoutKeyspace.fullyQualified) // For fully qualified statements, we always skip keyspace to avoid digest switching + return new ResultMessage.Prepared(hashWithoutKeyspace, cachedWithoutKeyspace); + + if (clientState.getRawKeyspace() != null && !cachedWithKeyspace.fullyQualified) // For non-fully qualified statements, we always include keyspace to avoid ambiguity + return new ResultMessage.Prepared(hashWithKeyspace, cachedWithKeyspace); + } + else // legacy caches, pre-CASSANDRA-15252 behaviour + { + return new ResultMessage.Prepared(hashWithKeyspace, cachedWithKeyspace); + } + } + else + { + // Make sure the missing one is going to be eventually re-prepared + evictPrepared(hashWithKeyspace); + evictPrepared(hashWithoutKeyspace); + } + } ParsedStatement.Prepared prepared = getStatement(queryString, clientState); prepared.rawCQLStatement = queryString; + int boundTerms = prepared.statement.getBoundTerms(); if (boundTerms > FBUtilities.MAX_UNSIGNED_SHORT) throw new InvalidRequestException(String.format("Too many markers(?). %d markers exceed the allowed maximum of %d", boundTerms, FBUtilities.MAX_UNSIGNED_SHORT)); assert boundTerms == prepared.boundNames.size(); - if (prepared.keyspace != null) + if (prepared.fullyQualified) { - // Edge-case of CASSANDRA-15252 in mixed-mode cluster. We accept that 15252 itself can manifest in a - // cluster that has both old and new nodes, but we would like to avoid a situation when the fix adds - // a new behaviour that can break which, in addition, can get triggered more frequently. - // If statement ID was generated on the old node _with_ use, when attempting to execute on the new node, - // we may fall into infinite loop. To break out of this loop, we put a prepared statement that client - // expects into cache, so that it could get PREPARED response on the second try. - ResultMessage.Prepared newBehavior = storePreparedStatement(queryString, null, prepared, forThrift); - ResultMessage.Prepared oldBehavior = clientState.getRawKeyspace() != null ? storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift) : newBehavior; - CassandraVersion minVersion = Gossiper.instance.getMinVersion(20, TimeUnit.MILLISECONDS); + ResultMessage.Prepared qualifiedWithoutKeyspace = storePreparedStatement(queryString, null, prepared, forThrift); + ResultMessage.Prepared qualifiedWithKeyspace = null; + if (clientState.getRawKeyspace() != null) + qualifiedWithKeyspace = storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift); - // Default to old behaviour in case we're not sure about the version. Even if we ever flip back to the old - // behaviour due to the gossip bug or incorrect version string, we'll end up with two re-prepare round-trips. - return minVersion != null && - ((minVersion.major == 3 && minVersion.minor == 0 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_30) >= 0) || - (minVersion.major == 3 && minVersion.minor != 0 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_3X) >= 0) || - (minVersion.major == 4 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_40) >= 0)) ? newBehavior : oldBehavior; + if (!newPreparedStatementBehaviour && qualifiedWithKeyspace != null) + return qualifiedWithKeyspace; + + return qualifiedWithoutKeyspace; } else { - return storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift); + clientState.warnAboutUseWithPreparedStatements(hashWithKeyspace, clientState.getRawKeyspace()); + + ResultMessage.Prepared nonQualifiedWithKeyspace = storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift); + ResultMessage.Prepared nonQualifiedWithNullKeyspace = storePreparedStatement(queryString, null, prepared, forThrift); + if (!useNewPreparedStatementBehaviour) + return nonQualifiedWithNullKeyspace; + + return nonQualifiedWithKeyspace; } } @@ -663,6 +741,11 @@ public class QueryProcessor implements QueryHandler thriftPreparedStatements.clear(); } + public static List getPreparedStatements() + { + return new ArrayList<>(preparedStatements.values()); + } + private static class MigrationSubscriber extends MigrationListener { private static void removeInvalidPreparedStatements(String ksName, String cfName) diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java index c551d42463..9176b1fc45 100644 --- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java +++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java @@ -380,6 +380,16 @@ public abstract class UntypedResultSet implements Iterable return data.get(column); } + public byte[] getByteArray(String column) + { + ByteBuffer buf = data.get(column); + byte[] arr = new byte[buf.remaining()]; + for (int i = 0; i < arr.length; i++) + arr[i] = buf.get(buf.position() + i); + + return arr; + } + public InetAddress getInetAddress(String column) { return InetAddressType.instance.compose(data.get(column)); diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index e4cca2a220..11a20f42ac 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -522,8 +522,12 @@ public class BatchStatement implements CQLStatement @Override public void prepareKeyspace(ClientState state) throws InvalidRequestException { + fullyQualified = true; for (ModificationStatement.Parsed statement : parsedStatements) + { statement.prepareKeyspace(state); + fullyQualified &= statement.fullyQualified; + } } public ParsedStatement.Prepared prepare(ClientState clientState) throws InvalidRequestException @@ -561,7 +565,7 @@ public class BatchStatement implements CQLStatement short[] partitionKeyBindIndexes = (haveMultipleCFs || batchStatement.statements.isEmpty())? null : boundNames.getPartitionKeyBindIndexes(batchStatement.statements.get(0).cfm); - return new ParsedStatement.Prepared(batchStatement, boundNames, partitionKeyBindIndexes, null); + return new ParsedStatement.Prepared(batchStatement, boundNames, partitionKeyBindIndexes, null, isFullyQualified()); } } diff --git a/src/java/org/apache/cassandra/cql3/statements/CFStatement.java b/src/java/org/apache/cassandra/cql3/statements/CFStatement.java index f3cdd7f64d..85b18ef426 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CFStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/CFStatement.java @@ -27,10 +27,11 @@ import org.apache.cassandra.exceptions.InvalidRequestException; public abstract class CFStatement extends ParsedStatement { protected final CFName cfName; - + protected Boolean fullyQualified; protected CFStatement(CFName cfName) { this.cfName = cfName; + this.fullyQualified = cfName != null ? cfName.hasKeyspace() : null; } public void prepareKeyspace(ClientState state) throws InvalidRequestException @@ -51,6 +52,13 @@ public abstract class CFStatement extends ParsedStatement cfName.setKeyspace(keyspace, true); } + public boolean isFullyQualified() + { + if (fullyQualified == null) + throw new IllegalStateException("Cannot determine whether or not the query was fully qualified"); + return fullyQualified; + } + public String keyspace() { assert cfName.hasKeyspace() : "The statement hasn't be prepared correctly"; diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 8376a0a89d..d328ec1d57 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -814,7 +814,7 @@ public abstract class ModificationStatement implements CQLStatement { VariableSpecifications boundNames = getBoundVariables(); ModificationStatement statement = prepare(boundNames, clientState); - return new ParsedStatement.Prepared(statement, boundNames, boundNames.getPartitionKeyBindIndexes(statement.cfm), statement.cfm.ksName); + return new ParsedStatement.Prepared(statement, boundNames, boundNames.getPartitionKeyBindIndexes(statement.cfm), statement.cfm.ksName, isFullyQualified()); } public ModificationStatement prepare(VariableSpecifications boundNames, ClientState clientState) diff --git a/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java b/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java index dcc476fbca..54a9dba6cb 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java @@ -64,31 +64,28 @@ public abstract class ParsedStatement @Nullable public final short[] partitionKeyBindIndexes; - + public final boolean fullyQualified; @Nullable public final String keyspace; - protected Prepared(CQLStatement statement, List boundNames, short[] partitionKeyBindIndexes, String keyspace) + protected Prepared(CQLStatement statement, List boundNames, short[] partitionKeyBindIndexes, String keyspace, boolean fullyQualified) { this.statement = statement; this.boundNames = boundNames; this.partitionKeyBindIndexes = partitionKeyBindIndexes; + this.rawCQLStatement = ""; + this.fullyQualified = fullyQualified; this.keyspace = keyspace; } - public Prepared(CQLStatement statement, VariableSpecifications names, short[] partitionKeyBindIndexes) + public Prepared(CQLStatement statement, VariableSpecifications names, short[] partitionKeyBindIndexes, String keyspace, boolean fullyQualified) { - this(statement, names, partitionKeyBindIndexes, null); - } - - public Prepared(CQLStatement statement, VariableSpecifications names, short[] partitionKeyBindIndexes, String keyspace) - { - this(statement, names.getSpecifications(), partitionKeyBindIndexes, keyspace); + this(statement, names.getSpecifications(), partitionKeyBindIndexes, keyspace, fullyQualified); } public Prepared(CQLStatement statement) { - this(statement, Collections.emptyList(), null, null); + this(statement, Collections.emptyList(), null, null,false); } } diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 632bf94f9c..aaa69191a5 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -1013,7 +1013,7 @@ public class SelectStatement implements CQLStatement protected ParsedStatement.Prepared prepare(SelectStatement stmt, VariableSpecifications boundNames, CFMetaData cfm) { - return new ParsedStatement.Prepared(stmt, boundNames, boundNames.getPartitionKeyBindIndexes(cfm), cfm.ksName); + return new ParsedStatement. Prepared(stmt, boundNames, boundNames.getPartitionKeyBindIndexes(cfm), cfm.ksName, isFullyQualified()); } /** diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index ec26a6953b..c037d38c8b 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -1581,14 +1581,38 @@ public final class SystemKeyspace preparedStatements.truncateBlockingWithoutSnapshot(); } - public static List> loadPreparedStatements() + public static int loadPreparedStatements(TriFunction onLoaded) { - String query = String.format("SELECT logged_keyspace, query_string FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PREPARED_STATEMENTS); + String query = String.format("SELECT prepared_id, logged_keyspace, query_string FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PREPARED_STATEMENTS); UntypedResultSet resultSet = executeOnceInternal(query); - List> r = new ArrayList<>(); + int counter = 0; for (UntypedResultSet.Row row : resultSet) - r.add(Pair.create(row.has("logged_keyspace") ? row.getString("logged_keyspace") : null, - row.getString("query_string"))); - return r; + { + if (onLoaded.accept(MD5Digest.wrap(row.getByteArray("prepared_id")), + row.getString("query_string"), + row.has("logged_keyspace") ? row.getString("logged_keyspace") : null)) + counter++; + } + return counter; } + + public static int loadPreparedStatement(MD5Digest digest, TriFunction onLoaded) + { + String query = String.format("SELECT prepared_id, logged_keyspace, query_string FROM %s.%s WHERE prepared_id = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, PREPARED_STATEMENTS); + UntypedResultSet resultSet = executeOnceInternal(query, digest.byteBuffer()); + int counter = 0; + for (UntypedResultSet.Row row : resultSet) + { + if (onLoaded.accept(MD5Digest.wrap(row.getByteArray("prepared_id")), + row.getString("query_string"), + row.has("logged_keyspace") ? row.getString("logged_keyspace") : null)) + counter++; + } + return counter; + } + + public static interface TriFunction { + D accept(A var1, B var2, C var3); + } + } diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 363db5fd3f..613b566f9a 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -1908,7 +1908,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } @Nullable - public CassandraVersion getMinVersion(int delay, TimeUnit timeUnit) + public CassandraVersion getMinVersion(long delay, TimeUnit timeUnit) { try { diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index d8bd165b77..a4edca7097 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -356,7 +356,7 @@ public class CassandraDaemon ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(SizeEstimatesRecorder.instance, 30, sizeRecorderInterval, TimeUnit.SECONDS); // Prepared statements - QueryProcessor.preloadPreparedStatement(); + QueryProcessor.instance.preloadPreparedStatements(); // Metrics String metricsReporterConfigFile = System.getProperty("cassandra.metricsReporterConfigFile"); diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java index 155fd6970e..52ab7938bf 100644 --- a/src/java/org/apache/cassandra/service/ClientState.java +++ b/src/java/org/apache/cassandra/service/ClientState.java @@ -53,6 +53,8 @@ import org.apache.cassandra.thrift.ThriftValidation; import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.CassandraVersion; +import org.apache.cassandra.utils.MD5Digest; /** * State related to a client connection. @@ -90,6 +92,7 @@ public class ClientState // Current user for the session private volatile AuthenticatedUser user; private volatile String keyspace; + private volatile boolean issuedPreparedStatementsUseWarning; /** * Force Compact Tables to be represented as CQL ones for the current client session (simulates @@ -457,4 +460,16 @@ public class ClientState { return user.getPermissions(resource); } + + public void warnAboutUseWithPreparedStatements(MD5Digest statementId, String preparedKeyspace) + { + if (!issuedPreparedStatementsUseWarning) + { + ClientWarn.instance.warn(String.format("`USE ` with prepared statements is considered to be an anti-pattern due to ambiguity in non-qualified table names. " + + "Please consider removing instances of `Session#setKeyspace()`, `Session#execute(\"USE \")` and `cluster.newSession()` from your code, and " + + "always use fully qualified table names (e.g. .). " + + "Keyspace used: %s, statement keyspace: %s, statement id: %s", getRawKeyspace(), preparedKeyspace, statementId)); + issuedPreparedStatementsUseWarning = true; + } + } } diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index d881e632f0..76b977cb6a 100644 --- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java @@ -17,7 +17,9 @@ */ package org.apache.cassandra.transport.messages; +import java.util.Objects; import java.util.UUID; +import java.util.concurrent.TimeUnit; import com.google.common.collect.ImmutableMap; import io.netty.buffer.ByteBuf; @@ -26,7 +28,9 @@ import org.apache.cassandra.cql3.CQLStatement; import org.apache.cassandra.cql3.ColumnSpecification; import org.apache.cassandra.cql3.QueryHandler; import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.statements.ModificationStatement; import org.apache.cassandra.cql3.statements.ParsedStatement; +import org.apache.cassandra.cql3.statements.SelectStatement; import org.apache.cassandra.exceptions.PreparedQueryNotFoundException; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; @@ -34,10 +38,13 @@ import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.transport.*; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.MD5Digest; +import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.UUIDGen; public class ExecuteMessage extends Message.Request { + private static final NoSpamLogger nospam = NoSpamLogger.getLogger(logger, 10, TimeUnit.MINUTES); + public static final Message.Codec codec = new Message.Codec() { public ExecuteMessage decode(ByteBuf body, ProtocolVersion version) @@ -93,6 +100,16 @@ public class ExecuteMessage extends Message.Request { QueryHandler handler = ClientState.getCQLQueryHandler(); ParsedStatement.Prepared prepared = handler.getPrepared(statementId); + if (prepared != null && !prepared.fullyQualified && !Objects.equals(state.getClientState().getRawKeyspace(), prepared.keyspace) + && (prepared.statement instanceof ModificationStatement || prepared.statement instanceof SelectStatement)) + { + state.getClientState().warnAboutUseWithPreparedStatements(statementId, prepared.keyspace); + String msg = String.format("Tried to execute a prepared unqalified statement on a keyspace it was not prepared on. " + + " Executing the resulting prepared statement will return unexpected results: %s (on keyspace %s, previously prepared on %s)", + statementId, state.getClientState().getRawKeyspace(), prepared.keyspace); + nospam.error(msg); + } + if (prepared == null) throw new PreparedQueryNotFoundException(statementId); diff --git a/test/distributed/org/apache/cassandra/distributed/test/PrepareBatchStatementsTest.java b/test/distributed/org/apache/cassandra/distributed/test/PrepareBatchStatementsTest.java new file mode 100644 index 0000000000..74aee98e61 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/PrepareBatchStatementsTest.java @@ -0,0 +1,96 @@ +/* + * 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 org.apache.cassandra.distributed.test; + +import java.util.List; +import java.util.stream.Collectors; + +import com.google.common.collect.Lists; +import org.junit.Test; + +import com.datastax.driver.core.Session; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.service.StorageService; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.junit.Assert.assertEquals; + +public class PrepareBatchStatementsTest extends TestBaseImpl +{ + @Test + public void testPreparedBatch() throws Exception + { + try (ICluster c = init(builder().withNodes(1) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .start())) + { + try (com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .build(); + Session s = cluster.connect()) + { + c.schemaChange(withKeyspace("CREATE KEYSPACE ks1 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};")); + c.schemaChange(withKeyspace("CREATE TABLE ks1.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + c.schemaChange(withKeyspace("CREATE KEYSPACE ks2 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};")); + c.schemaChange(withKeyspace("CREATE TABLE ks2.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + + String batch1 = "BEGIN BATCH\n" + + "UPDATE ks1.tbl SET v = ? where pk = ? and ck = ?;\n" + + "UPDATE ks2.tbl SET v = ? where pk = ? and ck = ?;\n" + + "APPLY BATCH;"; + String batch2 = "BEGIN BATCH\n" + + "INSERT INTO ks1.tbl (pk, ck, v) VALUES (?, ?, ?);\n" + + "INSERT INTO tbl (pk, ck, v) VALUES (?, ?, ?);\n" + + "APPLY BATCH;"; + + + s.prepare(batch1); + c.get(1).runOnInstance(() -> { + // no USE here, only a fully qualified batch - should get stored ONCE + List stmts = QueryProcessor.getPreparedStatements().stream().map(p -> p.rawCQLStatement).collect(Collectors.toList()); + assertEquals(Lists.newArrayList(batch1), stmts); + QueryProcessor.clearPreparedStatementsCache(); + }); + + s.execute("use ks2"); + s.prepare(batch1); + c.get(1).runOnInstance(() -> { + // after USE, fully qualified - should get stored twice! Once with null keyspace (new behaviour) once with ks2 keyspace (old behaviour) + List stmts = QueryProcessor.getPreparedStatements().stream().map(p -> p.rawCQLStatement).collect(Collectors.toList()); + assertEquals(Lists.newArrayList(batch1, batch1), stmts); + QueryProcessor.clearPreparedStatementsCache(); + }); + + s.prepare(batch2); + c.get(1).runOnInstance(() -> { + // after USE, should get stored twice, once with keyspace, once without + List stmts = QueryProcessor.getPreparedStatements().stream().map(p -> p.rawCQLStatement).collect(Collectors.toList()); + assertEquals(Lists.newArrayList(batch2, batch2), stmts); + QueryProcessor.clearPreparedStatementsCache(); + }); + } + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java new file mode 100644 index 0000000000..520e8cf7d8 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java @@ -0,0 +1,361 @@ +/* + * 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 org.apache.cassandra.distributed.test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.PreparedStatementHelper; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.exceptions.InvalidQueryException; +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.DynamicType; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.impl.RowUtil; +import org.apache.cassandra.utils.CassandraVersion; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.Throwables; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +public class ReprepareFuzzTest extends TestBaseImpl +{ + private static final Logger logger = LoggerFactory.getLogger(ReprepareFuzzTest.class); + + @Test + public void fuzzTest() throws Throwable + { + try (ICluster c = builder().withNodes(1) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .withInstanceInitializer(PrepareBehaviour::alwaysNewBehaviour) + .start()) + { + // Long string to make us invalidate caches occasionally + String veryLongString = "very"; + for (int i = 0; i < 2; i++) + veryLongString += veryLongString; + final String qualified = "SELECT pk as " + veryLongString + "%d, ck as " + veryLongString + "%d FROM ks%d.tbl"; + final String unqualified = "SELECT pk as " + veryLongString + "%d, ck as " + veryLongString + "%d FROM tbl"; + + int KEYSPACES = 3; + final int STATEMENTS_PER_KS = 3; + + for (int i = 0; i < KEYSPACES; i++) + { + c.schemaChange(withKeyspace("CREATE KEYSPACE ks" + i + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};")); + c.schemaChange(withKeyspace("CREATE TABLE ks" + i + ".tbl (pk int, ck int, PRIMARY KEY (pk, ck));")); + for (int j = 0; j < i; j++) + c.coordinator(1).execute("INSERT INTO ks" + i + ".tbl (pk, ck) VALUES (?, ?)", ConsistencyLevel.QUORUM, 1, j); + } + + List threads = new ArrayList<>(); + AtomicBoolean interrupt = new AtomicBoolean(false); + AtomicReference thrown = new AtomicReference<>(); + + int INFREQUENT_ACTION_COEF = 10; + + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); + for (int i = 0; i < FBUtilities.getAvailableProcessors() * 2; i++) + { + int seed = i; + threads.add(new Thread(() -> { + com.datastax.driver.core.Cluster cluster = null; + Session session = null; + + try + { + Random rng = new Random(seed); + int usedKsIdx = -1; + String usedKs = null; + Map, PreparedStatement> qualifiedStatements = new HashMap<>(); + Map, PreparedStatement> unqualifiedStatements = new HashMap<>(); + + cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .build(); + session = cluster.connect(); + while (!interrupt.get() && (System.nanoTime() < deadline)) + { + final int ks = rng.nextInt(KEYSPACES); + final int statementIdx = rng.nextInt(STATEMENTS_PER_KS); + final Pair statementId = Pair.create(ks, statementIdx); + + int v = rng.nextInt(INFREQUENT_ACTION_COEF + 1); + Action[] pool; + if (v == INFREQUENT_ACTION_COEF) + pool = infrequent; + else + pool = frequent; + + Action action = pool[rng.nextInt(pool.length)]; + switch (action) + { + case EXECUTE_QUALIFIED: + if (!qualifiedStatements.containsKey(statementId)) + continue; + + try + { + int counter = 0; + for (Iterator iter = RowUtil.toObjects(session.execute(qualifiedStatements.get(statementId).bind())); iter.hasNext(); ) + { + Object[] current = iter.next(); + int v0 = (int) current[0]; + int v1 = (int) current[1]; + Assert.assertEquals(v0, 1); + Assert.assertEquals(v1, counter++); + } + Assert.assertEquals(ks, counter); + } + catch (Throwable t) + { + if (t.getCause() != null && + t.getCause().getMessage().contains("Statement was prepared on keyspace")) + continue; + + throw t; + } + + break; + case EXECUTE_UNQUALIFIED: + if (!unqualifiedStatements.containsKey(statementId)) + continue; + + try + { + int counter = 0; + for (Iterator iter = RowUtil.toObjects(session.execute(unqualifiedStatements.get(statementId).bind())); iter.hasNext(); ) + { + Object[] current = iter.next(); + int v0 = (int) current[0]; + int v1 = (int) current[1]; + Assert.assertEquals(v0, 1); + Assert.assertEquals(v1, counter++); + } + Assert.assertEquals(unqualifiedStatements.get(statementId).getQueryKeyspace() + " " + usedKs + " " + statementId, + Integer.parseInt(unqualifiedStatements.get(statementId).getQueryKeyspace().replace("ks", "")), + counter); + + } + catch (Throwable t) + { + if (t.getCause() != null && + t.getCause().getMessage().contains("Statement was prepared on keyspace")) + continue; + + throw t; + } + + break; + case PREPARE_QUALIFIED: + { + String qs = String.format(qualified, statementIdx, statementIdx, ks); + String keyspace = "ks" + ks; + PreparedStatement preparedQualified = session.prepare(qs); + + // With prepared qualified, keyspace will be set to the keyspace of the statement when it was first executed + PreparedStatementHelper.assertHashWithoutKeyspace(preparedQualified, qs, keyspace); + qualifiedStatements.put(statementId, preparedQualified); + } + break; + case PREPARE_UNQUALIFIED: + try + { + String qs = String.format(unqualified, statementIdx, statementIdx, ks); + PreparedStatement preparedUnqalified = session.prepare(qs); + Assert.assertEquals(preparedUnqalified.getQueryKeyspace(), usedKs); + PreparedStatementHelper.assertHashWithKeyspace(preparedUnqalified, qs, usedKs); + unqualifiedStatements.put(Pair.create(usedKsIdx, statementIdx), preparedUnqalified); + } + catch (InvalidQueryException iqe) + { + if (!iqe.getMessage().contains("No keyspace has been")) + throw iqe; + } + catch (Throwable t) + { + if (usedKs == null) + { + // ignored + continue; + } + + throw t; + } + break; + case CLEAR_CACHES: + c.get(1).runOnInstance(() -> { + SystemKeyspace.loadPreparedStatements((id, query, keyspace) -> { + if (rng.nextBoolean()) + QueryProcessor.instance.evictPrepared(id); + return true; + }); + }); + break; + case RELOAD_FROM_TABLES: + c.get(1).runOnInstance(QueryProcessor::clearPreparedStatementsCache); + c.get(1).runOnInstance(() -> QueryProcessor.instance.preloadPreparedStatements()); + break; + case SWITCH_KEYSPACE: + usedKsIdx = ks; + usedKs = "ks" + ks; + session.execute("USE " + usedKs); + break; + case FORGET_PREPARED: + Map, PreparedStatement> toCleanup = rng.nextBoolean() ? qualifiedStatements : unqualifiedStatements; + Set> toDrop = new HashSet<>(); + for (Pair e : toCleanup.keySet()) + { + if (rng.nextBoolean()) + toDrop.add(e); + } + + for (Pair e : toDrop) + toCleanup.remove(e); + toDrop.clear(); + break; + case RECONNECT: + session.close(); + cluster.close(); + cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .build(); + session = cluster.connect(); + qualifiedStatements.clear(); + unqualifiedStatements.clear(); + usedKs = null; + usedKsIdx = -1; + break; + } + } + } + catch (Throwable t) + { + interrupt.set(true); + t.printStackTrace(); + while (true) + { + Throwable seen = thrown.get(); + Throwable merged = Throwables.merge(seen, t); + if (thrown.compareAndSet(seen, merged)) + break; + } + throw t; + } + finally + { + if (session != null) + session.close(); + if (cluster != null) + cluster.close(); + } + })); + } + + for (Thread thread : threads) + thread.start(); + + for (Thread thread : threads) + thread.join(); + + if (thrown.get() != null) + throw thrown.get(); + } + } + + private enum Action + { + EXECUTE_QUALIFIED, + EXECUTE_UNQUALIFIED, + PREPARE_QUALIFIED, + PREPARE_UNQUALIFIED, + CLEAR_CACHES, + FORGET_PREPARED, + RELOAD_FROM_TABLES, + SWITCH_KEYSPACE, + RECONNECT + } + + private static Action[] frequent = new Action[]{ Action.EXECUTE_QUALIFIED, + Action.EXECUTE_UNQUALIFIED, + Action.PREPARE_QUALIFIED, + Action.PREPARE_UNQUALIFIED, + Action.SWITCH_KEYSPACE}; + + private static Action[] infrequent = new Action[]{ Action.CLEAR_CACHES, + Action.FORGET_PREPARED, + Action.RELOAD_FROM_TABLES, + Action.RECONNECT + }; + + public static class PrepareBehaviour + { + static void alwaysNewBehaviour(ClassLoader cl, int nodeNumber) + { + DynamicType.Builder.MethodDefinition.ReceiverTypeDefinition klass = + new ByteBuddy().rebase(QueryProcessor.class) + .method(named("skipKeyspaceForQualifiedStatements")) + .intercept(MethodDelegation.to(AlwaysNewBehaviour.class)) + .method(named("useKeyspaceForNonQualifiedStatements")) + .intercept(MethodDelegation.to(AlwaysNewBehaviour.class)); + + klass.make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + } + + public static class AlwaysNewBehaviour + { + public static boolean skipKeyspaceForQualifiedStatements() + { + return true; + } + + public static boolean useKeyspaceForNonQualifiedStatements() + { + return true; + } + } +} diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index e09c4dfa7b..705fad8f1f 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -836,7 +836,7 @@ public abstract class CQLTester protected ResultMessage.Prepared prepare(String query) throws Throwable { - return QueryProcessor.prepare(formatQuery(query), ClientState.forInternalCalls(), false); + return QueryProcessor.instance.prepare(formatQuery(query), ClientState.forInternalCalls(), false); } protected UntypedResultSet execute(String query, Object... values) throws Throwable diff --git a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java index dd477ec17c..a35ed671db 100644 --- a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java +++ b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java @@ -92,7 +92,7 @@ public class PstmtPersistenceTest extends CQLTester Assert.assertNull(handler.getPrepared(stmtId)); // load prepared statements and validate that these still execute fine - QueryProcessor.preloadPreparedStatement(); + QueryProcessor.instance.preloadPreparedStatements(); validatePstmts(stmtIds, handler); @@ -189,6 +189,6 @@ public class PstmtPersistenceTest extends CQLTester private MD5Digest prepareStatement(String stmt, String keyspace, String table, ClientState clientState) { - return QueryProcessor.prepare(String.format(stmt, keyspace + "." + table), clientState, false).statementId; + return QueryProcessor.instance.prepare(String.format(stmt, keyspace + "." + table), clientState, false).statementId; } } diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java index adb69113b0..6a0c23cb12 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java @@ -1755,9 +1755,9 @@ public class SecondaryIndexTest extends CQLTester private ResultMessage.Prepared prepareStatement(String cql, boolean forThrift) { - return QueryProcessor.prepare(String.format(cql, KEYSPACE, currentTable()), - ClientState.forInternalCalls(), - forThrift); + return QueryProcessor.instance.prepare(String.format(cql, KEYSPACE, currentTable()), + ClientState.forInternalCalls(), + forThrift); } private void validateCell(Cell cell, ColumnDefinition def, ByteBuffer val, long timestamp) diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java index cac0fd32ee..8f8afc835a 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java @@ -171,16 +171,16 @@ public class UFTest extends CQLTester // drop it those statements should be removed from the cache in QueryProcessor. The other statements // should be unaffected. - ResultMessage.Prepared preparedSelect1 = QueryProcessor.prepare( + ResultMessage.Prepared preparedSelect1 = QueryProcessor.instance.prepare( String.format("SELECT key, %s(d) FROM %s.%s", fSin, KEYSPACE, currentTable()), ClientState.forInternalCalls(), false); - ResultMessage.Prepared preparedSelect2 = QueryProcessor.prepare( + ResultMessage.Prepared preparedSelect2 = QueryProcessor.instance.prepare( String.format("SELECT key FROM %s.%s", KEYSPACE, currentTable()), ClientState.forInternalCalls(), false); - ResultMessage.Prepared preparedInsert1 = QueryProcessor.prepare( + ResultMessage.Prepared preparedInsert1 = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, d) VALUES (?, %s(?))", KEYSPACE, currentTable(), fSin), ClientState.forInternalCalls(), false); - ResultMessage.Prepared preparedInsert2 = QueryProcessor.prepare( + ResultMessage.Prepared preparedInsert2 = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, d) VALUES (?, ?)", KEYSPACE, currentTable()), ClientState.forInternalCalls(), false); @@ -205,10 +205,10 @@ public class UFTest extends CQLTester Assert.assertEquals(1, Schema.instance.getFunctions(fSinName).size()); - preparedSelect1= QueryProcessor.prepare( + preparedSelect1= QueryProcessor.instance.prepare( String.format("SELECT key, %s(d) FROM %s.%s", fSin, KEYSPACE, currentTable()), ClientState.forInternalCalls(), false); - preparedInsert1 = QueryProcessor.prepare( + preparedInsert1 = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, d) VALUES (?, %s(?))", KEYSPACE, currentTable(), fSin), ClientState.forInternalCalls(), false); Assert.assertNotNull(QueryProcessor.instance.getPrepared(preparedSelect1.statementId)); @@ -271,7 +271,7 @@ public class UFTest extends CQLTester " key int PRIMARY KEY," + " val " + collectionType + ')'); - ResultMessage.Prepared prepared = QueryProcessor.prepare( + ResultMessage.Prepared prepared = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, val) VALUES (?, %s)", KEYSPACE, currentTable(), @@ -287,7 +287,7 @@ public class UFTest extends CQLTester " key int PRIMARY KEY," + " val tuple )"); - ResultMessage.Prepared prepared = QueryProcessor.prepare( + ResultMessage.Prepared prepared = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, val) VALUES (?, (%s(0.0)))", KEYSPACE, currentTable(), @@ -303,7 +303,7 @@ public class UFTest extends CQLTester createTable("CREATE TABLE %s (" + " key int PRIMARY KEY," + " val double)"); - ResultMessage.Prepared control = QueryProcessor.prepare( + ResultMessage.Prepared control = QueryProcessor.instance.prepare( String.format("INSERT INTO %s.%s (key, val) VALUES (?, ?)", KEYSPACE, currentTable()), @@ -754,8 +754,8 @@ public class UFTest extends CQLTester Assert.assertEquals(1, Schema.instance.getFunctions(fNameName).size()); - ResultMessage.Prepared prepared = QueryProcessor.prepare(String.format("SELECT key, %s(udt) FROM %s.%s", fName, KEYSPACE, currentTable()), - ClientState.forInternalCalls(), false); + ResultMessage.Prepared prepared = QueryProcessor.instance.prepare(String.format("SELECT key, %s(udt) FROM %s.%s", fName, KEYSPACE, currentTable()), + ClientState.forInternalCalls(), false); Assert.assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId)); // UT still referenced by table diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java index b8ef4d3d57..c455ae9112 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java @@ -1141,7 +1141,7 @@ public class AggregationTest extends CQLTester "SFUNC " + shortFunctionName(fState) + " " + "STYPE int"); - ResultMessage.Prepared prepared = QueryProcessor.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false); + ResultMessage.Prepared prepared = QueryProcessor.instance.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false); assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId)); execute("DROP AGGREGATE " + a + "(int)"); @@ -1153,7 +1153,7 @@ public class AggregationTest extends CQLTester "SFUNC " + shortFunctionName(fState) + " " + "STYPE int"); - prepared = QueryProcessor.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false); + prepared = QueryProcessor.instance.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false); assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId)); execute("DROP KEYSPACE " + otherKS + ";");