Inspect collection elements during write value conversion.

We now deeply introspect collection and map elements when obtaining write values to ensure proper UDT and tuple conversions.

Previously, collections containing UDT values did a pass-thru of values instead of applying UDT mapping.

Closes #1473

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java

@@ -955,7 +955,13 @@ private Object writeCollectionInternal(Collection<Object> source, ColumnType typ
 		ColumnType componentType = type.getRequiredComponentType();
 
 		for (Object element : source) {
-			converted.add(getWriteValue(element, componentType));
+
+			ColumnType elementType = componentType;
+			if (elementType.getType() == Object.class) {
+				elementType = getColumnTypeResolver().resolve(element);
+			}
+
+			converted.add(getWriteValue(element, elementType));
 		}
 
 		return converted;
@@ -969,7 +975,18 @@ private Object writeMapInternal(Map<Object, Object> source, ColumnType type) {
 		ColumnType valueType = type.getRequiredMapValueType();
 
 		for (Entry<Object, Object> entry : source.entrySet()) {
-			converted.put(getWriteValue(entry.getKey(), keyType), getWriteValue(entry.getValue(), valueType));
+
+			ColumnType elementKeyType = keyType;
+			if (elementKeyType.getType() == Object.class) {
+				elementKeyType = getColumnTypeResolver().resolve(entry.getKey());
+			}
+
+			ColumnType elementValueType = valueType;
+			if (elementValueType.getType() == Object.class) {
+				elementValueType = getColumnTypeResolver().resolve(entry.getValue());
+			}
+
+			converted.put(getWriteValue(entry.getKey(), elementKeyType), getWriteValue(entry.getValue(), elementValueType));
 		}
 
 		return converted;

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterUDTUnitTests.java

@@ -29,6 +29,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -213,6 +214,31 @@ void shouldWriteCompositeUdtPk() {
 				+ "VALUES ('foo',{zip:'69469',city:'Weinheim',streetlines:['Heckenpfad','14']})");
 	}
 
+	@Test // GH-1473
+	void shouldWriteMapCorrectly() {
+
+		Manufacturer manufacturer = new Manufacturer("foo", "bar");
+		AddressUserType addressUserType = prepareAddressUserType();
+
+		Map<Manufacturer, AddressUserType> value = Map.of(manufacturer, addressUserType);
+		Map<?, ?> writeValue = (Map<?, ?>) converter.convertToColumnType(value);
+		Map.Entry<?, ?> entry = writeValue.entrySet().iterator().next();
+
+		assertThat(entry.getKey()).isInstanceOf(UdtValue.class);
+		assertThat(entry.getValue()).isInstanceOf(UdtValue.class);
+	}
+
+	@Test // GH-1473
+	void shouldWriteSetCorrectly() {
+
+		AddressUserType addressUserType = prepareAddressUserType();
+
+		Set<AddressUserType> value = Set.of(addressUserType);
+		Set<?> writeValue = (Set<?>) converter.convertToColumnType(value);
+
+		assertThat(writeValue.iterator().next()).isInstanceOf(UdtValue.class);
+	}
+
 	private static AddressUserType prepareAddressUserType() {
 
 		AddressUserType addressUserType = new AddressUserType();

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQueryUnitTests.java

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -352,6 +353,27 @@ void bindsMappedUdtPropertyCorrectly() {
 		assertThat(actual.getPositionalValues().get(0)).isInstanceOf(UdtValue.class);
 	}
 
+	@Test // GH-1473
+	void bindsCollectionOfMappedUdtPropertyCorrectly() {
+
+		UserDefinedType addressType = UserDefinedTypeBuilder.forName("address").withField("city", DataTypes.TEXT)
+				.withField("country", DataTypes.TEXT).build();
+
+		when(userTypeResolver.resolveType(CqlIdentifier.fromCql("address"))).thenReturn(addressType);
+
+		StringBasedCassandraQuery cassandraQuery = getQueryMethod("findByMainAddress", Set.class);
+		CassandraParameterAccessor accessor = new ConvertingParameterAccessor(converter,
+				new CassandraParametersParameterAccessor(cassandraQuery.getQueryMethod(), Set.of(new AddressType())));
+
+		SimpleStatement actual = cassandraQuery.createQuery(accessor);
+
+		assertThat(actual.getQuery()).isEqualTo("SELECT * FROM person WHERE address=?;");
+		assertThat(actual.getPositionalValues().get(0)).isInstanceOf(Set.class);
+
+		Set<?> set = (Set<?>) actual.getPositionalValues().get(0);
+		assertThat(set.iterator().next()).isInstanceOf(UdtValue.class);
+	}
+
 	@Test // DATACASS-172
 	void bindsUdtValuePropertyCorrectly() {
 
@@ -467,6 +489,9 @@ private interface SampleRepository extends Repository<Person, String> {
 		@Query("SELECT * FROM person WHERE address=?0;")
 		Person findByMainAddress(AddressType address);
 
+		@Query("SELECT * FROM person WHERE address=?0;")
+		Person findByMainAddress(Set<AddressType> address);
+
 		@Query("SELECT * FROM person WHERE address=?0;")
 		Person findByMainAddress(UdtValue UdtValue);