diff --git a/CHANGES.txt b/CHANGES.txt index 0f381b8125..f0453dec97 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/doc/modules/cassandra/pages/cql/cql_singlefile.adoc b/doc/modules/cassandra/pages/cql/cql_singlefile.adoc index d99e12b8ed..4de76217b9 100644 --- a/doc/modules/cassandra/pages/cql/cql_singlefile.adoc +++ b/doc/modules/cassandra/pages/cql/cql_singlefile.adoc @@ -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 `` +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 diff --git a/src/java/org/apache/cassandra/cql3/QueryHandler.java b/src/java/org/apache/cassandra/cql3/QueryHandler.java index 7f0c650a37..ba6f4117f5 100644 --- a/src/java/org/apache/cassandra/cql3/QueryHandler.java +++ b/src/java/org/apache/cassandra/cql3/QueryHandler.java @@ -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(); } } } diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index c36337d41e..83ae821ba3 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 5f8f74ce6b..4b8cd29f0f 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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 onLoaded) + public static int loadPreparedStatements(TriFunction onLoaded) + { + return loadPreparedStatements(onLoaded, QueryProcessor.PRELOAD_PREPARED_STATEMENTS_FETCH_SIZE); + } + + public static int loadPreparedStatements(TriFunction 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; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/MixedModeFuzzTest.java b/test/distributed/org/apache/cassandra/distributed/test/MixedModeFuzzTest.java index 954280f0b8..609f94ce91 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/MixedModeFuzzTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/MixedModeFuzzTest.java @@ -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(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java index f56847f68c..9988234743 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ReprepareFuzzTest.java @@ -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; diff --git a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java index eea8ca1fd9..f6bf6cb7d5 100644 --- a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java +++ b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java @@ -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 preparedStatementLoadTimestamps = new ConcurrentHashMap<>(); + private static final ConcurrentMap 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> 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 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; } }