natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ensure prepared_statement INSERT timestamp precedes eviction DELETE 

Updates SystemKeyspace.writePreparedStatement to accept a timestamp
associated with the Prepared creation time. Using this timestamp
will ensure that an INSERT into system.prepared_statements will
always precede the timestamp for the same Prepared in
SystemKeyspace.removePreparedStatement.

This is needed because Caffeine 2.9.2 may evict an entry as soon
as it is inserted if the maximum weight of the cache is exceeded
causing the DELETE to be executed before the INSERT.

Additionally, any clusters currently experiencing a leaky
system.prepared_statements table from this bug may struggle to
bounce into a version with this fix as
SystemKeyspace.loadPreparedPreparedStatements currently does
not paginate the query to system.prepared_statements, causing heap
OOMs.  To fix this this patch adds pagination at 5000 rows and
aborts loading once the cache size is loaded. This should allow
nodes to come up and delete older prepared statements that may no
longer be used as the cache fills up (which should happen immediately).

This patch does not address the issue of Caffeine immediately evicting
a prepared statement, however it will prevent the
system.prepared_statements table from growing unbounded.  For most users
this should be adequate, as the cache should only be filled when there
are erroneously many unique prepared statements. In such a case we can
expect that clients will constantly prepare statements regardless
of whether or not the cache is evicting statements.

patch by Andy Tolbert; reviewed by Berenguer Blasi and Caleb Rackliffe for CASSANDRA-19703
This commit is contained in:
Andy Tolbert 2025-05-13 12:34:29 -05:00
parent c736d22cf8
commit d077f69553
8 changed files with 377 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGES.txt
View File

@ -1,5 +1,9 @@
4.0.18
4.0.19
* Ensure prepared_statement INSERT timestamp precedes eviction DELETE (CASSANDRA-19703)
* Gossip doesn't converge due to race condition when updating EndpointStates multiple fields (CASSANDRA-20659)
4.0.18
* Handle sstable metadata stats file getting a new mtime after compaction has finished (CASSANDRA-18119)
* Honor MAX_PARALLEL_TRANSFERS correctly (CASSANDRA-20532)
* Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. (CASSANDRA-20561)

22
doc/modules/cassandra/pages/cql/cql_singlefile.adoc
View File

@ -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

7
src/java/org/apache/cassandra/cql3/QueryHandler.java
View File

@ -65,6 +65,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)}.
@ -81,6 +87,7 @@ public interface QueryHandler
this.resultMetadataId = ResultSet.ResultMetadata.fromPrepared(statement).getResultMetadataId();
this.fullyQualified = fullyQualified;
this.keyspace = keyspace;
this.timestamp = ClientState.getTimestamp();
}
}
}

73
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger;
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.*;
@ -86,23 +87,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.getPreparedStatementsCacheSizeMB());
private static final AtomicInteger lastMinuteEvictionsCount = new AtomicInteger(0);
static
{
preparedStatements = Caffeine.newBuilder()
.executor(MoreExecutors.directExecutor())
.maximumWeight(capacityToBytes(DatabaseDescriptor.getPreparedStatementsCacheSizeMB()))
.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);
@ -116,6 +116,16 @@ public class QueryProcessor implements QueryHandler
DatabaseDescriptor.getPreparedStatementsCacheSizeMB());
}
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;
@ -140,6 +150,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
@ -154,17 +170,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;
}
@ -466,11 +483,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;
QueryPager pager = select.getQuery(makeInternalOptions(prepared.statement, values), FBUtilities.nowInSeconds()).getPager(null, ProtocolVersion.CURRENT);
SelectStatement select = (SelectStatement) statement;
int nowInSec = FBUtilities.nowInSeconds();
QueryPager pager = select.getQuery(makeInternalOptions(select, values), nowInSec).getPager(null, ProtocolVersion.CURRENT);
return UntypedResultSet.create(select, pager, pageSize);
}
@ -696,7 +735,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);

53
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -53,6 +53,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;
@ -78,6 +79,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
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;
@ -109,8 +111,10 @@ import org.apache.cassandra.utils.UUIDGen;
import static java.lang.String.format;
import static java.util.Collections.emptyMap;
import static java.util.Collections.singletonMap;
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.executeOnceInternal;
import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternalWithPaging;
public final class SystemKeyspace
{
@ -1615,11 +1619,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);
}
@ -1635,17 +1639,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;
}

7
test/distributed/org/apache/cassandra/distributed/test/MixedModeFuzzTest.java
View File

@ -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();

4
test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java
View File

@ -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;

240
test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java
View File

@ -21,31 +21,60 @@ 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.service.QueryState;
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 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
{
@ -102,7 +131,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
@ -140,12 +169,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;
}
}
@ -153,6 +194,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();
@ -176,7 +407,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;
}
}