DATACASS-306 - Create SASI indexes for annotated properties.

We now create SASI (SSTable Attached Secondary Index) indexes during session initialization for properties annotated with @SASI. Properties can be annotated with analyzer-annotations to configure analyzers. @Table public class Person { @Id String id; @SASI String names; @SASI @StandardAnalyzed(value = "de", enableStemming = true, normalization = Normalization.UPPERCASE, skipStopWords = true) String profession; @SASI @NonTokenizingAnalyzed String country; }
2017-07-21 09:36:51 +02:00
parent 4a0abd0f66
commit 9d3e12e3c4
12 changed files with 852 additions and 152 deletions
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGenerator.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGenerator.java
@@ -17,8 +17,14 @@ package org.springframework.data.cassandra.core.cql.generator;

 import static org.springframework.data.cassandra.core.cql.CqlStringUtils.*;

+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import 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;
+import org.springframework.util.StringUtils;

 /**
 * CQL generator for generating a {@code CREATE INDEX} statement.
@@ -63,8 +69,25 @@ public class CreateIndexCqlGenerator extends IndexNameCqlGenerator<CreateIndexSp
 			cql.append(" USING ").append("'").append(spec().getUsing()).append("'");
 		}

+		Map<String, String> options = spec().getOptions();
+
+		if (!options.isEmpty()) {
+
+			List<String> entries = new ArrayList<>(options.size());
+
+			options.forEach((key, value) -> entries
+					.add(String.format("'%s': '%s'", CqlStringUtils.escapeSingle(key), CqlStringUtils.escapeSingle(value))));
+
+			StringBuilder optionsCql = new StringBuilder(" WITH OPTIONS = ").append("{");
+			optionsCql.append(StringUtils.collectionToDelimitedString(entries, ", "));
+
+			optionsCql.append("}");
+			cql.append(optionsCql);
+		}
+
 		cql.append(";");

 		return cql;
 	}
+
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/keyspace/CreateIndexSpecification.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/keyspace/CreateIndexSpecification.java
@@ -17,6 +17,11 @@ package org.springframework.data.cassandra.core.cql.keyspace;

 import static org.springframework.data.cassandra.core.cql.CqlIdentifier.*;

+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -43,6 +48,8 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex

 	private String using;

+	private Map<String, String> options = new LinkedHashMap<>();
+
 	/**
 	 * Entry point into the {@link CreateIndexSpecification}'s fluent API to create a index. Convenient if imported
 	 * statically.
@@ -170,10 +177,14 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 	/**
 	 * Set a {@link ColumnFunction} such as {@code KEYS(…)}, {@code ENTRIES(…)}.
 	 *
+	 * @param columnFunction column function to apply, must not be {@literal null}.
 	 * @return this
 	 * @since 2.0
 	 */
 	public CreateIndexSpecification columnFunction(ColumnFunction columnFunction) {
+
+		Assert.notNull(columnFunction, "ColumnFunction must not be null");
+
 		this.columnFunction = columnFunction;
 		return this;
 	}
@@ -182,15 +193,43 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		return columnFunction;
 	}

+	/**
+	 * Configure a Index-creation options using key-value pairs.
+	 *
+	 * @param name option name.
+	 * @param value option value.
+	 * @return this
+	 * @since 2.0
+	 */
+	public CreateIndexSpecification withOption(String name, String value) {
+
+		this.options.put(name, value);
+		return this;
+	}
+
+	/**
+	 * @return index options map.
+	 */
+	public Map<String, String> getOptions() {
+		return Collections.unmodifiableMap(options);
+	}
+
 	/**
 	 * Sets the table name.
 	 *
+	 * @param tableName must not be {@literal null} or empty.
 	 * @return this
 	 */
 	public CreateIndexSpecification tableName(String tableName) {
 		return tableName(cqlId(tableName));
 	}

+	/**
+	 * Sets the table name.
+	 *
+	 * @param tableName must not be {@literal null}.
+	 * @return this
+	 */
 	public CreateIndexSpecification tableName(CqlIdentifier tableName) {

 		Assert.notNull(tableName, "CqlIdentifier must not be null");
@@ -207,10 +246,22 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		return tableName;
 	}

+	/**
+	 * Sets the column name.
+	 *
+	 * @param columnName must not be {@literal null} or empty.
+	 * @return this
+	 */
 	public CreateIndexSpecification columnName(String columnName) {
 		return columnName(cqlId(columnName));
 	}

+	/**
+	 * Sets the column name.
+	 *
+	 * @param columnName must not be {@literal null}.
+	 * @return this
+	 */
 	public CreateIndexSpecification columnName(CqlIdentifier columnName) {

 		Assert.notNull(columnName, "CqlIdentifier must not be null");
@@ -225,6 +276,30 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 	 * @since 2.0
 	 */
 	public enum ColumnFunction {
-		NONE, KEYS, VALUES, ENTRIES, FULL,
+
+		/**
+		 * Use the plain column value for indexing.
+		 */
+		NONE,
+
+		/**
+		 * Index keys for {@link Map} typed columns.
+		 */
+		KEYS,
+
+		/**
+		 * Index values for {@link Map} typed columns.
+		 */
+		VALUES,
+
+		/**
+		 * Index keys and values (entry-level indexing) for {@link Map} typed columns.
+		 */
+		ENTRIES,
+
+		/**
+		 * Index the entire {@link Collection}/{@link Map} as-is to match on whole collections/maps as predicate.
+		 */
+		FULL
 	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
@@ -19,8 +19,6 @@ 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;
@@ -442,82 +440,14 @@ public class CassandraMappingContext
 			if (property.isCompositePrimaryKey()) {
 				indexes.addAll(getCreateIndexSpecifications(tableName, getRequiredPersistentEntity(property)));
 			} else {
-				indexes.addAll(createIndexSpecifications(tableName, property));
+				indexes.addAll(IndexSpecificationFactory.createIndexSpecifications(property));
 			}
 		}
+
+		indexes.forEach(it -> it.tableName(entity.getTableName()));
 		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.
 	 *
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/IndexSpecificationFactory.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/IndexSpecificationFactory.java
@@ -0,0 +1,221 @@
+/*
+ * 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.mapping;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.AnnotatedParameterizedType;
+import java.lang.reflect.AnnotatedType;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
+import org.springframework.data.cassandra.core.mapping.SASI.NonTokenizingAnalyzed;
+import org.springframework.data.cassandra.core.mapping.SASI.Normalization;
+import org.springframework.data.cassandra.core.mapping.SASI.StandardAnalyzed;
+import org.springframework.data.mapping.MappingException;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Factory to create {@link org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification} based on
+ * index-annotated {@link CassandraPersistentProperty properties}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see Indexed
+ * @see SASI
+ */
+@SuppressWarnings("unchecked")
+class IndexSpecificationFactory {
+
+	private final static Map<Class<? extends Annotation>, CreateIndexConfigurer<? super Annotation>> INDEX_CONFIGURERS;
+
+	static {
+
+		Map<Class<? extends Annotation>, CreateIndexConfigurer<? extends Annotation>> configurers = new HashMap<>();
+
+		configurers.put(StandardAnalyzed.class, StandardAnalyzedConfigurer.INSTANCE);
+		configurers.put(NonTokenizingAnalyzed.class, NonTokenizingAnalyzedConfigurer.INSTANCE);
+
+		INDEX_CONFIGURERS = (Map) Collections.unmodifiableMap(configurers);
+	}
+
+	/**
+	 * Create a {@link List} of {@link CreateIndexSpecification} for a {@link CassandraPersistentProperty}. The resulting
+	 * specifications are configured according the index annotations but do not configure
+	 * {@link CreateIndexSpecification#tableName(String)}.
+	 *
+	 * @param property must not be {@literal null}.
+	 * @return {@link List} of {@link CreateIndexSpecification}.
+	 */
+	static List<CreateIndexSpecification> createIndexSpecifications(CassandraPersistentProperty property) {
+
+		List<CreateIndexSpecification> indexes = new ArrayList<>();
+
+		if (property.isAnnotationPresent(Indexed.class)) {
+
+			CreateIndexSpecification index = createIndexSpecification(property.findAnnotation(Indexed.class), property);
+
+			if (property.isMapLike()) {
+				index.entries();
+			}
+
+			indexes.add(index);
+		}
+
+		if (property.isAnnotationPresent(SASI.class)) {
+			indexes.add(createIndexSpecification(property.findAnnotation(SASI.class), property));
+		}
+
+		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) {
+					indexes.add(createIndexSpecification(keyIndex, property).keys());
+				}
+
+				if (valueIndex != null) {
+					indexes.add(createIndexSpecification(valueIndex, property).values());
+				}
+			}
+		}
+
+		return indexes;
+	}
+
+	private static CreateIndexSpecification createIndexSpecification(Indexed annotation,
+			CassandraPersistentProperty property) {
+
+		CreateIndexSpecification index;
+
+		if (StringUtils.hasText(annotation.value())) {
+			index = CreateIndexSpecification.createIndex(annotation.value());
+		} else {
+			index = CreateIndexSpecification.createIndex();
+		}
+
+		return index.columnName(property.getColumnName());
+	}
+
+	private static CreateIndexSpecification createIndexSpecification(SASI annotation,
+			CassandraPersistentProperty property) {
+		CreateIndexSpecification index;
+
+		if (StringUtils.hasText(annotation.value())) {
+			index = CreateIndexSpecification.createIndex(annotation.value());
+		} else {
+			index = CreateIndexSpecification.createIndex();
+		}
+
+		index.using("org.apache.cassandra.index.sasi.SASIIndex") //
+				.columnName(property.getColumnName()) //
+				.withOption("mode", annotation.indexMode().name());
+
+		long analyzerCount = INDEX_CONFIGURERS.keySet().stream().filter(property::isAnnotationPresent).count();
+
+		if (analyzerCount > 1) {
+			throw new IllegalStateException(
+					String.format("SASI indexed property %s must be annotated only with a single analyzer annotation", property));
+		}
+
+		for (Class<? extends Annotation> annotationType : INDEX_CONFIGURERS.keySet()) {
+
+			if (!property.isAnnotationPresent(annotationType)) {
+				continue;
+			}
+
+			Annotation analyzed = property.findAnnotation(annotationType);
+			INDEX_CONFIGURERS.get(annotationType).accept(analyzed, index);
+		}
+
+		return index;
+	}
+
+	interface CreateIndexConfigurer<T extends Annotation> extends BiConsumer<T, CreateIndexSpecification> {}
+
+	enum StandardAnalyzedConfigurer implements CreateIndexConfigurer<StandardAnalyzed> {
+
+		INSTANCE;
+
+		@Override
+		public void accept(StandardAnalyzed standardAnalyzed, CreateIndexSpecification index) {
+
+			index.withOption("analyzed", "true");
+			index.withOption("analyzer_class", "org.apache.cassandra.index.sasi.analyzer.StandardAnalyzer");
+			index.withOption("tokenization_enable_stemming", "" + standardAnalyzed.enableStemming());
+
+			if (standardAnalyzed.normalization() == Normalization.LOWERCASE) {
+				index.withOption("tokenization_normalize_lowercase", "true");
+			}
+
+			if (standardAnalyzed.normalization() == Normalization.UPPERCASE) {
+				index.withOption("tokenization_normalize_uppercase", "true");
+			}
+
+			if (StringUtils.hasText(standardAnalyzed.locale())) {
+				index.withOption("tokenization_locale", standardAnalyzed.locale());
+			}
+
+			if (!ObjectUtils.isEmpty(standardAnalyzed.skipStopWords())) {
+				index.withOption("tokenization_skip_stop_words", "" + standardAnalyzed.skipStopWords());
+			}
+		}
+	}
+
+	enum NonTokenizingAnalyzedConfigurer implements CreateIndexConfigurer<NonTokenizingAnalyzed> {
+
+		INSTANCE;
+
+		@Override
+		public void accept(NonTokenizingAnalyzed nonTokenizingAnalyzed, CreateIndexSpecification index) {
+
+			index.withOption("analyzed", "true");
+			index.withOption("analyzer_class", "org.apache.cassandra.index.sasi.analyzer.NonTokenizingAnalyzer");
+			index.withOption("case_sensitive", "" + nonTokenizingAnalyzed.caseSensitive());
+
+			if (nonTokenizingAnalyzed.normalization() == Normalization.LOWERCASE) {
+				index.withOption("normalize_lowercase", "true");
+			}
+
+			if (nonTokenizingAnalyzed.normalization() == Normalization.UPPERCASE) {
+				index.withOption("normalize_uppercase", "true");
+			}
+		}
+	}
+}
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/Indexed.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/Indexed.java
@@ -31,8 +31,8 @@ import java.lang.annotation.Target;
 * <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.
+ * Map types distinguish allows entry, key or value-level indexing with the restriction that only a single secondary
+ * index is allowed.
 *
 * <pre class="code">
 * &#64;Table
@@ -46,6 +46,7 @@ import java.lang.annotation.Target;
 * @author Alex Shvid
 * @author Matthew T. Adams
 * @author Mark Paluch
+ * @see Table
 */
@Retention(value = RetentionPolicy.RUNTIME)
@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE, ElementType.TYPE_USE })
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/SASI.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/SASI.java
@@ -0,0 +1,162 @@
+/*
+ * 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.mapping;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+
+/**
+ * Identifies a secondary index using SASI indexing on a single column.
+ * <p>
+ * SASI uses significantly using fewer memory, disk, and CPU resources. It enables querying with {@literal PREFIX} and
+ * {@literal CONTAINS} on strings, similar to the SQL implementation of {@literal LIKE = "foo*"} or
+ * {@literal LIKE = "*foo*"}.
+ *
+ * <pre class="code">
+ * &#64;Table
+ * class Person {
+ *
+ * 	&#64;SASI(analyzed = true, indexMode = CONTAINS) @StandardAnalyzed("en") String names; // allows LIKE queries
+ * 	&#64;SASI int age // allows age >= … queries;
+ * }
+ * </pre>
+ * <p>
+ * SASI indexing can apply an analyzer that is applied during indexing. Annotation-based indexing supports
+ * {@link StandardAnalyzed standard analyzer} and {@link NonTokenizingAnalyzed non-tokenizing analyzer}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see StandardAnalyzed
+ * @see NonTokenizingAnalyzed
+ */
+@Retention(value = RetentionPolicy.RUNTIME)
+@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+public @interface SASI {
+
+	/**
+	 * The name of the index. If {@literal null} or empty, then the index name will be generated by Cassandra and will be
+	 * unknown unless column metadata is used to discover the generated index name.
+	 */
+	String value() default "";
+
+	/**
+	 * The name of the index. If {@literal null} or empty, then the index name will be generated by Cassandra and will be
+	 * unknown unless column metadata is used to discover the generated index name.
+	 */
+	IndexMode indexMode() default IndexMode.PREFIX;
+
+	enum IndexMode {
+
+		/**
+		 * Allows prefix queries.
+		 */
+		PREFIX,
+
+		/**
+		 * Allows prefix, suffix and substring queries.
+		 */
+		CONTAINS,
+
+		/**
+		 * SPARSE mode is optimized for low-cardinality e.g. for indexed values having {@literal 5} or less corresponding
+		 * rows.
+		 */
+		SPARSE
+	}
+
+	/**
+	 * Apply standard analyzer to SASI indexing. This analyzer is used for analysis that involves stemming, case
+	 * normalization, case sensitivity, skipping common words like "and" and "the", and localization of the language used
+	 * to complete the analysis
+	 *
+	 * @see org.apache.cassandra.index.sasi.analyzer.StandardAnalyzer.
+	 */
+	@Retention(value = RetentionPolicy.RUNTIME)
+	@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+	@interface StandardAnalyzed {
+
+		/**
+		 * Defines the locale for tokenization.
+		 */
+		@AliasFor("locale")
+		String value() default "en";
+
+		/**
+		 * Defines the locale for tokenization.
+		 */
+		@AliasFor("value")
+		String locale() default "en";
+
+		/**
+		 * Enable stemming of input text to reduce words to their base form, for example
+		 * {@literal stemmer, stemming, stemmed} are based on {@literal stem}.
+		 */
+		boolean enableStemming() default false;
+
+		/**
+		 * Skips stop words from indexing.
+		 */
+		boolean skipStopWords() default false;
+
+		/**
+		 * Applies normalization to uppercase/lowercase.
+		 */
+		Normalization normalization() default Normalization.NONE;
+	}
+
+	/**
+	 * Apply non-tokenizing analyzer to SASI indexing. Use this analyzer for cases where the text is not analyzed, but
+	 * case normalization or sensitivity is required.
+	 *
+	 * @see org.apache.cassandra.index.sasi.analyzer.NonTokenizingAnalyzer
+	 */
+	@Retention(value = RetentionPolicy.RUNTIME)
+	@Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+	@interface NonTokenizingAnalyzed {
+
+		/**
+		 * Enable case-sensitive matching.
+		 */
+		boolean caseSensitive() default true;
+
+		/**
+		 * Applies normalization to uppercase/lowercase.
+		 */
+		Normalization normalization() default Normalization.NONE;
+	}
+
+	enum Normalization {
+
+		/**
+		 * Do not apply normalization.
+		 */
+		NONE,
+
+		/**
+		 * Normalize to lowercase.
+		 */
+		LOWERCASE,
+
+		/**
+		 * Normalize to uppercase.
+		 */
+		UPPERCASE;
+	}
+}
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGeneratorUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateIndexCqlGeneratorUnitTests.java
@@ -65,4 +65,14 @@ public class CreateIndexCqlGeneratorUnitTests {

 		assertThat(CreateIndexCqlGenerator.toCql(spec)).isEqualTo("CREATE INDEX IF NOT EXISTS ON mytable (column);");
 	}
