From 5b0e3f0c569fd716661f3da885fab37411ea97e7 Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Mon, 23 Mar 2020 10:58:37 +0100
Subject: [PATCH] DATACASS-743 - Introduce ColumnTypeResolver.

We now resolve column types by using ColumnTypeResolver that encapsulates type lookups that previously were spread across different areas of the converter. We consistently apply type resolution for singular properties and collection/Map-like properties.
---
 .../core/convert/CassandraColumnType.java     |  54 ++
 .../core/convert/CassandraConverter.java      |  28 +-
 .../cassandra/core/convert/ColumnType.java    | 258 +++++++++
 .../core/convert/ColumnTypeResolver.java      |  83 +++
 .../convert/DefaultCassandraColumnType.java   |  81 +++
 .../core/convert/DefaultColumnType.java       | 128 +++++
 .../convert/DefaultColumnTypeResolver.java    | 498 ++++++++++++++++++
 .../convert/MappingCassandraConverter.java    | 168 +++---
 .../cassandra/core/convert/QueryMapper.java   |  51 +-
 .../cassandra/core/convert/UpdateMapper.java  |  16 +-
 .../core/mapping/CassandraMappingContext.java |   4 +-
 .../query/ConvertingParameterAccessor.java    | 110 +---
 .../convert/ColumnTypeResolverUnitTests.java  | 219 ++++++++
 .../MappingCassandraConverterUnitTests.java   |  53 +-
 .../core/convert/QueryMapperUnitTests.java    |   2 +-
 .../ConvertingParameterAccessorUnitTests.java |  16 -
 16 files changed, 1525 insertions(+), 244 deletions(-)
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraColumnType.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnType.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolver.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultCassandraColumnType.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnType.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
 create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolverUnitTests.java

diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraColumnType.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraColumnType.java
new file mode 100644
index 000000000..b5688d2fe
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraColumnType.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+
+/**
+ * Descriptor for a Cassandra column type exposing a {@link DataType}.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+public interface CassandraColumnType extends ColumnType {
+
+	/**
+	 * Returns the {@link DataType} associated with this column type.
+	 *
+	 * @return
+	 */
+	DataType getDataType();
+
+	/**
+	 * Returns whether the associated {@link DataType} is a {@link TupleType}.
+	 *
+	 * @return
+	 */
+	default boolean isTupleType() {
+		return getDataType() instanceof TupleType;
+	}
+
+	/**
+	 * Returns whether the associated {@link DataType} is a {@link UserDefinedType}.
+	 *
+	 * @return
+	 */
+	default boolean isUserDefinedType() {
+		return getDataType() instanceof UserDefinedType;
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraConverter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraConverter.java
index daf6df034..30e037e22 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraConverter.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraConverter.java
@@ -23,6 +23,7 @@ import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.convert.EntityConverter;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Central Cassandra specific converter interface from Object to Row.
@@ -47,6 +48,15 @@ public interface CassandraConverter
 	@Override
 	CassandraMappingContext getMappingContext();
 
+	/**
+	 * Returns the {@link ColumnTypeResolver} to resolve {@link ColumnType} for properties, {@link TypeInformation}, and
+	 * {@code values}.
+	 *
+	 * @return the {@link ColumnTypeResolver}
+	 * @since 3.0
+	 */
+	ColumnTypeResolver getColumnTypeResolver();
+
 	/**
 	 * Returns the Id for an entity. It can return:
 	 * <ul>
@@ -82,7 +92,23 @@ public interface CassandraConverter
 	 * @return the result of the conversion.
 	 * @since 1.5
 	 */
-	Object convertToColumnType(Object value, TypeInformation<?> typeInformation);
+	default Object convertToColumnType(Object value, TypeInformation<?> typeInformation) {
+
+		Assert.notNull(value, "Value must not be null");
+		Assert.notNull(typeInformation, "TypeInformation must not be null");
+
+		return convertToColumnType(value, getColumnTypeResolver().resolve(typeInformation));
+	}
+
+	/**
+	 * Converts the given object into a value Cassandra will be able to store natively in a column.
+	 *
+	 * @param value {@link Object} to convert; must not be {@literal null}.
+	 * @param typeDescriptor {@link ColumnType} used to describe the object type; must not be {@literal null}.
+	 * @return the result of the conversion.
+	 * @since 3.0
+	 */
+	Object convertToColumnType(Object value, ColumnType typeDescriptor);
 
 	/**
 	 * Converts and writes a {@code source} object into a {@code sink} using the given {@link CassandraPersistentEntity}.
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnType.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnType.java
new file mode 100644
index 000000000..572d2e43e
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnType.java
@@ -0,0 +1,258 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.data.UdtValue;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+
+/**
+ * Interface to access column type information. The {@link CassandraColumnType} subtype exposes Cassandra-specific
+ * {@link DataType} information.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+public interface ColumnType {
+
+	/**
+	 * Creates a {@link ColumnType} for a {@link Class}.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return
+	 */
+	static ColumnType create(Class<?> type) {
+		return create(ClassTypeInformation.from(type));
+	}
+
+	/**
+	 * Creates a {@link ColumnType} for a {@link TypeInformation}.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return
+	 */
+	static ColumnType create(TypeInformation<?> type) {
+		return new DefaultColumnType(type);
+	}
+
+	/**
+	 * Creates a {@link ColumnType} for a {@link Class} and {@link DataType}.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @param dataType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType create(Class<?> type, DataType dataType) {
+		return new DefaultCassandraColumnType(ClassTypeInformation.from(type), dataType);
+	}
+
+	/**
+	 * Creates a List {@link ColumnType} given its {@link ColumnType component type}.
+	 *
+	 * @param componentType must not be {@literal null}.
+	 * @return
+	 */
+	static ColumnType listOf(ColumnType componentType) {
+
+		if (componentType instanceof CassandraColumnType) {
+			return listOf((CassandraColumnType) componentType);
+		}
+
+		return new DefaultColumnType(ClassTypeInformation.LIST, componentType);
+	}
+
+	/**
+	 * Creates a List {@link ColumnType} given its {@link CassandraColumnType component type}.
+	 *
+	 * @param componentType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType listOf(CassandraColumnType componentType) {
+		return new DefaultCassandraColumnType(ClassTypeInformation.LIST, DataTypes.listOf(componentType.getDataType()),
+				componentType);
+	}
+
+	/**
+	 * Creates a Set {@link ColumnType} given its {@link ColumnType component type}.
+	 *
+	 * @param componentType must not be {@literal null}.
+	 * @return
+	 */
+	static ColumnType setOf(ColumnType componentType) {
+
+		if (componentType instanceof CassandraColumnType) {
+			return setOf((CassandraColumnType) componentType);
+		}
+
+		return new DefaultColumnType(ClassTypeInformation.SET, componentType);
+	}
+
+	/**
+	 * Creates a Set {@link ColumnType} given its {@link CassandraColumnType component type}.
+	 *
+	 * @param componentType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType setOf(CassandraColumnType componentType) {
+		return new DefaultCassandraColumnType(ClassTypeInformation.SET, DataTypes.setOf(componentType.getDataType()),
+				componentType);
+	}
+
+	/**
+	 * Creates a Map {@link ColumnType} given its {@link ColumnType key and value types}.
+	 *
+	 * @param keyType must not be {@literal null}.
+	 * @param valueType must not be {@literal null}.
+	 * @return
+	 */
+	static ColumnType mapOf(ColumnType keyType, ColumnType valueType) {
+		return new DefaultColumnType(ClassTypeInformation.MAP, keyType, valueType);
+	}
+
+	/**
+	 * Creates a Map {@link CassandraColumnType} given its {@link CassandraColumnType key and value types}.
+	 *
+	 * @param keyType must not be {@literal null}.
+	 * @param valueType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType mapOf(CassandraColumnType keyType, CassandraColumnType valueType) {
+		return new DefaultCassandraColumnType(ClassTypeInformation.MAP,
+				DataTypes.mapOf(keyType.getDataType(), valueType.getDataType()), keyType, valueType);
+	}
+
+	/**
+	 * Creates a UDT {@link CassandraColumnType} given its {@link UserDefinedType Cassandra type}.
+	 *
+	 * @param dataType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType udtOf(UserDefinedType dataType) {
+		return new DefaultCassandraColumnType(UdtValue.class, dataType);
+	}
+
+	/**
+	 * Creates a Tuple {@link CassandraColumnType} given its {@link TupleType Cassandra type}.
+	 *
+	 * @param dataType must not be {@literal null}.
+	 * @return
+	 */
+	static CassandraColumnType tupleOf(TupleType dataType) {
+		return new DefaultCassandraColumnType(TupleValue.class, dataType);
+	}
+
+	/**
+	 * Returns the Java type of the column.
+	 *
+	 * @return
+	 */
+	Class<?> getType();
+
+	/**
+	 * Returns whether the type can be considered a collection, which means it's a container of elements, e.g. a
+	 * {@link java.util.Collection} and {@link java.lang.reflect.Array} or anything implementing {@link Iterable}. If this
+	 * returns {@literal true} you can expect {@link #getComponentType()} to return a non-{@literal null} value.
+	 *
+	 * @return
+	 */
+	boolean isCollectionLike();
+
+	/**
+	 * Returns whether the property is a {@link java.util.List}. If this returns {@literal true} you can expect
+	 * {@link #getComponentType()} to return something not {@literal null}.
+	 *
+	 * @return
+	 */
+	boolean isList();
+
+	/**
+	 * Returns whether the property is a {@link java.util.Set}. If this returns {@literal true} you can expect
+	 * {@link #getComponentType()} to return something not {@literal null}.
+	 *
+	 * @return
+	 */
+	boolean isSet();
+
+	/**
+	 * Returns whether the property is a {@link java.util.Map}. If this returns {@literal true} you can expect
+	 * {@link #getComponentType()} as well as {@link #getMapValueType()} to return something not {@literal null}.
+	 *
+	 * @return
+	 */
+	boolean isMap();
+
+	/**
+	 * Returns the component type for {@link java.util.Collection}s or the key type for {@link java.util.Map}s.
+	 *
+	 * @return
+	 */
+	@Nullable
+	ColumnType getComponentType();
+
+	/**
+	 * Returns the component type for {@link java.util.Collection}s, the key type for {@link java.util.Map}s or the single
+	 * generic type if available. Throws {@link IllegalStateException} if the component value type cannot be resolved.
+	 *
+	 * @return
+	 * @throws IllegalStateException if the component type cannot be resolved, e.g. if a raw type is used or the type is
+	 *           not generic in the first place.
+	 */
+	default ColumnType getRequiredComponentType() {
+
+		ColumnType columnType = getComponentType();
+
+		if (columnType == null) {
+			throw new IllegalStateException("Type has no component type");
+		}
+
+		return columnType;
+	}
+
+	/**
+	 * Returns the map value type in case the underlying type is a {@link java.util.Map}.
+	 *
+	 * @return
+	 */
+	@Nullable
+	ColumnType getMapValueType();
+
+	/**
+	 * Returns the map value type in case the underlying type is a {@link java.util.Map}. or throw
+	 * {@link IllegalStateException} if the map value type cannot be resolved.
+	 *
+	 * @return
+	 * @throws IllegalStateException if the map value type cannot be resolved, usually due to the current
+	 *           {@link java.util.Map} type being a raw one.
+	 */
+	default ColumnType getRequiredMapValueType() {
+
+		ColumnType columnType = getMapValueType();
+
+		if (columnType == null) {
+			throw new IllegalStateException("Type has no map value type");
+		}
+
+		return columnType;
+	}
+
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolver.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolver.java
new file mode 100644
index 000000000..932a1d30a
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolver.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Resolves {@link ColumnType} for properties, {@link TypeInformation}, and {@code values}.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+public interface ColumnTypeResolver {
+
+	/**
+	 * Resolve a {@link CassandraColumnType} from a {@link CassandraPersistentProperty}. Considers
+	 * {@link CassandraType}-annotated properties.
+	 *
+	 * @param property must not be {@literal null}.
+	 * @return
+	 * @see CassandraType
+	 * @see CassandraPersistentProperty
+	 */
+	default CassandraColumnType resolve(CassandraPersistentProperty property) {
+
+		Assert.notNull(property, "Property must not be null");
+
+		if (property.isAnnotationPresent(CassandraType.class)) {
+			return resolve(property.getRequiredAnnotation(CassandraType.class));
+		}
+
+		return resolve(property.getTypeInformation());
+	}
+
+	/**
+	 * Resolve a {@link CassandraColumnType} from {@link TypeInformation}. Considers potentially registered custom
+	 * converters and simple type rules.
+	 *
+	 * @param typeInformation must not be {@literal null}.
+	 * @return
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder
+	 * @see CassandraCustomConversions
+	 */
+	CassandraColumnType resolve(TypeInformation<?> typeInformation);
+
+	/**
+	 * Resolve a {@link CassandraColumnType} from a {@link CassandraType} annotation.
+	 *
+	 * @param annotation must not be {@literal null}.
+	 * @return
+	 * @see CassandraType
+	 * @see CassandraPersistentProperty
+	 */
+	CassandraColumnType resolve(CassandraType annotation);
+
+	/**
+	 * Resolve a {@link ColumnType} from a {@code value}. Considers potentially registered custom converters and simple
+	 * type rules.
+	 *
+	 * @param value
+	 * @return
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder
+	 * @see CassandraCustomConversions
+	 */
+	ColumnType resolve(@Nullable Object value);
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultCassandraColumnType.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultCassandraColumnType.java
new file mode 100644
index 000000000..e9cbf1c7c
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultCassandraColumnType.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import java.util.stream.Collectors;
+
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+
+/**
+ * Default {@link CassandraColumnType} implementation.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class DefaultCassandraColumnType extends DefaultColumnType implements CassandraColumnType {
+
+	private final DataType dataType;
+
+	DefaultCassandraColumnType(Class<?> type, DataType dataType, ColumnType... parameters) {
+		this(ClassTypeInformation.from(type), dataType, parameters);
+	}
+
+	DefaultCassandraColumnType(TypeInformation<?> typeInformation, DataType dataType, ColumnType... parameters) {
+		super(typeInformation, parameters);
+
+		this.dataType = dataType;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.CassandraColumnType#getDataType()
+	 */
+	public DataType getDataType() {
+		return dataType;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.DefaultColumnType#toString()
+	 */
+	@Override
+	public String toString() {
+
+		StringBuilder builder = new StringBuilder();
+
+		if (isTupleType()) {
+			builder.append("Tuple: ");
+		}
+
+		if (isUserDefinedType()) {
+			builder.append("UDT: ").append(((UserDefinedType) getDataType()).getName());
+		}
+
+		builder.append(getType().getName()).append(" [").append(getDataType()).append("]");
+
+		if (getParameters().isEmpty()) {
+			return builder.toString();
+		}
+
+		builder.append("<").append(getParameters().stream().map(Object::toString).collect(Collectors.toList())).append(">");
+
+		return builder.toString();
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnType.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnType.java
new file mode 100644
index 000000000..7c0291d48
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnType.java
@@ -0,0 +1,128 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+
+/**
+ * Default {@link ColumnType} implementation.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class DefaultColumnType implements ColumnType {
+
+	public static final DefaultColumnType OBJECT = new DefaultColumnType(ClassTypeInformation.OBJECT);
+
+	private final TypeInformation<?> typeInformation;
+	private final List<ColumnType> parameters;
+
+	DefaultColumnType(TypeInformation<?> typeInformation, ColumnType... parameters) {
+		this.typeInformation = typeInformation;
+		this.parameters = Arrays.asList(parameters);
+	}
+
+	List<ColumnType> getParameters() {
+		return parameters;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#getType()
+	 */
+	@Override
+	public Class<?> getType() {
+		return typeInformation.getType();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#isCollectionLike()
+	 */
+	@Override
+	public boolean isCollectionLike() {
+		return typeInformation.isCollectionLike();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#isList()
+	 */
+	@Override
+	public boolean isList() {
+		return List.class.isAssignableFrom(typeInformation.getType());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#isSet()
+	 */
+	@Override
+	public boolean isSet() {
+		return Set.class.isAssignableFrom(typeInformation.getType());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#isMap()
+	 */
+	@Override
+	public boolean isMap() {
+		return typeInformation.isMap();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#getComponentType()
+	 */
+	@Nullable
+	@Override
+	public ColumnType getComponentType() {
+		return !parameters.isEmpty() ? parameters.get(0) : null;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnType#getMapValueType()
+	 */
+	@Nullable
+	@Override
+	public ColumnType getMapValueType() {
+		return parameters.size() > 1 ? parameters.get(1) : null;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+
+		if (parameters.isEmpty()) {
+			return getType().getName();
+		}
+
+		return String.format("%s<%s>", getType().getName(),
+				parameters.stream().map(Object::toString).collect(Collectors.toList()));
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
new file mode 100644
index 000000000..04fe301c8
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
@@ -0,0 +1,498 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.StreamSupport;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
+import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.mapping.MappingException;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.data.UdtValue;
+import com.datastax.oss.driver.api.core.detach.AttachmentPoint;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.ListType;
+import com.datastax.oss.driver.api.core.type.MapType;
+import com.datastax.oss.driver.api.core.type.SetType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+
+/**
+ * Default {@link ColumnTypeResolver} implementation backed by {@link CustomConversions} and {@link CodecRegistry}.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class DefaultColumnTypeResolver implements ColumnTypeResolver {
+
+	private final CassandraMappingContext mappingContext;
+	private final UserTypeResolver userTypeResolver;
+	private final CodecRegistry codecRegistry;
+	private CustomConversions customConversions;
+
+	public DefaultColumnTypeResolver(CassandraMappingContext mappingContext) {
+		this.mappingContext = mappingContext;
+		this.userTypeResolver = mappingContext.getUserTypeResolver();
+		this.codecRegistry = mappingContext.getCodecRegistry();
+		this.customConversions = mappingContext.getCustomConversions();
+	}
+
+	public void setCustomConversions(CustomConversions customConversions) {
+		this.customConversions = customConversions;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnTypeResolver#resolve(org.springframework.data.util.TypeInformation)
+	 */
+	@Override
+	public CassandraColumnType resolve(TypeInformation<?> typeInformation) {
+
+		Optional<Class<?>> writeTarget = customConversions.getCustomWriteTarget(typeInformation.getType());
+
+		return writeTarget.map(it -> {
+			return createCassandraTypeDescriptor(tryResolve(it), ClassTypeInformation.from(it));
+		}).orElseGet(() -> {
+
+			if (typeInformation.getType().isEnum()) {
+				return ColumnType.create(String.class, DataTypes.TEXT);
+			}
+
+			return createCassandraTypeDescriptor(typeInformation);
+		});
+	}
+
+	private DataType tryResolve(Class<?> type) {
+
+		if (TupleValue.class.isAssignableFrom(type)) {
+			return DataTypes.tupleOf();
+		}
+
+		if (UdtValue.class.isAssignableFrom(type)) {
+			return UnknownUserDefinedType.INSTANCE;
+		}
+
+		return codecRegistry.codecFor(type).getCqlType();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnTypeResolver#resolve(org.springframework.data.cassandra.core.mapping.CassandraType)
+	 */
+	@Override
+	public CassandraColumnType resolve(CassandraType annotation) {
+
+		CassandraType.Name type = annotation.type();
+
+		switch (type) {
+			case MAP:
+				assertTypeArguments(annotation.typeArguments().length, 2);
+
+				CassandraColumnType keyType = createCassandraTypeDescriptor(
+						CassandraSimpleTypeHolder.getDataTypeFor(annotation.typeArguments()[0]));
+				CassandraColumnType valueType = createCassandraTypeDescriptor(
+						CassandraSimpleTypeHolder.getDataTypeFor(annotation.typeArguments()[1]));
+
+				return ColumnType.mapOf(keyType, valueType);
+
+			case LIST:
+			case SET:
+				assertTypeArguments(annotation.typeArguments().length, 1);
+
+				DataType componentType = annotation.typeArguments()[0] == CassandraType.Name.UDT
+						? getUserType(annotation.userTypeName())
+						: CassandraSimpleTypeHolder.getDataTypeFor(annotation.typeArguments()[0]);
+
+				if (type == CassandraType.Name.SET) {
+					return ColumnType.setOf(createCassandraTypeDescriptor(componentType));
+				}
+
+				return ColumnType.listOf(createCassandraTypeDescriptor(componentType));
+
+			case TUPLE:
+
+				DataType[] dataTypes = Arrays.stream(annotation.typeArguments()).map(CassandraSimpleTypeHolder::getDataTypeFor)
+						.toArray(DataType[]::new);
+
+				return ColumnType.tupleOf(DataTypes.tupleOf(dataTypes));
+			case UDT:
+
+				return createCassandraTypeDescriptor(getUserType(annotation.userTypeName()));
+			default:
+				return createCassandraTypeDescriptor(CassandraSimpleTypeHolder.getDataTypeFor(type));
+		}
+	}
+
+	/*
+	 *  (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.ColumnTypeResolver#resolve(java.lang.Object)
+	 */
+	@Override
+	public ColumnType resolve(@Nullable Object value) {
+
+		if (value != null) {
+
+			ClassTypeInformation<?> typeInformation = ClassTypeInformation.from(value.getClass());
+
+			Optional<Class<?>> writeTarget = customConversions.getCustomWriteTarget(typeInformation.getType());
+
+			return writeTarget.map(it -> {
+				return (ColumnType) createCassandraTypeDescriptor(tryResolve(it), typeInformation);
+			}).orElseGet(() -> {
+
+				if (typeInformation.getType().isEnum()) {
+					return ColumnType.create(String.class, DataTypes.TEXT);
+				}
+
+				if (value instanceof Map) {
+					return ColumnType.mapOf(DefaultColumnType.OBJECT, DefaultColumnType.OBJECT);
+				}
+
+				if (value instanceof List) {
+					return ColumnType.listOf(DefaultColumnType.OBJECT);
+				}
+
+				if (value instanceof Set) {
+					return ColumnType.listOf(DefaultColumnType.OBJECT);
+				}
+
+				if (value instanceof UdtValue) {
+					return ColumnType.udtOf(((UdtValue) value).getType());
+				}
+
+				if (value instanceof TupleValue) {
+					return ColumnType.tupleOf(((TupleValue) value).getType());
+				}
+
+				BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(typeInformation);
+
+				if (persistentEntity != null) {
+
+					if (persistentEntity.isUserDefinedType() || persistentEntity.isTupleType()) {
+						return resolve(persistentEntity.getTypeInformation());
+					}
+				}
+
+				return ColumnType.create(typeInformation.getType());
+			});
+		}
+
+		return DefaultColumnType.OBJECT;
+	}
+
+	private CassandraColumnType createCassandraTypeDescriptor(DataType dataType) {
+
+		GenericType<Object> javaType = codecRegistry.codecFor(dataType).getJavaType();
+		return ColumnType.create(javaType.getRawType(), dataType);
+	}
+
+	private CassandraColumnType createCassandraTypeDescriptor(DataType dataType, TypeInformation<?> typeInformation) {
+
+		if (typeInformation.isCollectionLike() || typeInformation.isMap()) {
+
+			if (dataType instanceof ListType) {
+
+				TypeInformation<?> component = typeInformation.getComponentType();
+				DataType elementType = ((ListType) dataType).getElementType();
+
+				if (component != null) {
+					return ColumnType.listOf(createCassandraTypeDescriptor(elementType, component));
+				}
+
+				Class<?> componentType = resolveToJavaType(elementType);
+				return ColumnType.listOf(ColumnType.create(componentType, elementType));
+			}
+
+			if (dataType instanceof SetType) {
+
+				TypeInformation<?> component = typeInformation.getComponentType();
+				DataType elementType = ((SetType) dataType).getElementType();
+
+				if (component != null) {
+					return ColumnType.setOf(createCassandraTypeDescriptor(elementType, component));
+				}
+
+				Class<?> componentType = resolveToJavaType(elementType);
+				return ColumnType.setOf(ColumnType.create(componentType, elementType));
+			}
+
+			if (dataType instanceof MapType) {
+
+				TypeInformation<?> mapKeyType = typeInformation.getComponentType();
+				TypeInformation<?> mapValueType = typeInformation.getMapValueType();
+
+				MapType mapType = (MapType) dataType;
+
+				CassandraColumnType keyDescriptor = null;
+				CassandraColumnType valueDescriptor = null;
+
+				if (mapKeyType != null) {
+					keyDescriptor = createCassandraTypeDescriptor(mapType.getKeyType(), mapKeyType);
+				}
+
+				if (mapValueType != null) {
+					valueDescriptor = createCassandraTypeDescriptor(mapType.getValueType(), mapValueType);
+				}
+
+				if (keyDescriptor == null) {
+					keyDescriptor = ColumnType.create(resolveToJavaType(mapType.getKeyType()), mapType.getKeyType());
+				}
+
+				if (valueDescriptor == null) {
+					valueDescriptor = ColumnType.create(resolveToJavaType(mapType.getValueType()), mapType.getValueType());
+				}
+
+				return ColumnType.mapOf(keyDescriptor, valueDescriptor);
+			}
+		}
+
+		return new DefaultCassandraColumnType(typeInformation, dataType);
+	}
+
+	private CassandraColumnType createCassandraTypeDescriptor(TypeInformation<?> typeInformation) {
+
+		if (List.class.isAssignableFrom(typeInformation.getType())) {
+			return ColumnType.listOf(resolve(typeInformation.getRequiredComponentType()));
+		}
+
+		if (Set.class.isAssignableFrom(typeInformation.getType())) {
+			return ColumnType.setOf(resolve(typeInformation.getRequiredComponentType()));
+		}
+
+		if (typeInformation.isMap()) {
+			return ColumnType.mapOf(resolve(typeInformation.getRequiredComponentType()),
+					resolve(typeInformation.getRequiredMapValueType()));
+		}
+
+		BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(typeInformation);
+
+		if (persistentEntity != null) {
+
+			if (persistentEntity.isUserDefinedType()) {
+				return new DefaultCassandraColumnType(typeInformation, persistentEntity.getUserType());
+			}
+
+			if (persistentEntity.isTupleType()) {
+
+				DataType[] componentTypes = StreamSupport.stream(persistentEntity.spliterator(), false) //
+						.map(this::resolve) //
+						.map(CassandraColumnType::getDataType) //
+						.toArray(DataType[]::new);
+
+				return new DefaultCassandraColumnType(typeInformation, DataTypes.tupleOf(componentTypes));
+			}
+
+			return new UnresolvableCassandraType(typeInformation);
+		}
+
+		return new DefaultCassandraColumnType(typeInformation, tryResolve(typeInformation.getType()));
+	}
+
+	private Class<?> resolveToJavaType(DataType dataType) {
+		TypeCodec<Object> codec = codecRegistry.codecFor(dataType);
+		return codec.getJavaType().getRawType();
+	}
+
+	private DataType getUserType(String userTypeName) {
+
+		UserDefinedType type = userTypeResolver.resolveType(CqlIdentifier.fromCql(userTypeName));
+
+		if (type == null) {
+			throw new IllegalArgumentException(String.format("Cannot resolve UserDefinedType for [%s]", userTypeName));
+		}
+
+		return type;
+	}
+
+	private void assertTypeArguments(int args, int expected) {
+
+		if (args != expected) {
+			throw new InvalidDataAccessApiUsageException(
+					String.format("Expected [%d] type arguments actual was [%d]", expected, args));
+		}
+	}
+
+	enum UnknownUserDefinedType implements com.datastax.oss.driver.api.core.type.UserDefinedType {
+		INSTANCE;
+
+		UnknownUserDefinedType() {}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#getKeyspace()
+		 */
+		@Override
+		public CqlIdentifier getKeyspace() {
+			return null;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#getName()
+		 */
+		@Override
+		public CqlIdentifier getName() {
+			return CqlIdentifier.fromCql("unknown");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#isFrozen()
+		 */
+		@Override
+		public boolean isFrozen() {
+			return false;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#getFieldNames()
+		 */
+		@Override
+		public List<CqlIdentifier> getFieldNames() {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#firstIndexOf(com.datastax.oss.driver.api.core.CqlIdentifier)
+		 */
+		@Override
+		public int firstIndexOf(CqlIdentifier id) {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#firstIndexOf(java.lang.String)
+		 */
+		@Override
+		public int firstIndexOf(String name) {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#getFieldTypes()
+		 */
+		@Override
+		public List<DataType> getFieldTypes() {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#copy(boolean)
+		 */
+		@Override
+		public com.datastax.oss.driver.api.core.type.UserDefinedType copy(boolean newFrozen) {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#newValue()
+		 */
+		@Override
+		public UdtValue newValue() {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#newValue(java.lang.Object[])
+		 */
+		@Override
+		public UdtValue newValue(Object... fields) {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.type.UserDefinedType#getAttachmentPoint()
+		 */
+		@Override
+		public AttachmentPoint getAttachmentPoint() {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.detach.Detachable#isDetached()
+		 */
+		@Override
+		public boolean isDetached() {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see com.datastax.oss.driver.api.core.detach.Detachable#attach(com.datastax.oss.driver.api.core.detach.AttachmentPoint)
+		 */
+		@Override
+		public void attach(AttachmentPoint attachmentPoint) {
+			throw new UnsupportedOperationException(
+					"This implementation should only be used internally, this is likely a driver bug");
+		}
+	}
+
+	static class UnresolvableCassandraType extends DefaultCassandraColumnType {
+
+		public UnresolvableCassandraType(TypeInformation<?> type, ColumnType... parameters) {
+			super(type, null, parameters);
+		}
+
+		public UnresolvableCassandraType(Class<?> type, ColumnType... parameters) {
+			super(type, null, parameters);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.convert.DefaultCassandraColumnType#getDataType()
+		 */
+		@Override
+		public DataType getDataType() {
+			throw new MappingException(String.format("Cannot resolve DataType for %s", getType().getName()));
+		}
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
index fc38fad9a..d8c84b0ec 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
@@ -41,9 +41,9 @@ import org.springframework.data.cassandra.core.mapping.BasicMapId;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
-import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.cassandra.core.mapping.MapId;
 import org.springframework.data.cassandra.core.mapping.MapIdentifiable;
+import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.PreferredConstructor;
@@ -68,7 +68,6 @@ import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.data.TupleValue;
 import com.datastax.oss.driver.api.core.data.UdtValue;
-import com.datastax.oss.driver.api.core.type.DataType;
 import com.datastax.oss.driver.api.core.type.TupleType;
 import com.datastax.oss.driver.api.core.type.UserDefinedType;
 import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
@@ -96,6 +95,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 	private SpELContext spELContext;
 
+	private final DefaultColumnTypeResolver cassandraTypeResolver;
+
 	/**
 	 * Create a new {@link MappingCassandraConverter} with a {@link CassandraMappingContext}.
 	 */
@@ -106,8 +107,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		CassandraCustomConversions conversions = new CassandraCustomConversions(Collections.emptyList());
 
 		this.mappingContext = newDefaultMappingContext(conversions);
-		this.setCustomConversions(conversions);
 		this.spELContext = new SpELContext(RowReaderPropertyAccessor.INSTANCE);
+		this.cassandraTypeResolver = new DefaultColumnTypeResolver(mappingContext);
+		this.setCustomConversions(conversions);
 	}
 
 	/**
@@ -121,9 +123,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		Assert.notNull(mappingContext, "CassandraMappingContext must not be null");
 
-		this.setCustomConversions(mappingContext.getCustomConversions());
 		this.mappingContext = mappingContext;
 		this.spELContext = new SpELContext(RowReaderPropertyAccessor.INSTANCE);
+		this.cassandraTypeResolver = new DefaultColumnTypeResolver(mappingContext);
+		this.setCustomConversions(mappingContext.getCustomConversions());
 	}
 
 	private static ConversionService newConversionService() {
@@ -140,6 +143,15 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return mappingContext;
 	}
 
+	@Override
+	public void setCustomConversions(CustomConversions conversions) {
+		super.setCustomConversions(conversions);
+
+		if (this.cassandraTypeResolver != null) {
+			this.cassandraTypeResolver.setCustomConversions(conversions);
+		}
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
 	 */
@@ -172,6 +184,14 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return this.mappingContext;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.CassandraConverter#getColumnTypeResolver()
+	 */
+	@Override
+	public ColumnTypeResolver getColumnTypeResolver() {
+		return this.cassandraTypeResolver;
+	}
+
 	/**
 	 * Create a new {@link ConvertingPropertyAccessor} for the given {@link Object source} and
 	 * {@link CassandraPersistentEntity entity}.
@@ -344,17 +364,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return convertToColumnType(obj, ClassTypeInformation.from(obj.getClass()));
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.core.convert.CassandraConverter#convertToColumnType(java.lang.Object, org.springframework.data.util.TypeInformation)
-	 */
 	@Override
-	public Object convertToColumnType(Object value, TypeInformation<?> typeInformation) {
-
-		Assert.notNull(value, "Value must not be null");
-		Assert.notNull(typeInformation, "TypeInformation must not be null");
-
+	public Object convertToColumnType(Object value, ColumnType columnType) {
 		// noinspection ConstantConditions
-		return value.getClass().isArray() ? value : getWriteValue(value, typeInformation);
+		return value.getClass().isArray() ? value : getWriteValue(value, columnType);
 	}
 
 	@Override
@@ -362,6 +375,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		Assert.notNull(source, "Value must not be null");
 
+		// TODO
 		Class<?> beanClassLoaderClass = transformClassToBeanClassLoaderClass(source.getClass());
 
 		CassandraPersistentEntity<?> entity = getMappingContext().getRequiredPersistentEntity(beanClassLoaderClass);
@@ -397,22 +411,14 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		for (CassandraPersistentProperty property : entity) {
 
-			Object value = getWriteValue(property, accessor);
-
-			if (log.isDebugEnabled()) {
-				log.debug("doWithProperties Property.type {}, Property.value {}", property.getType().getName(), value);
-			}
-
-			if (!property.isWritable()) {
-				continue;
-			}
-
 			if (property.isCompositePrimaryKey()) {
 
 				if (log.isDebugEnabled()) {
 					log.debug("Property is a compositeKey");
 				}
 
+				Object value = accessor.getProperty(property);
+
 				if (value == null) {
 					continue;
 				}
@@ -424,6 +430,16 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 				continue;
 			}
 
+			Object value = getWriteValue(property, accessor);
+
+			if (log.isDebugEnabled()) {
+				log.debug("doWithProperties Property.type {}, Property.value {}", property.getType().getName(), value);
+			}
+
+			if (!property.isWritable()) {
+				continue;
+			}
+
 			if (log.isDebugEnabled()) {
 				log.debug("Adding map.entry [{}] - [{}]", property.getRequiredColumnName(), value);
 			}
@@ -508,7 +524,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 						"MapId contains references [%s] that is an unknown property of [%s]", entry.getKey(), entity.getName()));
 			}
 
-			Object writeValue = getWriteValue(entry.getValue(), persistentProperty.getTypeInformation());
+			Object writeValue = getWriteValue(entry.getValue(), cassandraTypeResolver.resolve(persistentProperty));
 
 			sink.put(persistentProperty.getRequiredColumnName(), writeValue);
 		}
@@ -623,38 +639,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 */
 	private Class<?> getTargetType(CassandraPersistentProperty property) {
 		return getCustomConversions().getCustomWriteTarget(property.getType())
-				.orElseGet(() -> determineTargetType(property));
-	}
-
-	private Class<?> determineTargetType(CassandraPersistentProperty property) {
-
-		if (property.isAnnotationPresent(CassandraType.class)) {
-			return getPropertyTargetType(property);
-		}
-
-		if (property.isCompositePrimaryKey() || property.isCollectionLike()
-				|| getCustomConversions().isSimpleType(property.getType())) {
-
-			return property.getType();
-		}
-
-		return getPropertyTargetType(property);
-	}
-
-	private Class<?> getPropertyTargetType(CassandraPersistentProperty property) {
-
-		if (property.isCollectionLike() || property.isMapLike()) {
-			return property.getType();
-		}
-
-		DataType dataType = getMappingContext().getDataType(property);
-		if (dataType instanceof UserDefinedType || dataType instanceof TupleType) {
-			return property.getType();
-		}
-
-		TypeCodec<Object> codec = getCodecRegistry().codecFor(dataType);
-
-		return codec.getJavaType().getRawType();
+				.orElseGet(() -> cassandraTypeResolver.resolve(property).getType());
 	}
 
 	/**
@@ -668,8 +653,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	@Nullable
 	@SuppressWarnings("unchecked")
 	private <T> T getWriteValue(CassandraPersistentProperty property, ConvertingPropertyAccessor propertyAccessor) {
-		return (T) getWriteValue(propertyAccessor.getProperty(property, (Class<T>) determineTargetType(property)),
-				property.getTypeInformation());
+
+		ColumnType cassandraTypeDescriptor = cassandraTypeResolver.resolve(property);
+
+		return (T) getWriteValue(propertyAccessor.getProperty(property, cassandraTypeDescriptor.getType()),
+				cassandraTypeDescriptor);
 	}
 
 	/**
@@ -677,18 +665,18 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 * conversion of collection element types.
 	 *
 	 * @param value the value, may be {@literal null}.
-	 * @param typeInformation the type information.
+	 * @param columnType the type information.
 	 * @return the return value, may be {@literal null}.
 	 */
 	@Nullable
 	@SuppressWarnings("unchecked")
-	private Object getWriteValue(@Nullable Object value, @Nullable TypeInformation<?> typeInformation) {
+	private Object getWriteValue(@Nullable Object value, ColumnType columnType) {
 
 		if (value == null) {
 			return null;
 		}
 
-		Class<?> requestedTargetType = typeInformation != null ? typeInformation.getType() : Object.class;
+		Class<?> requestedTargetType = columnType.getType();
 
 		if (getCustomConversions().hasCustomWriteTarget(value.getClass(), requestedTargetType)) {
 
@@ -711,35 +699,34 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return getPotentiallyConvertedSimpleValue(value, requestedTargetType);
 		}
 
-		TypeInformation<?> type = typeInformation != null ? typeInformation
-				: ClassTypeInformation.from((Class) value.getClass());
-
 		if (value instanceof Collection) {
-			return writeCollectionInternal((Collection<Object>) value, type);
+			return writeCollectionInternal((Collection<Object>) value, columnType);
 		}
 
 		if (value instanceof Map) {
-			return writeMapInternal((Map<Object, Object>) value, type);
+			return writeMapInternal((Map<Object, Object>) value, columnType);
 		}
 
+		TypeInformation<?> type = ClassTypeInformation.from((Class) value.getClass());
 		TypeInformation<?> actualType = type.getRequiredActualType();
-
 		BasicCassandraPersistentEntity<?> entity = getMappingContext().getPersistentEntity(actualType.getType());
 
-		if (entity != null) {
+		if (entity != null && columnType instanceof CassandraColumnType) {
 
-			if (entity.isTupleType()) {
+			CassandraColumnType cassandraType = (CassandraColumnType) columnType;
 
-				TupleValue tupleValue = getMappingContext().getTupleType(entity).newValue();
+			if (entity.isTupleType() && cassandraType.isTupleType()) {
+
+				TupleValue tupleValue = ((TupleType) cassandraType.getDataType()).newValue();
 
 				write(value, tupleValue, entity);
 
 				return tupleValue;
 			}
 
-			if (entity.isUserDefinedType()) {
+			if (entity.isUserDefinedType() && cassandraType.isUserDefinedType()) {
 
-				UdtValue udtValue = entity.getUserType().newValue();
+				UdtValue udtValue = ((UserDefinedType) cassandraType.getDataType()).newValue();
 
 				write(value, udtValue, entity);
 
@@ -750,28 +737,27 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return value;
 	}
 
-	private Object writeCollectionInternal(Collection<Object> source, TypeInformation<?> type) {
+	private Object writeCollectionInternal(Collection<Object> source, ColumnType type) {
 
 		Collection<Object> converted = CollectionFactory.createCollection(getCollectionType(type), source.size());
-
-		TypeInformation<?> actualType = type.getRequiredActualType();
+		ColumnType componentType = type.getRequiredComponentType();
 
 		for (Object element : source) {
-			converted.add(convertToColumnType(element, actualType));
+			converted.add(getWriteValue(element, componentType));
 		}
 
 		return converted;
 	}
 
-	private Object writeMapInternal(Map<Object, Object> source, TypeInformation<?> type) {
+	private Object writeMapInternal(Map<Object, Object> source, ColumnType type) {
 
 		Map<Object, Object> converted = CollectionFactory.createMap(type.getType(), source.size());
 
-		TypeInformation<?> keyType = type.getRequiredComponentType();
-		TypeInformation<?> valueType = type.getRequiredMapValueType();
+		ColumnType keyType = type.getRequiredComponentType();
+		ColumnType valueType = type.getRequiredMapValueType();
 
 		for (Entry<Object, Object> entry : source.entrySet()) {
-			converted.put(convertToColumnType(entry.getKey(), keyType), convertToColumnType(entry.getValue(), valueType));
+			converted.put(getWriteValue(entry.getKey(), keyType), getWriteValue(entry.getValue(), valueType));
 		}
 
 		return converted;
@@ -782,7 +768,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 *
 	 * @param value may be {@literal null}.
 	 * @param requestedTargetType must not be {@literal null}.
-	 * @see CassandraType
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraType
 	 */
 	@SuppressWarnings("unchecked")
 	@Nullable
@@ -840,24 +826,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return getConversionService().convert(value, target);
 	}
 
-	private static 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();
-		}
-
+	private static Class<?> getCollectionType(ColumnType type) {
 		return type.getType();
 	}
 
@@ -1101,4 +1070,5 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return parent.getSource();
 		}
 	}
+
 }
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/QueryMapper.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/QueryMapper.java
index 661274b6f..a83c20bf9 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/QueryMapper.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/QueryMapper.java
@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core.convert;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -43,8 +44,6 @@ import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.PropertyPath;
 import org.springframework.data.mapping.PropertyReferenceException;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.util.ClassTypeInformation;
-import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -132,10 +131,9 @@ public class QueryMapper {
 			Predicate predicate = criteriaDefinition.getPredicate();
 
 			Object value = predicate.getValue();
+			ColumnType typeDescriptor = getColumnType(field, value, predicate.getOperator());
 
-			Object mappedValue = value != null
-					? getConverter().convertToColumnType(value, getTypeInformation(field, value))
-					: null;
+			Object mappedValue = value != null ? getConverter().convertToColumnType(value, typeDescriptor) : null;
 
 			Predicate mappedPredicate = new Predicate(predicate.getOperator(), mappedValue);
 
@@ -255,8 +253,8 @@ public class QueryMapper {
 
 			field.getProperty().ifPresent(seen::add);
 
-			columns.getSelector(column).filter(selector -> selector instanceof ColumnSelector).ifPresent(
-					columnSelector -> getCqlIdentifier(column, field).ifPresent(columnNames::add));
+			columns.getSelector(column).filter(selector -> selector instanceof ColumnSelector)
+					.ifPresent(columnSelector -> getCqlIdentifier(column, field).ifPresent(columnNames::add));
 		}
 
 		if (columns.isEmpty()) {
@@ -331,22 +329,47 @@ public class QueryMapper {
 
 	Field createPropertyField(@Nullable CassandraPersistentEntity<?> entity, ColumnName key) {
 
-		return Optional.ofNullable(entity)
-				.<Field> map(e -> new MetadataBackedField(key, e, getMappingContext()))
+		return Optional.ofNullable(entity).<Field> map(e -> new MetadataBackedField(key, e, getMappingContext()))
 				.orElseGet(() -> new Field(key));
 	}
 
-	TypeInformation<?> getTypeInformation(Field field, @Nullable Object value) {
+	ColumnType getColumnType(Field field, @Nullable Object value, @Nullable CriteriaDefinition.Operator operator) {
+
+		ColumnType typeDescriptor;
+		if (field.getProperty().isPresent()) {
+			typeDescriptor = converter.getColumnTypeResolver().resolve(field.getProperty().get());
+		} else {
+
+			typeDescriptor = converter.getColumnTypeResolver().resolve(value);
+		}
 
 		if (field.getProperty().isPresent()) {
-			return field.getProperty().get().getTypeInformation();
+
+			CassandraPersistentProperty property = field.getProperty().get();
+
+			if (property.isCollectionLike()) {
+				if (operator == CriteriaDefinition.Operators.CONTAINS) {
+					typeDescriptor = typeDescriptor.getRequiredComponentType();
+				}
+			}
+
+			if (property.isMapLike()) {
+
+				if (operator == CriteriaDefinition.Operators.CONTAINS_KEY) {
+					typeDescriptor = typeDescriptor.getRequiredComponentType();
+				}
+
+				if (operator == CriteriaDefinition.Operators.CONTAINS) {
+					typeDescriptor = typeDescriptor.getRequiredMapValueType();
+				}
+			}
 		}
 
-		if (value != null) {
-			return ClassTypeInformation.from(value.getClass());
+		if (value instanceof Collection && operator == CriteriaDefinition.Operators.IN) {
+			typeDescriptor = ColumnType.listOf(typeDescriptor);
 		}
 
-		return ClassTypeInformation.OBJECT;
+		return typeDescriptor;
 	}
 
 	/**
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/UpdateMapper.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/UpdateMapper.java
index 407701c4f..1ee03f3f5 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/UpdateMapper.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/UpdateMapper.java
@@ -149,7 +149,7 @@ public class UpdateMapper extends QueryMapper {
 			return new SetAtKeyOp(field.getMappedKey(), mappedKey, mappedValue);
 		}
 
-		TypeInformation<?> typeInformation = getTypeInformation(field, rawValue);
+		ColumnType descriptor = getColumnType(field, rawValue, null);
 
 		if (updateOp instanceof SetAtIndexOp) {
 
@@ -158,12 +158,12 @@ public class UpdateMapper extends QueryMapper {
 			Assert.state(op.getValue() != null,
 					() -> String.format("SetAtIndexOp for %s attempts to set null", field.getProperty()));
 
-			Object mappedValue = getConverter().convertToColumnType(op.getValue(), typeInformation);
+			Object mappedValue = getConverter().convertToColumnType(op.getValue(), descriptor);
 
 			return new SetAtIndexOp(field.getMappedKey(), op.getIndex(), mappedValue);
 		}
 
-		if (rawValue instanceof Collection && typeInformation.isCollectionLike()) {
+		if (rawValue instanceof Collection && descriptor.isCollectionLike()) {
 
 			Collection<?> collection = (Collection) rawValue;
 
@@ -180,7 +180,7 @@ public class UpdateMapper extends QueryMapper {
 			}
 		}
 
-		Object mappedValue = rawValue == null ? null : getConverter().convertToColumnType(rawValue, typeInformation);
+		Object mappedValue = rawValue == null ? null : getConverter().convertToColumnType(rawValue, descriptor);
 
 		return new SetOp(field.getMappedKey(), mappedValue);
 	}
@@ -188,8 +188,8 @@ public class UpdateMapper extends QueryMapper {
 	private AssignmentOp getMappedUpdateOperation(Field field, RemoveOp updateOp) {
 
 		Object value = updateOp.getValue();
-		TypeInformation<?> typeInformation = getTypeInformation(field, value);
-		Object mappedValue = getConverter().convertToColumnType(value, typeInformation);
+		ColumnType descriptor = getColumnType(field, value, null);
+		Object mappedValue = getConverter().convertToColumnType(value, descriptor);
 
 		return new RemoveOp(field.getMappedKey(), mappedValue);
 	}
@@ -198,8 +198,8 @@ public class UpdateMapper extends QueryMapper {
 	private AssignmentOp getMappedUpdateOperation(Field field, AddToOp updateOp) {
 
 		Iterable<Object> value = updateOp.getValue();
-		TypeInformation<?> typeInformation = getTypeInformation(field, value);
-		Collection<Object> mappedValue = (Collection) getConverter().convertToColumnType(value, typeInformation);
+		ColumnType descriptor = getColumnType(field, value, null);
+		Collection<Object> mappedValue = (Collection) getConverter().convertToColumnType(value, descriptor);
 
 		if (field.getProperty().isPresent()) {
 
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
index 29d0b2d0b..d6f3f77a4 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
@@ -284,7 +284,7 @@ public class CassandraMappingContext
 	}
 
 	@Nullable
-	protected UserTypeResolver getUserTypeResolver() {
+	public UserTypeResolver getUserTypeResolver() {
 		return this.userTypeResolver;
 	}
 
@@ -753,7 +753,6 @@ public class CassandraMappingContext
 
 		DataType keyType = getDataTypeWithUserTypeFactory(keyTypeInformation, dataTypeProvider, () -> {
 
-			keyTypeInformation.getType();
 			DataType type = CassandraSimpleTypeHolder.getDataTypeFor(keyTypeInformation.getType());
 
 			if (type != null) {
@@ -765,7 +764,6 @@ public class CassandraMappingContext
 
 		DataType valueType = getDataTypeWithUserTypeFactory(valueTypeInformation, dataTypeProvider, () -> {
 
-			valueTypeInformation.getType();
 			DataType type = CassandraSimpleTypeHolder.getDataTypeFor(valueTypeInformation.getType());
 
 			if (type != null) {
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java
index e9df23c44..74f36d6cc 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java
@@ -15,15 +15,14 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.Optional;
 import java.util.Set;
 
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
-import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
-import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -32,11 +31,6 @@ import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.type.DataType;
-import com.datastax.oss.driver.api.core.type.DataTypes;
-import com.datastax.oss.driver.api.core.type.ListType;
-import com.datastax.oss.driver.api.core.type.MapType;
-import com.datastax.oss.driver.api.core.type.SetType;
-import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
 import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
 
 /**
@@ -55,14 +49,11 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 
 	private final CassandraParameterAccessor delegate;
 
-	private final CodecRegistry codecRegistry;
-
 	ConvertingParameterAccessor(CassandraConverter converter, CassandraParameterAccessor delegate,
 			CodecRegistry codecRegistry) {
 
 		this.converter = converter;
 		this.delegate = delegate;
-		this.codecRegistry = codecRegistry;
 	}
 
 	/* (non-Javadoc)
@@ -103,7 +94,7 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 	 */
 	@Override
 	public Object getBindableValue(int index) {
-		return potentiallyConvert(index, this.delegate.getBindableValue(index));
+		return potentiallyConvert(index, this.delegate.getBindableValue(index), null);
 	}
 
 	/* (non-Javadoc)
@@ -164,107 +155,24 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 
 	@SuppressWarnings("unchecked")
 	@Nullable
-	private Object potentiallyConvert(int index, @Nullable Object bindableValue) {
+	Object potentiallyConvert(int index, @Nullable Object bindableValue, @Nullable CassandraPersistentProperty property) {
 
 		if (bindableValue == null) {
 			return null;
 		}
 
-		return this.converter.convertToColumnType(bindableValue, findTypeInformation(index, bindableValue, null));
-	}
-
-	@SuppressWarnings("unchecked")
-	@Nullable
-	private Object potentiallyConvert(int index, @Nullable Object bindableValue, CassandraPersistentProperty property) {
-
-		return (bindableValue == null ? null
-				: this.converter.convertToColumnType(bindableValue, findTypeInformation(index, bindableValue, property)));
-	}
-
-	private TypeInformation<?> findTypeInformation(int index, Object bindableValue,
-			@Nullable CassandraPersistentProperty property) {
-
-		if (this.delegate.findCassandraType(index) != null) {
-
-			TypeCodec<?> typeCodec = codecRegistry.codecFor(getDataType(index, property));
-
-			if (typeCodec.getJavaType().getType() instanceof Class<?>) {
-				return ClassTypeInformation.from((Class<?>) typeCodec.getJavaType().getType());
-			}
-
-			return ClassTypeInformation.from(typeCodec.getJavaType().getRawType());
-		}
-
-		if (property == null) {
-			return ClassTypeInformation.from(bindableValue.getClass());
-		}
-
-		return property.getTypeInformation();
-	}
-
-	/**
-	 * Return the {@link DataType} based on annotated parameters with {@link CassandraType}, the
-	 * {@link CassandraPersistentProperty} type or the declared parameter type.
-	 *
-	 * @param index index of parameter.
-	 * @param property {@link CassandraPersistentProperty}.
-	 * @return the {@link DataType}
-	 */
-	DataType getDataType(int index, @Nullable CassandraPersistentProperty property) {
-
 		CassandraType cassandraType = this.delegate.findCassandraType(index);
 
 		if (cassandraType != null) {
-			return CassandraSimpleTypeHolder.getDataTypeFor(cassandraType.type());
+			this.converter.convertToColumnType(bindableValue, converter.getColumnTypeResolver().resolve(cassandraType));
 		}
 
-		CassandraMappingContext mappingContext = converter.getMappingContext();
-		TypeInformation<?> typeInformation = ClassTypeInformation.from(getParameterType(index));
-
-		if (property == null) {
-			return mappingContext.getDataType(typeInformation.getType());
+		if (property != null && ((property.isCollectionLike() && bindableValue instanceof Collection)
+				|| (!property.isCollectionLike() && !(bindableValue instanceof Collection)))) {
+			return this.converter.convertToColumnType(bindableValue, converter.getColumnTypeResolver().resolve(property));
 		}
 
-		return getDataType(mappingContext, typeInformation, property);
-	}
-
-	private DataType getDataType(CassandraMappingContext mappingContext, TypeInformation<?> typeInformation,
-			CassandraPersistentProperty property) {
-
-		DataType dataType = mappingContext.getDataType(property);
-
-		if (property.isCollectionLike() && !typeInformation.isCollectionLike()) {
-			if (dataType instanceof ListType) {
-
-				ListType collectionType = (ListType) dataType;
-				return collectionType.getElementType();
-			}
-
-			if (dataType instanceof SetType) {
-
-				SetType collectionType = (SetType) dataType;
-				return collectionType.getElementType();
-			}
-		}
-
-		if (!property.isCollectionLike() && typeInformation.isCollectionLike()) {
-
-			if (typeInformation.isAssignableFrom(SET)) {
-				return DataTypes.setOf(dataType);
-			}
-
-			return DataTypes.listOf(dataType);
-		}
-
-		if (property.isMap()) {
-			if (dataType instanceof MapType) {
-
-				MapType collectionType = (MapType) dataType;
-				return collectionType.getKeyType();
-			}
-		}
-
-		return mappingContext.getDataType(property);
+		return this.converter.convertToColumnType(bindableValue, converter.getColumnTypeResolver().resolve(bindableValue));
 	}
 
 	/**
@@ -301,7 +209,7 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 		 */
 		@Nullable
 		public Object next() {
-			return potentiallyConvert(this.index++, this.delegate.next());
+			return potentiallyConvert(this.index++, this.delegate.next(), null);
 		}
 
 		/*
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolverUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolverUnitTests.java
new file mode 100644
index 000000000..e810f96dd
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/ColumnTypeResolverUnitTests.java
@@ -0,0 +1,219 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 org.springframework.data.cassandra.core.convert;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Test;
+
+import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.util.ClassTypeInformation;
+
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+
+/**
+ * Unit tests for {@link DefaultColumnTypeResolver}.
+ *
+ * @author Mark Paluch
+ */
+public class ColumnTypeResolverUnitTests {
+
+	CassandraMappingContext mappingContext = new CassandraMappingContext();
+	ColumnTypeResolver resolver = new DefaultColumnTypeResolver(mappingContext);
+
+	@Test // DATACASS-743
+	public void shouldResolveSimpleType() {
+
+		assertThat(resolver.resolve("foo").getType()).isEqualTo(String.class);
+		assertThat(resolver.resolve(ClassTypeInformation.from(String.class)).getType()).isEqualTo(String.class);
+		assertThat(resolver.resolve(ClassTypeInformation.from(String.class)).getDataType()).isEqualTo(DataTypes.TEXT);
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("name")).getType()).isEqualTo(String.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("name")).getDataType()).isEqualTo(DataTypes.TEXT);
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveEnumType() {
+
+		assertThat(resolver.resolve(MyEnum.INSTANCE).getType()).isEqualTo(String.class);
+		assertThat(resolver.resolve(ClassTypeInformation.from(MyEnum.class)).getType()).isEqualTo(String.class);
+		assertThat(resolver.resolve(ClassTypeInformation.from(MyEnum.class)).getDataType()).isEqualTo(DataTypes.TEXT);
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumAsString")).getType())
+				.isEqualTo(String.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumAsString")).getDataType())
+				.isEqualTo(DataTypes.TEXT);
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderCassandraType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumAsInt")).getType()).isEqualTo(Integer.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumAsInt")).getDataType())
+				.isEqualTo(DataTypes.INT);
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveSimpleListType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("names")).getType()).isEqualTo(List.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("names")).getDataType())
+				.isEqualTo(DataTypes.listOf(DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveListOfEnumType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumsAsString")).getType()).isEqualTo(List.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumsAsString")).getDataType())
+				.isEqualTo(DataTypes.listOf(DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderListWithCassandraType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumsAsInt")).getType()).isEqualTo(List.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumsAsInt")).getDataType())
+				.isEqualTo(DataTypes.listOf(DataTypes.INT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveSimpleSetType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("nameSet")).getType()).isEqualTo(Set.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("nameSet")).getDataType())
+				.isEqualTo(DataTypes.setOf(DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveSetOfEnumType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumSetAsString")).getType())
+				.isEqualTo(Set.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumSetAsString")).getDataType())
+				.isEqualTo(DataTypes.setOf(DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderSetWithCassandraType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumSetAsInt")).getType()).isEqualTo(Set.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumSetAsInt")).getDataType())
+				.isEqualTo(DataTypes.setOf(DataTypes.INT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveSimpleMapType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("nameMap")).getType()).isEqualTo(Map.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("nameMap")).getDataType())
+				.isEqualTo(DataTypes.mapOf(DataTypes.TEXT, DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldResolveMapOfEnumType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumMapAsString")).getType())
+				.isEqualTo(Map.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumMapAsString")).getDataType())
+				.isEqualTo(DataTypes.mapOf(DataTypes.TEXT, DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderMapWithCassandraType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumMapAsInt")).getType()).isEqualTo(Map.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("enumMapAsInt")).getDataType())
+				.isEqualTo(DataTypes.mapOf(DataTypes.INT, DataTypes.TEXT));
+	}
+
+	@Test // DATACASS-743
+	public void shouldReportEmptyTupleType() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("tupleValue")).getType())
+				.isEqualTo(TupleValue.class);
+		assertThat(resolver.resolve(entity.getRequiredPersistentProperty("tupleValue")).getDataType())
+				.isEqualTo(DataTypes.tupleOf());
+	}
+
+	static class Person {
+
+		String name;
+
+		MyEnum enumAsString;
+
+		@CassandraType(type = CassandraType.Name.INT) MyEnum enumAsInt;
+
+		List<String> names;
+
+		List<MyEnum> enumsAsString;
+
+		@CassandraType(type = CassandraType.Name.LIST, typeArguments = CassandraType.Name.INT) List<MyEnum> enumsAsInt;
+
+		Set<String> nameSet;
+
+		EnumSet<MyEnum> enumSetAsString;
+
+		@CassandraType(type = CassandraType.Name.SET, typeArguments = CassandraType.Name.INT) EnumSet<MyEnum> enumSetAsInt;
+
+		Map<String, String> nameMap;
+
+		Map<MyEnum, MyEnum> enumMapAsString;
+
+		@CassandraType(type = CassandraType.Name.MAP,
+				typeArguments = { CassandraType.Name.INT, CassandraType.Name.TEXT }) Map<MyEnum, MyEnum> enumMapAsInt;
+
+		TupleValue tupleValue;
+	}
+
+	enum MyEnum {
+		INSTANCE;
+	}
+}
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterUnitTests.java
index eb91eeb08..323b2d407 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterUnitTests.java
@@ -123,7 +123,7 @@ public class MappingCassandraConverterUnitTests {
 
 		mappingCassandraConverter.write(enumToOrdinalMapping, insert);
 
-		assertThat(getValues(insert)).contains(Integer.valueOf(Condition.USED.ordinal()));
+		assertThat(getValues(insert)).contains(Condition.USED.ordinal());
 	}
 
 	@Test // DATACASS-255, DATACASS-652
@@ -970,6 +970,45 @@ public class MappingCassandraConverterUnitTests {
 		assertThat(result.firstname).isEqualTo("fn");
 	}
 
+	@Test // DATACASS-743
+	public void shouldConsiderCassandraTypeOnList() {
+
+		TypeWithConvertedCollections value = new TypeWithConvertedCollections();
+		value.conditionList = Arrays.asList(Condition.MINT, Condition.USED);
+
+		Map<CqlIdentifier, Object> insert = new LinkedHashMap<>();
+
+		this.mappingCassandraConverter.write(value, insert);
+
+		assertThat(insert).containsEntry(CqlIdentifier.fromCql("conditionlist"), Arrays.asList(0, 1));
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderCassandraTypeOnSet() {
+
+		TypeWithConvertedCollections value = new TypeWithConvertedCollections();
+		value.conditionSet = new LinkedHashSet<>(Arrays.asList(Condition.MINT, Condition.USED));
+
+		Map<CqlIdentifier, Object> insert = new LinkedHashMap<>();
+
+		this.mappingCassandraConverter.write(value, insert);
+
+		assertThat(insert).containsEntry(CqlIdentifier.fromCql("conditionset"), new LinkedHashSet<>(Arrays.asList(0, 1)));
+	}
+
+	@Test // DATACASS-743
+	public void shouldConsiderCassandraTypeOnMap() {
+
+		TypeWithConvertedCollections value = new TypeWithConvertedCollections();
+		value.conditionMap = Collections.singletonMap(Condition.MINT, Condition.USED);
+
+		Map<CqlIdentifier, Object> insert = new LinkedHashMap<>();
+
+		this.mappingCassandraConverter.write(value, insert);
+
+		assertThat(insert).containsEntry(CqlIdentifier.fromCql("conditionmap"), Collections.singletonMap(0, 1));
+	}
+
 	private static List<Object> getValues(Map<CqlIdentifier, Object> statement) {
 		return new ArrayList<>(statement.values());
 	}
@@ -1194,6 +1233,18 @@ public class MappingCassandraConverterUnitTests {
 		@ReadOnlyProperty String computedName;
 	}
 
+	public static class TypeWithConvertedCollections {
+
+		@CassandraType(type = CassandraType.Name.LIST,
+				typeArguments = CassandraType.Name.INT) List<Condition> conditionList;
+
+		@CassandraType(type = CassandraType.Name.SET, typeArguments = CassandraType.Name.INT) Set<Condition> conditionSet;
+
+		@CassandraType(type = CassandraType.Name.MAP,
+				typeArguments = { CassandraType.Name.INT, CassandraType.Name.INT }) Map<Condition, Condition> conditionMap;
+
+	}
+
 	static class WithValue {
 
 		final @Id String id;
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/QueryMapperUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/QueryMapperUnitTests.java
index 364d495a6..1011247fe 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/QueryMapperUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/QueryMapperUnitTests.java
@@ -216,7 +216,7 @@ public class QueryMapperUnitTests {
 	@Test // DATACASS-343
 	public void shouldMapCollectionApplyingUdtValueCollectionConversion() {
 
-		Query query = Query.query(Criteria.where("addresses").in(new Address("21 Jump-Street")));
+		Query query = Query.query(Criteria.where("address").in(new Address("21 Jump-Street")));
 
 		Filter mappedObject = queryMapper.getMappedObject(query, persistentEntity);
 
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
index 0a29c0a23..7c537148d 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
@@ -31,10 +31,7 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
-import org.springframework.data.cassandra.core.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.oss.driver.api.core.type.DataTypes;
 import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
@@ -111,7 +108,6 @@ public class ConvertingParameterAccessorUnitTests {
 
 		when(mockParameterAccessor.iterator())
 				.thenReturn((Iterator) Collections.singletonList(Collections.singletonList(localDate)).iterator());
-		when(mockProperty.getTypeInformation()).thenReturn((TypeInformation) ClassTypeInformation.LIST);
 
 		PotentiallyConvertingIterator iterator = (PotentiallyConvertingIterator) convertingParameterAccessor.iterator();
 		Object converted = iterator.nextConverted(mockProperty);
@@ -122,16 +118,4 @@ public class ConvertingParameterAccessorUnitTests {
 
 		assertThat(list.get(0)).isInstanceOf(LocalDate.class);
 	}
-
-	@Test // DATACASS-7, DATACASS-506
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	public void shouldProvideTypeBasedOnPropertyType() {
-
-		when(mockProperty.getDataType()).thenReturn(DataTypes.TEXT);
-		when(mockProperty.isAnnotationPresent(CassandraType.class)).thenReturn(true);
-		when(mockProperty.getRequiredAnnotation(CassandraType.class)).thenReturn(mock(CassandraType.class));
-		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) String.class);
-
-		assertThat(convertingParameterAccessor.getDataType(0, mockProperty)).isEqualTo(DataTypes.TEXT);
-	}
 }