DATACASS-357 - Move parameter conversion to MappingCassandraConverter.

Previously, query method parameter conversion was handled separately. This was duplicate code and the code additionally converted arguments into property types regardless the further usage. Collection arguments (e.g. for IN query usage) could be converted into the property type (List of String converted into String).

 We now handle collection conversion and single element conversion separately so collections are no longer converted into the property's type. Collection elements are now inspected individually regarding their type/simple type conversion.

 This change also considers enum types as simple types with a distinct conversion of the enum value into a Cassandra value (numeric, character). The change in enum value handling reduces the scope of the conversion service usage and prevents accidental conversion.

Original pull request: #89.

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

@@ -19,6 +19,7 @@ import org.springframework.data.cassandra.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
 import org.springframework.data.convert.EntityConverter;
+import org.springframework.data.util.TypeInformation;
 
 /**
  * Central Cassandra specific converter interface from Object to Row.
@@ -62,6 +63,16 @@ public interface CassandraConverter
 	 */
 	void write(Object source, Object sink, CassandraPersistentEntity<?> entity);
 
+	/**
+	 * Converts the given object into one Cassandra will be able to store natively in a column.
+	 *
+	 * @param obj can be {@literal null}.
+	 * @param typeInformation must not be {@literal null}.
+	 * @return
+	 * @since 1.5
+	 */
+	Object convertToCassandraColumn(Object obj, TypeInformation<?> typeInformation);
+
 	/**
 	 * Returns the {@link CustomConversions} registered in the {@link CassandraConverter}.
 	 * 

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

@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.springframework.data.cassandra.convert;
 
 import java.util.ArrayList;
@@ -135,9 +134,7 @@ public class CustomConversions {
 	 * @return
 	 */
 	public boolean isSimpleType(Class<?> type) {
-
-		// Enums have no native Cassandra support
-		return (!type.isEnum() && simpleTypeHolder.isSimpleType(type));
+		return simpleTypeHolder.isSimpleType(type);
 	}
 
 	/**

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

@@ -15,8 +15,7 @@
  */
 package org.springframework.data.cassandra.convert;
 
-import static org.springframework.data.cassandra.repository.support.BasicMapId.Entry;
-import static org.springframework.data.cassandra.repository.support.BasicMapId.id;
+import static org.springframework.data.cassandra.repository.support.BasicMapId.*;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -36,6 +35,7 @@ import org.springframework.data.cassandra.mapping.BasicCassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
+import org.springframework.data.cassandra.mapping.CassandraType;
 import org.springframework.data.cassandra.repository.MapId;
 import org.springframework.data.cassandra.repository.MapIdentifiable;
 import org.springframework.data.convert.EntityInstantiator;
@@ -247,9 +247,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return;
 		}
 
-		Object obj = getReadValue(property, valueProvider);
-
-		propertyAccessor.setProperty(property, obj);
+		propertyAccessor.setProperty(property, getReadValue(valueProvider, property));
 	}
 
 	@SuppressWarnings("unused")
@@ -273,6 +271,25 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		throw new MappingException("Unknown row object " + ObjectUtils.nullSafeClassName(row));
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.convert.CassandraConverter#convertToCassandraColumn(java.lang.Object, org.springframework.data.util.TypeInformation)
+	 */
+	@Override
+	public Object convertToCassandraColumn(Object obj, TypeInformation<?> typeInformation) {
+
+		Assert.notNull(typeInformation, "TypeInformation must not be null!");
+
+		if (obj == null) {
+			return null;
+		}
+
+		if (obj.getClass().isArray()) {
+			return obj;
+		}
+
+		return getWriteValue(obj, typeInformation);
+	}
+
 	@Override
 	public void write(Object source, Object sink) {
 
@@ -458,8 +475,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		Class<?> targetType = getTargetType(idProperty);
 
 		if (conversionService.canConvert(id.getClass(), targetType)) {
-			return Collections.singleton(QueryBuilder.eq(idProperty.getColumnName().toCql(),
-				conversionService.convert(id, targetType)));
+			return Collections.singleton(
+					QueryBuilder.eq(idProperty.getColumnName().toCql(), getPotentiallyConvertedSimpleValue(id, targetType)));
 		}
 
 		return Collections.singleton(QueryBuilder.eq(idProperty.getColumnName().toCql(), id));
@@ -514,7 +531,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 						entry.getKey(), entity.getName()));
 
 			clauses.add(QueryBuilder.eq(persistentProperty.getColumnName().toCql(),
-					getWriteValue(persistentProperty, entry.getValue())));
+					getWriteValue(entry.getValue(), persistentProperty.getTypeInformation())));
 		}
 
 		return clauses;