+
+	@Test // DATACASS-306
+	public void createIndexWithOptions() {
+
+		CreateIndexSpecification spec = CreateIndexSpecification.createIndex().tableName("mytable").columnName("column")
+				.withOption("foo", "b'a'r").withOption("type", "PREFIX");
+
+		assertThat(CreateIndexCqlGenerator.toCql(spec))
+				.isEqualTo("CREATE INDEX ON mytable (column) WITH OPTIONS = {'foo': 'b''a''r', 'type': 'PREFIX'};");
+	}
 }
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java
@@ -29,8 +29,6 @@ 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;
@@ -316,13 +314,6 @@ public class CassandraMappingContextUnitTests {
 		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"));
@@ -331,34 +322,6 @@ public class CassandraMappingContextUnitTests {
 		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() {

@@ -385,32 +348,6 @@ public class CassandraMappingContextUnitTests {
 		@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
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/IndexCreationIntegrationTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/IndexCreationIntegrationTests.java
@@ -0,0 +1,110 @@
+/*
+ * 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.mapping;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.junit.Assume.*;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.data.annotation.Id;
+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.keyspace.CreateIndexSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
+import org.springframework.data.cassandra.core.mapping.SASI.StandardAnalyzed;
+import org.springframework.data.cassandra.support.CassandraVersion;
+import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
+import org.springframework.data.util.Version;
+
+import com.datastax.driver.core.TableMetadata;
+
+/**
+ * Integration tests usin {@link CassandraMappingContext} and {@link CreateIndexSpecification} to integratively verify
+ * index creation.
+ *
+ * @author Mark Paluch
+ */
+public class IndexCreationIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
+
+	private CassandraMappingContext mappingContext = new CassandraMappingContext();
+	private Version cassandraVersion;
+
+	@Before
+	public void before() {
+
+		cassandraVersion = CassandraVersion.get(session);
+
+		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(Version.parse("3.4")));
+	}
+
+	@Test
+	public void shouldCreateSecondaryIndex() throws InterruptedException {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(WithSecondaryIndex.class);
+		CreateTableSpecification createTable = mappingContext.getCreateTableSpecificationFor(entity);
+		List<CreateIndexSpecification> createIndexes = mappingContext.getCreateIndexSpecificationsFor(entity);
+
+		session.execute(CreateTableCqlGenerator.toCql(createTable));
+		createIndexes.forEach(it -> session.execute(CreateIndexCqlGenerator.toCql(it)));
+
+		Thread.sleep(500); // index creation is async so we do poor man's sync to await completion
+
+		TableMetadata metadata = getMetadata(createTable.getName().toCql());
+
+		assertThat(metadata.getIndex("firstname_index")).isNotNull();
+		assertThat(metadata.getIndex("withsecondaryindex_map_idx")).isNotNull();
+	}
+
+	@Test
+	public void shouldCreateSasiIndex() throws InterruptedException {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(WithSasiIndex.class);
+		CreateTableSpecification createTable = mappingContext.getCreateTableSpecificationFor(entity);
+		List<CreateIndexSpecification> createIndexes = mappingContext.getCreateIndexSpecificationsFor(entity);
+
+		session.execute(CreateTableCqlGenerator.toCql(createTable));
+		createIndexes.forEach(it -> session.execute(CreateIndexCqlGenerator.toCql(it)));
+
+		Thread.sleep(500); // index creation is async so we do poor man's sync to await completion
+
+		TableMetadata metadata = getMetadata(createTable.getName().toCql());
+		assertThat(metadata.getIndex("withsasiindex_firstname_idx")).isNotNull();
+	}
+
+	private TableMetadata getMetadata(String tableName) {
+		return session.getCluster().getMetadata().getKeyspace(session.getLoggedKeyspace()).getTable(tableName);
+	}
+
+	static class WithSecondaryIndex {
+
+		@Id String id;
+
+		@Indexed("firstname_index") String firstname;
+
+		Map<String, @Indexed String> map;
+	}
+
+	static class WithSasiIndex {
+
+		@Id String id;
+
+		@SASI @StandardAnalyzed String firstname;
+	}
+}
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/IndexSpecificationFactoryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/IndexSpecificationFactoryUnitTests.java
@@ -0,0 +1,205 @@
+/*
+ * 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.mapping;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Test;
+import org.springframework.data.annotation.AccessType;
+import org.springframework.data.annotation.AccessType.Type;
+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.CreateIndexSpecification.ColumnFunction;
+import org.springframework.data.cassandra.core.mapping.SASI.NonTokenizingAnalyzed;
+import org.springframework.data.cassandra.core.mapping.SASI.Normalization;
+import org.springframework.data.cassandra.core.mapping.SASI.StandardAnalyzed;
+
+/**
+ * Unit tests for {@link IndexSpecificationFactory}.
+ *
+ * @author Mark Paluch
+ */
+public class IndexSpecificationFactoryUnitTests {
+
+	CassandraMappingContext mappingContext = new CassandraMappingContext();
+
+	@Test // DATACASS-213
+	public void createIndexShouldConsiderAnnotatedProperties() {
+
+		CreateIndexSpecification firstname = createIndexFor(IndexedType.class, "firstname");
+
+		assertThat(firstname.getColumnName()).isEqualTo(CqlIdentifier.cqlId("first_name"));
+		assertThat(firstname.getTableName()).isNull();
+		assertThat(firstname.getName()).isEqualTo(CqlIdentifier.cqlId("my_index"));
+		assertThat(firstname.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+
+		CreateIndexSpecification entries = createIndexFor(IndexedType.class, "entries");
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getTableName()).isNull();
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.ENTRIES);
+	}
+
+	@Test // DATACASS-213
+	public void createMapKeyIndexShouldConsiderAnnotatedAccessors() {
+
+		CreateIndexSpecification entries = createIndexFor(IndexedMapKeyProperty.class, "entries");
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.KEYS);
+	}
+
+	@Test // DATACASS-213
+	public void createMapValueIndexShouldConsiderAnnotatedAccessors() {
+
+		CreateIndexSpecification entries = createIndexFor(MapValueIndexProperty.class, "entries");
+
+		assertThat(entries.getColumnName()).isEqualTo(CqlIdentifier.cqlId("entries"));
+		assertThat(entries.getName()).isNull();
+		assertThat(entries.getColumnFunction()).isEqualTo(ColumnFunction.VALUES);
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForSimpleSasiShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "simpleSasi");
+
+		assertThat(simpleSasi.getColumnName()).isEqualTo(CqlIdentifier.cqlId("simplesasi"));
+		assertThat(simpleSasi.isCustom()).isTrue();
+		assertThat(simpleSasi.getUsing()).isEqualTo("org.apache.cassandra.index.sasi.SASIIndex");
+		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(simpleSasi.getOptions()).containsEntry("mode", "PREFIX").doesNotContainKeys("analyzed",
+				"analyzer_class");
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForStandardAnalyzedSasiShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "sasiStandard");
+
+		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(simpleSasi.getOptions()).containsEntry("mode", "PREFIX") //
+				.containsEntry("analyzed", "true") //
+				.containsEntry("tokenization_skip_stop_words", "false") //
+				.containsEntry("analyzer_class", "org.apache.cassandra.index.sasi.analyzer.StandardAnalyzer") //
+				.containsEntry("tokenization_locale", "de");
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForStandardAnalyzedSasiWithOptionsShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "sasiStandardWithOptions");
+
+		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(simpleSasi.getOptions()).containsEntry("tokenization_skip_stop_words", "true") //
+				.containsEntry("tokenization_locale", "de") //
+				.containsEntry("tokenization_enable_stemming", "true") //
+				.containsEntry("tokenization_normalize_uppercase", "true") //
+				.doesNotContainKey("tokenization_normalize_lowercase");
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForStandardAnalyzedSasiWithLowercaseShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "sasiStandardLowercase");
+
+		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(simpleSasi.getOptions()).containsEntry("tokenization_normalize_lowercase", "true") //
+				.doesNotContainKey("tokenization_normalize_uppercase");
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForNonTokenizingAnalyzedSasiShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "sasiNontokenizing");
+
+		assertThat(simpleSasi.getOptions()).containsEntry("mode", "PREFIX") //
+				.containsEntry("analyzed", "true") //
+				.containsEntry("case_sensitive", "true") //
+				.containsEntry("analyzer_class", "org.apache.cassandra.index.sasi.analyzer.NonTokenizingAnalyzer") //
+				.doesNotContainKeys("normalize_lowercase", "normalize_uppercase");
+	}
+
+	@Test // DATACASS-306
+	public void createIndexForNonTokenizingAnalyzedSasiWithLowercaseShouldApplyIndexOptions() {
+
+		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "sasiNontokenizingLowercase");
+
+		assertThat(simpleSasi.getOptions()).containsEntry("normalize_lowercase", "true") //
+				.containsEntry("case_sensitive", "false") //
+				.doesNotContainKey("normalize_uppercase");
+
+	}
+
+	private CreateIndexSpecification createIndexFor(Class<?> type, String property) {
+		return IndexSpecificationFactory.createIndexSpecifications(getProperty(type, property)).stream().findFirst()
+				.orElse(null);
+	}
+
+	private CassandraPersistentProperty getProperty(Class<?> type, String property) {
+		return mappingContext.getRequiredPersistentEntity(type).getRequiredPersistentProperty(property);
+	}
+
+	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;
+
+		@SASI String simpleSasi;
+		@SASI @StandardAnalyzed("de") String sasiStandard;
+		@SASI @StandardAnalyzed(value = "de", enableStemming = true, normalization = Normalization.UPPERCASE,
+				skipStopWords = true) String sasiStandardWithOptions;
+
+		@SASI @StandardAnalyzed(normalization = Normalization.LOWERCASE) String sasiStandardLowercase;
+
+		@SASI @NonTokenizingAnalyzed String sasiNontokenizing;
+
+		@SASI @NonTokenizingAnalyzed(caseSensitive = false,
+				normalization = Normalization.LOWERCASE) String sasiNontokenizingLowercase;
+	}
+
+	@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) {}
+	}
+}
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java
@@ -35,6 +35,8 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
 import org.springframework.data.cassandra.config.SchemaAction;
 import org.springframework.data.cassandra.core.CassandraOperations;
+import org.springframework.data.cassandra.core.cql.generator.CreateIndexCqlGenerator;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
 import org.springframework.data.cassandra.domain.AddressType;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.QueryDerivationIntegrationTests.PersonRepository.NumberOfChildren;
@@ -140,7 +142,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC
 	@Test // DATACASS-172
 	public void shouldFindByMappedUdt() throws InterruptedException {

-		template.getCqlOperations().execute("CREATE INDEX IF NOT EXISTS person_main_address ON person (mainaddress);");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("person_main_address")
+				.ifNotExists().tableName("person").columnName("mainaddress");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -153,7 +158,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC
 	@Test // DATACASS-172
 	public void shouldFindByMappedUdtStringQuery() throws InterruptedException {

-		template.getCqlOperations().execute("CREATE INDEX IF NOT EXISTS person_main_address ON person (mainaddress);");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("person_main_address")
+				.ifNotExists().tableName("person").columnName("mainaddress");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -186,8 +194,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC

 		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(Version.parse("3.4")));

-		template.getCqlOperations().execute(
-				"CREATE CUSTOM INDEX IF NOT EXISTS fn_starts_with ON person (nickname) USING 'org.apache.cassandra.index.sasi.SASIIndex';");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("fn_starts_with").ifNotExists()
+				.tableName("person").columnName("nickname").using("org.apache.cassandra.index.sasi.SASIIndex");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -204,8 +214,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC
 	@Test // DATACASS-7
 	public void shouldFindByNumberOfChildren() throws Exception {

-		template.getCqlOperations()
-				.execute("CREATE INDEX IF NOT EXISTS person_number_of_children ON person (numberofchildren);");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("person_number_of_children")
+				.ifNotExists().tableName("person").columnName("numberofchildren");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -218,7 +230,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC
 	@Test // DATACASS-7
 	public void shouldFindByLocalDate() throws InterruptedException {

-		template.getCqlOperations().execute("CREATE INDEX IF NOT EXISTS person_created_date ON person (createddate);");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("person_created_date")
+				.ifNotExists().tableName("person").columnName("createddate");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -248,8 +263,10 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC

 		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(Version.parse("3.4")));

-		template.getCqlOperations().execute(
-				"CREATE CUSTOM INDEX IF NOT EXISTS fn_starts_with ON person (nickname) USING 'org.apache.cassandra.index.sasi.SASIIndex';");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("fn_starts_with").ifNotExists()
+				.tableName("person").columnName("nickname").using("org.apache.cassandra.index.sasi.SASIIndex");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
@@ -265,9 +282,11 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC

 		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(Version.parse("3.4")));

-		template.getCqlOperations().execute(
-				"CREATE CUSTOM INDEX IF NOT EXISTS fn_contains ON person (nickname) USING 'org.apache.cassandra.index.sasi.SASIIndex'\n"
-						+ "WITH OPTIONS = { 'mode': 'CONTAINS' };");
+		CreateIndexSpecification indexSpecification = CreateIndexSpecification.createIndex("fn_contains").ifNotExists()
+				.tableName("person").columnName("nickname").using("org.apache.cassandra.index.sasi.SASIIndex")
+				.withOption("mode", "CONTAINS");
+
+		template.getCqlOperations().execute(CreateIndexCqlGenerator.toCql(indexSpecification));

 		// Give Cassandra some time to build the index
 		Thread.sleep(500);
--- a/src/main/asciidoc/reference/mapping.adoc
+++ b/src/main/asciidoc/reference/mapping.adoc
@@ -246,6 +246,7 @@ 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.
+* `@SASI` - applied at the field level. Allows SASI index creation during 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
@@ -350,9 +351,12 @@ See the <<cassandra.connectors,configuration chapter>> for how to configure a `U

 ==== Index creation

-You can annotate particular entity properties with `@Indexed` if you whish to create secondary indexes on application
+You can annotate particular entity properties with `@Indexed` or `@SASI` 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.

+You can configure a SASI index to apply an analyzer such as `StandardAnalyzer` or `NonTokenizingAnalyzer` via
+`@StandardAnalyzed` respective `@NonTokenizingAnalyzed`.
+
 Map types distinguish between `ENTRY`, `KEYS` and `VALUES` indexes. Index creation derives the index type from the
 annotated element:

@@ -366,6 +370,9 @@ public class Person {
  @Id
  private String key;

+  @SASI @StandardAnalyzed
+  private String names;
+
  @Indexed("indexed_map")
  private Map<String, String> entries;