DATACASS-213 - Create secondary indexes for annotated properties.

We now create secondary indexes for annotated properties via @Indexed. Index creation is part of schema creation that is executed after Session initialization and table creation.

We support plain secondary indexes and key/value/entry indexes for map columns. Index creation is useful for rapid development but should not be used in large setups or at least with care to not impact performance in a negative way.

@Table
public class Person {

  @Id
  private String key;

  @Indexed("name_index")
  private String name;

  private Map<@Indexed String, String> keys;
}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBean.java

@@ -165,6 +165,7 @@ public class CassandraSessionFactoryBean extends CassandraCqlSessionFactoryBean
 
 		schemaCreator.createUserTypes(ifNotExists);
 		schemaCreator.createTables(ifNotExists);
+		schemaCreator.createIndexes(ifNotExists);
 	}
 
 	/**

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

@@ -26,8 +26,10 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.cql.generator.CreateIndexCqlGenerator;
 import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGenerator;
 import org.springframework.data.cassandra.core.cql.generator.CreateUserTypeCqlGenerator;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateUserTypeSpecification;
 import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
@@ -94,6 +96,32 @@ public class CassandraPersistentEntitySchemaCreator {
 				.collect(Collectors.toList());
 	}
 
+	/**
+	 * Create indexes from types known to {@link CassandraMappingContext}.
+	 *
+	 * @param ifNotExists {@literal true} to create tables using {@code IF NOT EXISTS}.
+	 */
+	public void createIndexes(boolean ifNotExists) {
+
+		createIndexSpecifications(ifNotExists).forEach(specification -> cassandraAdminOperations.getCqlOperations()
+				.execute(CreateIndexCqlGenerator.toCql(specification)));
+	}
+
+	/**
+	 * Create {@link List} of {@link CreateIndexSpecification}.
+	 *
+	 * @param ifNotExists {@literal true} to create indexes using {@code IF NOT EXISTS}.
+	 * @return {@link List} of {@link CreateIndexSpecification}.
+	 */
+	protected List<CreateIndexSpecification> createIndexSpecifications(boolean ifNotExists) {
+
+		return mappingContext.getTableEntities() //
+				.stream() //
+				.flatMap(entity -> mappingContext.getCreateIndexSpecificationsFor(entity).stream()) //
+				.peek(it -> it.ifNotExists(ifNotExists)) //
+				.collect(Collectors.toList());
+	}
+
 	/**
 	 * Create user types from types known to {@link CassandraMappingContext}.
 	 *
@@ -133,8 +161,8 @@ public class CassandraPersistentEntitySchemaCreator {
 			List<CqlIdentifier> ordered = new ArrayList<>(seen);
 			Collections.reverse(ordered);
 
-			specifications.addAll(ordered.stream()
-					.filter(created::add).map(identifier -> mappingContext
+			specifications.addAll(ordered
+					.stream().filter(created::add).map(identifier -> mappingContext
 							.getCreateUserTypeSpecificationFor(byTableName.get(identifier)).ifNotExists(ifNotExists))
 					.collect(Collectors.toList()));
 		});

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGenerator.java

@@ -18,12 +18,14 @@ package org.springframework.data.cassandra.core.cql.generator;
 import static org.springframework.data.cassandra.core.cql.CqlStringUtils.*;
 
 import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification.ColumnFunction;
 
 /**
  * CQL generator for generating a {@code CREATE INDEX} statement.
  *
  * @author Matthew T. Adams
  * @author David Webb
+ * @author Mark Paluch
  */
 public class CreateIndexCqlGenerator extends IndexNameCqlGenerator<CreateIndexSpecification> {
 
@@ -40,13 +42,25 @@ public class CreateIndexCqlGenerator extends IndexNameCqlGenerator<CreateIndexSp
 
 		cql = noNull(cql);
 
-		cql.append("CREATE").append(spec().isCustom() ? " CUSTOM" : "").append(" INDEX ")
-				.append(spec().getIfNotExists() ? "IF NOT EXISTS " : "")
-				.append(spec().getName() == null ? "" : spec().getName()).append(" ON ").append(spec().getTableName())
-				.append(" (").append(spec().getColumnName()).append(")");
+		cql.append("CREATE").append(spec().isCustom() ? " CUSTOM" : "").append(" INDEX")
+				.append(spec().getIfNotExists() ? " IF NOT EXISTS" : "");
+
+		if (spec().getName() != null) {
+			cql.append(" ").append(spec().getName());
+		}
+
+		cql.append(" ON ").append(spec().getTableName()).append(" (");
+
+		if (spec().getColumnFunction() != ColumnFunction.NONE) {
+			cql.append(spec().getColumnFunction().name()).append("(").append(spec().getColumnName()).append(")");
+		} else {
+			cql.append(spec().getColumnName());
+		}
+
+		cql.append(")");
 
 		if (spec().isCustom()) {
-			cql.append(" USING ").append(spec().getUsing());
+			cql.append(" USING ").append("'").append(spec().getUsing()).append("'");
 		}
 
 		cql.append(";");

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/keyspace/CreateIndexSpecification.java

@@ -26,14 +26,21 @@ import org.springframework.util.StringUtils;
  *
  * @author Matthew T. Adams
  * @author David Webb
+ * @author Mark Paluch
  */
 public class CreateIndexSpecification extends IndexNameSpecification<CreateIndexSpecification>
 		implements IndexDescriptor {
 
 	private boolean ifNotExists = false;
+
 	private boolean custom = false;
+
 	private CqlIdentifier tableName;
+
 	private CqlIdentifier columnName;
+
+	private ColumnFunction columnFunction = ColumnFunction.NONE;
+
 	private String using;
 
 	/**
@@ -120,6 +127,61 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		return columnName;
 	}
 
+	/**
+	 * Causes the inclusion of an {@code KEYS} clause.
+	 *
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification keys() {
+		return columnFunction(ColumnFunction.KEYS);
+	}
+
+	/**
+	 * Causes the inclusion of an {@code VALUES} clause.
+	 *
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification values() {
+		return columnFunction(ColumnFunction.VALUES);
+	}
+
+	/**
+	 * Causes the inclusion of an {@code ENTRIES} clause.
+	 *
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification entries() {
+		return columnFunction(ColumnFunction.ENTRIES);
+	}
+
+	/**
+	 * Causes the inclusion of an {@code FULL} clause.
+	 *
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification full() {
+		return columnFunction(ColumnFunction.FULL);
+	}
+
+	/**
+	 * Set a {@link ColumnFunction} such as {@code KEYS(…)}, {@code ENTRIES(…)}.
+	 *
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification columnFunction(ColumnFunction columnFunction) {
+		this.columnFunction = columnFunction;
+		return this;
+	}
+
+	public ColumnFunction getColumnFunction() {
+		return columnFunction;
+	}
+
 	/**
 	 * Sets the table name.
 	 *
@@ -156,4 +218,13 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		this.columnName = columnName;
 		return this;
 	}
+
+	/**
+	 * Column functions to specify indexing behavior.
+	 *
+	 * @since 2.0
+	 */
+	public enum ColumnFunction {
+		NONE, KEYS, VALUES, ENTRIES, FULL,
+	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/keyspace/CreateTableSpecification.java

@@ -21,9 +21,12 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier;
  * Builder class to construct a {@code CREATE TABLE} specification.
  *
  * @author Matthew T. Adams
+ * @author Mark Paluch
  */
 public class CreateTableSpecification extends TableSpecification<CreateTableSpecification> {
 
+	private boolean ifNotExists = false;
+
 	/**
 	 * Entry point into the {@link CreateTableSpecification}'s fluent API to create a table. Convenient if imported
 	 * statically.
@@ -48,8 +51,6 @@ public class CreateTableSpecification extends TableSpecification<CreateTableSpec
 		return new CreateTableSpecification().name(name);
 	}
 
-	private boolean ifNotExists = false;
-
 	@Override
 	public CreateTableSpecification name(CqlIdentifier name) {
 		return (CreateTableSpecification) super.name(name);

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

@@ -48,9 +48,14 @@ public class BasicCassandraPersistentEntityMetadataVerifier implements Cassandra
 
 		List<MappingException> exceptions = new ArrayList<>();
 
-		final List<CassandraPersistentProperty> idProperties = new ArrayList<>();
-		final List<CassandraPersistentProperty> partitionKeyColumns = new ArrayList<>();
-		final List<CassandraPersistentProperty> primaryKeyColumns = new ArrayList<>();
+		List<CassandraPersistentProperty> idProperties = new ArrayList<>();
+		List<CassandraPersistentProperty> partitionKeyColumns = new ArrayList<>();
+		List<CassandraPersistentProperty> primaryKeyColumns = new ArrayList<>();
+
+		// @Indexed not allowed on type level
+		if (entity.isAnnotationPresent(Indexed.class)) {
+			exceptions.add(new MappingException("@Indexed cannot be used on entity classes"));
+		}
 
 		// Ensure entity is not both a @Table(@Persistent) and a @PrimaryKeyClass
 		if (entity.isCompositePrimaryKey()) {

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

@@ -17,6 +17,11 @@ package org.springframework.data.cassandra.core.mapping;
 
 import static org.springframework.data.cassandra.core.cql.CqlIdentifier.*;
 
+import java.lang.annotation.Annotation;
+import java.lang.reflect.AnnotatedParameterizedType;
+import java.lang.reflect.AnnotatedType;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -37,6 +42,7 @@ import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.model.AnnotationBasedPersistentProperty;
 import org.springframework.data.mapping.model.Property;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.util.Optionals;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.util.Assert;
@@ -422,4 +428,40 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 	public boolean isMapLike() {
 		return ClassUtils.isAssignable(Map.class, getType());
 	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#findAnnotatedType(java.lang.Class)
+	 */
+	@Override
+	public AnnotatedType findAnnotatedType(Class<? extends Annotation> annotationType) {
+
+		return Optionals.toStream(Optional.ofNullable(getField()).map(Field::getAnnotatedType), //
+				Optional.ofNullable(getGetter()).map(Method::getAnnotatedReturnType), //
+				Optional.ofNullable(getSetter()).map(it -> it.getParameters()[0].getAnnotatedType())) //
+				.filter(it -> hasAnnotation(it, annotationType, getTypeInformation())) //
+				.findFirst() //
+				.orElse(null);
+	}
+
+	private static boolean hasAnnotation(AnnotatedType type, Class<? extends Annotation> annotationType,
+			TypeInformation<?> typeInformation) {
+
+		if (AnnotatedElementUtils.hasAnnotation(type, annotationType)) {
+			return true;
+		}
+
+		AnnotatedParameterizedType parameterizedType = (AnnotatedParameterizedType) type;
+		AnnotatedType[] arguments = parameterizedType.getAnnotatedActualTypeArguments();
+
+		if (typeInformation.isCollectionLike() && arguments.length == 1) {
+			return AnnotatedElementUtils.hasAnnotation(arguments[0], annotationType);
+		}
+
+		if (typeInformation.isMap() && arguments.length == 2) {
+			return AnnotatedElementUtils.hasAnnotation(arguments[0], annotationType)
+					|| AnnotatedElementUtils.hasAnnotation(arguments[1], annotationType);
+		}
+
+		return false;
+	}
 }

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

@@ -19,6 +19,9 @@ import static org.springframework.data.cassandra.core.cql.CqlIdentifier.*;
 import static org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification.*;
 import static org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder.*;
 
+import java.lang.reflect.AnnotatedParameterizedType;
+import java.lang.reflect.AnnotatedType;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -36,6 +39,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateUserTypeSpecification;
 import org.springframework.data.cassandra.core.mapping.UserTypeUtil.FrozenLiteralDataType;
@@ -371,7 +375,7 @@ public class CassandraMappingContext
 
 		Assert.notNull(entity, "CassandraPersistentEntity must not be null");
 
-		final CreateTableSpecification specification = createTable().name(entity.getTableName());
+		CreateTableSpecification specification = createTable().name(entity.getTableName());
 
 		for (CassandraPersistentProperty property : entity) {
 
@@ -416,6 +420,104 @@ public class CassandraMappingContext
 		return specification;
 	}
 
+	/**
+	 * @param entity must not be {@literal null}.
+	 * @return
+	 * @since 2.0
+	 */
+	public List<CreateIndexSpecification> getCreateIndexSpecificationsFor(CassandraPersistentEntity<?> entity) {
+
+		Assert.notNull(entity, "CassandraPersistentEntity must not be null");
+
+		return getCreateIndexSpecifications(entity.getTableName(), entity);
+	}
+
+	private List<CreateIndexSpecification> getCreateIndexSpecifications(CqlIdentifier tableName,
+			CassandraPersistentEntity<?> entity) {
+
+		List<CreateIndexSpecification> indexes = new ArrayList<>();
+
+		for (CassandraPersistentProperty property : entity) {
+
+			if (property.isCompositePrimaryKey()) {
+				indexes.addAll(getCreateIndexSpecifications(tableName, getRequiredPersistentEntity(property)));
+			} else {
+				indexes.addAll(createIndexSpecifications(tableName, property));
+			}
+		}
+		return indexes;
+	}
+
+	private List<CreateIndexSpecification> createIndexSpecifications(CqlIdentifier tableName,
+			CassandraPersistentProperty property) {
+
+		List<CreateIndexSpecification> indexes = new ArrayList<>();
+
+		if (property.isAnnotationPresent(Indexed.class)) {
+
+			Indexed annotation = property.findAnnotation(Indexed.class);
+			CreateIndexSpecification index = createIndexSpecification(annotation, tableName, property);
+
+			if (property.isMapLike()) {
+				index.entries();
+			}
+
+			indexes.add(index);
+		}
+
+		if (property.isMapLike()) {
+
+			AnnotatedType type = property.findAnnotatedType(Indexed.class);
+
+			if (type instanceof AnnotatedParameterizedType) {
+
+				AnnotatedParameterizedType parameterizedType = (AnnotatedParameterizedType) type;
+				AnnotatedType[] typeArgs = parameterizedType.getAnnotatedActualTypeArguments();
+
+				Indexed keyIndex = typeArgs.length == 2 ? AnnotatedElementUtils.getMergedAnnotation(typeArgs[0], Indexed.class)
+						: null;
+				Indexed valueIndex = typeArgs.length == 2
+						? AnnotatedElementUtils.getMergedAnnotation(typeArgs[1], Indexed.class)
+						: null;
+
+				if ((!indexes.isEmpty() && (keyIndex != null || valueIndex != null))
+						|| (keyIndex != null && valueIndex != null)) {
+					throw new MappingException("Multiple index declarations for " + property
+							+ " found. A map index must be either declared for entries, keys or values.");
+				}
+
+				if (keyIndex != null) {
+
+					CreateIndexSpecification index = createIndexSpecification(keyIndex, tableName, property);
+					index.keys();
+					indexes.add(index);
+				}
+
+				if (valueIndex != null) {
+					CreateIndexSpecification index = createIndexSpecification(valueIndex, tableName, property);
+					index.values();
+					indexes.add(index);
+				}
+			}
+		}
+
+		return indexes;
+	}
+
+	private CreateIndexSpecification createIndexSpecification(Indexed annotation, CqlIdentifier tableName,
+			CassandraPersistentProperty property) {
+
+		CreateIndexSpecification index;
+
+		if (StringUtils.hasText(annotation.value())) {
+			index = CreateIndexSpecification.createIndex(annotation.value());
+		} else {
+			index = CreateIndexSpecification.createIndex();
+		}
+
+		return index.tableName(tableName).columnName(property.getColumnName());
+	}
+
 	/**
 	 * Returns a {@link CreateUserTypeSpecification} for the given entity, including all mapping information.
 	 *

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

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
+import java.lang.annotation.Annotation;
+import java.lang.reflect.AnnotatedType;
+
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -101,4 +104,15 @@ public interface CassandraPersistentProperty
 	 * @return a boolean indicating whether this property type is a {@link java.util.Map}.
 	 */
 	boolean isMapLike();
+
+	/**
+	 * Find an {@link AnnotatedType} by {@code annotationType} derived from the property type. Annotated type is looked up
+	 * by introspecting property field/accessors. Collection/Map-like types are introspected for type annotations within
+	 * type arguments.
+	 *
+	 * @param annotationType must not be {@literal null}.
+	 * @return the annotated type or {@literal null}.
+	 * @since 2.0
+	 */
+	AnnotatedType findAnnotatedType(Class<? extends Annotation> annotationType);
 }

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

@@ -21,13 +21,34 @@ import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
 /**
- * Identifies a secondary index in the table on a single, non-key column.
+ * Identifies a secondary index in the table on a single column.
+ * <p>
+ * The following columns of a {@link Table} type can be annotated with {@link Indexed}:
+ * <ul>
+ * <li>Scalar data types</li>
+ * <li>User-defined types</li>
+ * <li>Collection types</li>
+ * <li>Map type</li>
+ * </ul>
+ * <p>
+ * Map types distinguish between the column function applied before indexing. Maps support entry, key or value-level
+ * indexing with the restriction that only a single secondary index is allowed.
+ *
+ * <pre class="code">
+ * &#64;Table
+ * class Person {
+ *
+ * 	Map<@Indexed String, String> indexedKey; // allows CONTAINS KEY queries
+ * 	Map<String, @Indexed String> indexedValue; // allows CONTAINS queries
+ * }
+ * </pre>
  *
  * @author Alex Shvid
  * @author Matthew T. Adams
+ * @author Mark Paluch
  */
 @Retention(value = RetentionPolicy.RUNTIME)
-@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE, ElementType.TYPE_USE })
 public @interface Indexed {
 
 	/**

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

@@ -51,6 +51,11 @@ public class PrimaryKeyClassEntityMetadataVerifier implements CassandraPersisten
 
 		Class<?> entityType = entity.getType();
 
+		// @Indexed not allowed on type level
+		if (entity.isAnnotationPresent(Indexed.class)) {
+			exceptions.add(new MappingException("@Indexed cannot be used on primary key classes"));
+		}
+
 		// Ensure entity is not both a @Table(@Persistent) and a @PrimaryKey
 		if (entity.isAnnotationPresent(Table.class)) {
 			exceptions.add(new MappingException(String.format("Entity cannot be of type @%s and @%s",

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java

@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.springframework.data.cassandra.config;
 
 import static org.assertj.core.api.Assertions.*;
@@ -30,7 +29,6 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
-import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Session;
@@ -62,10 +60,6 @@ public class CassandraSessionFactoryBeanUnitTests {
 		factoryBean.setCluster(mockCluster);
 	}
 
-	protected CqlIdentifier newCqlIdentifier(String id) {
-		return new CqlIdentifier(id, false);
-	}
-
 	@Test // DATACASS-219
 	public void afterPropertiesSetPerformsSchemaAction() throws Exception {
 
@@ -99,7 +93,7 @@ public class CassandraSessionFactoryBeanUnitTests {
 	}
 
 	private void performSchemaActionCallsCreateTableWithArgumentsMatchingTheSchemaAction(SchemaAction schemaAction,
-			final boolean dropTables, final boolean dropUnused, final boolean ifNotExists) {
+			boolean dropTables, boolean dropUnused, boolean ifNotExists) {
 
 		doAnswer(invocationOnMock -> {
 			assertThat(invocationOnMock.<Boolean> getArgument(0)).isEqualTo(dropTables);

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

@@ -28,8 +28,11 @@ import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.core.cql.CqlOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.Indexed;
+import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
 
 /**
@@ -115,6 +118,19 @@ public class CassandraPersistentEntitySchemaCreatorUnitTests {
 		verifyTypesGetCreatedInOrderFor("universetype", "moontype", "planettype");
 	}
 
+	@Test // DATACASS-213
+	public void createsIndexes() {
+
+		context.getPersistentEntity(IndexedEntity.class);
+
+		CassandraPersistentEntitySchemaCreator schemaCreator = new CassandraPersistentEntitySchemaCreator(context,
+				adminOperations);
+
+		schemaCreator.createIndexes(false);
+
+		verify(operations).execute("CREATE INDEX ON indexedentity (firstname);");
+	}
+
 	private void verifyTypesGetCreatedInOrderFor(String... typenames) {
 
 		InOrder inOrder = Mockito.inOrder(operations);
@@ -149,4 +165,11 @@ public class CassandraPersistentEntitySchemaCreatorUnitTests {
 	static class SpaceAgencyType {
 		List<AstronautType> astronauts;
 	}
+
+	@Table
+	static class IndexedEntity {
+
+		@Id String id;
+		@Indexed String firstName;
+	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGeneratorIntegrationTests.java

@@ -1,71 +0,0 @@
-/*
- * Copyright 2017 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.cql.generator;
-
-import org.junit.Test;
-import org.springframework.data.cassandra.core.cql.generator.CreateIndexCqlGeneratorUnitTests.BasicTest;
-import org.springframework.data.cassandra.core.cql.generator.CreateIndexCqlGeneratorUnitTests.CreateIndexTest;
-import org.springframework.data.cassandra.support.CqlDataSet;
-import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
-
-/**
- * Integration tests that reuse unit tests.
- *
- * @author Matthew T. Adams
- * @author Oliver Gierke
- * @author Mark Paluch
- */
-public class CreateIndexCqlGeneratorIntegrationTests {
-
-	/**
-	 * Integration test base class that knows how to do everything except instantiate the concrete unit test type T.
-	 *
-	 * @author Matthew T. Adams
-	 * @param <T> The concrete unit test class to which this integration test corresponds.
-	 */
-	public static abstract class Base<T extends CreateIndexTest> extends AbstractKeyspaceCreatingIntegrationTest {
-		T unit;
-
-		public abstract T unit();
-
-		@Test
-		public void test() {
-			unit = unit();
-			unit.prepare();
-
-			session.execute(unit.cql);
-
-			CqlIndexSpecificationAssertions.assertIndex(unit.specification, keyspace, session);
-		}
-	}
-
-	public static class BasicIntegrationTest extends Base<BasicTest> {
-
-		public BasicIntegrationTest() {
-
-			cassandraRule.before(
-					CqlDataSet.fromClassPath("integration/cql/generator/CreateIndexCqlGeneratorIntegrationTests-BasicTest.cql")
-							.executeIn(this.keyspace));
-		}
-
-		@Override
-		public BasicTest unit() {
-			return new BasicTest();
-		}
-
-	}
-
-}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGeneratorUnitTests.java

@@ -25,54 +25,44 @@ import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecifica
  *
  * @author Matthew T. Adams
  * @author David Webb
+ * @author Mark Paluch
  */
 public class CreateIndexCqlGeneratorUnitTests {
 
-	/**
-	 * Asserts that the preamble is first & correctly formatted in the given CQL string.
-	 */
-	public static void assertPreamble(String indexName, String tableName, String cql) {
-		assertThat(cql.startsWith("CREATE INDEX " + indexName + " ON " + tableName)).isTrue();
+	@Test // DATACASS-213
+	public void createIndex() {
+
+		CreateIndexSpecification spec = CreateIndexSpecification.createIndex().name("myindex").tableName("mytable")
+				.columnName("column");
+
+		assertThat(CreateIndexCqlGenerator.toCql(spec)).isEqualTo("CREATE INDEX myindex ON mytable (column);");
 	}
 
-	/**
-	 * Asserts that the given list of columns definitions are contained in the given CQL string properly.
-	 *
-	 * @param columnName IE, "(foo)"
-	 */
-	public static void assertColumn(String columnName, String cql) {
-		assertThat(cql.contains("(" + columnName + ")")).isTrue();
+	@Test // DATACASS-213
+	public void createCustomIndex() {
+
+		CreateIndexSpecification spec = CreateIndexSpecification.createIndex().name("myindex").tableName("mytable")
+				.columnName("column").using("indexclass");
+
+		assertThat(CreateIndexCqlGenerator.toCql(spec))
+				.isEqualTo("CREATE CUSTOM INDEX myindex ON mytable (column) USING 'indexclass';");
 	}
 
-	/**
-	 * Convenient base class that other test classes can use so as not to repeat the generics declarations or
-	 * {@link #generator()} method.
-	 */
-	public static abstract class CreateIndexTest
-			extends AbstractIndexOperationCqlGeneratorTest<CreateIndexSpecification, CreateIndexCqlGenerator> {
+	@Test // DATACASS-213
+	public void createIndexOnKeys() {
 
-		public CreateIndexCqlGenerator generator() {
-			return new CreateIndexCqlGenerator(specification);
-		}
+		CreateIndexSpecification spec = CreateIndexSpecification.createIndex().tableName("mytable").keys()
+				.columnName("column");
+
+		assertThat(CreateIndexCqlGenerator.toCql(spec)).isEqualTo("CREATE INDEX ON mytable (KEYS(column));");
 	}
 
-	public static class BasicTest extends CreateIndexTest {
+	@Test // DATACASS-213
+	public void createIndexIfNotExists() {
 
-		public String name = "myindex";
-		public String tableName = "mytable";
-		public String column1 = "column1";
+		CreateIndexSpecification spec = CreateIndexSpecification.createIndex().tableName("mytable").columnName("column")
+				.ifNotExists();
 
-		public CreateIndexSpecification specification() {
-			return CreateIndexSpecification.createIndex().name(name).tableName(tableName).columnName(column1);
-		}
-
-		@Test
-		public void test() {
-			prepare();
-
-			assertPreamble(name, tableName, cql);
-			assertColumn(column1, cql);
-		}
+		assertThat(CreateIndexCqlGenerator.toCql(spec)).isEqualTo("CREATE INDEX IF NOT EXISTS ON mytable (column);");
 	}
-
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/IndexLifecycleCqlGeneratorIntegrationTests.java

@@ -1,61 +0,0 @@
-/*
- * Copyright 2017 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.cql.generator;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
-
-/**
- * Integration tests that reuse unit tests.
- *
- * @author Matthew T. Adams
- * @author Oliver Gierke
- * @author Mark Paluch
- */
-public class IndexLifecycleCqlGeneratorIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
-
-	private static final Logger log = LoggerFactory.getLogger(IndexLifecycleCqlGeneratorIntegrationTests.class);
-
-	@Before
-	public void setUp() throws Exception {
-		execute("integration/cql/generator/CreateIndexCqlGeneratorIntegrationTests-BasicTest.cql", this.keyspace);
-	}
-
-	@Test
-	public void lifecycleTest() {
-
-		CreateIndexCqlGeneratorUnitTests.BasicTest createTest = new CreateIndexCqlGeneratorUnitTests.BasicTest();
-		DropIndexCqlGeneratorUnitTests.BasicTest dropTest = new DropIndexCqlGeneratorUnitTests.BasicTest();
-		DropIndexCqlGeneratorUnitTests.IfExistsTest dropIfExists = new DropIndexCqlGeneratorUnitTests.IfExistsTest();
-
-		createTest.prepare();
-		dropTest.prepare();
-		dropIfExists.prepare();
-
-		log.info(createTest.cql);
-		session.execute(createTest.cql);
-
-		CqlIndexSpecificationAssertions.assertIndex(createTest.specification, keyspace, session);
-
-		log.info(dropTest.cql);
-		session.execute(dropTest.cql);
-
-		CqlIndexSpecificationAssertions.assertNoIndex(createTest.specification, keyspace, session);
-	}
-}

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

@@ -107,6 +107,17 @@ public class BasicCassandraPersistentEntityMetadataVerifierUnitTests {
 		}
 	}
 
+	@Test // DATACASS-213
+	public void shouldFailOnIndexedEntity() {
+
+		try {
+			verifier.verify(context.getRequiredPersistentEntity(InvalidIndexedPerson.class));
+			fail("Missing MappingException");
+		} catch (MappingException e) {
+			assertThat(e).hasMessageContaining("@Indexed cannot be used on entity classes");
+		}
+	}
+
 	interface MyInterface {}
 
 	static class NonPersistentClass {
@@ -126,6 +137,15 @@ public class BasicCassandraPersistentEntityMetadataVerifierUnitTests {
 		String lastName;
 	}
 
+	@Table
+	@Indexed
+	static class InvalidIndexedPerson {
+
+		@Id String id;
+
+		String firstName;
+	}
+
 	@Table
 	static class Animal {
 

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

@@ -23,17 +23,22 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.annotation.AccessType;
+import org.springframework.data.annotation.AccessType.Type;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.core.convert.CassandraCustomConversions;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.core.cql.Ordering;
 import org.springframework.data.cassandra.core.cql.PrimaryKeyType;
 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.convert.WritingConverter;
 import org.springframework.data.mapping.MappingException;
@@ -298,6 +303,133 @@ public class CassandraMappingContextUnitTests {
 		@PrimaryKey PrimaryKeyWithOrderedClusteredColumns key;
 	}
 
+	@Test // DATACASS-213
+	public void createIndexShouldConsiderAnnotatedProperties() {
+
+		List<CreateIndexSpecification> specifications = mappingContext
+				.getCreateIndexSpecificationsFor(mappingContext.getRequiredPersistentEntity(IndexedType.class));
+
+		CreateIndexSpecification firstname = getSpecificationFor("first_name", specifications);
+
+		assertThat(firstname.getColumnName()).isEqualTo(CqlIdentifier.cqlId("first_name"));
+		assertThat(firstname.getTableName()).isEqualTo(CqlIdentifier.cqlId("indexedtype"));
+		assertThat(firstname.getName()).isEqualTo(CqlIdentifier.cqlId("my_index"));
+		assertThat(firstname.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+
+		CreateIndexSpecification entries = getSpecificationFor("entries", specifications);
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getTableName()).isEqualTo(CqlIdentifier.cqlId("indexedtype"));
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.ENTRIES);
+
+		CreateIndexSpecification phoneNumbers = getSpecificationFor("phoneNumbers", specifications);
+
+		assertThat(phoneNumbers.getColumnName()).isEqualTo(CqlIdentifier.cqlId("phoneNumbers"));
+		assertThat(phoneNumbers.getTableName()).isEqualTo(CqlIdentifier.cqlId("indexedtype"));
+		assertThat(phoneNumbers.getName()).isNull();
+		assertThat(phoneNumbers.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+	}
+
+	@Test // DATACASS-213
+	public void createMapKeyIndexShouldConsiderAnnotatedAccessors() {
+
+		List<CreateIndexSpecification> specifications = mappingContext
+				.getCreateIndexSpecificationsFor(mappingContext.getRequiredPersistentEntity(IndexedMapKeyProperty.class));
+
+		CreateIndexSpecification entries = getSpecificationFor("entries", specifications);
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getTableName()).isEqualTo(CqlIdentifier.cqlId("indexedmapkeyproperty"));
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.KEYS);
+	}
+
+	@Test // DATACASS-213
+	public void createMapValueIndexShouldConsiderAnnotatedAccessors() {
+
+		List<CreateIndexSpecification> specifications = mappingContext
+				.getCreateIndexSpecificationsFor(mappingContext.getRequiredPersistentEntity(MapValueIndexProperty.class));
+
+		CreateIndexSpecification entries = getSpecificationFor("entries", specifications);
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getTableName()).isEqualTo(CqlIdentifier.cqlId("mapvalueindexproperty"));
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.VALUES);
+	}
+
+	@Test // DATACASS-213
+	public void createIndexForClusteredPrimaryKeyShouldConsiderAnnotatedAccessors() {
+
+		List<CreateIndexSpecification> specifications = mappingContext
+				.getCreateIndexSpecificationsFor(mappingContext.getRequiredPersistentEntity(CompositeKeyEntity.class));
+
+		CreateIndexSpecification entries = getSpecificationFor("last_name", specifications);
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("last_name"));
+		assertThat(entries.getTableName()).isEqualTo(CqlIdentifier.cqlId("compositekeyentity"));
+		assertThat(entries.getName()).isEqualTo(CqlIdentifier.cqlId("my_index"));
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+	}
+
+	private static CreateIndexSpecification getSpecificationFor(String column,
+			List<CreateIndexSpecification> specifications) {
+
+		return specifications.stream().filter(it -> it.getColumnName().equals(CqlIdentifier.cqlId(column))).findFirst()
+				.orElseThrow(() -> new NoSuchElementException(column));
+	}
+
+	static class IndexedType {
+
+		@PrimaryKeyColumn("first_name") @Indexed("my_index") String firstname;
+
+		@Indexed List<String> phoneNumbers;
+
+		@Indexed Map<String, String> entries;
+
+		Map<@Indexed String, String> keys;
+
+		Map<String, @Indexed String> values;
+	}
+
+	@AccessType(Type.PROPERTY)
+	static class IndexedMapKeyProperty {
+
+		public Map<@Indexed String, String> getEntries() {
+			return null;
+		}
+
+		public void setEntries(Map<String, String> entries) {}
+	}
+
+	@AccessType(Type.PROPERTY)
+	static class MapValueIndexProperty {
+
+		public Map<String, String> getEntries() {
+			return null;
+		}
+
+		public void setEntries(Map<String, @Indexed String> entries) {}
+	}
+
+	@PrimaryKeyClass
+	static class CompositeKeyWithIndex {
+
+		@PrimaryKeyColumn(value = "first_name", type = PrimaryKeyType.PARTITIONED) String firstname;
+		@PrimaryKeyColumn("last_name") @Indexed("my_index") String lastname;
+	}
+
+	static class CompositeKeyEntity {
+
+		@PrimaryKey CompositeKeyWithIndex key;
+	}
+
+	static class InvalidMapIndex {
+
+		@Indexed Map<@Indexed String, String> mixed;
+	}
+
 	@Test // DATACASS-296
 	public void shouldCreatePersistentEntityIfNoConversionRegistered() {
 
@@ -542,5 +674,4 @@ public class CassandraMappingContextUnitTests {
 			return "serialized";
 		}
 	}
-
 }

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

