mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.0' into cassandra-3.11
This commit is contained in:
Alex Petrov
commit
5d0aeb1996
|
|
@ -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));
|
||||
}
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String, String> 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<ParsedStatement.Prepared> getPreparedStatements()
|
||||
{
|
||||
return new ArrayList<>(preparedStatements.values());
|
||||
}
|
||||
|
||||
private static class MigrationSubscriber extends MigrationListener
|
||||
{
|
||||
private static void removeInvalidPreparedStatements(String ksName, String cfName)
|
||||
|
|
|
|||
|
|
@ -380,6 +380,16 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
|
|||
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));
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<ColumnSpecification> boundNames, short[] partitionKeyBindIndexes, String keyspace)
|
||||
protected Prepared(CQLStatement statement, List<ColumnSpecification> 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.<ColumnSpecification>emptyList(), null, null);
|
||||
this(statement, Collections.<ColumnSpecification>emptyList(), null, null,false);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1581,14 +1581,38 @@ public final class SystemKeyspace
|
|||
preparedStatements.truncateBlockingWithoutSnapshot();
|
||||
}
|
||||
|
||||
public static List<Pair<String, String>> loadPreparedStatements()
|
||||
public static int loadPreparedStatements(TriFunction<MD5Digest, String, String, Boolean> 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<Pair<String, String>> 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<MD5Digest, String, String, Boolean> 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<A, B, C, D> {
|
||||
D accept(A var1, B var2, C var3);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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 <keyspace>` 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(<keyspace>)`, `Session#execute(\"USE <keyspace>\")` and `cluster.newSession(<keyspace>)` from your code, and " +
|
||||
"always use fully qualified table names (e.g. <keyspace>.<table>). " +
|
||||
"Keyspace used: %s, statement keyspace: %s, statement id: %s", getRawKeyspace(), preparedKeyspace, statementId));
|
||||
issuedPreparedStatementsUseWarning = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<ExecuteMessage> codec = new Message.Codec<ExecuteMessage>()
|
||||
{
|
||||
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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<IInvokableInstance> 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<String> 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<String> 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<String> stmts = QueryProcessor.getPreparedStatements().stream().map(p -> p.rawCQLStatement).collect(Collectors.toList());
|
||||
assertEquals(Lists.newArrayList(batch2, batch2), stmts);
|
||||
QueryProcessor.clearPreparedStatementsCache();
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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<IInvokableInstance> 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<Thread> threads = new ArrayList<>();
|
||||
AtomicBoolean interrupt = new AtomicBoolean(false);
|
||||
AtomicReference<Throwable> 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<Pair<Integer, Integer>, PreparedStatement> qualifiedStatements = new HashMap<>();
|
||||
Map<Pair<Integer, Integer>, 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<Integer, Integer> 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<Object[]> 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<Object[]> 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<Pair<Integer, Integer>, PreparedStatement> toCleanup = rng.nextBoolean() ? qualifiedStatements : unqualifiedStatements;
|
||||
Set<Pair<Integer, Integer>> toDrop = new HashSet<>();
|
||||
for (Pair<Integer, Integer> e : toCleanup.keySet())
|
||||
{
|
||||
if (rng.nextBoolean())
|
||||
toDrop.add(e);
|
||||
}
|
||||
|
||||
for (Pair<Integer, Integer> 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<QueryProcessor> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<double> )");
|
||||
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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 + ";");
|
||||
|
|
|
|||
Loading…
Reference in New Issue