fix: invalidate prepared cache on UDT changes

Match changed UDTs by keyspace and type name instead of full field equality. Result and variable metadata can contain a different field shape after ALTER TYPE, so full UDT equality is too strict and leaves affected prepared statements cached.

Ignore type-create events that do not have an old type, add a unit regression for nested UDT matching, and make PreparedStatementCachingIT retain cached futures so it validates invalidation rather than racing weak-value collection.

Author: dkropachev

core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java

@@ -83,12 +83,12 @@ protected CqlPrepareAsyncProcessor(
         });
   }
 
-  private static boolean typeMatches(UserDefinedType oldType, DataType typeToCheck) {
+  static boolean typeMatches(UserDefinedType oldType, DataType typeToCheck) {
 
     switch (typeToCheck.getProtocolCode()) {
       case ProtocolConstants.DataType.UDT:
         UserDefinedType udtType = (UserDefinedType) typeToCheck;
-        return udtType.equals(oldType)
+        return sameTypeName(udtType, oldType)
             ? true
             : Iterables.any(udtType.getFieldTypes(), (testType) -> typeMatches(oldType, testType));
       case ProtocolConstants.DataType.LIST:
@@ -110,7 +110,14 @@ private static boolean typeMatches(UserDefinedType oldType, DataType typeToCheck
     }
   }
 
+  private static boolean sameTypeName(UserDefinedType left, UserDefinedType right) {
+    return left.getKeyspace().equals(right.getKeyspace()) && left.getName().equals(right.getName());
+  }
+
   private void onTypeChanged(TypeChangeEvent event) {
+    if (event.oldType == null) {
+      return;
+    }
     for (Map.Entry<PrepareRequest, CompletableFuture<PreparedStatement>> entry :
         this.cache.asMap().entrySet()) {
 

core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessorTest.java

@@ -21,6 +21,9 @@
 
 import com.datastax.oss.driver.api.core.cql.PrepareRequest;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+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.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -85,4 +88,22 @@ public void should_return_defensive_copy_when_future_is_in_flight() throws Excep
     returned.toCompletableFuture().cancel(false);
     assertThat(inFlight.isCancelled()).isFalse();
   }
+
+  @Test
+  public void should_match_udt_by_name_when_field_definitions_differ() {
+    UserDefinedType oldType =
+        new UserDefinedTypeBuilder("ks", "test_type_2")
+            .withField("c", DataTypes.INT)
+            .withField("d", DataTypes.TEXT)
+            .build();
+    UserDefinedType resultType =
+        new UserDefinedTypeBuilder("ks", "test_type_2")
+            .withField("c", DataTypes.INT)
+            .withField("d", DataTypes.TEXT)
+            .withField("i", DataTypes.BLOB)
+            .build();
+
+    assertThat(CqlPrepareAsyncProcessor.typeMatches(oldType, DataTypes.listOf(resultType)))
+        .isTrue();
+  }
 }

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/PreparedStatementCachingIT.java

@@ -18,12 +18,14 @@
 package com.datastax.oss.driver.core.cql;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import com.codahale.metrics.Gauge;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
 import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.cql.PrepareRequest;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.metrics.DefaultSessionMetric;
 import com.datastax.oss.driver.api.core.session.ProgrammaticArguments;
@@ -34,10 +36,12 @@
 import com.datastax.oss.driver.api.testinfra.session.SessionUtils;
 import com.datastax.oss.driver.categories.IsolatedTests;
 import com.datastax.oss.driver.internal.core.context.DefaultDriverContext;
+import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
 import com.datastax.oss.driver.internal.core.cql.CqlPrepareAsyncProcessor;
 import com.datastax.oss.driver.internal.core.cql.CqlPrepareSyncProcessor;
 import com.datastax.oss.driver.internal.core.metadata.schema.events.TypeChangeEvent;
 import com.datastax.oss.driver.internal.core.session.BuiltInRequestProcessors;
+import com.datastax.oss.driver.internal.core.session.DefaultSession;
 import com.datastax.oss.driver.internal.core.session.RequestProcessor;
 import com.datastax.oss.driver.internal.core.session.RequestProcessorRegistry;
 import com.datastax.oss.driver.shaded.guava.common.cache.RemovalListener;
@@ -53,6 +57,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -117,6 +122,9 @@ private static class TestCqlPrepareAsyncProcessor extends CqlPrepareAsyncProcess
     private static final Logger LOG =
         LoggerFactory.getLogger(PreparedStatementCachingIT.TestCqlPrepareAsyncProcessor.class);
 
+    private final Set<CompletableFuture<PreparedStatement>> retainedCacheValues =
+        ConcurrentHashMap.newKeySet();
+
     private static RemovalListener<Object, Object> buildCacheRemoveCallback(
         @NonNull Optional<DefaultDriverContext> context) {
       return (evt) -> {
@@ -133,10 +141,25 @@ private static RemovalListener<Object, Object> buildCacheRemoveCallback(
     }
 
     public TestCqlPrepareAsyncProcessor(@NonNull Optional<DefaultDriverContext> context) {
-      // Default CqlPrepareAsyncProcessor uses weak values here as well.  We avoid doing so
-      // to prevent cache entries from unexpectedly disappearing mid-test.
+      // Default CqlPrepareAsyncProcessor uses weak values. Retain the cached futures so this test
+      // validates type-change invalidation instead of racing cache value collection.
       super(context, builder -> builder.removalListener(buildCacheRemoveCallback(context)));
     }
+
+    @Override
+    public CompletionStage<PreparedStatement> process(
+        PrepareRequest request,
+        DefaultSession session,
+        InternalDriverContext context,
+        String sessionLogPrefix) {
+      CompletionStage<PreparedStatement> stage =
+          super.process(request, session, context, sessionLogPrefix);
+      CompletableFuture<PreparedStatement> cachedValue = cache.getIfPresent(request);
+      if (cachedValue != null) {
+        retainedCacheValues.add(cachedValue);
+      }
+      return stage;
+    }
   }
 
   private static class TestDefaultDriverContext extends DefaultDriverContext {
@@ -222,11 +245,11 @@ private void invalidationTestInner(
       assertThat(getPreparedCacheSize(session)).isEqualTo(0);
       setupTestSchema.accept(session);
 
-      session.prepare(preparedStmtQueryType1);
-      ByteBuffer queryId2 = session.prepare(preparedStmtQueryType2).getId();
+      PreparedStatement statement1 = session.prepare(preparedStmtQueryType1);
+      PreparedStatement statement2 = session.prepare(preparedStmtQueryType2);
+      ByteBuffer queryId2 = statement2.getId();
       assertThat(getPreparedCacheSize(session)).isEqualTo(2);
 
-      CountDownLatch preparedStmtCacheRemoveLatch = new CountDownLatch(1);
       CountDownLatch typeChangeEventLatch = new CountDownLatch(expectedChangedTypes.size());
 
       DefaultDriverContext ctx = (DefaultDriverContext) session.getContext();
@@ -260,7 +283,6 @@ private void invalidationTestInner(
                   removedQueryEventError.set(
                       Optional.of("Unable to set reference for PS removal event"));
                 }
-                preparedStmtCacheRemoveLatch.countDown();
               });
 
       // alter test_type_2 to trigger cache invalidation and above events
@@ -270,17 +292,20 @@ private void invalidationTestInner(
       assertThat(Uninterruptibles.awaitUninterruptibly(typeChangeEventLatch, 10, TimeUnit.SECONDS))
           .withFailMessage("typeChangeEventLatch did not trigger before timeout")
           .isTrue();
-      assertThat(
-              Uninterruptibles.awaitUninterruptibly(
-                  preparedStmtCacheRemoveLatch, 10, TimeUnit.SECONDS))
-          .withFailMessage("preparedStmtCacheRemoveLatch did not trigger before timeout")
-          .isTrue();
 
-      /* Okay, the latch triggered so cache processing should now be done.  Let's validate :allthethings: */
-      assertThat(changedTypes.keySet()).isEqualTo(expectedChangedTypes);
-      assertThat(removedQueryIds.get()).isNotEmpty().get().isEqualTo(queryId2);
+      await()
+          .atMost(30, TimeUnit.SECONDS)
+          .untilAsserted(() -> assertThat(getPreparedCacheSize(session)).isEqualTo(1));
+
+      assertThat(session.prepare(preparedStmtQueryType1)).isSameAs(statement1);
       assertThat(getPreparedCacheSize(session)).isEqualTo(1);
 
+      assertThat(session.prepare(preparedStmtQueryType2)).isNotSameAs(statement2);
+      assertThat(getPreparedCacheSize(session)).isEqualTo(2);
+
+      assertThat(changedTypes.keySet()).isEqualTo(expectedChangedTypes);
+      removedQueryIds.get().ifPresent(queryId -> assertThat(queryId).isEqualTo(queryId2));
+
       // check no errors were seen in callback (and report those as fail msgs)
       // if something is broken these may still succeed due to timing
       // but shouldn't intermittently fail if the code is working properly