DATACASS-456 - Refactor CassandraPersistentProperty to single-column.

CassandraPersistentProperty can now map only to a single column. Previously, CassandraPersistentProperty covered composite primary key mapping which required the mapping context in BasicCassandraPersistentProperty to resolve composite primary key entities. Composite primary keys are now resolved upfront without the need of indirection through CassandraPersistentProperty.

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

@@ -409,11 +409,10 @@ public class CassandraTemplate implements CassandraOperations {
 				String.format("Entity class [%s] has no primary key", entityClass.getName())));
 
 		if (idProperty.isCompositePrimaryKey()) {
-			String typeName = idProperty.getCompositePrimaryKeyEntity().getType().getName();
 
 			throw new IllegalArgumentException(
 					String.format("Entity class [%s] uses a composite primary key class [%s] which this method can't support",
-							entityClass.getName(), typeName));
+							entityClass.getName(), idProperty.getType().getName()));
 		}
 
 		Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());

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

@@ -207,8 +207,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		if (property.isCompositePrimaryKey()) {
 
-			CassandraPersistentEntity<?> keyEntity = property.getCompositePrimaryKeyEntity();
-
+			CassandraPersistentEntity<?> keyEntity = mappingContext.getRequiredPersistentEntity(property);
 			Optional<Object> optionalKey = propertyAccessor.getProperty(property);
 
 			if (!optionalKey.isPresent()) {
@@ -216,8 +215,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			// now recurse on using the key this time
-			optionalKey.ifPresent(key -> readProperties(property.getCompositePrimaryKeyEntity(), valueProvider,
-					getConvertingAccessor(key, keyEntity)));
+			optionalKey.ifPresent(key -> readProperties(keyEntity, valueProvider, getConvertingAccessor(key, keyEntity)));
 
 			// now that the key's properties have been populated, set the key property on the entity
 			propertyAccessor.setProperty(property, optionalKey);
@@ -338,12 +336,14 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			if (property.isCompositePrimaryKey()) {
+
 				if (log.isDebugEnabled()) {
 					log.debug("Property is a compositeKey");
 				}
 
-				writeMapFromWrapper(getConvertingAccessor(value.orElse(null), property.getCompositePrimaryKeyEntity()), insert,
-						property.getCompositePrimaryKeyEntity());
+				CassandraPersistentEntity<?> compositePrimaryKey = mappingContext.getRequiredPersistentEntity(property);
+				writeMapFromWrapper(getConvertingAccessor(value.orElse(null), compositePrimaryKey), insert,
+						compositePrimaryKey);
 
 				return;
 			}
@@ -368,12 +368,14 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			if (property.isCompositePrimaryKey()) {
+
 				if (log.isDebugEnabled()) {
 					log.debug("Property is a compositeKey");
 				}
 
-				writeInsertFromWrapper(getConvertingAccessor(value.orElse(null), property.getCompositePrimaryKeyEntity()),
-						insert, property.getCompositePrimaryKeyEntity());
+				CassandraPersistentEntity<?> compositePrimaryKey = mappingContext.getRequiredPersistentEntity(property);
+				writeInsertFromWrapper(getConvertingAccessor(value.orElse(null), compositePrimaryKey), insert,
+						compositePrimaryKey);
 
 				return;
 			}
@@ -402,8 +404,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			Optional<Object> value = getWriteValue(property, accessor);
 
 			if (property.isCompositePrimaryKey()) {
-				CassandraPersistentEntity<?> keyEntity = property.getCompositePrimaryKeyEntity();
-				writeUpdateFromWrapper(getConvertingAccessor(value.orElse(null), keyEntity), update, keyEntity);
+
+				CassandraPersistentEntity<?> compositePrimaryKey = mappingContext.getRequiredPersistentEntity(property);
+
+				writeUpdateFromWrapper(getConvertingAccessor(value.orElse(null), compositePrimaryKey), update,
+						compositePrimaryKey);
 				return;
 			}
 
@@ -447,6 +452,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		});
 	}
 
+	@SuppressWarnings("unchecked")
 	private Collection<Clause> getWhereClauses(Object source, CassandraPersistentEntity<?> entity) {
 
 		Assert.notNull(source, "Id source must not be null");
@@ -463,7 +469,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 			// FIXME: Generics
 			CassandraPersistentEntity<?> whereEntity = optionalCompositeIdProperty //
-					.map(CassandraPersistentProperty::getCompositePrimaryKeyEntity) //
+					.map(it -> (CassandraPersistentEntity<?>) mappingContext.getRequiredPersistentEntity(it)) //
 					.orElse((CassandraPersistentEntity) entity);
 
 			return getWhereClauses((MapId) id, whereEntity);
@@ -480,8 +486,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 					.orElseThrow(() -> new InvalidDataAccessApiUsageException(
 							String.format("Cannot use [%s] as composite Id for [%s]", id, entity.getName())));
 
-			return getWhereClauses(getConvertingAccessor(id, compositeIdProperty.getCompositePrimaryKeyEntity()),
-					compositeIdProperty.getCompositePrimaryKeyEntity());
+			CassandraPersistentEntity<?> compositePrimaryKey = mappingContext
+					.getRequiredPersistentEntity(compositeIdProperty);
+			return getWhereClauses(getConvertingAccessor(id, compositePrimaryKey), compositePrimaryKey);
 		}
 
 		Class<?> targetType = getTargetType(idProperty);
@@ -588,9 +595,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	protected <T> Class<T> transformClassToBeanClassLoaderClass(Class<T> entity) {
 		try {
 			return (Class<T>) ClassUtils.forName(entity.getName(), beanClassLoader);
-		} catch (ClassNotFoundException e) {
-			return entity;
-		} catch (LinkageError e) {
+		} catch (ClassNotFoundException | LinkageError e) {
 			return entity;
 		}
 	}
