mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-4.0' into cassandra-4.1
* cassandra-4.0: Ensure prepared_statement INSERT timestamp precedes eviction DELETE
This commit is contained in:
Andy Tolbert
commit
9c308ce2f3
|
|
@ -1,3 +1,8 @@
|
|||
4.1.10
|
||||
Merged from 4.0:
|
||||
* Ensure prepared_statement INSERT timestamp precedes eviction DELETE (CASSANDRA-19703)
|
||||
|
||||
|
||||
4.1.9
|
||||
* Grant permission on keyspaces system_views and system_virtual_schema not possible (CASSANDRA-20171)
|
||||
* Fix mixed mode paxos ttl commit hang (CASSANDRA-20514)
|
||||
|
|
|
|||
|
|
@ -239,6 +239,28 @@ provide values for `LIMIT`, `TIMESTAMP`, and `TTL` clauses. If anonymous
|
|||
bind markers are used, the names for the query parameters will be
|
||||
`[limit]`, `[timestamp]`, and `[ttl]`, respectively.
|
||||
|
||||
===== Prepared Statement Caching
|
||||
|
||||
Prepared Statements are cached by cassandra in-memory using a
|
||||
https://github.com/ben-manes/caffeine[Caffeine]-managed cache which
|
||||
can be configured using
|
||||
xref:managing/configuration/cass_yaml_file.adoc#_prepared_statements_cache_size[`prepared_statements_cache_size`].
|
||||
The cache is also persisted to the `system.prepared_statements` table
|
||||
so it can be preloaded into memory on startup.
|
||||
|
||||
To ensure optimal performance, it's important to use a bind `<variable>`
|
||||
for *all non-constant values* in your CQL statements. If you include
|
||||
literal values directly in the query instead, each variation will be
|
||||
treated as a unique statement that must be prepared and cached
|
||||
separately. This will soon overflow the prepared statement cache,
|
||||
which is small by design.
|
||||
|
||||
When the cache reaches its maximum size, older or less frequently
|
||||
used statements are
|
||||
https://github.com/ben-manes/caffeine/wiki/Eviction[evicted],
|
||||
leading to additional overhead as previously prepared statements must
|
||||
be re-prepared.
|
||||
|
||||
[[dataDefinition]]
|
||||
=== Data Definition
|
||||
|
||||
|
|
|
|||
|
|
@ -66,6 +66,12 @@ public interface QueryHandler
|
|||
|
||||
public final MD5Digest resultMetadataId;
|
||||
|
||||
/**
|
||||
* Timestamp of when this prepared statement was created. Used in QueryProcessor.preparedStatements cache
|
||||
* to ensure that the deletion timestamp always succeeds the insert timestamp.
|
||||
*/
|
||||
public final long timestamp;
|
||||
|
||||
/**
|
||||
* Contains the CQL statement source if the statement has been "regularly" perpared via
|
||||
* {@link QueryHandler#prepare(String, ClientState, Map)}.
|
||||
|
|
@ -82,6 +88,7 @@ public interface QueryHandler
|
|||
this.resultMetadataId = ResultSet.ResultMetadata.fromPrepared(statement).getResultMetadataId();
|
||||
this.fullyQualified = fullyQualified;
|
||||
this.keyspace = keyspace;
|
||||
this.timestamp = ClientState.getTimestamp();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import java.util.stream.Collectors;
|
|||
|
||||
import com.github.benmanes.caffeine.cache.Cache;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import com.github.benmanes.caffeine.cache.RemovalCause;
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.base.Predicate;
|
||||
import com.google.common.collect.*;
|
||||
|
|
@ -101,23 +102,22 @@ public class QueryProcessor implements QueryHandler
|
|||
// counters. Callers of processStatement are responsible for correctly notifying metrics
|
||||
public static final CQLMetrics metrics = new CQLMetrics();
|
||||
|
||||
// Paging size to use when preloading prepared statements.
|
||||
public static final int PRELOAD_PREPARED_STATEMENTS_FETCH_SIZE = 5000;
|
||||
|
||||
// Size of the prepared statement cache in bytes.
|
||||
public static long PREPARED_STATEMENT_CACHE_SIZE_BYTES = capacityToBytes(DatabaseDescriptor.getPreparedStatementsCacheSizeMiB());
|
||||
|
||||
private static final AtomicInteger lastMinuteEvictionsCount = new AtomicInteger(0);
|
||||
|
||||
static
|
||||
{
|
||||
preparedStatements = Caffeine.newBuilder()
|
||||
.executor(ImmediateExecutor.INSTANCE)
|
||||
.maximumWeight(capacityToBytes(DatabaseDescriptor.getPreparedStatementsCacheSizeMiB()))
|
||||
.maximumWeight(PREPARED_STATEMENT_CACHE_SIZE_BYTES)
|
||||
.weigher(QueryProcessor::getSizeOfPreparedStatementForCache)
|
||||
.removalListener((key, prepared, cause) -> {
|
||||
MD5Digest md5Digest = (MD5Digest) key;
|
||||
if (cause.wasEvicted())
|
||||
{
|
||||
metrics.preparedStatementsEvicted.inc();
|
||||
lastMinuteEvictionsCount.incrementAndGet();
|
||||
SystemKeyspace.removePreparedStatement(md5Digest);
|
||||
}
|
||||
}).build();
|
||||
.removalListener((key, prepared, cause) -> evictPreparedStatement(key, cause))
|
||||
.build();
|
||||
|
||||
ScheduledExecutors.scheduledTasks.scheduleAtFixedRate(() -> {
|
||||
long count = lastMinuteEvictionsCount.getAndSet(0);
|
||||
|
|
@ -131,6 +131,16 @@ public class QueryProcessor implements QueryHandler
|
|||
DatabaseDescriptor.getPreparedStatementsCacheSizeMiB());
|
||||
}
|
||||
|
||||
private static void evictPreparedStatement(MD5Digest key, RemovalCause cause)
|
||||
{
|
||||
if (cause.wasEvicted())
|
||||
{
|
||||
metrics.preparedStatementsEvicted.inc();
|
||||
lastMinuteEvictionsCount.incrementAndGet();
|
||||
SystemKeyspace.removePreparedStatement(key);
|
||||
}
|
||||
}
|
||||
|
||||
private static long capacityToBytes(long cacheSizeMB)
|
||||
{
|
||||
return cacheSizeMB * 1024 * 1024;
|
||||
|
|
@ -155,6 +165,12 @@ public class QueryProcessor implements QueryHandler
|
|||
}
|
||||
|
||||
public void preloadPreparedStatements()
|
||||
{
|
||||
preloadPreparedStatements(PRELOAD_PREPARED_STATEMENTS_FETCH_SIZE);
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
public int preloadPreparedStatements(int pageSize)
|
||||
{
|
||||
int count = SystemKeyspace.loadPreparedStatements((id, query, keyspace) -> {
|
||||
try
|
||||
|
|
@ -169,17 +185,18 @@ public class QueryProcessor implements QueryHandler
|
|||
// 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.get(computeId(query, null), (ignored_) -> prepared);
|
||||
return true;
|
||||
return prepared;
|
||||
}
|
||||
catch (RequestValidationException 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;
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}, pageSize);
|
||||
logger.info("Preloaded {} prepared statements", count);
|
||||
return count;
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -556,12 +573,33 @@ public class QueryProcessor implements QueryHandler
|
|||
public static UntypedResultSet executeInternalWithPaging(String query, int pageSize, Object... values)
|
||||
{
|
||||
Prepared prepared = prepareInternal(query);
|
||||
if (!(prepared.statement instanceof SelectStatement))
|
||||
|
||||
return executeInternalWithPaging(prepared.statement, pageSize, values);
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes with a non-prepared statement using paging. Generally {@link #executeInternalWithPaging(String, int, Object...)}
|
||||
* should be used instead of this, but this may be used in niche cases like
|
||||
* {@link SystemKeyspace#loadPreparedStatement(MD5Digest, SystemKeyspace.TriFunction)} where prepared statements are
|
||||
* being loaded into {@link #preparedStatements} so it doesn't make sense to prepare a statement in this context.
|
||||
*/
|
||||
public static UntypedResultSet executeOnceInternalWithPaging(String query, int pageSize, Object... values)
|
||||
{
|
||||
QueryState queryState = internalQueryState();
|
||||
CQLStatement statement = parseStatement(query, queryState.getClientState());
|
||||
statement.validate(queryState.getClientState());
|
||||
|
||||
return executeInternalWithPaging(statement, pageSize, values);
|
||||
}
|
||||
|
||||
private static UntypedResultSet executeInternalWithPaging(CQLStatement statement, int pageSize, Object... values)
|
||||
{
|
||||
if (!(statement instanceof SelectStatement))
|
||||
throw new IllegalArgumentException("Only SELECTs can be paged");
|
||||
|
||||
SelectStatement select = (SelectStatement)prepared.statement;
|
||||
SelectStatement select = (SelectStatement) statement;
|
||||
int nowInSec = FBUtilities.nowInSeconds();
|
||||
QueryPager pager = select.getQuery(makeInternalOptionsWithNowInSec(prepared.statement, nowInSec, values), nowInSec).getPager(null, ProtocolVersion.CURRENT);
|
||||
QueryPager pager = select.getQuery(makeInternalOptionsWithNowInSec(select, nowInSec, values), nowInSec).getPager(null, ProtocolVersion.CURRENT);
|
||||
return UntypedResultSet.create(select, pager, pageSize);
|
||||
}
|
||||
|
||||
|
|
@ -800,7 +838,7 @@ public class QueryProcessor implements QueryHandler
|
|||
|
||||
Prepared previous = preparedStatements.get(statementId, (ignored_) -> prepared);
|
||||
if (previous == prepared)
|
||||
SystemKeyspace.writePreparedStatement(keyspace, statementId, queryString);
|
||||
SystemKeyspace.writePreparedStatement(keyspace, statementId, queryString, prepared.timestamp);
|
||||
|
||||
ResultSet.PreparedMetadata preparedMetadata = ResultSet.PreparedMetadata.fromPrepared(prepared.statement);
|
||||
ResultSet.ResultMetadata resultMetadata = ResultSet.ResultMetadata.fromPrepared(prepared.statement);
|
||||
|
|
|
|||
|
|
@ -56,6 +56,7 @@ import org.slf4j.Logger;
|
|||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.cql3.QueryHandler.Prepared;
|
||||
import org.apache.cassandra.cql3.QueryProcessor;
|
||||
import org.apache.cassandra.cql3.UntypedResultSet;
|
||||
import org.apache.cassandra.cql3.functions.AggregateFcts;
|
||||
|
|
@ -87,6 +88,7 @@ import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
|
|||
import org.apache.cassandra.io.util.DataInputBuffer;
|
||||
import org.apache.cassandra.io.util.DataOutputBuffer;
|
||||
import org.apache.cassandra.io.util.File;
|
||||
import org.apache.cassandra.io.util.FileUtils;
|
||||
import org.apache.cassandra.io.util.RebufferingInputStream;
|
||||
import org.apache.cassandra.locator.IEndpointSnitch;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
|
|
@ -130,9 +132,11 @@ import static java.util.Collections.singletonMap;
|
|||
import static java.util.concurrent.TimeUnit.MICROSECONDS;
|
||||
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
|
||||
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.PREPARED_STATEMENT_CACHE_SIZE_BYTES;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeInternalWithNowInSec;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternalWithPaging;
|
||||
import static org.apache.cassandra.service.paxos.Commit.latest;
|
||||
import static org.apache.cassandra.utils.CassandraVersion.NULL_VERSION;
|
||||
import static org.apache.cassandra.utils.CassandraVersion.UNREADABLE_VERSION;
|
||||
|
|
@ -1842,11 +1846,11 @@ public final class SystemKeyspace
|
|||
}
|
||||
}
|
||||
|
||||
public static void writePreparedStatement(String loggedKeyspace, MD5Digest key, String cql)
|
||||
public static void writePreparedStatement(String loggedKeyspace, MD5Digest key, String cql, long timestamp)
|
||||
{
|
||||
executeInternal(format("INSERT INTO %s (logged_keyspace, prepared_id, query_string) VALUES (?, ?, ?)",
|
||||
executeInternal(format("INSERT INTO %s (logged_keyspace, prepared_id, query_string) VALUES (?, ?, ?) USING TIMESTAMP ?",
|
||||
PreparedStatements.toString()),
|
||||
loggedKeyspace, key.byteBuffer(), cql);
|
||||
loggedKeyspace, key.byteBuffer(), cql, timestamp);
|
||||
logger.debug("stored prepared statement for logged keyspace '{}': '{}'", loggedKeyspace, cql);
|
||||
}
|
||||
|
||||
|
|
@ -1862,17 +1866,50 @@ public final class SystemKeyspace
|
|||
preparedStatements.truncateBlockingWithoutSnapshot();
|
||||
}
|
||||
|
||||
public static int loadPreparedStatements(TriFunction<MD5Digest, String, String, Boolean> onLoaded)
|
||||
public static int loadPreparedStatements(TriFunction<MD5Digest, String, String, Prepared> onLoaded)
|
||||
{
|
||||
return loadPreparedStatements(onLoaded, QueryProcessor.PRELOAD_PREPARED_STATEMENTS_FETCH_SIZE);
|
||||
}
|
||||
|
||||
public static int loadPreparedStatements(TriFunction<MD5Digest, String, String, Prepared> onLoaded, int pageSize)
|
||||
{
|
||||
String query = String.format("SELECT prepared_id, logged_keyspace, query_string FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PREPARED_STATEMENTS);
|
||||
UntypedResultSet resultSet = executeOnceInternal(query);
|
||||
UntypedResultSet resultSet = executeOnceInternalWithPaging(query, pageSize);
|
||||
int counter = 0;
|
||||
|
||||
// As the cache size may be briefly exceeded before statements are evicted, we allow loading 110% the cache size
|
||||
// to avoid logging early.
|
||||
long preparedBytesLoadThreshold = (long) (PREPARED_STATEMENT_CACHE_SIZE_BYTES * 1.1);
|
||||
long preparedBytesLoaded = 0L;
|
||||
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))
|
||||
Prepared prepared = onLoaded.accept(MD5Digest.wrap(row.getByteArray("prepared_id")),
|
||||
row.getString("query_string"),
|
||||
row.has("logged_keyspace") ? row.getString("logged_keyspace") : null);
|
||||
if (prepared != null)
|
||||
{
|
||||
counter++;
|
||||
preparedBytesLoaded += Math.max(0, prepared.pstmntSize);
|
||||
|
||||
if (preparedBytesLoaded > preparedBytesLoadThreshold)
|
||||
{
|
||||
// In the event that we detect that we have loaded more bytes than the cache size return early to
|
||||
// prevent an indefinite startup time. This is almost certainly caused by the prepared statement cache
|
||||
// leaking (CASSANDRA-19703) which should not recur after being on a version running this code.
|
||||
// In such a case it's better to warn and continue startup than to continually page over millions of
|
||||
// prepared statements that would be immediately evicted.
|
||||
logger.warn("Detected prepared statement cache filling up during preload after preparing {} " +
|
||||
"statements (loaded {} with prepared_statements_cache_size being {}). " +
|
||||
"This could be an indication that prepared statements leaked prior to CASSANDRA-19703 " +
|
||||
"being fixed. Returning early to prevent indefinite startup. " +
|
||||
"Consider truncating {}.{} to clear out leaked prepared statements.",
|
||||
counter,
|
||||
FileUtils.stringifyFileSize(preparedBytesLoaded),
|
||||
FileUtils.stringifyFileSize(PREPARED_STATEMENT_CACHE_SIZE_BYTES),
|
||||
SchemaConstants.SYSTEM_KEYSPACE_NAME, PREPARED_STATEMENTS);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return counter;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ import net.bytebuddy.dynamic.DynamicType;
|
|||
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
|
||||
import net.bytebuddy.implementation.MethodDelegation;
|
||||
import org.apache.cassandra.cql3.CQLStatement;
|
||||
import org.apache.cassandra.cql3.QueryHandler;
|
||||
import org.apache.cassandra.cql3.QueryHandler.Prepared;
|
||||
import org.apache.cassandra.cql3.QueryProcessor;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.distributed.api.ConsistencyLevel;
|
||||
|
|
@ -268,9 +268,10 @@ public class MixedModeFuzzTest extends TestBaseImpl
|
|||
|
||||
c.get(nodeWithFix.get() ? 1 : 2).runOnInstance(() -> {
|
||||
SystemKeyspace.loadPreparedStatements((id, query, keyspace) -> {
|
||||
Prepared prepared = QueryProcessor.instance.getPrepared(id);
|
||||
if (rng.nextBoolean())
|
||||
QueryProcessor.instance.evictPrepared(id);
|
||||
return true;
|
||||
return prepared;
|
||||
});
|
||||
});
|
||||
break;
|
||||
|
|
@ -450,7 +451,7 @@ public class MixedModeFuzzTest extends TestBaseImpl
|
|||
if (existing != null)
|
||||
return existing;
|
||||
|
||||
QueryHandler.Prepared prepared = QueryProcessor.parseAndPrepare(queryString, clientState, false);
|
||||
Prepared prepared = QueryProcessor.parseAndPrepare(queryString, clientState, false);
|
||||
CQLStatement statement = prepared.statement;
|
||||
|
||||
int boundTerms = statement.getBindVariables().size();
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ 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.QueryHandler.Prepared;
|
||||
import org.apache.cassandra.cql3.QueryProcessor;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.distributed.api.ConsistencyLevel;
|
||||
|
|
@ -226,9 +227,10 @@ public class ReprepareFuzzTest extends TestBaseImpl
|
|||
case CLEAR_CACHES:
|
||||
c.get(1).runOnInstance(() -> {
|
||||
SystemKeyspace.loadPreparedStatements((id, query, keyspace) -> {
|
||||
Prepared prepared = QueryProcessor.instance.getPrepared(id);
|
||||
if (rng.nextBoolean())
|
||||
QueryProcessor.instance.evictPrepared(id);
|
||||
return true;
|
||||
return prepared;
|
||||
});
|
||||
});
|
||||
break;
|
||||
|
|
|
|||
|
|
@ -21,33 +21,62 @@ import java.net.InetSocketAddress;
|
|||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.apache.cassandra.db.ReadQuery;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.db.marshal.Int32Type;
|
||||
import org.apache.cassandra.db.marshal.UTF8Type;
|
||||
import org.apache.cassandra.schema.SchemaConstants;
|
||||
import org.apache.cassandra.schema.SchemaKeyspaceTables;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.ClientState;
|
||||
import org.apache.cassandra.transport.Dispatcher;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.MD5Digest;
|
||||
import org.jboss.byteman.contrib.bmunit.BMRule;
|
||||
import org.jboss.byteman.contrib.bmunit.BMRules;
|
||||
import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
|
||||
|
||||
import static java.util.Collections.emptyMap;
|
||||
import static org.apache.cassandra.service.QueryState.forInternalCalls;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
@RunWith(BMUnitRunner.class)
|
||||
public class PstmtPersistenceTest extends CQLTester
|
||||
{
|
||||
private static final CompletableFuture<?>[] futureArray = new CompletableFuture[0];
|
||||
|
||||
private static final ConcurrentMap<MD5Digest, Long> preparedStatementLoadTimestamps = new ConcurrentHashMap<>();
|
||||
private static final ConcurrentMap<MD5Digest, Long> preparedStatementRemoveTimestamps = new ConcurrentHashMap<>();
|
||||
|
||||
// page size passed to preloadPreparedStatements
|
||||
private static final int PRELOAD_PAGE_SIZE = 100;
|
||||
|
||||
// recorded page invocations in preloadPreparedStatements
|
||||
private static final AtomicInteger pageInvocations = new AtomicInteger();
|
||||
|
||||
@Before
|
||||
public void setUp()
|
||||
{
|
||||
preparedStatementLoadTimestamps.clear();
|
||||
preparedStatementRemoveTimestamps.clear();
|
||||
|
||||
QueryProcessor.clearPreparedStatements(false);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testCachedPreparedStatements() throws Throwable
|
||||
{
|
||||
|
|
@ -104,7 +133,7 @@ public class PstmtPersistenceTest extends CQLTester
|
|||
Assert.assertNotNull(prepared);
|
||||
}
|
||||
|
||||
// add anther prepared statement and sync it to table
|
||||
// add another prepared statement and sync it to table
|
||||
prepareStatement(statement2, "foo", "bar", clientState);
|
||||
|
||||
// statement1 will have two statements prepared because of `setKeyspace` usage
|
||||
|
|
@ -142,12 +171,24 @@ public class PstmtPersistenceTest extends CQLTester
|
|||
|
||||
createTable("CREATE TABLE %s (key int primary key, val int)");
|
||||
|
||||
long initialEvicted = numberOfEvictedStatements();
|
||||
|
||||
for (int cnt = 1; cnt < 10000; cnt++)
|
||||
{
|
||||
prepareStatement("INSERT INTO %s (key, val) VALUES (?, ?) USING TIMESTAMP " + cnt, clientState);
|
||||
|
||||
if (numberOfEvictedStatements() > 0)
|
||||
if (numberOfEvictedStatements() - initialEvicted > 0)
|
||||
{
|
||||
assertEquals("Number of statements in table and in cache don't match", numberOfStatementsInMemory(), numberOfStatementsOnDisk());
|
||||
|
||||
// prepare more statements to trigger more evictions
|
||||
for (int cnt2 = cnt + 1; cnt2 < cnt + 10; cnt2++)
|
||||
prepareStatement("INSERT INTO %s (key, val) VALUES (?, ?) USING TIMESTAMP " + cnt2, clientState);
|
||||
|
||||
// each new prepared statement should have caused an eviction
|
||||
assertEquals("eviction count didn't increase by the expected number", 10, numberOfEvictedStatements() - initialEvicted);
|
||||
assertEquals("Number of statements in memory (expected) and table (actual) don't match", numberOfStatementsInMemory(), numberOfStatementsOnDisk());
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
|
@ -155,6 +196,196 @@ public class PstmtPersistenceTest extends CQLTester
|
|||
fail("Prepared statement eviction does not work");
|
||||
}
|
||||
|
||||
@Test
|
||||
@BMRules(rules= {
|
||||
@BMRule(name = "CaptureWriteTimestamps",
|
||||
targetClass = "SystemKeyspace",
|
||||
targetMethod = "writePreparedStatement(String, MD5Digest, String, long)",
|
||||
targetLocation = "AT INVOKE executeInternal",
|
||||
action = "org.apache.cassandra.cql3.PstmtPersistenceTest.preparedStatementLoadTimestamps.put($key, $timestamp);"
|
||||
),
|
||||
@BMRule(name = "CaptureEvictTimestamps",
|
||||
targetClass = "QueryProcessor",
|
||||
targetMethod = "evictPreparedStatement(MD5Digest, RemovalCause)",
|
||||
action = "org.apache.cassandra.cql3.PstmtPersistenceTest.preparedStatementRemoveTimestamps.put($key, org.apache.cassandra.service.ClientState.getTimestamp());"
|
||||
)
|
||||
})
|
||||
public void testAsyncPstmtInvalidation() throws Throwable
|
||||
{
|
||||
ClientState clientState = ClientState.forInternalCalls();
|
||||
createTable("CREATE TABLE %s (key int primary key, val int)");
|
||||
|
||||
// prepare statements concurrently in a thread pool to exercise bug encountered in CASSANDRA-19703 where
|
||||
// delete from table occurs before the insert due to early eviction.
|
||||
final ExecutorService executor = Executors.newFixedThreadPool(10);
|
||||
|
||||
long initialEvicted = numberOfEvictedStatements();
|
||||
try
|
||||
{
|
||||
int initialMaxStatementsToPrepare = 10000;
|
||||
int maxStatementsToPrepare = initialMaxStatementsToPrepare;
|
||||
boolean hasEvicted = false;
|
||||
int concurrency = 100;
|
||||
List<CompletableFuture<MD5Digest>> prepareFutures = new ArrayList<>(concurrency);
|
||||
|
||||
for (int cnt = 1; cnt <= maxStatementsToPrepare; cnt++)
|
||||
{
|
||||
final int localCnt = cnt;
|
||||
prepareFutures.add(CompletableFuture.supplyAsync(() -> prepareStatement("INSERT INTO %s (key, val) VALUES (?, ?) USING TIMESTAMP " + localCnt, clientState), executor));
|
||||
|
||||
if (prepareFutures.size() == concurrency)
|
||||
{
|
||||
// Await completion of current inflight futures
|
||||
CompletableFuture.allOf(prepareFutures.toArray(futureArray)).get(10, TimeUnit.SECONDS);
|
||||
prepareFutures.clear();
|
||||
}
|
||||
|
||||
// Once we've detected evictions, prepare as many statements as we've prepared so far to initialMaxStatementsToPrepare and then stop.
|
||||
if (!hasEvicted && numberOfEvictedStatements() - initialEvicted > 0)
|
||||
{
|
||||
maxStatementsToPrepare = Math.min(cnt * 2, initialMaxStatementsToPrepare);
|
||||
hasEvicted = true;
|
||||
}
|
||||
}
|
||||
|
||||
long evictedStatements = numberOfEvictedStatements() - initialEvicted;
|
||||
assertNotEquals("Should have evicted some prepared statements", 0, evictedStatements);
|
||||
|
||||
// Recorded prepared statement removals should match metrics
|
||||
assertEquals("Actual evicted statements does not match metrics", evictedStatements, preparedStatementRemoveTimestamps.size());
|
||||
|
||||
// For each prepared statement evicted, assert the time it was deleted is greater than the timestamp
|
||||
// used for when it was loaded.
|
||||
for (Map.Entry<MD5Digest, Long> evictedStatementEntry : preparedStatementRemoveTimestamps.entrySet())
|
||||
{
|
||||
MD5Digest key = evictedStatementEntry.getKey();
|
||||
long deletionTimestamp = evictedStatementEntry.getValue();
|
||||
long insertionTimestamp = preparedStatementLoadTimestamps.get(key);
|
||||
|
||||
assertTrue(String.format("Expected deletion timestamp for prepared statement (%d) to be greater than insertion timestamp (%d)",
|
||||
deletionTimestamp, insertionTimestamp),
|
||||
deletionTimestamp > insertionTimestamp);
|
||||
}
|
||||
|
||||
// ensure the number of statements on disk match the number in memory, if number of statements on disk eclipses in memory, there was a leak.
|
||||
assertEquals("Number of statements in memory (expected) and table (actual) don't match", numberOfStatementsInMemory(), numberOfStatementsOnDisk());
|
||||
}
|
||||
finally
|
||||
{
|
||||
executor.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Invoked whenever paging happens in testPreloadPreparedStatements, increments PAGE_INVOCATIONS when we detect
|
||||
* paging happening in the path of QueryProcessor.preloadPreparedStatements with the expected page size.
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
private static void nextPageReadQuery(ReadQuery query, int pageSize)
|
||||
{
|
||||
TableMetadata metadata = query.metadata();
|
||||
if (metadata.keyspace.equals(SchemaConstants.SYSTEM_KEYSPACE_NAME) &&
|
||||
metadata.name.equals(SystemKeyspace.PREPARED_STATEMENTS) &&
|
||||
pageSize == PRELOAD_PAGE_SIZE)
|
||||
{
|
||||
for (StackTraceElement stackTraceElement : Thread.currentThread().getStackTrace())
|
||||
{
|
||||
if (stackTraceElement.getClassName().equals(QueryProcessor.class.getName()) && stackTraceElement.getMethodName().equals("preloadPreparedStatements"))
|
||||
{
|
||||
pageInvocations.incrementAndGet();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@BMRule(name = "CapturePageInvocations",
|
||||
targetClass = "PartitionRangeQueryPager",
|
||||
targetMethod = "nextPageReadQuery(int)",
|
||||
action = "org.apache.cassandra.cql3.PstmtPersistenceTest.nextPageReadQuery($this.query, $pageSize)")
|
||||
public void testPreloadPreparedStatements() throws Throwable
|
||||
{
|
||||
ClientState clientState = ClientState.forInternalCalls();
|
||||
createTable("CREATE TABLE %s (key int primary key, val int)");
|
||||
|
||||
// Prepare more statements than the paging size to ensure paging works properly.
|
||||
int statementsToPrepare = 750;
|
||||
|
||||
for (int cnt = 1; cnt <= statementsToPrepare; cnt++)
|
||||
{
|
||||
prepareStatement("INSERT INTO %s (key, val) VALUES (?, ?) USING TIMESTAMP " + cnt, clientState);
|
||||
}
|
||||
|
||||
// Capture how many statements are in memory before clearing cache.
|
||||
long statementsInMemory = numberOfStatementsInMemory();
|
||||
long statementsOnDisk = numberOfStatementsOnDisk();
|
||||
assertEquals(statementsOnDisk, statementsInMemory);
|
||||
|
||||
// Drop prepared statements from cache only and ensure the cache empties out.
|
||||
QueryProcessor.clearPreparedStatements(true);
|
||||
assertEquals(0, numberOfStatementsInMemory());
|
||||
|
||||
// Load prepared statements and ensure the cache size matches max
|
||||
QueryProcessor.instance.preloadPreparedStatements(PRELOAD_PAGE_SIZE);
|
||||
|
||||
long statementsInMemoryAfterLoading = numberOfStatementsInMemory();
|
||||
// Ensure size of cache matches statements that were on disk before preload
|
||||
assertEquals("Statements prepared - evicted (expected) does not match statements in memory (actual)",
|
||||
statementsOnDisk, statementsInMemoryAfterLoading);
|
||||
|
||||
// Number of statements on disk shold match memory
|
||||
assertEquals(statementsInMemoryAfterLoading, numberOfStatementsOnDisk());
|
||||
|
||||
// Ensure only executed the expected amount of pages.
|
||||
int expectedPageInvocations = (int) Math.ceil(statementsInMemoryAfterLoading / (double) PRELOAD_PAGE_SIZE);
|
||||
assertEquals(expectedPageInvocations, pageInvocations.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPreloadPreparedStatementsUntilCacheFull()
|
||||
{
|
||||
QueryHandler handler = ClientState.getCQLQueryHandler();
|
||||
ClientState clientState = ClientState.forInternalCalls();
|
||||
createTable("CREATE TABLE %s (key int primary key, val int)");
|
||||
|
||||
// Fill up and clear the prepared statement cache several times to load up the system.prepared_statements table.
|
||||
// This simulates a 'leak' of prepared statements akin to CASSANDRA-19703 as the system.prepared_statements
|
||||
// table is able to grow to a larger size than the in memory prepared statement cache. In such a case we
|
||||
// should detect a possible leak and defer paging indefinitely by returning early in preloadPreparedStatements.
|
||||
int statementsLoadedWhenFull = -1;
|
||||
long accumulatedSize = 0;
|
||||
// load enough prepared statements to fill the cache 5 times.
|
||||
for (int cnt = 0; accumulatedSize < QueryProcessor.PREPARED_STATEMENT_CACHE_SIZE_BYTES * 5; cnt++)
|
||||
{
|
||||
MD5Digest id = prepareStatement("INSERT INTO %s (key, val) VALUES (?, ?) USING TIMESTAMP " + cnt, clientState);
|
||||
QueryHandler.Prepared prepared = handler.getPrepared(id);
|
||||
assertTrue(prepared.pstmntSize > -1);
|
||||
accumulatedSize += prepared.pstmntSize;
|
||||
if (statementsLoadedWhenFull == -1 && accumulatedSize > QueryProcessor.PREPARED_STATEMENT_CACHE_SIZE_BYTES)
|
||||
{
|
||||
statementsLoadedWhenFull = cnt;
|
||||
}
|
||||
// clear cache repeatedly to avoid eviction.
|
||||
QueryProcessor.clearPreparedStatements(true);
|
||||
}
|
||||
|
||||
|
||||
int preloadedStatements = QueryProcessor.instance.preloadPreparedStatements(PRELOAD_PAGE_SIZE);
|
||||
|
||||
// Should have loaded as many statements as we detected were loaded before cache would be full.
|
||||
assertTrue(String.format("Preloaded %d statements, expected at least %d",
|
||||
preloadedStatements, statementsLoadedWhenFull),
|
||||
preloadedStatements > statementsLoadedWhenFull);
|
||||
|
||||
// We should only expect to load how many statements we were able to load before filling the cache
|
||||
// + a buffer of 110%, set to 1.5x just to deal with sensitivity of detecting cache filling up.
|
||||
int atMostPreloadedExpected = (int) (statementsLoadedWhenFull * 1.5);
|
||||
assertTrue(String.format("Preloaded %d statements, but only expected that we'd load at most %d",
|
||||
preloadedStatements, atMostPreloadedExpected),
|
||||
preloadedStatements <= atMostPreloadedExpected);
|
||||
}
|
||||
|
||||
private long numberOfStatementsOnDisk() throws Throwable
|
||||
{
|
||||
UntypedResultSet.Row row = execute("SELECT COUNT(*) FROM " + SchemaConstants.SYSTEM_KEYSPACE_NAME + '.' + SystemKeyspace.PREPARED_STATEMENTS).one();
|
||||
|
|
@ -178,7 +409,6 @@ public class PstmtPersistenceTest extends CQLTester
|
|||
|
||||
private MD5Digest prepareStatement(String stmt, String keyspace, String table, ClientState clientState)
|
||||
{
|
||||
System.out.println(stmt + String.format(stmt, keyspace + "." + table));
|
||||
return QueryProcessor.instance.prepare(String.format(stmt, keyspace + "." + table), clientState).statementId;
|
||||
return QueryProcessor.instance.prepare(String.format(stmt, keyspace + '.' + table), clientState).statementId;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue