fix: match UDT names for prepared cache invalidation

Prepared statement cache invalidation was comparing UDTs with UserDefinedType.equals(), which includes field definitions. That is too strict for schema-change invalidation: a prepared statement can carry a UDT shape from result metadata that is not field-equal to the schema event's old type, especially when the UDT is nested in a collection.

Invalidate by UDT identity instead: keyspace and type name are enough to know that a prepared statement references the changed type. Created-type events still have no old type, so ignore them before matching.

Add a unit regression for a collection UDT whose field definitions differ but keyspace and type name match.

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();
+  }
 }