DATACASS-487 - Support map columns with mapped and converted key/value types.

We now support map columns that use mapped user-defined types and converted, non-primitive types in their keys and values. Map columns can be used for schema generation, column and key/value types are derived from the declared types by inspecting whether they are either UDTs or they can be converted by a custom registered converter.

class Supplier {

  @Id String id;

  Map<Manufacturer, List<Currency>> currencies;
}

@UserDefinedType
class Manufacturer {
  String name;
}

class UDTToCurrencyConverter implements Converter<UDTValue, Currency> {
 // …
}

class CurrencyToUDTConverter implements Converter<Currency, UDTValue> {
  // …
}

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

@@ -113,14 +113,8 @@ public class ColumnReader {
 		}
 
 		// Map
-		if (collectionTypes.size() == 2) {
-
-			DataType keyType = collectionTypes.get(0);
-			TypeCodec<Object> keyTypeCodec = codecRegistry.codecFor(keyType);
-
-			DataType valueType = collectionTypes.get(1);
-			TypeCodec<Object> valueTypeCodec = codecRegistry.codecFor(valueType);
-			return row.getMap(i, keyTypeCodec.getJavaType().getRawType(), valueTypeCodec.getJavaType().getRawType());
+		if (type.getName() == Name.MAP) {
+			return row.getObject(i);
 		}
 
 		throw new IllegalStateException("Unknown Collection type encountered. Valid collections are Set, List and Map.");

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.cassandra.core.convert;
 
+import lombok.AllArgsConstructor;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -22,8 +24,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
-import lombok.AllArgsConstructor;
-
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.context.ApplicationContext;
@@ -54,9 +56,6 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ObjectUtils;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.Row;
@@ -746,20 +745,15 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		TypeInformation<?> type = typeInformation != null ? typeInformation
 				: ClassTypeInformation.from((Class) value.getClass());
 
-		TypeInformation<?> actualType = type.getRequiredActualType();
-
 		if (value instanceof Collection) {
-
-			Collection<Object> original = (Collection<Object>) value;
-			Collection<Object> converted = CollectionFactory.createCollection(getCollectionType(type), original.size());
-
-			for (Object element : original) {
-				converted.add(convertToColumnType(element, actualType));
-			}
-
-			return converted;
+			return writeCollectionInternal((Collection<Object>) value, type);
 		}
 
+		if (value instanceof Map) {
+			return writeMapInternal((Map<Object, Object>) value, type);
+		}
+
+		TypeInformation<?> actualType = type.getRequiredActualType();
 		BasicCassandraPersistentEntity<?> entity = getMappingContext().getPersistentEntity(actualType.getType());
 
 		if (entity != null && entity.isUserDefinedType()) {
@@ -774,6 +768,34 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return value;
 	}
 
+	private Object writeCollectionInternal(Collection<Object> source, TypeInformation<?> type) {
+
+		Collection<Object> converted = CollectionFactory.createCollection(getCollectionType(type), source.size());
+		TypeInformation<?> actualType = type.getRequiredActualType();
+
+		for (Object element : source) {
+			converted.add(convertToColumnType(element, actualType));
+		}
+
+		return converted;
+	}
+
+	private Object writeMapInternal(Map<Object, Object> source, TypeInformation<?> type) {
+
+		Map<Object, Object> converted = CollectionFactory.createMap(type.getType(), source.size());
+
+		TypeInformation<?> keyType = type.getRequiredComponentType();
+		TypeInformation<?> valueType = type.getRequiredMapValueType();
+
+		for (Entry<Object, Object> entry : source.entrySet()) {
+
+			Object key = convertToColumnType(entry.getKey(), keyType);
+			converted.put(key, convertToColumnType(entry.getValue(), valueType));
+		}
+
+		return converted;
+	}
+
 	/**
 	 * Performs special enum handling or simply returns the value as is.
 	 *
@@ -876,46 +898,57 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return null;
 		}
 
-		if (getCustomConversions().hasCustomWriteTarget(property.getActualType()) && property.isCollectionLike()) {
+		return convertReadValue(value, property.getTypeInformation());
+	}
+
+	@Nullable
+	private Object convertReadValue(Object value, TypeInformation<?> typeInformation) {
+
+		if (getCustomConversions().hasCustomWriteTarget(typeInformation.getRequiredActualType().getType())
+				&& typeInformation.isCollectionLike()) {
 
 			if (value instanceof Collection) {
 
 				Collection<Object> original = (Collection<Object>) value;
 
-				Collection<Object> converted = CollectionFactory.createCollection(property.getType(), original.size());
+				Collection<Object> converted = CollectionFactory.createCollection(typeInformation.getType(), original.size());
 
 				for (Object element : original) {
-					converted.add(getConversionService().convert(element, property.getActualType()));
+					converted.add(getConversionService().convert(element, typeInformation.getRequiredActualType().getType()));
 				}
 
 				return converted;
 			}
 		}
 
-		if (property.isCollectionLike() && value instanceof Collection) {
-			return readCollectionOrArray(property.getTypeInformation(), (Collection<?>) value);
+		if (typeInformation.isCollectionLike() && value instanceof Collection) {
+			return readCollectionOrArrayInternal((Collection<?>) value, typeInformation);
+		}
+
+		if (typeInformation.isMap() && value instanceof Map) {
+			return readMapInternal((Map<Object, Object>) value, typeInformation);
 		}
 
 		BasicCassandraPersistentEntity<?> persistentEntity = getMappingContext()
-				.getPersistentEntity(property.getActualType());
+				.getPersistentEntity(typeInformation.getRequiredActualType());
 
 		if (persistentEntity != null && persistentEntity.isUserDefinedType() && value instanceof UDTValue) {
 			return readEntityFromUdt(persistentEntity, (UDTValue) value);
 		}
 
-		return getPotentiallyConvertedSimpleRead(value, property.getType());
+		return getPotentiallyConvertedSimpleRead(value, typeInformation.getType());
 	}
 
 	/**
 	 * Reads the given {@link Collection} into a collection of the given {@link TypeInformation}.
 	 *
-	 * @param targetType must not be {@literal null}.
 	 * @param sourceValue must not be {@literal null}.
+	 * @param targetType must not be {@literal null}.
 	 * @return the converted {@link Collection} or array, will never be {@literal null}.
 	 */
 	@Nullable
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private Object readCollectionOrArray(TypeInformation<?> targetType, Collection<?> sourceValue) {
+	private Object readCollectionOrArrayInternal(Collection<?> sourceValue, TypeInformation<?> targetType) {
 
 		Assert.notNull(targetType, "Target type must not be null!");
 
@@ -949,6 +982,44 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return getPotentiallyConvertedSimpleRead(items, targetType.getType());
 	}
 
+	/**
+	 * Reads the given {@link Map} into a map of the given {@link TypeInformation}.
+	 *
+	 * @param sourceValue must not be {@literal null}.
+	 * @param targetType must not be {@literal null}.
+	 * @return the converted {@link Collection} or array, will never be {@literal null}.
+	 */
+	@Nullable
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private Object readMapInternal(Map<Object, Object> sourceValue, TypeInformation<?> targetType) {
+
+		Assert.notNull(targetType, "Target type must not be null!");
+
+		TypeInformation<?> keyType = targetType.getComponentType();
+		TypeInformation<?> valueType = targetType.getMapValueType();
+
+		Class<?> rawKeyType = keyType != null ? keyType.getType() : null;
+		Map<Object, Object> map = CollectionFactory.createMap(targetType.getType(), rawKeyType, sourceValue.size());
+
+		if (sourceValue.isEmpty()) {
+			return map;
+		}
+
+		for (Entry<Object, Object> entry : sourceValue.entrySet()) {
+
+			Object key = entry.getKey();
+
+			if (rawKeyType != null && !rawKeyType.isAssignableFrom(key.getClass())) {
+				key = convertReadValue(key, keyType);
+			}
+
+			Object value = entry.getValue();
+			map.put(key, convertReadValue(value, valueType));
+		}
+
+		return map;
+	}
+
 	private TypeCodec<Object> getCodec(CassandraPersistentProperty property) {
 		return CodecRegistry.DEFAULT_INSTANCE.codecFor(mappingContext.getDataType(property));
 	}

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

@@ -130,7 +130,7 @@ public class UpdateMapper extends QueryMapper {
 			Optional<? extends TypeInformation<?>> typeInformation = field.getProperty()
 					.map(PersistentProperty::getTypeInformation);
 
-			Optional<TypeInformation<?>> keyType = typeInformation.map(TypeInformation::getActualType);
+			Optional<TypeInformation<?>> keyType = typeInformation.map(TypeInformation::getComponentType);
 			Optional<TypeInformation<?>> valueType = typeInformation.map(TypeInformation::getMapValueType);
 
 			Object mappedKey = keyType.map(typeInfo -> getConverter().convertToColumnType(op.getKey(), typeInfo))
@@ -219,7 +219,7 @@ public class UpdateMapper extends QueryMapper {
 		Optional<? extends TypeInformation<?>> typeInformation = field.getProperty()
 				.map(PersistentProperty::getTypeInformation);
 
-		Optional<TypeInformation<?>> keyType = typeInformation.map(TypeInformation::getActualType);
+		Optional<TypeInformation<?>> keyType = typeInformation.map(TypeInformation::getComponentType);
 		Optional<TypeInformation<?>> valueType = typeInformation.map(TypeInformation::getMapValueType);
 
 		Map<Object, Object> result = new LinkedHashMap<>(updateOp.getValue().size(), 1);

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java

@@ -29,6 +29,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
 
 import org.springframework.beans.BeansException;
@@ -90,7 +91,7 @@ public class CassandraMappingContext
 
 	private @Nullable ClassLoader beanClassLoader;
 
-	// useful caches
+	// caches
 	private final Map<CqlIdentifier, Set<CassandraPersistentEntity<?>>> entitySetsByTableName = new HashMap<>();
 	private final Set<BasicCassandraPersistentEntity<?>> userDefinedTypes = new HashSet<>();
 	private final Set<BasicCassandraPersistentEntity<?>> tableEntities = new HashSet<>();
@@ -542,7 +543,7 @@ public class CassandraMappingContext
 					throw new MappingException(String.format("User type [%s] not found", userTypeName));
 				}
 
-				DataType dataType = getUserDataType(property, userType);
+				DataType dataType = getUserDataType(property.getTypeInformation(), userType);
 
 				if (dataType != null) {
 					return dataType;
@@ -552,54 +553,84 @@ public class CassandraMappingContext
 			return property.getDataType();
 		}
 
-		BasicCassandraPersistentEntity<?> persistentEntity = getPersistentEntity(property.getActualType());
+		return getDataTypeWithUserTypeFactory(property.getTypeInformation(), dataTypeProvider, property::getDataType);
+	}
+
+	private DataType getDataTypeWithUserTypeFactory(TypeInformation<?> typeInformation, DataTypeProvider dataTypeProvider,
+			Supplier<DataType> fallback) {
+
+		BasicCassandraPersistentEntity<?> persistentEntity = getPersistentEntity(typeInformation.getRequiredActualType());
 
 		if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
 
-			DataType dataType = getUserDataType(property, dataTypeProvider.getDataType(persistentEntity));
+			DataType dataType = getUserDataType(typeInformation, dataTypeProvider.getDataType(persistentEntity));
 
 			if (dataType != null) {
 				return dataType;
 			}
 		}
 
-		return this.customConversions.getCustomWriteTarget(property.getType())
-				.map(CassandraSimpleTypeHolder::getDataTypeFor)
-				.orElseGet(() -> this.customConversions
-						.getCustomWriteTarget(property.getActualType())
-						.filter(it -> !property.isMapLike())
-						.map(it -> {
+		Optional<DataType> customWriteTarget = customConversions.getCustomWriteTarget(typeInformation.getType())
+				.map(CassandraSimpleTypeHolder::getDataTypeFor);
 
-							if (property.isCollectionLike()) {
-								if (List.class.isAssignableFrom(property.getType())) {
-									return DataType.list(getDataTypeFor(it));
-								}
+		DataType dataType = customWriteTarget.orElseGet(() -> {
 
-								if (Set.class.isAssignableFrom(property.getType())) {
-									return DataType.set(getDataTypeFor(it));
-								}
+			return customConversions.getCustomWriteTarget(typeInformation.getRequiredActualType().getType()) //
+					.filter(it -> !typeInformation.isMap()) //
+					.map(it -> {
+
+						if (typeInformation.isCollectionLike()) {
+
+							if (List.class.isAssignableFrom(typeInformation.getType())) {
+								return DataType.list(getDataTypeFor(it));
 							}
 
-							return getDataTypeFor(it);
-
-						}).orElseGet(() -> {
-
-							if (property.isMapLike()) {
-
-								Class<?> keyType = property.getComponentType();
-								Class<?> valueType = property.getMapValueType();
-
-								return DataType.map(getDataType(keyType, dataTypeProvider),
-										getDataType(valueType, dataTypeProvider));
+							if (Set.class.isAssignableFrom(typeInformation.getType())) {
+								return DataType.set(getDataTypeFor(it));
 							}
+						}
 
-							return property.getDataType();
+						return getDataTypeFor(it);
+					}).orElse(null);
+		});
 
-						}));
+		if (dataType != null) {
+			return dataType;
+		}
+
+		return typeInformation.isMap() ? getMapDataType(typeInformation, dataTypeProvider) : fallback.get();
+	}
+
+	private DataType getMapDataType(TypeInformation<?> typeInformation, DataTypeProvider dataTypeProvider) {
+
+		TypeInformation<?> keyTypeInformation = typeInformation.getComponentType();
+		TypeInformation<?> valueTypeInformation = typeInformation.getMapValueType();
+
+		DataType keyType = getDataTypeWithUserTypeFactory(keyTypeInformation, dataTypeProvider, () -> {
+
+			DataType type = getDataTypeFor(keyTypeInformation.getType());
+			if (type != null) {
+				return type;
+			}
+
+			throw new MappingException("Cannot resolve key type for " + typeInformation + ".");
+		});
+
+		DataType valueType = getDataTypeWithUserTypeFactory(valueTypeInformation, dataTypeProvider, () -> {
+
+			DataType type = getDataTypeFor(valueTypeInformation.getType());
+			if (type != null) {
+				return type;
+			}
+
+			throw new MappingException("Cannot resolve value type for " + typeInformation + ".");
+		});
+
+		return DataType.map(keyType, valueType);
 	}
 
 	@Nullable
-	private DataType getUserDataType(CassandraPersistentProperty property, DataType elementType) {
+	private DataType getUserDataType(TypeInformation<?> property, DataType elementType) {
 
 		if (property.isCollectionLike()) {
 
@@ -612,7 +643,7 @@ public class CassandraMappingContext
 			}
 		}
 
-		return !(property.isCollectionLike() || property.isMapLike()) ? elementType : null;
+		return !(property.isCollectionLike() || property.isMap()) ? elementType : null;
 	}
 
 	/**

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

@@ -25,6 +25,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.Currency;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Before;
@@ -116,12 +117,17 @@ public class MappingCassandraConverterUDTIntegrationTests extends AbstractSpring
 			session.execute("CREATE TYPE IF NOT EXISTS engine (manufacturer FROZEN<manufacturer>);");
 			session.execute("CREATE TABLE car (id text PRIMARY KEY, engine FROZEN<engine>);");
 
+			session.execute("DROP TABLE IF EXISTS supplier;");
+			session.execute(
+					"CREATE TABLE supplier (id text PRIMARY KEY, acceptedCurrencies frozen<map<manufacturer, list<currency>>>);");
+
 		} else {
 
 			session.execute("TRUNCATE addressbook;");
 			session.execute("TRUNCATE bank;");
 			session.execute("TRUNCATE money;");
 			session.execute("TRUNCATE car;");
+			session.execute("TRUNCATE supplier;");
 		}
 	}
 
@@ -379,8 +385,6 @@ public class MappingCassandraConverterUDTIntegrationTests extends AbstractSpring
 	@Test // DATACASS-172, DATACASS-400
 	public void shouldWriteNestedUdt() {
 
-		session.execute("INSERT INTO car (id, engine)  VALUES ('1',  {manufacturer: {name:'a good one'}});");
-
 		Engine engine = new Engine(new Manufacturer("a good one"));
 
 		Car car = new Car("1", engine);
@@ -392,6 +396,22 @@ public class MappingCassandraConverterUDTIntegrationTests extends AbstractSpring
 				.isEqualTo("INSERT INTO car (engine,id) VALUES ({manufacturer:{name:'a good one'}},'1');");
 	}
 
+	@Test // DATACASS-487
+	public void shouldReadUdtInMap() {
+
+		session.execute("INSERT INTO supplier (id,acceptedCurrencies) VALUES ('1',"
+				+ "{{name:'a good one'}:[{currency:'EUR'},{currency:'USD'}]});");
+
+		ResultSet resultSet = session.execute("SELECT * FROM supplier");
+		Supplier supplier = converter.read(Supplier.class, resultSet.one());
+
+		assertThat(supplier.getAcceptedCurrencies()).isNotEmpty();
+
+		List<Currency> currencies = supplier.getAcceptedCurrencies().get(new Manufacturer("a good one"));
+
+		assertThat(currencies).contains(Currency.getInstance("EUR"), Currency.getInstance("USD"));
+	}
+
 	@Table
 	@Getter
 	@AllArgsConstructor
@@ -435,12 +455,21 @@ public class MappingCassandraConverterUDTIntegrationTests extends AbstractSpring
 	}
 
 	@UserDefinedType
-	@Getter
+	@Data
 	@AllArgsConstructor
 	private static class Manufacturer {
 		String name;
 	}
 
+	@Table
+	@Data
+	@AllArgsConstructor
+	private static class Supplier {
+
+		@Id String id;
+		Map<Manufacturer, List<Currency>> acceptedCurrencies;
+	}
+
 	@Data
 	@Table
 	public static class AddressBook {

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

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 static org.mockito.Mockito.*;
+import static org.springframework.data.cassandra.test.util.RowMockUtil.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.UserDefinedType;
+import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
+import org.springframework.data.cassandra.support.UserTypeBuilder;
+import org.springframework.data.cassandra.test.util.RowMockUtil;
+
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.UDTValue;
+import com.datastax.driver.core.UserType;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+
+/**
+ * Unit tests for UDT through {@link MappingCassandraConverter}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.Silent.class) // there are some unused stubbings in RowMockUtil but they're used in other
+																					// tests
+public class MappingCassandraConverterUDTUnitTests {
+
+	@Rule public final ExpectedException expectedException = ExpectedException.none();
+	@Mock UserTypeResolver userTypeResolver;
+
+	UserType manufacturer = UserTypeBuilder.forName("manufacturer").withField("name", DataType.varchar()).build();
+	UserType currency = UserTypeBuilder.forName("mycurrency").withField("currency", DataType.varchar()).build();
+
+	Row rowMock;
+
+	CassandraMappingContext mappingContext;
+	MappingCassandraConverter mappingCassandraConverter;
+
+	@Before
+	public void setUp() {
+
+		mappingContext = new CassandraMappingContext();
+		mappingContext.setUserTypeResolver(userTypeResolver);
+
+		mappingCassandraConverter = new MappingCassandraConverter(mappingContext);
+		mappingCassandraConverter.afterPropertiesSet();
+
+		when(userTypeResolver.resolveType(CqlIdentifier.of("manufacturer"))).thenReturn(manufacturer);
+		when(userTypeResolver.resolveType(CqlIdentifier.of("currency"))).thenReturn(currency);
+	}
+
+	@Test // DATACASS-487
+	public void shouldReadMappedUdtInMap() {
+
+		UDTValue key = manufacturer.newValue().setString("name", "a good one");
+		UDTValue value1 = currency.newValue().setString("currency", "EUR");
+		UDTValue value2 = currency.newValue().setString("currency", "USD");
+
+		Map<UDTValue, List<UDTValue>> map = new HashMap<>();
+		map.put(key, Arrays.asList(value1, value2));
+
+		rowMock = RowMockUtil
+				.newRowMock(column("acceptedCurrencies", map, DataType.map(manufacturer, DataType.list(currency))));
+
+		Supplier supplier = mappingCassandraConverter.read(Supplier.class, rowMock);
+
+		assertThat(supplier.getAcceptedCurrencies()).isNotEmpty();
+
+		List<Currency> currencies = supplier.getAcceptedCurrencies().get(new Manufacturer("a good one"));
+		assertThat(currencies).contains(new Currency("EUR"), new Currency("USD"));
+	}
+
+	@Test // DATACASS-487
+	public void shouldWriteMappedUdtInMap() {
+
+		Map<Manufacturer, List<Currency>> currencies = Collections.singletonMap(new Manufacturer("a good one"),
+				Arrays.asList(new Currency("EUR"), new Currency("USD")));
+
+		Supplier supplier = new Supplier(currencies);
+		Insert insert = QueryBuilder.insertInto("table");
+		mappingCassandraConverter.write(supplier, insert);
+
+		assertThat(insert.toString()).contains("VALUES ({{name:'a good one'}:[{currency:'EUR'},{currency:'USD'}]}");
+	}
+
+	@UserDefinedType
+	@Data
+	@AllArgsConstructor
+	private static class Manufacturer {
+		String name;
+	}
+
+	@UserDefinedType
+	@Data
+	@AllArgsConstructor
+	private static class Currency {
+		String currency;
+	}
+
+	@Data
+	@AllArgsConstructor
+	private static class Supplier {
+		Map<Manufacturer, List<Currency>> acceptedCurrencies;
+	}
+}

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

@@ -33,15 +33,7 @@ import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 
 import org.junit.Before;
 import org.junit.Rule;
@@ -910,6 +902,46 @@ public class MappingCassandraConverterUnitTests {
 		assertThat(delete.toString()).isEqualTo("DELETE FROM foo WHERE first_name='first' AND lastname='last';");
 	}
 
+	@Test // DATACASS-487
+	public void shouldReadConvertedMap() {
+
+		LocalDate date1 = LocalDate.fromYearMonthDay(2018, 1, 1);
+		LocalDate date2 = LocalDate.fromYearMonthDay(2019, 1, 1);
+		Map<String, List<LocalDate>> times = Collections.singletonMap("Europe/Paris", Arrays.asList(date1, date2));
+		rowMock = RowMockUtil.newRowMock(
+				RowMockUtil.column("times", times, DataType.map(DataType.varchar(), DataType.list(DataType.date()))));
+
+		TypeWithConvertedMap converted = mappingCassandraConverter.read(TypeWithConvertedMap.class, rowMock);
+
+		assertThat(converted.times).containsKeys(ZoneId.of("Europe/Paris"));
+
+		List<java.time.LocalDate> convertedTimes = converted.times.get(ZoneId.of("Europe/Paris"));
+		assertThat(convertedTimes).hasSize(2).hasOnlyElementsOfType(java.time.LocalDate.class);
+	}
+
+	@Test // DATACASS-487
+	public void shouldWriteConvertedMap() {
+
+		java.time.LocalDate date1 = java.time.LocalDate.of(2018, 1, 1);
+		java.time.LocalDate date2 = java.time.LocalDate.of(2019, 1, 1);
+
+		TypeWithConvertedMap typeWithConvertedMap = new TypeWithConvertedMap();
+		typeWithConvertedMap.times = Collections.singletonMap(ZoneId.of("Europe/Paris"), Arrays.asList(date1, date2));
+
+		Insert insert = QueryBuilder.insertInto("table");
+
+		mappingCassandraConverter.write(typeWithConvertedMap, insert);
+
+		List<Object> values = getValues(insert);
+		assertThat(values).hasSize(1);
+		assertThat(values.get(0)).isInstanceOf(Map.class);
+
+		Map<String, List<LocalDate>> map = (Map) values.get(0);
+
+		assertThat(map).containsKey("Europe/Paris");
+		assertThat(map.get("Europe/Paris")).hasOnlyElementsOfType(LocalDate.class);
+	}
+
 	@SuppressWarnings("unchecked")
 	private static <T> List<T> getListValue(Insert statement) {
 
@@ -1163,4 +1195,12 @@ public class MappingCassandraConverterUnitTests {
 
 		ZoneId zoneId;
 	}
+
+	@Table
+	public static class TypeWithConvertedMap {
+
+		@PrimaryKey private String id;
+
+		Map<ZoneId, List<java.time.LocalDate>> times;
+	}
 }

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

@@ -25,6 +25,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Currency;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import org.junit.Before;
@@ -75,7 +76,7 @@ public class QueryMapperUnitTests {
 	UserType userType = UserTypeBuilder.forName("address").withField("street", DataType.varchar()).build();
 
 	@Before
-	public void before() throws Exception {
+	public void before() {
 
 		CassandraCustomConversions customConversions = new CassandraCustomConversions(
 				Collections.singletonList(CurrencyConverter.INSTANCE));
@@ -212,6 +213,34 @@ public class QueryMapperUnitTests {
 		assertThat(mappedCriteriaDefinition.getPredicate().getValue().toString()).isEqualTo("[{street:'21 Jump-Street'}]");
 	}
 
+	@Test // DATACASS-487
+	public void shouldMapUdtMapContainsKey() {
+
+		Query query = Query.query(Criteria.where("relocations").containsKey(new Address("21 Jump-Street")));
+
+		Filter mappedObject = queryMapper.getMappedObject(query, persistentEntity);
+
+		CriteriaDefinition mappedCriteriaDefinition = mappedObject.iterator().next();
+
+		assertThat(mappedCriteriaDefinition.getPredicate().getOperator()).isEqualTo(Operators.CONTAINS_KEY);
+		assertThat(mappedCriteriaDefinition.getPredicate().getValue()).isInstanceOf(UDTValue.class);
+		assertThat(mappedCriteriaDefinition.getPredicate().getValue().toString()).isEqualTo("{street:'21 Jump-Street'}");
+	}
+
+	@Test // DATACASS-487
+	public void shouldMapUdtMapContains() {
+
+		Query query = Query.query(Criteria.where("relocations").contains(new Address("21 Jump-Street")));
+
+		Filter mappedObject = queryMapper.getMappedObject(query, persistentEntity);
+
+		CriteriaDefinition mappedCriteriaDefinition = mappedObject.iterator().next();
+
+		assertThat(mappedCriteriaDefinition.getPredicate().getOperator()).isEqualTo(Operators.CONTAINS);
+		assertThat(mappedCriteriaDefinition.getPredicate().getValue()).isInstanceOf(UDTValue.class);
+		assertThat(mappedCriteriaDefinition.getPredicate().getValue().toString()).isEqualTo("{street:'21 Jump-Street'}");
+	}
+
 	@Test // DATACASS-343
 	public void shouldMapPropertyToColumnName() {
 
@@ -300,6 +329,7 @@ public class QueryMapperUnitTests {
 
 		Address address;
 		List<Address> addresses;
+		Map<Address, Address> relocations;
 		Currency currency;
 		State state;
 

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

@@ -16,6 +16,10 @@
 package org.springframework.data.cassandra.core.convert;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
 
 import java.util.Collections;
 import java.util.Currency;
@@ -29,11 +33,17 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.UserDefinedType;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.core.query.Update;
+import org.springframework.data.cassandra.support.UserTypeBuilder;
+
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.UserType;
 
 /**
  * Unit tests for {@link UpdateMapper}.
@@ -51,9 +61,10 @@ public class UpdateMapperUnitTests {
 	@Mock UserTypeResolver userTypeResolver;
 
 	Currency currency = Currency.getInstance("EUR");
+	UserType manufacturer = UserTypeBuilder.forName("manufacturer").withField("name", DataType.varchar()).build();
 
 	@Before
-	public void before() throws Exception {
+	public void before() {
 
 		CassandraCustomConversions customConversions = new CassandraCustomConversions(
 				Collections.singletonList(CurrencyConverter.INSTANCE));
@@ -68,6 +79,8 @@ public class UpdateMapperUnitTests {
 		updateMapper = new UpdateMapper(cassandraConverter);
 
 		persistentEntity = mappingContext.getRequiredPersistentEntity(Person.class);
+
+		when(userTypeResolver.resolveType(CqlIdentifier.of("manufacturer"))).thenReturn(manufacturer);
 	}
 
 	@Test // DATACASS-343
@@ -79,6 +92,18 @@ public class UpdateMapperUnitTests {
 		assertThat(update.toString()).isEqualTo("first_name = 'foo'");
 	}
 
+	@Test // DATACASS-487
+	public void shouldReplaceUdtMap() {
+
+		Manufacturer manufacturer = new Manufacturer("foobar");
+		Map<Manufacturer, Currency> map = Collections.singletonMap(manufacturer, currency);
+
+		Update update = updateMapper.getMappedObject(Update.empty().set("manufacturers", map), persistentEntity);
+
+		assertThat(update.getUpdateOperations()).hasSize(1);
+		assertThat(update.toString()).isEqualTo("manufacturers = {{name:'foobar'}:'Euro'}");
+	}
+
 	@Test // DATACASS-343
 	public void shouldCreateSetAtIndexUpdate() {
 
@@ -97,6 +122,17 @@ public class UpdateMapperUnitTests {
 		assertThat(update.toString()).isEqualTo("map['baz'] = 'Euro'");
 	}
 
+	@Test // DATACASS-487
+	public void shouldCreateSetAtUdtKeyUpdate() {
+
+		Manufacturer manufacturer = new Manufacturer("foobar");
+		Update update = updateMapper.getMappedObject(Update.empty().set("manufacturers").atKey(manufacturer).to(currency),
+				persistentEntity);
+
+		assertThat(update.getUpdateOperations()).hasSize(1);
+		assertThat(update.toString()).isEqualTo("manufacturers[{name:'foobar'}] = 'Euro'");
+	}
+
 	@Test // DATACASS-343
 	public void shouldAddToMap() {
 
@@ -106,6 +142,17 @@ public class UpdateMapperUnitTests {
 		assertThat(update.toString()).isEqualTo("map = map + {'foo':'Euro'}");
 	}
 
+	@Test // DATACASS-487
+	public void shouldAddUdtToMap() {
+
+		Manufacturer manufacturer = new Manufacturer("foobar");
+		Update update = updateMapper.getMappedObject(Update.empty().addTo("manufacturers").entry(manufacturer, currency),
+				persistentEntity);
+
+		assertThat(update.getUpdateOperations()).hasSize(1);
+		assertThat(update.toString()).isEqualTo("manufacturers = manufacturers + {{name:'foobar'}:'Euro'}");
+	}
+
 	@Test // DATACASS-343
 	public void shouldPrependAllToList() {
 
@@ -178,10 +225,18 @@ public class UpdateMapperUnitTests {
 		List<Currency> list;
 		@Column("set_col") Set<Currency> set;
 		Map<String, Currency> map;
+		Map<Manufacturer, Currency> manufacturers;
 		Currency currency;
 
 		Integer number;
 
 		@Column("first_name") String firstName;
 	}
+
+	@Data
+	@UserDefinedType
+	@AllArgsConstructor
+	static class Manufacturer {
+		String name;
+	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java

@@ -41,6 +41,7 @@ import org.springframework.data.cassandra.core.cql.keyspace.ColumnSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification.ColumnFunction;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
+import org.springframework.data.cassandra.support.UserTypeBuilder;
 import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.util.ClassTypeInformation;
@@ -71,7 +72,7 @@ public class CassandraMappingContextUnitTests {
 	}
 
 	@Test
-	public void testgetRequiredPersistentEntityOfTransientType() {
+	public void testGetRequiredPersistentEntityOfTransientType() {
 		this.mappingContext.getRequiredPersistentEntity(Transient.class);
 	}
 
@@ -81,6 +82,7 @@ public class CassandraMappingContextUnitTests {
 	public void testGetExistingPersistentEntityHappyPath() {
 
 		TableMetadata tableMetadata = mock(TableMetadata.class);
+
 		when(tableMetadata.getName()).thenReturn(X.class.getSimpleName().toLowerCase());
 
 		mappingContext.getRequiredPersistentEntity(X.class);
@@ -244,6 +246,25 @@ public class CassandraMappingContextUnitTests {
 		assertThat(kind.getOrdering()).isEqualTo(Ordering.ASCENDING);
 	}
 
+	@Test // DATACASS-487
+	public void shouldCreateTableForMappedAndConvertedColumn() {
+
+		UserType mappedudt = UserTypeBuilder.forName("mappedudt").withField("foo", DataType.ascii()).build();
+		mappingContext.setUserTypeResolver(typeName -> mappedudt);
+		mappingContext.setCustomConversions(
+				new CassandraCustomConversions(Collections.singletonList(HumanToStringConverter.INSTANCE)));
+
+		CassandraPersistentEntity<?> persistentEntity = mappingContext
+				.getRequiredPersistentEntity(WithMapOfMixedTypes.class);
+
+		CreateTableSpecification tableSpecification = mappingContext.getCreateTableSpecificationFor(persistentEntity);
+
+		assertThat(tableSpecification.getColumns()).hasSize(2);
+		ColumnSpecification column = tableSpecification.getColumns().get(1);
+
+		assertThat(column.getType().toString()).isEqualTo("map<frozen<mappedudt>, list<text>>");
+	}
+
 	@Table
 	private static class PrimaryKeyOnPropertyWithPrimaryKeyClass {
 
@@ -545,6 +566,7 @@ public class CassandraMappingContextUnitTests {
 	public void shouldNotRetainInvalidEntitiesInCache() {
 
 		TableMetadata tableMetadata = mock(TableMetadata.class);
+
 		when(tableMetadata.getName())
 				.thenReturn(InvalidEntityWithIdAndPrimaryKeyColumn.class.getSimpleName().toLowerCase());
 
@@ -563,32 +585,27 @@ public class CassandraMappingContextUnitTests {
 
 	@Table
 	private static class InvalidEntityWithIdAndPrimaryKeyColumn {
-
 		@Id String foo;
 		@PrimaryKeyColumn String bar;
 	}
 
 	@Table
 	static class EntityWithComplexPrimaryKeyColumn {
-
 		@PrimaryKeyColumn(ordinal = 0, type = PrimaryKeyType.PARTITIONED) Object complexObject;
 	}
 
 	@Table
 	static class EntityWithComplexId {
-
 		@Id Object complexObject;
 	}
 
 	@PrimaryKeyClass
 	static class PrimaryKeyClassWithComplexId {
-
 		@PrimaryKeyColumn(ordinal = 0, type = PrimaryKeyType.PARTITIONED) Object complexObject;
 	}
 
 	@Table
 	static class EntityWithPrimaryKeyClassWithComplexId {
-
 		@Id PrimaryKeyClassWithComplexId primaryKeyClassWithComplexId;
 	}
 
@@ -609,14 +626,17 @@ public class CassandraMappingContextUnitTests {
 
 	@Table
 	private static class WithUdt {
-
 		@Id String id;
-
 		@CassandraType(type = DataType.Name.UDT, userTypeName = "mappedudt") UDTValue udtValue;
-
 		@CassandraType(type = DataType.Name.UDT, userTypeName = "NestedType") Nested nested;
 	}
 
+	@Table
+	private static class WithMapOfMixedTypes {
+		@Id String id;
+		Map<MappedUdt, List<Human>> people;
+	}
+
 	enum HumanToStringConverter implements Converter<Human, String> {
 
 		INSTANCE;
@@ -629,16 +649,13 @@ public class CassandraMappingContextUnitTests {
 
 	@Table
 	private static class TypeWithCustomConvertedMap {
-
 		@Id String id;
 		Map<String, Collection<String>> stringMap;
-
 		@CassandraType(type = Name.ASCII) Map<String, Collection<String>> blobMap;
 	}
 
 	@Table
 	private static class TypeWithListOfHumans {
-
 		@Id String id;
 		List<Human> humans;
 	}
@@ -657,7 +674,6 @@ public class CassandraMappingContextUnitTests {
 	@UserDefinedType(value = "NestedType")
 	public static class Nested {
 		String s1;
-
 		@CassandraType(type = Name.UDT, userTypeName = "AnotherNestedType") AnotherNested anotherNested;
 	}
 

src/main/asciidoc/new-features.adoc

@@ -9,6 +9,7 @@ This chapter summarizes changes and new features for each release.
 * Template API extended with `count(…)` and `exists(…)` methods accepting `Query`.
 * <<cassandra.template.query.fluent-template-api,Fluent API>> for CRUD operations.
 * Cassandra Tuple support via `TupleValue`.
+* Support for `map` columns using User-defined/converted types.
 
 [[new-features.2-0-0]]
 == What's new in Spring Data for Apache Cassandra 2.0