@@ -613,17 +630,19 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return conversions.getCustomWriteTarget(property.getType());
 		}
 
-		if (conversions.isSimpleType(property.getType())) {
+		if (property.findAnnotation(CassandraType.class) != null) {
+			return getPropertyTargetType(property);
+		}
+
+		if (property.isCompositePrimaryKey() || conversions.isSimpleType(property.getType())
+				|| property.isCollectionLike()) {
 			return property.getType();
 		}
 
-		if (property.isCompositePrimaryKey()) {
-			return property.getType();
-		}
+		return getPropertyTargetType(property);
+	}
 
-		if (property.isCollectionLike()) {
-			return property.getType();
-		}
+	private Class<?> getPropertyTargetType(CassandraPersistentProperty property) {
 
 		DataType dataType = mappingContext.getDataType(property);
 
@@ -646,78 +665,127 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 */
 	@SuppressWarnings("unchecked")
 	private Object getWriteValue(CassandraPersistentProperty property, ConvertingPropertyAccessor accessor) {
-		return getWriteValue(property, accessor.getProperty(property, getTargetType(property)));
+		return getWriteValue(accessor.getProperty(property, getTargetType(property)), property.getTypeInformation());
 	}
 
 	/**
-	 * Retrieve the value to write for the given {@link CassandraPersistentProperty} from
-	 * {@link ConvertingPropertyAccessor} and perform optionally a conversion of collection element types.
+	 * Retrieve the value from {@code value} applying the given {@link TypeInformation} and perform optionally a
+	 * conversion of collection element types.
 	 *
-	 * @param property the property.
-	 * @param value the value
+	 * @param value the value, may be {@literal null}.
+	 * @param typeInformation the type information.
 	 * @return the return value, may be {@literal null}.
 	 */
 	@SuppressWarnings("unchecked")
-	private Object getWriteValue(CassandraPersistentProperty property, Object value) {
+	private Object getWriteValue(Object value, TypeInformation<?> typeInformation) {
 
-		if (value != null) {
+		if (value == null) {
+			return null;
+		}
 
-			if (conversions.hasCustomWriteTarget(property.getActualType()) && property.isCollectionLike()) {
-				Class<?> customWriteTarget = conversions.getCustomWriteTarget(property.getActualType());
+		if (conversions.isSimpleType(value.getClass())) {
+			// Doesn't need conversion
+			return getPotentiallyConvertedSimpleValue(value, typeInformation.getType());
+		}
 
-				if (Collection.class.isAssignableFrom(property.getType()) && value instanceof Collection) {
+		if (getCustomConversions().hasCustomWriteTarget(value.getClass())) {
+			return getConversionService().convert(value, getCustomConversions().getCustomWriteTarget(value.getClass()));
+		}
 
-					Collection<Object> original = (Collection<Object>) value;
-					Collection<Object> converted = CollectionFactory.createCollection(property.getType(), original.size());
+		TypeInformation<?> type = typeInformation != null ? typeInformation : ClassTypeInformation.from(value.getClass());
+		TypeInformation<?> actualType = type.getActualType();
 
-					for (Object o : original) {
-						converted.add(getConversionService().convert(o, customWriteTarget));
-					}
+		if (value instanceof Collection) {
 
-					return converted;
-				}
+			Collection<Object> original = (Collection<Object>) value;
+			Collection<Object> converted = CollectionFactory.createCollection(getCollectionType(type), original.size());
+
+			for (Object o : original) {
+				converted.add(convertToCassandraColumn(o, actualType));
 			}
 
-			CassandraPersistentEntity<?> persistentEntity = getMappingContext().getPersistentEntity(property.getActualType());
+			return converted;
+		}
 
-			if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
+		CassandraPersistentEntity<?> persistentEntity = getMappingContext().getPersistentEntity(actualType.getType());
 
-				if (property.isCollectionLike() && value instanceof Collection) {
-					Collection<Object> original = (Collection<Object>) value;
+		if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
 
-					Collection<Object> converted = CollectionFactory.createCollection(property.getType(), original.size());
+			UDTValue udtValue = persistentEntity.getUserType().newValue();
+			write(value, udtValue, persistentEntity);
 
-					for (Object element : original) {
-						if (element instanceof UDTValue) {
-							converted.add(element);
-						} else {
-							converted.add(getWriteValue(property, element));
-						}
-					}
-					return converted;
-				}
-
-				UDTValue udtValue = persistentEntity.getUserType().newValue();
-				write(value, udtValue, persistentEntity);
-
-				return udtValue;
-			}
+			return udtValue;
 		}
 
 		return value;
 	}
 
+	/**
+	 * Checks whether we have a custom conversion registered for the given value into an arbitrary simple Cassandra type.
+	 * Returns the converted value if so. If not, we perform special enum handling or simply return the value as is.
+	 *
+	 * @param value may be {@literal null}.
+	 * @param requestedTargetType must not be {@literal null}.
+	 * @return
+	 * @see CassandraType
+	 */
+	private Object getPotentiallyConvertedSimpleValue(Object value, Class<?> requestedTargetType) {
+
+		if (value == null) {
+			return null;
+		}
+
+		if (conversions.hasCustomWriteTarget(value.getClass(), requestedTargetType)) {
+			return conversionService.convert(value, conversions.getCustomWriteTarget(value.getClass(), requestedTargetType));
+		}
+
+		// Cassandra has no default enum handling - convert it either to string
+		// or - if requested - to a different type
+		if (Enum.class.isAssignableFrom(value.getClass())) {
+
+			if (requestedTargetType != null && !requestedTargetType.isEnum()
+					&& conversionService.canConvert(value.getClass(), requestedTargetType)) {
+				return conversionService.convert(value, requestedTargetType);
+			}
+
+			return ((Enum<?>) value).name();
+		}
+
+		return value;
+	}
+
+	private Class<?> getCollectionType(TypeInformation<?> type) {
+
+		if (type.getType().isInterface()) {
+			return type.getType();
+		}
+
+		if (ClassTypeInformation.LIST.isAssignableFrom(type)) {
+			return ClassTypeInformation.LIST.getType();
+		}
+
+		if (ClassTypeInformation.SET.isAssignableFrom(type)) {
+			return ClassTypeInformation.SET.getType();
+		}
+
+		if (!type.isCollectionLike()) {
+			return ClassTypeInformation.LIST.getType();
+		}
+
+		return type.getType();
+	}
+
 	/**
 	 * Retrieve the value to read for the given {@link CassandraPersistentProperty} from
 	 * {@link BasicCassandraRowValueProvider} and perform optionally a conversion of collection element types.
 	 *
-	 * @param property the property.
 	 * @param row the row.
+	 * @param property the property.
 	 * @return the return value, may be {@literal null}.
 	 */
 	@SuppressWarnings("unchecked")
-	private Object getReadValue(CassandraPersistentProperty property,
-			PropertyValueProvider<CassandraPersistentProperty> row) {
+	private Object getReadValue(PropertyValueProvider<CassandraPersistentProperty> row,
+			CassandraPersistentProperty property) {
 
 		Object obj = row.getPropertyValue(property);
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java

@@ -15,16 +15,11 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.Set;
 
-import org.springframework.core.CollectionFactory;
-import org.springframework.core.convert.ConversionService;
 import org.springframework.data.cassandra.convert.CassandraConverter;
-import org.springframework.data.cassandra.convert.CustomConversions;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
-import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.mapping.CassandraSimpleTypeHolder;
 import org.springframework.data.cassandra.mapping.CassandraType;
@@ -37,7 +32,6 @@ import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.DataType.CollectionType;
 import com.datastax.driver.core.TypeCodec;
-import com.datastax.driver.core.UDTValue;
 
 /**
  * Custom {@link org.springframework.data.repository.query.ParameterAccessor} that uses a {@link CassandraConverter} to
@@ -147,85 +141,27 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 			return null;
 		}
 
-		if (bindableValue.getClass().isArray()) {
-			return bindableValue;
-		}
-
-		if (property == null && getCustomConversions().hasCustomWriteTarget(bindableValue.getClass())) {
-			return converter.getConversionService().convert(bindableValue,
-					getCustomConversions().getCustomWriteTarget(bindableValue.getClass()));
-		}
-
-		// TODO: Polishing necessary
-		DataType parameterType = getDataType(index, property);
-
-		if (parameterType != null) {
-
-			if (property != null && getCustomConversions().hasCustomWriteTarget(property.getActualType())
-					&& property.isCollectionLike()) {
-
-				Class<?> customWriteTarget = getCustomConversions().getCustomWriteTarget(property.getActualType());
-
-				if (Collection.class.isAssignableFrom(property.getType()) && bindableValue instanceof Collection) {
-
-					Collection<Object> original = (Collection<Object>) bindableValue;
-					Collection<Object> converted = CollectionFactory.createCollection(property.getType(), original.size());
-
-					for (Object element : original) {
-						converted.add(getConversionService().convert(element, customWriteTarget));
-					}
-
-					return converted;
-				}
-			}
-
-			if (property != null) {
-				CassandraPersistentEntity<?> persistentEntity =
-						converter.getMappingContext().getPersistentEntity(property.getActualType());
-
-				if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
-					return toUDTValue(bindableValue, persistentEntity);
-				}
-			}
-
-			TypeCodec<?> cassandraType = CodecRegistry.DEFAULT_INSTANCE.codecFor(parameterType);
-
-			if (cassandraType.getJavaType().getRawType().isAssignableFrom(bindableValue.getClass())) {
-				return bindableValue;
-			}
-
-			return converter.getConversionService().convert(bindableValue, cassandraType.getJavaType().getRawType());
-		}
-
-		CassandraPersistentEntity<?> persistentEntity =
-				converter.getMappingContext().getPersistentEntity(bindableValue.getClass());
-
-		if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
-			return toUDTValue(bindableValue, persistentEntity);
-		}
-
-		return bindableValue;
+		return converter.convertToCassandraColumn(bindableValue, findTypeInformation(index, bindableValue, property));
 	}
 
-	private UDTValue toUDTValue(Object bindableValue, CassandraPersistentEntity<?> persistentEntity) {
+	private TypeInformation<?> findTypeInformation(int index, Object bindableValue,
+			CassandraPersistentProperty property) {
 
-		if (bindableValue instanceof UDTValue) {
-			return (UDTValue) bindableValue;
+		if (delegate.findCassandraType(index) != null) {
+
+			TypeCodec<?> typeCodec = CodecRegistry.DEFAULT_INSTANCE.codecFor(getDataType(index, property));
+			if (typeCodec.getJavaType().getType() instanceof Class<?>) {
+				return ClassTypeInformation.from((Class<?>) typeCodec.getJavaType().getType());
+			}
+
+			return ClassTypeInformation.from(typeCodec.getJavaType().getRawType());
 		}
 
-		UDTValue udtValue = persistentEntity.getUserType().newValue();
+		if (property == null) {
+			return ClassTypeInformation.from(bindableValue.getClass());
+		}
 
-		converter.write(bindableValue, udtValue, persistentEntity);
-
-		return udtValue;
-	}
-
-	private CustomConversions getCustomConversions() {
-		return converter.getCustomConversions();
-	}
-
-	private ConversionService getConversionService() {
-		return converter.getConversionService();
+		return property.getTypeInformation();
 	}
 
 	/**

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

@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.springframework.data.cassandra.convert;
 
 import static org.assertj.core.api.Assertions.*;
@@ -41,6 +40,20 @@ import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.DataType.Name;
+import com.datastax.driver.core.LocalDate;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.querybuilder.Assignment;
+import com.datastax.driver.core.querybuilder.BuiltStatement;
+import com.datastax.driver.core.querybuilder.Clause;
+import com.datastax.driver.core.querybuilder.Delete;
+import com.datastax.driver.core.querybuilder.Delete.Where;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.driver.core.querybuilder.Update.Assignments;
+
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -68,21 +81,6 @@ import org.springframework.data.cassandra.mapping.Table;
 import org.springframework.data.util.Version;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import com.datastax.driver.core.ColumnDefinitions;
-import com.datastax.driver.core.DataType;
-import com.datastax.driver.core.DataType.Name;
-import com.datastax.driver.core.LocalDate;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.querybuilder.Assignment;
-import com.datastax.driver.core.querybuilder.BuiltStatement;
-import com.datastax.driver.core.querybuilder.Clause;
-import com.datastax.driver.core.querybuilder.Delete;
-import com.datastax.driver.core.querybuilder.Delete.Where;
-import com.datastax.driver.core.querybuilder.Insert;
-import com.datastax.driver.core.querybuilder.QueryBuilder;
-import com.datastax.driver.core.querybuilder.Update;
-import com.datastax.driver.core.querybuilder.Update.Assignments;
-
 /**
  * Unit tests for {@link MappingCassandraConverter}.
  *
@@ -97,9 +95,7 @@ public class MappingCassandraConverterUnitTests {
 
 	@Rule public final ExpectedException expectedException = ExpectedException.none();
 
-	@Mock ColumnDefinitions columnDefinitionsMock;
-
-	@Mock Row rowMock;
+	@Mock private Row rowMock;
 
 	private CassandraMappingContext mappingContext;
 	private MappingCassandraConverter mappingCassandraConverter;

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

@@ -33,6 +33,8 @@ import org.springframework.data.cassandra.mapping.BasicCassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.mapping.CassandraType;
 import org.springframework.data.cassandra.repository.query.ConvertingParameterAccessor.PotentiallyConvertingIterator;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
 
 import com.datastax.driver.core.DataType;
 
@@ -45,9 +47,11 @@ import com.datastax.driver.core.DataType;
 @RunWith(MockitoJUnitRunner.class)
 public class ConvertingParameterAccessorUnitTests {
 
-	@Mock CassandraParameterAccessor mockParameterAccessor;
+	@Mock
+	CassandraParameterAccessor mockParameterAccessor;
 
-	@Mock CassandraPersistentProperty mockProperty;
+	@Mock
+	CassandraPersistentProperty mockProperty;
 
 	ConvertingParameterAccessor convertingParameterAccessor;
 
@@ -55,6 +59,7 @@ public class ConvertingParameterAccessorUnitTests {
 
 	@Before
 	public void setUp() {
+
 		this.converter = new MappingCassandraConverter(new BasicCassandraMappingContext());
 		this.converter.afterPropertiesSet();
 		this.convertingParameterAccessor = new ConvertingParameterAccessor(converter, mockParameterAccessor);
@@ -77,6 +82,7 @@ public class ConvertingParameterAccessorUnitTests {
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldReturnNativeBindableValue() {
+
 		when(mockParameterAccessor.getBindableValue(0)).thenReturn("hello");
 		when(mockParameterAccessor.getDataType(0)).thenReturn(DataType.varchar());
 		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) String.class);
@@ -95,6 +101,7 @@ public class ConvertingParameterAccessorUnitTests {
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldReturnConvertedBindableValue() {
+
 		LocalDate localDate = LocalDate.of(2010, 7, 4);
 
 		when(mockParameterAccessor.getBindableValue(0)).thenReturn(localDate);
@@ -110,6 +117,7 @@ public class ConvertingParameterAccessorUnitTests {
 	 */
 	@Test
 	public void shouldReturnDataTypeProvidedByDelegate() {
+
 		when(mockParameterAccessor.getDataType(0)).thenReturn(DataType.varchar());
 
 		assertThat(convertingParameterAccessor.getDataType(0)).isEqualTo(DataType.varchar());
@@ -122,15 +130,12 @@ public class ConvertingParameterAccessorUnitTests {
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldConvertCollections() {
+
 		LocalDate localDate = LocalDate.of(2010, 7, 4);
 
 		when(mockParameterAccessor.iterator())
 				.thenReturn((Iterator) Collections.singletonList(Collections.singletonList(localDate)).iterator());
-		when(mockParameterAccessor.getDataType(0)).thenReturn(DataType.list(DataType.date()));
-		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) List.class);
-		when(mockProperty.getType()).thenReturn((Class) List.class);
-		when(mockProperty.getActualType()).thenReturn((Class) LocalDate.class);
-		when(mockProperty.isCollectionLike()).thenReturn(true);
+		when(mockProperty.getTypeInformation()).thenReturn((TypeInformation) ClassTypeInformation.LIST);
 
 		PotentiallyConvertingIterator iterator = (PotentiallyConvertingIterator) convertingParameterAccessor.iterator();
 		Object converted = iterator.nextConverted(mockProperty);
@@ -148,6 +153,7 @@ public class ConvertingParameterAccessorUnitTests {
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldProvideTypeBasedOnValue() {
+
 		when(mockParameterAccessor.getDataType(0)).thenReturn(null);
 		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) LocalDate.class);
 
@@ -160,10 +166,10 @@ public class ConvertingParameterAccessorUnitTests {
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldProvideTypeBasedOnPropertyType() {
+
 		when(mockProperty.getDataType()).thenReturn(DataType.varchar());
 		when(mockProperty.findAnnotation(CassandraType.class)).thenReturn(mock(CassandraType.class));
 		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) String.class);
-		when(mockParameterAccessor.getDataType(0)).thenReturn(null);
 
 		assertThat(convertingParameterAccessor.getDataType(0, mockProperty)).isEqualTo(DataType.varchar());
 	}

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

@@ -19,6 +19,8 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
 import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 
 import org.junit.Before;
@@ -117,6 +119,17 @@ public class PartTreeCassandraQueryUnitTests {
 		assertThat(query).isEqualTo("SELECT * FROM person;");
 	}
 
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-357">DATACASS-357</a>
+	 */
+	@Test
+	public void shouldDeriveFieldInCollectionQuery() {
+		String query = deriveQueryFromMethod("findByFirstnameIn", new Class[] { Collection.class },
+				Arrays.asList("Hank", "Walter"));
+
+		assertThat(query).isEqualTo("SELECT * FROM person WHERE firstname IN ('Hank','Walter');");
+	}
+
 	/**
 	 * @see <a href="https://jira.spring.io/browse/DATACASS-172">DATACASS-172</a>
 	 */
@@ -145,13 +158,34 @@ public class PartTreeCassandraQueryUnitTests {
 		assertThat(query).isEqualTo("SELECT * FROM person WHERE mainaddress={};");
 	}
 
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-357">DATACASS-357</a>
+	 */
+	@Test
+	public void shouldDeriveUdtInCollectionQuery() {
+
+		when(userTypeResolverMock.resolveType(CqlIdentifier.cqlId("address"))).thenReturn(userTypeMock);
+		when(userTypeMock.newValue()).thenReturn(udtValueMock);
+
+		String query = deriveQueryFromMethod("findByMainAddressIn", new Class[] { Collection.class },
+				Collections.singleton(udtValueMock));
+
+		assertThat(query).isEqualTo("SELECT * FROM person WHERE mainaddress IN ({});");
+	}
+
 	private String deriveQueryFromMethod(String method, Object... args) {
+
 		Class<?>[] types = new Class<?>[args.length];
 
 		for (int i = 0; i < args.length; i++) {
 			types[i] = ClassUtils.getUserClass(args[i].getClass());
 		}
 
+		return deriveQueryFromMethod(method, types, args);
+	}
+
+	private String deriveQueryFromMethod(String method, Class<?>[] types, Object... args) {
+
 		PartTreeCassandraQuery partTreeQuery = createQueryForMethod(method, types);
 
 		CassandraParameterAccessor accessor = new CassandraParametersParameterAccessor(partTreeQuery.getQueryMethod(),
@@ -193,6 +227,10 @@ public class PartTreeCassandraQueryUnitTests {
 
 		Person findByMainAddress(UDTValue udtValue);
 
+		Person findByMainAddressIn(Collection<Address> address);
+
+		Person findByFirstnameIn(Collection<String> firstname);
+
 		PersonProjection findPersonProjectedBy();
 
 		<T> T findDynamicallyProjectedBy(Class<T> type);