From d077f695534b3db57b1573ee6b2bd0463c3883b3 Mon Sep 17 00:00:00 2001 From: Andy Tolbert <6889771+tolbertam@users.noreply.github.com> Date: Tue, 13 May 2025 12:34:29 -0500 Subject: [PATCH] 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 --- CHANGES.txt | 6 +- .../cassandra/pages/cql/cql_singlefile.adoc | 22 ++ .../apache/cassandra/cql3/QueryHandler.java | 7 + .../apache/cassandra/cql3/QueryProcessor.java | 73 ++++-- .../apache/cassandra/db/SystemKeyspace.java | 53 +++- .../distributed/test/MixedModeFuzzTest.java | 7 +- .../distributed/test/ReprepareFuzzTest.java | 4 +- .../cassandra/cql3/PstmtPersistenceTest.java | 240 +++++++++++++++++- 8 files changed, 377 insertions(+), 35 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 676522b9bf..9a036b2119 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/doc/modules/cassandra/pages/cql/cql_singlefile.adoc b/doc/modules/cassandra/pages/cql/cql_singlefile.adoc index 89ed359b73..1044975e35 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 e0480a6e4b..638b87018c 100644 --- a/src/java/org/apache/cassandra/cql3/QueryHandler.java +++ b/src/java/org/apache/cassandra/cql3/QueryHandler.java @@ -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(); } } } diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index c1045548b8..861222890a 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 7655257759..56dd03a3c4 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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 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 df4a554e2e..829f7d04a1 100644 --- a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java +++ b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java @@ -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 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 { @@ -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> 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(); @@ -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; } }