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