@@ -894,7 +899,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		Class<?> rawComponentType = componentType.map(TypeInformation::getType).orElse((Class) List.class);
 
 		collectionType = Collection.class.isAssignableFrom(collectionType) ? collectionType : List.class;
-		Collection<Object> items = targetType.getType().isArray() ? new ArrayList<Object>()
+		Collection<Object> items = targetType.getType().isArray() ? new ArrayList<>()
 				: CollectionFactory.createCollection(collectionType, rawComponentType, sourceValue.size());
 
 		if (sourceValue.isEmpty()) {

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

@@ -103,19 +103,6 @@ public class QueryMapper {
 		return this.mappingContext;
 	}
 
-	/**
-	 * Return {@link ColumnSelector}s for all columns of {@link CassandraPersistentEntity}.
-	 *
-	 * @param entity must not be {@literal null}.
-	 * @return {@link ColumnSelector}s for all columns of {@link CassandraPersistentEntity}.
-	 */
-	public List<Selector> getColumns(CassandraPersistentEntity<?> entity) {
-
-		return entity.getPersistentProperties() //
-				.flatMap(p -> p.getColumnNames().stream()).map(ColumnSelector::from) //
-				.collect(Collectors.toList());
-	}
-
 	/**
 	 * Map a {@link Filter} with a {@link CassandraPersistentEntity type hint}. Filter mapping translates property names
 	 * to column names and maps {@link Predicate} values to simple Cassandra values.
@@ -137,6 +124,11 @@ public class QueryMapper {
 
 			Predicate predicate = criteriaDefinition.getPredicate();
 
+			field.getProperty().filter(CassandraPersistentProperty::isCompositePrimaryKey).ifPresent(it -> {
+				throw new IllegalArgumentException(
+						"Cannot use composite primary key directly. Reference a property of the composite primary key");
+			});
+
 			Optional<Object> value = Optional.ofNullable(predicate.getValue());
 			TypeInformation<?> typeInformation = getTypeInformation(field, value);
 			Optional<Object> mappedValue = getConverter().convertToColumnType(value, typeInformation);
@@ -179,22 +171,26 @@ public class QueryMapper {
 		}
 
 		if (columns.isEmpty()) {
-
-			entity.doWithProperties((PropertyHandler<CassandraPersistentProperty>) property -> {
-
-				if (property.isCompositePrimaryKey()) {
-					for (CqlIdentifier cqlIdentifier : property.getColumnNames()) {
-						selectors.add(ColumnSelector.from(cqlIdentifier.toCql()));
-					}
-				} else {
-					selectors.add(ColumnSelector.from(property.getColumnName().toCql()));
-				}
-			});
+			addColumns(entity, selectors);
 		}
 
 		return selectors;
 	}
 
+	private void addColumns(CassandraPersistentEntity<?> entity, List<Selector> selectors) {
+
+		entity.doWithProperties((PropertyHandler<CassandraPersistentProperty>) property -> {
+
+			if (property.isCompositePrimaryKey()) {
+
+				CassandraPersistentEntity<?> primaryKeyEntity = mappingContext.getRequiredPersistentEntity(property);
+				addColumns(primaryKeyEntity, selectors);
+			} else {
+				selectors.add(ColumnSelector.from(property.getColumnName().toCql()));
+			}
+		});
+	}
+
 	private Selector getMappedSelector(Selector selector, CqlIdentifier cqlIdentifier) {
 
 		if (selector instanceof ColumnSelector) {
@@ -311,7 +307,14 @@ public class QueryMapper {
 
 		try {
 			if (field.getProperty().isPresent()) {
-				return field.getProperty().map(CassandraPersistentProperty::getColumnName);
+				return field.getProperty().map(cassandraPersistentProperty -> {
+
+					if (cassandraPersistentProperty.isCompositePrimaryKey()) {
+						throw new IllegalArgumentException(
+								"Cannot use composite primary key directly. Reference a property of the composite primary key");
+					}
+					return cassandraPersistentProperty.getColumnName();
+				});
 			}
 
 			if (column.getColumnName().isPresent()) {

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

@@ -17,9 +17,7 @@ package org.springframework.data.cassandra.core.mapping;
 
 import static org.springframework.data.cql.core.CqlIdentifier.*;
 
-import java.util.ArrayList;
 import java.util.Comparator;
-import java.util.List;
 import java.util.Optional;
 
 import org.springframework.beans.BeansException;
@@ -32,7 +30,6 @@ import org.springframework.data.cql.core.CqlIdentifier;
 import org.springframework.data.cql.support.exception.UnsupportedCassandraOperationException;
 import org.springframework.data.mapping.Association;
 import org.springframework.data.mapping.AssociationHandler;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.BasicPersistentEntity;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.util.TypeInformation;
@@ -60,8 +57,6 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 
 	private CassandraPersistentEntityMetadataVerifier verifier = DEFAULT_VERIFIER;
 
-	private MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
-
 	private ApplicationContext context;
 
 	private StandardEvaluationContext spelContext;
@@ -76,35 +71,21 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 	 * @param typeInformation must not be {@literal null}.
 	 */
 	public BasicCassandraPersistentEntity(TypeInformation<T> typeInformation) {
-		this(typeInformation, null, DEFAULT_VERIFIER);
+		this(typeInformation, DEFAULT_VERIFIER);
 	}
 
 	/**
 	 * Create a new {@link BasicCassandraPersistentEntity} with the given {@link TypeInformation}. Will default the table
 	 * name to the entity's simple type name.
 	 *
-	 * @param typeInformation
+	 * @param typeInformation must not be {@literal null}.
+	 * @param verifier must not be {@literal null}.
 	 */
 	public BasicCassandraPersistentEntity(TypeInformation<T> typeInformation,
-			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext) {
-		this(typeInformation, mappingContext, DEFAULT_VERIFIER);
-	}
-
-	/**
-	 * Create a new {@link BasicCassandraPersistentEntity} with the given {@link TypeInformation}. Will default the table
-	 * name to the entity's simple type name.
-	 *
-	 * @param typeInformation
-	 */
-	public BasicCassandraPersistentEntity(TypeInformation<T> typeInformation,
-			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext,
 			CassandraPersistentEntityMetadataVerifier verifier) {
 
-		// FIXME: Constructor with comparator, no optionality here
 		super(typeInformation, PROPERTY_COMPARATOR);
 
-		this.mappingContext = mappingContext;
-
 		setVerifier(verifier);
 	}
 
@@ -141,41 +122,12 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 		return findAnnotation(PrimaryKeyClass.class).isPresent();
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentEntity#getCompositePrimaryKeyProperties()
-	 */
-	@Override
-	public List<CassandraPersistentProperty> getCompositePrimaryKeyProperties() {
-
-		List<CassandraPersistentProperty> properties = new ArrayList<>();
-
-		Assert.state(isCompositePrimaryKey(),
-				String.format("[%s] does not represent a composite primary key class", this.getType().getName()));
-
-		addCompositePrimaryKeyProperties(this, properties);
-
-		return properties;
-	}
-
-	protected void addCompositePrimaryKeyProperties(CassandraPersistentEntity<?> compositePrimaryKeyEntity,
-			final List<CassandraPersistentProperty> properties) {
-
-		compositePrimaryKeyEntity.getPersistentProperties().forEach(property -> {
-
-			if (property.isCompositePrimaryKey()) {
-				addCompositePrimaryKeyProperties(property.getCompositePrimaryKeyEntity(), properties);
-			} else {
-				properties.add(property);
-			}
-
-		});
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mapping.model.BasicPersistentEntity#verify()
 	 */
 	@Override
 	public void verify() throws MappingException {
+
 		super.verify();
 
 		if (verifier != null) {
@@ -225,14 +177,6 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 		}
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentEntity#getMappingContext()
-	 */
-	@Override
-	public MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext() {
-		return mappingContext;
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentEntity#setTableName(org.springframework.cassandra.core.cql.CqlIdentifier)
 	 */

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

@@ -17,13 +17,10 @@ package org.springframework.data.cassandra.core.mapping;
 
 import static org.springframework.data.cql.core.CqlIdentifier.*;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -36,12 +33,10 @@ import org.springframework.data.cql.core.CqlIdentifier;
 import org.springframework.data.cql.core.Ordering;
 import org.springframework.data.cql.core.PrimaryKeyType;
 import org.springframework.data.mapping.Association;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.AnnotationBasedPersistentProperty;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.mapping.model.Property;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
-import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.util.Assert;
@@ -71,12 +66,12 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 	/**
 	 * Whether this property has been explicitly instructed to force quote column names.
 	 */
-	private Boolean forceQuote;
+	private Optional<Boolean> forceQuote = Optional.empty();
 
 	/**
 	 * An unmodifiable list of this property's column names.
 	 */
-	private List<CqlIdentifier> columnNames;
+	private CqlIdentifier columnName;
 
 	/**
 	 * Create a new {@link BasicCassandraPersistentProperty}.
@@ -133,40 +128,17 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return (CassandraPersistentEntity<?>) super.getOwner();
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isCompositePrimaryKey()
-	 */
-	@Override
-	public boolean isCompositePrimaryKey() {
-		return (AnnotatedElementUtils.findMergedAnnotation(getType(), PrimaryKeyClass.class) != null);
-	}
-
-	/**
-	 * @return
-	 */
-	public Class<?> getCompositePrimaryKeyType() {
-		return (isCompositePrimaryKey() ? getType() : null);
-	}
-
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getCompositePrimaryKeyTypeInformation()
-	 */
-	@Override
-	public TypeInformation<?> getCompositePrimaryKeyTypeInformation() {
-		return (isCompositePrimaryKey() ? ClassTypeInformation.from(getCompositePrimaryKeyType()) : null);
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getColumnName()
 	 */
 	@Override
 	public CqlIdentifier getColumnName() {
 
-		List<CqlIdentifier> columnNames = getColumnNames();
+		if (this.columnName == null) {
+			this.columnName = determineColumnName();
+		}
 
-		Assert.state(columnNames.size() == 1, String.format("Property [%s] has no single column mapping", getName()));
-
-		return columnNames.get(0);
+		return this.columnName;
 	}
 
 	/* (non-Javadoc)
@@ -177,9 +149,6 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return findAnnotation(PrimaryKeyColumn.class).map(PrimaryKeyColumn::ordering);
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getDataType()
-	 */
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getDataType()
 	 */
@@ -206,6 +175,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		}
 
 		if (isMap()) {
+
 			List<TypeInformation<?>> args = getTypeInformation().getTypeArguments();
 
 			ensureTypeArguments(args.size(), 2);
@@ -214,6 +184,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		}
 
 		if (isCollectionLike()) {
+
 			List<TypeInformation<?>> args = getTypeInformation().getTypeArguments();
 
 			ensureTypeArguments(args.size(), 1);
@@ -285,13 +256,19 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 	}
 
 	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isClusterKeyColumn()
+	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isCompositePrimaryKey()
 	 */
 	@Override
-	public boolean isClusterKeyColumn() {
+	public boolean isCompositePrimaryKey() {
+		return (AnnotatedElementUtils.findMergedAnnotation(getType(), PrimaryKeyClass.class) != null);
+	}
 
-		return findAnnotation(PrimaryKeyColumn.class)
-				.filter(primaryKeyColumn -> PrimaryKeyType.CLUSTERED.equals(primaryKeyColumn.type())).isPresent();
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isPrimaryKeyColumn()
+	 */
+	@Override
+	public boolean isPrimaryKeyColumn() {
+		return isAnnotationPresent(PrimaryKeyColumn.class);
 	}
 
 	/* (non-Javadoc)
@@ -305,11 +282,13 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 	}
 
 	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isPrimaryKeyColumn()
+	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#isClusterKeyColumn()
 	 */
 	@Override
-	public boolean isPrimaryKeyColumn() {
-		return isAnnotationPresent(PrimaryKeyColumn.class);
+	public boolean isClusterKeyColumn() {
+
+		return findAnnotation(PrimaryKeyColumn.class)
+				.filter(primaryKeyColumn -> PrimaryKeyType.CLUSTERED.equals(primaryKeyColumn.type())).isPresent();
 	}
 
 	protected DataType getDataTypeFor(DataType.Name dataTypeName) {
@@ -327,16 +306,6 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 
 	protected DataType getDataTypeFor(Class<?> javaType) {
 
-		Optional<CassandraPersistentEntity<?>> optionalEntity = getOwner().getMappingContext().getPersistentEntity(javaType)
-				.map(CassandraPersistentEntity.class::cast);
-
-		Optional<CassandraPersistentEntity<?>> udtEntity = optionalEntity
-				.filter(CassandraPersistentEntity::isUserDefinedType);
-
-		if (udtEntity.isPresent()) {
-			return udtEntity.map(CassandraPersistentEntity::getUserType).get();
-		}
-
 		DataType dataType = CassandraSimpleTypeHolder.getDataTypeFor(javaType);
 
 		if (dataType == null) {
@@ -356,52 +325,36 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		}
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getColumnNames()
-	 */
-	@Override
-	public List<CqlIdentifier> getColumnNames() {
-
-		columnNames = (columnNames != null ? columnNames : Collections.unmodifiableList(determineColumnNames()));
-
-		return columnNames;
-	}
-
-	protected List<CqlIdentifier> determineColumnNames() {
-
-		List<CqlIdentifier> columnNames = new ArrayList<>();
+	private CqlIdentifier determineColumnName() {
 
 		if (isCompositePrimaryKey()) { // then the id type has @PrimaryKeyClass
-			addCompositePrimaryKeyColumnNames(getCompositePrimaryKeyEntity(), columnNames);
-		} else { // else we're dealing with a single-column field
-			String defaultName = getName(); // TODO: replace with naming strategy class
-			String overriddenName;
-			boolean forceQuote;
-
-			if (isIdProperty()) { // then the id is of a simple type (since it's not a composite primary key)
-				Optional<PrimaryKey> optionalPrimaryKey = findAnnotation(PrimaryKey.class);
-				overriddenName = optionalPrimaryKey.map(PrimaryKey::value).orElse("");
-				forceQuote = optionalPrimaryKey.map(PrimaryKey::forceQuote).orElse(false);
-
-			} else if (isPrimaryKeyColumn()) { // then it's a simple type
-				Optional<PrimaryKeyColumn> optionalPrimaryKey = findAnnotation(PrimaryKeyColumn.class);
-				overriddenName = optionalPrimaryKey.map(PrimaryKeyColumn::value).orElse("");
-				forceQuote = optionalPrimaryKey.map(PrimaryKeyColumn::forceQuote).orElse(false);
-
-			} else { // then it's a vanilla column with the assumption that it's mapped to a single column
-				Optional<Column> optionalColumn = findAnnotation(Column.class);
-				overriddenName = optionalColumn.map(Column::value).orElse("");
-				forceQuote = optionalColumn.map(Column::forceQuote).orElse(false);
-			}
-
-			columnNames.add(createColumnName(defaultName, overriddenName, forceQuote));
-
+			return null;
 		}
 
-		return columnNames;
+		String defaultName = getName(); // TODO: replace with naming strategy class
+		String overriddenName;
+		boolean forceQuote;
+
+		if (isIdProperty()) { // then the id is of a simple type (since it's not a composite primary key)
+			Optional<PrimaryKey> optionalPrimaryKey = findAnnotation(PrimaryKey.class);
+			overriddenName = optionalPrimaryKey.map(PrimaryKey::value).orElse("");
+			forceQuote = optionalPrimaryKey.map(PrimaryKey::forceQuote).orElse(false);
+
+		} else if (isPrimaryKeyColumn()) { // then it's a simple type
+			Optional<PrimaryKeyColumn> optionalPrimaryKey = findAnnotation(PrimaryKeyColumn.class);
+			overriddenName = optionalPrimaryKey.map(PrimaryKeyColumn::value).orElse("");
+			forceQuote = optionalPrimaryKey.map(PrimaryKeyColumn::forceQuote).orElse(false);
+
+		} else { // then it's a vanilla column with the assumption that it's mapped to a single column
+			Optional<Column> optionalColumn = findAnnotation(Column.class);
+			overriddenName = optionalColumn.map(Column::value).orElse("");
+			forceQuote = optionalColumn.map(Column::forceQuote).orElse(false);
+		}
+
+		return createColumnName(defaultName, overriddenName, forceQuote);
 	}
 
-	protected CqlIdentifier createColumnName(String defaultName, String overriddenName, boolean forceQuote) {
+	private CqlIdentifier createColumnName(String defaultName, String overriddenName, boolean forceQuote) {
 
 		String name = defaultName;
 
@@ -412,47 +365,15 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return cqlId(name, forceQuote);
 	}
 
-	protected void addCompositePrimaryKeyColumnNames(CassandraPersistentEntity<?> compositePrimaryKeyEntity,
-			final List<CqlIdentifier> columnNames) {
-
-		compositePrimaryKeyEntity.getPersistentProperties().forEach(property -> {
-			if (property.isCompositePrimaryKey()) {
-				addCompositePrimaryKeyColumnNames(property.getCompositePrimaryKeyEntity(), columnNames);
-			} else {
-				columnNames.add(property.getColumnName());
-			}
-		});
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#setColumnName(org.springframework.cassandra.core.cql.CqlIdentifier)
 	 */
 	@Override
 	public void setColumnName(CqlIdentifier columnName) {
 
-		Assert.notNull(columnName, "columnName must not be null");
+		Assert.notNull(columnName, "ColumnName must not be null");
 
-		setColumnNames(Collections.singletonList(columnName));
-	}
-
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#setColumnNames(java.util.List)
-	 */
-	@Override
-	public void setColumnNames(List<CqlIdentifier> columnNames) {
-
-		Assert.notNull(columnNames, "List of column names must not be null");
-
-		// force calculation of columnNames if not known yet
-		getColumnNames();
-
-		Assert.state(this.columnNames.size() == columnNames.size(),
-				String.format(
-						"Property [%s] of entity [%s] is mapped to [%s] column%s, but given column name list has size [%s]",
-						getName(), getOwner().getType().getName(), this.columnNames.size(), this.columnNames.size() == 1 ? "" : "s",
-						columnNames.size()));
-
-		this.columnNames = Collections.unmodifiableList(new ArrayList<>(columnNames));
+		this.columnName = columnName;
 	}
 
 	/* (non-Javadoc)
@@ -461,45 +382,17 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 	@Override
 	public void setForceQuote(boolean forceQuote) {
 
-		if (this.forceQuote != null && this.forceQuote == forceQuote) {
-			return;
-		} else {
-			this.forceQuote = forceQuote;
+		boolean changed = !this.forceQuote.isPresent() || this.forceQuote.filter(v -> v != forceQuote).isPresent();
+
+		this.forceQuote = Optional.of(forceQuote);
+
+		if (changed) {
+
+			CqlIdentifier columnName = getColumnName();
+			if (columnName != null) {
+				setColumnName(cqlId(columnName.getUnquoted(), forceQuote));
+			}
 		}
-
-		List<CqlIdentifier> columnNames = getColumnNames() //
-				.stream() //
-				.map(CqlIdentifier::getUnquoted) //
-				.map(name -> cqlId(name, forceQuote)) //
-				.collect(Collectors.toList());
-
-		setColumnNames(columnNames);
-	}
-
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getCompositePrimaryKeyProperties()
-	 */
-	@Override
-	public List<CassandraPersistentProperty> getCompositePrimaryKeyProperties() {
-
-		Assert.state(isCompositePrimaryKey(),
-				String.format("[%s] does not represent a composite primary key property", getName()));
-
-		return getCompositePrimaryKeyEntity().getCompositePrimaryKeyProperties();
-	}
-
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.mapping.CassandraPersistentProperty#getCompositePrimaryKeyEntity()
-	 */
-	@Override
-	public CassandraPersistentEntity<?> getCompositePrimaryKeyEntity() {
-
-		MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext = getOwner()
-				.getMappingContext();
-
-		Assert.state(mappingContext != null, "CassandraMappingContext needed");
-
-		return mappingContext.getRequiredPersistentEntity(getCompositePrimaryKeyTypeInformation());
 	}
 
 	/* (non-Javadoc)

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

@@ -254,10 +254,6 @@ public class CassandraMappingContext
 			entities.add(entity);
 
 			if (!entity.isUserDefinedType()) {
-				if (entity.isCompositePrimaryKey()) {
-					primaryKeyEntities.add(entity);
-				}
-
 				entity.findAnnotation(Table.class).ifPresent(table -> tableEntities.add(entity));
 			}
 		});
@@ -286,10 +282,10 @@ public class CassandraMappingContext
 		BasicCassandraPersistentEntity<T> entity;
 
 		if (userDefinedType != null) {
-			entity = new CassandraUserTypePersistentEntity<>(typeInformation, this, verifier, userTypeResolver);
+			entity = new CassandraUserTypePersistentEntity<>(typeInformation, verifier, userTypeResolver);
 
 		} else {
-			entity = new BasicCassandraPersistentEntity<>(typeInformation, this, verifier);
+			entity = new BasicCassandraPersistentEntity<>(typeInformation, verifier);
 		}
 
 		if (context != null) {
@@ -442,10 +438,6 @@ public class CassandraMappingContext
 	private DataType getDataTypeWithUserTypeFactory(CassandraPersistentProperty property,
 			DataTypeProvider dataTypeProvider) {
 
-		if (property.isCompositePrimaryKey()) {
-			return property.getDataType();
-		}
-
 		if (property.findAnnotation(CassandraType.class).isPresent()) {
 			return property.getDataType();
 		}
@@ -464,6 +456,7 @@ public class CassandraMappingContext
 		return customConversions.getCustomWriteTarget(property.getType()) //
 				.map(CassandraSimpleTypeHolder::getDataTypeFor) //
 				.orElseGet(() -> customConversions.getCustomWriteTarget(property.getActualType()) //
+						.filter(it -> !property.isMapLike()) //
 						.map(it -> {
 
 							if (property.isCollectionLike()) {
@@ -478,9 +471,26 @@ public class CassandraMappingContext
 							}
 
 							return getDataTypeFor(it);
-						}).orElseGet(property::getDataType)
+						}).orElseGet(() -> {
 
-		);
+							if (property.isMapLike()) {
+
+								Class<?> keyType = property.getComponentType().get();
+								Class<?> valueType = property.getMapValueType().get();
+								return DataType.map(getDataType(keyType, dataTypeProvider), getDataType(valueType, dataTypeProvider));
+							}
+
+							return property.getDataType();
+						}));
+
+	}
+
+	private DataType getDataType(Class<?> type, DataTypeProvider dataTypeProvider) {
+
+		return getPersistentEntity(type) //
+				.filter(CassandraPersistentEntity::isUserDefinedType) //
+				.map(dataTypeProvider::getDataType) //
+				.orElseGet(() -> getDataType(type));
 	}
 
 	private DataType getUserDataType(CassandraPersistentProperty property, DataTypeProvider dataTypeProvider,

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

@@ -15,13 +15,10 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
-import java.util.List;
-
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.data.cql.core.CqlIdentifier;
 import org.springframework.data.mapping.PersistentEntity;
-import org.springframework.data.mapping.context.MappingContext;
 
 import com.datastax.driver.core.UserType;
 
@@ -40,9 +37,6 @@ public interface CassandraPersistentEntity<T>
 	 */
 	boolean isCompositePrimaryKey();
 
-	// TODO: return rather a Stream, rename to "getPrimaryKeyProperties"
-	List<CassandraPersistentProperty> getCompositePrimaryKeyProperties();
-
 	/**
 	 * Returns the table name to which the entity shall be persisted.
 	 */
@@ -76,8 +70,6 @@ public interface CassandraPersistentEntity<T>
 	 */
 	UserType getUserType();
 
-	// TODO: Review if that's required or it can be handled in a different way
-	MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext();
 
 	ApplicationContext getApplicationContext();
 }

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

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
-import java.util.List;
 import java.util.Optional;
 
 import org.springframework.context.ApplicationContextAware;
@@ -24,7 +23,6 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cql.core.CqlIdentifier;
 import org.springframework.data.cql.core.Ordering;
 import org.springframework.data.mapping.PersistentProperty;
-import org.springframework.data.util.TypeInformation;
 
 import com.datastax.driver.core.DataType;
 
@@ -41,40 +39,10 @@ public interface CassandraPersistentProperty
 		extends PersistentProperty<CassandraPersistentProperty>, ApplicationContextAware {
 
 	/**
-	 * Whether the property is a composite primary key.
-	 */
-	boolean isCompositePrimaryKey();
-
-	/**
-	 * Returns a {@link CassandraPersistentEntity} representing the composite primary key class of this entity, or null if
-	 * this class does not use a composite primary key.
-	 */
-	CassandraPersistentEntity<?> getCompositePrimaryKeyEntity();
-
-	/**
-	 * Returns a {@link TypeInformation} representing the type of the composite primary key class of this entity, or null
-	 * if this class does not use a composite primary key.
-	 */
-	TypeInformation<?> getCompositePrimaryKeyTypeInformation();
-
-	/**
-	 * Gets the list of composite primary key properties that this composite primary key field is a placeholder for.
-	 */
-	List<CassandraPersistentProperty> getCompositePrimaryKeyProperties();
-
-	/**
-	 * The name of the single column to which the property is persisted. This is a convenience method when the caller
-	 * knows that the property is mapped to a single column. Throws {@link IllegalStateException} if this property is
-	 * mapped to multiple columns.
+	 * The name of the single column to which the property is persisted.
 	 */
 	CqlIdentifier getColumnName();
 
-	/**
-	 * The names of the columns to which the property is persisted if this is a composite primary key property. Never
-	 * returns null.
-	 */
-	List<CqlIdentifier> getColumnNames();
-
 	/**
 	 * The ordering (ascending or descending) for the column. Valid only for primary key columns; returns null for
 	 * non-primary key columns.
@@ -96,14 +64,9 @@ public interface CassandraPersistentProperty
 	boolean isIndexed();
 
 	/**
-	 * Whether the property is a partition key column.
+	 * Whether the property is a composite primary key.
 	 */
-	boolean isPartitionKeyColumn();
-
-	/**
-	 * Whether the property is a cluster key column.
-	 */
-	boolean isClusterKeyColumn();
+	boolean isCompositePrimaryKey();
 
 	/**
 	 * Whether the property is a partition key column or a cluster key column
@@ -113,6 +76,16 @@ public interface CassandraPersistentProperty
 	 */
 	boolean isPrimaryKeyColumn();
 
+	/**
+	 * Whether the property is a partition key column.
+	 */
+	boolean isPartitionKeyColumn();
+
+	/**
+	 * Whether the property is a cluster key column.
+	 */
+	boolean isClusterKeyColumn();
+
 	@Override
 	CassandraPersistentEntity<?> getOwner();
 
@@ -120,7 +93,7 @@ public interface CassandraPersistentProperty
 	 * Whether to force-quote the column names of this property.
 	 *
 	 * @param forceQuote
-	 * @see CassandraPersistentProperty#getColumnNames()
+	 * @see CassandraPersistentProperty#getColumnName()
 	 */
 	void setForceQuote(boolean forceQuote);
 
@@ -129,18 +102,10 @@ public interface CassandraPersistentProperty
 	 * property is not mapped by a single column, throws {@link IllegalStateException}. If the given column name is null,
 	 * {@link IllegalArgumentException} is thrown.
 	 *
-	 * @param columnName
+	 * @param columnName must not be {@literal null}.
 	 */
 	void setColumnName(CqlIdentifier columnName);
 
-	/**
-	 * Sets this property's column names to the collection given. The given collection must have the same size as this
-	 * property's current list of column names, and must contain no {@code null} elements.
-	 *
-	 * @param columnName
-	 */
-	void setColumnNames(List<CqlIdentifier> columnNames);
-
 	/**
 	 * Returns whether the property is a {@link java.util.Map}.
 	 *

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

@@ -18,7 +18,6 @@ package org.springframework.data.cassandra.core.mapping;
 import java.util.Optional;
 
 import org.springframework.data.cql.core.CqlIdentifier;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
@@ -45,15 +44,13 @@ public class CassandraUserTypePersistentEntity<T> extends BasicCassandraPersiste
 	 * Create a new {@link CassandraUserTypePersistentEntity}.
 	 *
 	 * @param typeInformation must not be {@literal null}.
-	 * @param mappingContext must not be {@literal null}.
 	 * @param verifier must not be {@literal null}.
 	 * @param resolver must not be {@literal null}.
 	 */
 	public CassandraUserTypePersistentEntity(TypeInformation<T> typeInformation,
-			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext,
 			CassandraPersistentEntityMetadataVerifier verifier, UserTypeResolver resolver) {
 
-		super(typeInformation, mappingContext, verifier);
+		super(typeInformation, verifier);
 
 		Assert.notNull(resolver, "UserTypeResolver must not be null");
 

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

@@ -16,18 +16,16 @@
 package org.springframework.data.cassandra.core.mapping;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.springframework.data.cql.core.CqlIdentifier.*;
+
+import lombok.EqualsAndHashCode;
 
 import java.io.Serializable;
 import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 
 import org.junit.Before;
 import org.junit.Test;
-import org.springframework.data.cql.core.CqlIdentifier;
 import org.springframework.data.cql.core.PrimaryKeyType;
 import org.springframework.data.cql.core.keyspace.ColumnSpecification;
 import org.springframework.data.cql.core.keyspace.CreateTableSpecification;
@@ -48,51 +46,20 @@ public class CassandraCompositePrimaryKeyUnitTests {
 	private static final CassandraSimpleTypeHolder SIMPLE_TYPE_HOLDER = new CassandraSimpleTypeHolder();
 
 	@PrimaryKeyClass
-	static class Key implements Serializable {
+	@EqualsAndHashCode
+	static class CompositeKey implements Serializable {
 
 		private static final long serialVersionUID = 1L;
 
 		@PrimaryKeyColumn(ordinal = 0, type = PrimaryKeyType.PARTITIONED) String z;
 
 		@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.CLUSTERED) String a;
-
-		@Override
-		public int hashCode() {
-			final int prime = 31;
-			int result = 1;
-			result = prime * result + ((a == null) ? 0 : a.hashCode());
-			result = prime * result + ((z == null) ? 0 : z.hashCode());
-			return result;
-		}
-
-		@Override
-		public boolean equals(Object obj) {
-			if (this == obj)
-				return true;
-			if (obj == null)
-				return false;
-			if (getClass() != obj.getClass())
-				return false;
-			Key other = (Key) obj;
-			if (a == null) {
-				if (other.a != null)
-					return false;
-			} else if (!a.equals(other.a))
-				return false;
-			if (z == null) {
-				if (other.z != null)
-					return false;
-			} else if (!z.equals(other.z))
-				return false;
-			return true;
-		}
-
 	}
 
 	@Table
-	static class Thing {
+	static class TypeWithCompositeKey {
 
-		@PrimaryKey Key id;
+		@PrimaryKey CompositeKey id;
 
 		Date time;
 
@@ -100,36 +67,29 @@ public class CassandraCompositePrimaryKeyUnitTests {
 	}
 
 	CassandraMappingContext context;
-	CassandraPersistentEntity<?> thing;
+
+	CassandraPersistentEntity<?> entity;
+
 	CassandraPersistentEntity<?> key;
 
 	@Before
 	public void setup() {
+
 		context = new CassandraMappingContext();
-		thing = context.getRequiredPersistentEntity(ClassTypeInformation.from(Thing.class));
-		key = context.getRequiredPersistentEntity(ClassTypeInformation.from(Key.class));
+		entity = context.getRequiredPersistentEntity(ClassTypeInformation.from(TypeWithCompositeKey.class));
+		key = context.getRequiredPersistentEntity(ClassTypeInformation.from(CompositeKey.class));
 	}
 
 	@Test
 	public void validateMappingInfo() {
 
-		Field field = ReflectionUtils.findField(Thing.class, "id");
-		CassandraPersistentProperty property = new BasicCassandraPersistentProperty(Property.of(field), thing,
+		Field field = ReflectionUtils.findField(TypeWithCompositeKey.class, "id");
+		CassandraPersistentProperty property = new BasicCassandraPersistentProperty(Property.of(field), entity,
 				SIMPLE_TYPE_HOLDER);
 		assertThat(property.isIdProperty()).isTrue();
 		assertThat(property.isCompositePrimaryKey()).isTrue();
 
-		List<CqlIdentifier> expectedColumnNames = Arrays.asList(cqlId("z"), cqlId("a"));
-		assertThat(expectedColumnNames.equals(property.getColumnNames())).isTrue();
-
-		List<CqlIdentifier> actualColumnNames = new ArrayList<>();
-		List<CassandraPersistentProperty> properties = property.getCompositePrimaryKeyProperties();
-		for (CassandraPersistentProperty p : properties) {
-			actualColumnNames.addAll(p.getColumnNames());
-		}
-		assertThat(expectedColumnNames.equals(actualColumnNames)).isTrue();
-
-		CreateTableSpecification spec = context.getCreateTableSpecificationFor(thing);
+		CreateTableSpecification spec = context.getCreateTableSpecificationFor(entity);
 
 		List<ColumnSpecification> partitionKeyColumns = spec.getPartitionKeyColumns();
 		assertThat(partitionKeyColumns).hasSize(1);

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

@@ -93,11 +93,6 @@ public class CassandraMappingContextUnitTests {
 		assertThat(idProperty).hasValueSatisfying(actual -> {
 
 			assertThat(actual.getColumnName().toCql()).isEqualTo("foo");
-
-			List<CqlIdentifier> columnNames = actual.getColumnNames();
-
-			assertThat(columnNames).hasSize(1);
-			assertThat(columnNames.get(0).toCql()).isEqualTo("foo");
 		});
 	}
 
@@ -178,7 +173,6 @@ public class CassandraMappingContextUnitTests {
 						.isTrue();
 
 		assertThat(primaryKeyClass.isCompositePrimaryKey()).isTrue();
-		assertThat(primaryKeyClass.getCompositePrimaryKeyProperties()).hasSize(2);
 
 		CassandraPersistentProperty firstname = primaryKeyClass.getRequiredPersistentProperty("firstname");
 

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

@@ -38,7 +38,6 @@ import org.springframework.data.util.ClassTypeInformation;
 @RunWith(MockitoJUnitRunner.class)
 public class CassandraUserTypePersistentEntityUnitTests {
 
-	@Mock CassandraMappingContext mappingContextMock;
 	@Mock UserTypeResolver userTypeResolverMock;
 
 	@Test // DATACASS-172
@@ -85,8 +84,7 @@ public class CassandraUserTypePersistentEntityUnitTests {
 	}
 
 	private <T> CassandraUserTypePersistentEntity<T> getEntity(Class<T> entityClass) {
-		return new CassandraUserTypePersistentEntity<>(ClassTypeInformation.from(entityClass), mappingContextMock, null,
-				userTypeResolverMock);
+		return new CassandraUserTypePersistentEntity<>(ClassTypeInformation.from(entityClass), null, userTypeResolverMock);
 	}
 
 	@UserDefinedType

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

@@ -19,8 +19,6 @@ import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.cql.core.CqlIdentifier.*;
 
 import java.io.Serializable;
-import java.util.Arrays;
-import java.util.List;
 
 import org.junit.Test;
 import org.springframework.data.cql.core.CqlIdentifier;
@@ -110,12 +108,9 @@ public class ForceQuotedPropertiesSimpleUnitTests {
 		CassandraPersistentProperty stringOne = key.getRequiredPersistentProperty("stringOne");
 
 		assertThat(stringZero.getColumnName().toCql()).isEqualTo("\"stringZero\"");
+		assertThat(stringZero.getColumnName()).isEqualTo(quotedCqlId("stringZero"));
 		assertThat(stringOne.getColumnName().toCql()).isEqualTo("\"stringOne\"");
-
-		List<CqlIdentifier> names = Arrays.asList(quotedCqlId("stringZero"), quotedCqlId("stringOne"));
-		CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(ImplicitComposite.class);
-
-		assertThat(entity.getRequiredPersistentProperty("primaryKey").getColumnNames()).isEqualTo(names);
+		assertThat(stringOne.getColumnName()).isEqualTo(quotedCqlId("stringOne"));
 	}
 
 	@PrimaryKeyClass
@@ -148,11 +143,6 @@ public class ForceQuotedPropertiesSimpleUnitTests {
 		assertThat(stringOne.getColumnName()).isEqualTo(CqlIdentifier.cqlId("stringOne"));
 		assertThat(stringZero.getColumnName().toCql()).isEqualTo("stringzero");
 		assertThat(stringOne.getColumnName().toCql()).isEqualTo("stringone");
-
-		List<CqlIdentifier> names = Arrays.asList(cqlId("stringZero"), cqlId("stringOne"));
-		CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(DefaultComposite.class);
-
-		assertThat(entity.getRequiredPersistentProperty("primaryKey").getColumnNames()).isEqualTo(names);
 	}
 
 	@PrimaryKeyClass
@@ -187,11 +177,6 @@ public class ForceQuotedPropertiesSimpleUnitTests {
 
 		assertThat(stringZero.getColumnName().toCql()).isEqualTo("\"" + EXPLICIT_KEY_0 + "\"");
 		assertThat(stringOne.getColumnName().toCql()).isEqualTo("\"" + EXPLICIT_KEY_1 + "\"");
-
-		List<CqlIdentifier> names = Arrays.asList(quotedCqlId(EXPLICIT_KEY_0), quotedCqlId(EXPLICIT_KEY_1));
-		CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(ExplicitComposite.class);
-
-		assertThat(entity.getRequiredPersistentProperty("primaryKey").getColumnNames()).isEqualTo(names);
 	}
 
 	@PrimaryKeyClass

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

@@ -269,7 +269,7 @@ public class CassandraQueryCreatorUnitTests {
 		assertThat(query).isEqualTo("SELECT * FROM key WHERE firstname='Walter';");
 	}
 
-	@Test(expected = IllegalStateException.class) // DATACASS-7
+	@Test(expected = IllegalArgumentException.class) // DATACASS-7
 	public void createsFindByPrimaryKey2PartCorrectly() {
 		createQuery("findByKey", TypeWithCompositeId.class, new Key());
 	}