@@ -142,6 +142,17 @@ public class PrimaryKeyClassEntityMetadataVerifierUnitTests {
 		}
 	}
 
+	@Test // DATACASS-213
+	public void shouldFailForIndexedPrimaryKey() {
+
+		try {
+			verifier.verify(getEntity(InvalidIndexedPrimaryKeyType.class));
+			fail("Missing MappingException");
+		} catch (MappingException e) {
+			assertThat(e).hasMessageContaining("@Indexed cannot be used on primary key classes");
+		}
+	}
+
 	private CassandraPersistentEntity<?> getEntity(Class<?> entityClass) {
 		return context.getRequiredPersistentEntity(entityClass);
 	}
@@ -236,6 +247,13 @@ public class PrimaryKeyClassEntityMetadataVerifierUnitTests {
 		@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 0) String pk;
 	}
 
+	@PrimaryKeyClass
+	@Indexed
+	static class InvalidIndexedPrimaryKeyType {
+
+		@PrimaryKeyColumn(ordinal = 0, type = PrimaryKeyType.PARTITIONED) String species;
+	}
+
 	private static class NoOpVerifier implements CassandraPersistentEntityMetadataVerifier {
 
 		@Override

src/main/asciidoc/new-features.adoc

@@ -7,6 +7,7 @@
 * CRUD repository interface renaming: `CassandraRepository` using `MapId` is now renamed to `MapIdCassandraRepository`. `TypedIdCassandraRepository` is renamed to `CassandraRepository`.
 * Lightweight transactions via `InsertOptions` and `UpdateOptions` using the Template API.
 * Merge of Spring CQL into Spring Data Cassandra.
+* Index creation on application startup via `@Indexed`.
 
 [[new-features.1-5-0]]
 == What's new in Spring Data for Apache Cassandra 1.5

src/main/asciidoc/reference/cassandra.adoc

@@ -672,7 +672,7 @@ These entity classes can be used to create Cassandra table specifications and us
 Schema creation is tied to `Session` initialization with `SchemaAction`. Following actions are supported:
 
 * `SchemaAction.NONE`: No tables/types will be created or dropped. This is the default setting.
-* `SchemaAction.CREATE`: Create tables and user-defined types from entities annotated with `@Table` and types annotated with `@UserDefinedType`. Existing tables/types will cause an error if the type is attempted to be created.
+* `SchemaAction.CREATE`: Create tables, indexesm and user-defined types from entities annotated with `@Table` and types annotated with `@UserDefinedType`. Existing tables/types will cause an error if the type is attempted to be created.
 * `SchemaAction.CREATE_IF_NOT_EXISTS`: Like `SchemaAction.CREATE` but with `IF NOT EXISTS` applied. Existing tables/types won't cause any errors but may remain stale.
 * `SchemaAction.RECREATE`: Drops and recreate existing tables and types that are known to be used. Tables and types that are not configured in the application are not dropped.
 * `SchemaAction.RECREATE_DROP_UNUSED`: Drop all tables and types and recreate only known tables and types.

src/main/asciidoc/reference/mapping.adoc

@@ -245,6 +245,7 @@ be referenced with `@PrimaryKey`.
 where it is applied from being stored in the database.
 * `@Column` - applied at the field level. Describes the column name as it will be represented in the Cassandra table
 thus allowing the name to be different than the field name of the class.
+* `@Indexed` - applied at the field level. Describes the index to be created at session initialization.
 * `@CassandraType` - applied at the field level to specify a Cassandra data type. Types are derived from
 the declaration by default.
 * `@UserDefinedType` - applied at the type level to specify a Cassandra user-defined data type (UDT). Types are derived
@@ -286,6 +287,7 @@ public class Person {
   private String firstName;
 
   @Column(forceQuote = true)
+  @Indexed
   private String lastName;
 
   private Address address;
@@ -299,7 +301,7 @@ public class Person {
   @CassandraType(type = Name.SET, typeArguments = Name.BIGINT)
   private Set<Long> timestamps;
 
-  private Map<String, InetAddress> sessions;
+  private Map<@Indexed String, InetAddress> sessions;
 
   public Person(Integer ssn) {
     this.ssn = ssn;
@@ -346,6 +348,37 @@ public class Address {
 NOTE: Working with User-Defined Types requires a `UserTypeResolver` configured with the mapping context.
 See the <<cassandra.connectors,configuration chapter>> for how to configure a `UserTypeResolver`.
 
+==== Index creation
+
+You can annotate particular entity properties with `@Indexed` if you whish to create secondary indexes on application
+startup. Index creation will create simple secondary indexes for scalar types, user-defined, and collection types.
+
+Map types distinguish between `ENTRY`, `KEYS` and `VALUES` indexes. Index creation derives the index type from the
+annotated element:
+
+.Variants of map indexing
+====
+[source,java]
+----
+@Table
+public class Person {
+
+  @Id
+  private String key;
+
+  @Indexed("indexed_map")
+  private Map<String, String> entries;
+
+  private Map<@Indexed String, String> keys;
+
+  private Map<String, @Indexed String> values;
+
+  // …
+}
+----
+====
+
+WARNING: Index creation on session initialization may have a severe performance impact on application startup.
 
 [[cassandra.mapping.explicit-converters]]
 === Overriding Mapping with explicit Converters