From e4b6fd1d8648592c85c1faf01485d45b712ca3db Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 14 Aug 2026 13:30:10 +0200 Subject: [PATCH 1/2] fix: anchor the prepare cache entry on the prepared statement The prepare cache holds its values weakly, so an entry can be collected while the application is still using the statement it produced, costing a re-PREPARE round trip on the next prepare() of the same query. Callers routinely keep only the PreparedStatement, not the CompletionStage the processor hands back, so nothing keeps the cached future reachable. Store the cached future on the statement itself. The cache holds the future weakly, the future references the statement, and the statement references the future back, so the cycle survives exactly as long as the application holds the statement and becomes collectible as a whole once it does not. Expose this through PrepareCacheAnchor, an internal hook interface in the same vein as RequestRoutingTypeAccessor, rather than casting to DefaultPreparedStatement: a third-party PreparedStatement can opt into anchoring by implementing it. Co-Authored-By: Claude Opus 5 (1M context) --- .../core/cql/CqlPrepareAsyncProcessor.java | 6 ++ .../core/cql/DefaultPreparedStatement.java | 18 ++++- .../internal/core/cql/PrepareCacheAnchor.java | 43 +++++++++++ .../cql/CqlPrepareAsyncProcessorTest.java | 61 ++++++++++++++++ .../cql/DefaultPreparedStatementTest.java | 41 +---------- .../core/cql/PreparedStatementTestHelper.java | 71 +++++++++++++++++++ 6 files changed, 199 insertions(+), 41 deletions(-) create mode 100644 core/src/main/java/com/datastax/oss/driver/internal/core/cql/PrepareCacheAnchor.java create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/cql/PreparedStatementTestHelper.java diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java index 570d98d0aa0..6b4fa4cf23a 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java @@ -164,6 +164,12 @@ public CompletionStage process( mine.completeExceptionally(error); cache.invalidate(request); // Make sure failure isn't cached indefinitely } else { + // Anchor the cache entry on the statement, so that it survives for as long as + // the application holds the statement. The cache holds its values weakly, and + // callers routinely keep only the statement, not the future we return here. + if (preparedStatement instanceof PrepareCacheAnchor) { + ((PrepareCacheAnchor) preparedStatement).setPrepareCacheAnchor(mine); + } mine.complete(preparedStatement); } }); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java index 652e3f50af7..3b1f1d18360 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatement.java @@ -53,12 +53,14 @@ import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import net.jcip.annotations.ThreadSafe; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @ThreadSafe -public class DefaultPreparedStatement implements PreparedStatement, RequestRoutingTypeAccessor { +public class DefaultPreparedStatement + implements PreparedStatement, RequestRoutingTypeAccessor, PrepareCacheAnchor { private static final Logger LOGGER = LoggerFactory.getLogger(DefaultPreparedStatement.class); private static final Splitter SPACE_SPLITTER = Splitter.onPattern("\\s+"); private static final Splitter COMMA_SPLITTER = Splitter.onPattern(","); @@ -87,6 +89,15 @@ public class DefaultPreparedStatement implements PreparedStatement, RequestRouti @Nullable private final RequestRoutingType requestRoutingType; private volatile boolean skipMetadata; + /** + * Retains the prepare cache entry that produced this statement. Never read: it exists purely so + * that the entry, which the cache holds weakly, stays reachable for as long as this statement is. + * + * @see PrepareCacheAnchor + */ + @SuppressWarnings("unused") + private volatile CompletableFuture prepareCacheAnchor; + public DefaultPreparedStatement( ByteBuffer id, String query, @@ -144,6 +155,11 @@ public DefaultPreparedStatement( query, resultMetadataId, resultSetDefinitions, this.executionProfileForBoundStatements); } + @Override + public void setPrepareCacheAnchor(@Nullable CompletableFuture anchor) { + this.prepareCacheAnchor = anchor; + } + @NonNull @Override public ByteBuffer getId() { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/PrepareCacheAnchor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/PrepareCacheAnchor.java new file mode 100644 index 00000000000..12835603914 --- /dev/null +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/PrepareCacheAnchor.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.cql; + +import com.datastax.oss.driver.api.core.cql.PreparedStatement; +import edu.umd.cs.findbugs.annotations.Nullable; +import java.util.concurrent.CompletableFuture; + +/** + * Internal hook allowing a {@link PreparedStatement} to keep its prepare cache entry reachable. + * + *

{@link CqlPrepareAsyncProcessor} caches prepare futures with weak values, so an entry can be + * collected while the application still holds the resulting statement, causing a needless + * re-PREPARE. Storing the cached future on the statement ties the entry's lifetime to the + * statement's: the cache holds a weak reference to the future, the future references the statement, + * and the statement references the future back. The cycle stays reachable while the application + * holds the statement, and becomes collectible as a whole once it does not. + * + *

Implementations only need to retain the reference; the anchor is never read back. + */ +public interface PrepareCacheAnchor { + + /** + * Retains the prepare cache entry for this statement, preventing its weak-value eviction for as + * long as this statement is reachable. + */ + void setPrepareCacheAnchor(@Nullable CompletableFuture anchor); +} diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java index 8bde253dfac..817bd8340c3 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java @@ -17,6 +17,7 @@ */ package com.datastax.oss.driver.internal.core.cql; +import static com.datastax.oss.driver.internal.core.cql.PreparedStatementTestHelper.newPreparedStatement; import static org.assertj.core.api.Assertions.assertThat; import com.datastax.oss.driver.api.core.cql.PrepareRequest; @@ -25,6 +26,7 @@ import com.datastax.oss.driver.api.core.type.UserDefinedType; import com.datastax.oss.driver.internal.core.type.UserDefinedTypeBuilder; import com.datastax.oss.driver.shaded.guava.common.cache.Cache; +import java.lang.ref.WeakReference; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; @@ -106,4 +108,63 @@ public void should_match_udt_by_name_when_field_definitions_differ() { assertThat(CqlPrepareAsyncProcessor.typeMatches(oldType, DataTypes.listOf(resultType))) .isTrue(); } + + /** + * The anchor is what keeps a weakly-held cache entry alive: while the application holds the + * statement, the entry survives GC even though nothing else references the cached future. + */ + @Test + public void should_keep_cache_entry_alive_via_prepared_statement_anchor() throws Exception { + PrepareRequest request = new DefaultPrepareRequest("SELECT 1"); + + // The only surviving reference is the statement; the future stays in the callee's frame. + DefaultPreparedStatement ps = anchorNewEntry(request); + + collectGarbage(); + + assertThat(cache.getIfPresent(request)).isNotNull(); + assertThat(cache.getIfPresent(request).get()).isSameAs(ps); + } + + /** + * The reverse: once the statement becomes unreachable the whole cycle is collectible, so the + * anchor cannot turn the cache into a leak. + */ + @Test + public void should_evict_cache_entry_when_prepared_statement_is_unreachable() throws Exception { + PrepareRequest request = new DefaultPrepareRequest("SELECT 1"); + + // Wrapping in a WeakReference lets us drop the statement without keeping it in a local. + WeakReference ps = new WeakReference<>(anchorNewEntry(request)); + + collectGarbage(); + + assertThat(ps.get()) + .as("statement was not collected, so the cache assertion below proves nothing") + .isNull(); + assertThat(cache.getIfPresent(request)).isNull(); + } + + /** + * Reproduces what {@link CqlPrepareAsyncProcessor#process} does on a successful prepare: cache + * the future, anchor it on the resulting statement, then complete it. The future is deliberately + * a local of this method, so it becomes unreachable as soon as this frame returns. + */ + private DefaultPreparedStatement anchorNewEntry(PrepareRequest request) { + CompletableFuture cachedFuture = new CompletableFuture<>(); + cache.put(request, cachedFuture); + + DefaultPreparedStatement ps = newPreparedStatement(); + ps.setPrepareCacheAnchor(cachedFuture); + cachedFuture.complete(ps); + return ps; + } + + private void collectGarbage() throws InterruptedException { + for (int i = 0; i < 10; i++) { + System.gc(); + Thread.sleep(50); + cache.cleanUp(); + } + } } diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatementTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatementTest.java index 7ec6232fea7..c08980d29a4 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatementTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/DefaultPreparedStatementTest.java @@ -17,17 +17,12 @@ */ package com.datastax.oss.driver.internal.core.cql; +import static com.datastax.oss.driver.internal.core.cql.PreparedStatementTestHelper.newPreparedStatement; import static org.assertj.core.api.Assertions.assertThat; -import com.datastax.oss.driver.api.core.ConsistencyLevel; import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; -import com.datastax.oss.driver.api.core.DefaultProtocolVersion; import com.datastax.oss.driver.api.core.RequestRoutingType; import com.datastax.oss.driver.api.core.cql.BoundStatement; -import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; -import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry; -import com.datastax.oss.protocol.internal.util.Bytes; -import java.util.Collections; import org.junit.Test; public class DefaultPreparedStatementTest { @@ -75,38 +70,4 @@ public void should_keep_detected_lwt_routing_type_after_bound_consistency_overri assertThat(boundStatement.getRequestRoutingType()).isEqualTo(RequestRoutingType.LWT); } - - private DefaultPreparedStatement newPreparedStatement( - ConsistencyLevel consistencyLevel, - ConsistencyLevel serialConsistencyLevel, - RequestRoutingType requestRoutingType) { - ColumnDefinitions variableDefinitions = - DefaultColumnDefinitions.valueOf(Collections.emptyList()); - return new DefaultPreparedStatement( - Bytes.fromHexString("0x"), - "SELECT * FROM test.foo WHERE pk = ?", - variableDefinitions, - Collections.emptyList(), - null, - null, - null, - null, - Collections.emptyMap(), - null, - null, - null, - null, - null, - Collections.emptyMap(), - null, - null, - null, - Integer.MIN_VALUE, - consistencyLevel, - serialConsistencyLevel, - false, - CodecRegistry.DEFAULT, - DefaultProtocolVersion.DEFAULT, - requestRoutingType); - } } diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/PreparedStatementTestHelper.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/PreparedStatementTestHelper.java new file mode 100644 index 00000000000..949f1f3d6a8 --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/PreparedStatementTestHelper.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.cql; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.DefaultProtocolVersion; +import com.datastax.oss.driver.api.core.RequestRoutingType; +import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; +import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry; +import com.datastax.oss.protocol.internal.util.Bytes; +import java.util.Collections; + +/** Builds minimally-valid {@link DefaultPreparedStatement} instances for tests. */ +public class PreparedStatementTestHelper { + + /** Returns a statement with no consistency levels and no routing type configured. */ + public static DefaultPreparedStatement newPreparedStatement() { + return newPreparedStatement(null, null, null); + } + + public static DefaultPreparedStatement newPreparedStatement( + ConsistencyLevel consistencyLevel, + ConsistencyLevel serialConsistencyLevel, + RequestRoutingType requestRoutingType) { + ColumnDefinitions variableDefinitions = + DefaultColumnDefinitions.valueOf(Collections.emptyList()); + return new DefaultPreparedStatement( + Bytes.fromHexString("0x"), + "SELECT * FROM test.foo WHERE pk = ?", + variableDefinitions, + Collections.emptyList(), + null, + null, + null, + null, + Collections.emptyMap(), + null, + null, + null, + null, + null, + Collections.emptyMap(), + null, + null, + null, + Integer.MIN_VALUE, + consistencyLevel, + serialConsistencyLevel, + false, + CodecRegistry.DEFAULT, + DefaultProtocolVersion.DEFAULT, + requestRoutingType); + } + + private PreparedStatementTestHelper() {} +} From bbbb62d318a5f88f94aa9453c295b137ff9c259f Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 14 Aug 2026 13:30:27 +0200 Subject: [PATCH 2/2] refactor: always return a defensive copy from the prepare cache PR #892 returned the cached future itself once it was completed, so that a caller holding the returned stage would keep the weakly-held entry reachable. The anchor added in the previous commit ties the entry's lifetime to the statement instead, which covers that case and the far more common one where the caller keeps only the statement. That leaves the shortcut with no liveness value and one drawback: the cached future is handed to callers, who can still overwrite it through obtrudeValue/obtrudeException. Restore the unconditional defensive copy. Co-Authored-By: Claude Opus 5 (1M context) --- .../core/cql/CqlPrepareAsyncProcessor.java | 5 ----- .../core/cql/CqlPrepareAsyncProcessorTest.java | 15 +++++++++------ 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java index 6b4fa4cf23a..dc426980848 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java @@ -175,11 +175,6 @@ public CompletionStage process( }); } } - // If the future is already completed, return it directly to maintain a strong reference - // in the cache and avoid premature GC with weakValues() (ScyllaDB PR #892). - if (result.isDone()) { - return result; - } // Return a defensive copy. So if a client cancels its request, the cache won't be impacted // nor a potential concurrent request. return result.thenApply(x -> x); // copy() is available only since Java 9 diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java index 817bd8340c3..0f23cbc89df 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java @@ -50,12 +50,12 @@ public void setup() { } /** - * When the cached future is already completed, process() should return the exact same instance - * (identity). This ensures callers hold a strong reference to the cached CF, preventing - * weak-value eviction under GC pressure. + * process() always hands out a defensive copy, including for an already-completed entry: keeping + * the entry alive is the anchor's job, so the cached future never needs to be exposed. A caller + * that obtrudes on its copy must not corrupt what the cache holds. */ @Test - public void should_return_cached_future_directly_when_already_completed() throws Exception { + public void should_return_defensive_copy_when_future_is_already_completed() throws Exception { PrepareRequest request = new DefaultPrepareRequest("SELECT 1"); PreparedStatement ps = Mockito.mock(PreparedStatement.class); @@ -63,10 +63,13 @@ public void should_return_cached_future_directly_when_already_completed() throws CompletableFuture completed = CompletableFuture.completedFuture(ps); cache.put(request, completed); - // process() should return the exact same object CompletionStage returned = processor.process(request, null, null, "test"); - assertThat(returned).isSameAs(completed); + assertThat(returned).isNotSameAs(completed); + assertThat(returned.toCompletableFuture().get()).isSameAs(ps); + + returned.toCompletableFuture().obtrudeValue(null); + assertThat(completed.get()).isSameAs(ps); } /**