DATACASS-146 - Enhance Repository methods to accept QueryOptions as arguments.

We now support Repository query methods with query options. Query options can be passed either as an additional parameter to a Repository query method or applied with annotation. Annotation-based query options are supported via @Consistency. A query options parameter has precedence over the annotation if a method declares both, an annotation-based consistency level and accepts a query options parameter. interface SampleRepository extends Repository<Person, String> { @Query("SELECT * FROM person WHERE lastname = ?0;") @Consistency(ConsistencyLevel.LOCAL_ONE) Person findByLastname(String lastname); @Consistency(ConsistencyLevel.LOCAL_ONE) Person findByAge(int age); Person findByAge(int age, QueryOptions options); } SampleRepository repository = …; repository.findByAge(42, QueryOptions.builder().fetchSize(44).build());
2017-08-29 17:48:59 +02:00
parent d00ff35b19
commit 0ee5190f09
23 changed files with 674 additions and 255 deletions
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/Consistency.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/Consistency.java
@@ -0,0 +1,45 @@
+/*
+ * 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.repository;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.data.annotation.QueryAnnotation;
+
+import com.datastax.driver.core.ConsistencyLevel;
+
+/**
+ * Annotation to declare a {@link ConsistencyLevel} for CQL queries executed through query methods.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see org.springframework.data.cassandra.core.cql.QueryOptions
+ */
+@Target({ ElementType.ANNOTATION_TYPE, ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@QueryAnnotation
+public @interface Consistency {
+
+	/**
+	 * @return the {@link ConsistencyLevel} applied to the query executed using a query method.
+	 */
+	ConsistencyLevel value();
+}
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java
@@ -15,12 +15,6 @@
 */
 package org.springframework.data.cassandra.repository.query;

-import lombok.RequiredArgsConstructor;
-
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.CollectionExecution;
@@ -29,15 +23,11 @@ import org.springframework.data.cassandra.repository.query.CassandraQueryExecuti
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.ResultSetQuery;
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.SingleEntityExecution;
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.StreamExecution;
-import org.springframework.data.convert.CustomConversions;
-import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
-import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;

 import com.datastax.driver.core.Statement;

@@ -47,16 +37,10 @@ import com.datastax.driver.core.Statement;
 * @author Mark Paluch
 * @author John Blum
 */
-public abstract class AbstractCassandraQuery implements RepositoryQuery {
-
-	protected static Logger log = LoggerFactory.getLogger(AbstractCassandraQuery.class);
-
-	private final CassandraQueryMethod queryMethod;
+public abstract class AbstractCassandraQuery extends CassandraRepositoryQuerySupport {

 	private final CassandraOperations operations;

-	private final EntityInstantiators instantiators;
-
 	/**
 	 * Create a new {@link AbstractCassandraQuery} from the given {@link CassandraQueryMethod} and
 	 * {@link CassandraOperations}.
@@ -66,32 +50,17 @@ public abstract class AbstractCassandraQuery implements RepositoryQuery {
 	 */
 	public AbstractCassandraQuery(CassandraQueryMethod queryMethod, CassandraOperations operations) {

-		Assert.notNull(queryMethod, "CassandraQueryMethod must not be null");
+		super(queryMethod);
+
 		Assert.notNull(operations, "CassandraOperations must not be null");

-		this.queryMethod = queryMethod;
 		this.operations = operations;
-		this.instantiators = new EntityInstantiators();
 	}

-	/* (non-Javadoc) */
-	private EntityInstantiators getEntityInstantiators() {
-		return this.instantiators;
-	}
-
-	/* (non-Javadoc) */
 	protected CassandraOperations getOperations() {
 		return this.operations;
 	}

-	/* (non-Javadoc)
-	 * @see org.springframework.data.repository.query.RepositoryQuery#getQueryMethod()
-	 */
-	@Override
-	public CassandraQueryMethod getQueryMethod() {
-		return this.queryMethod;
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.query.RepositoryQuery#execute(java.lang.Object[])
 	 */
@@ -117,6 +86,13 @@ public abstract class AbstractCassandraQuery implements RepositoryQuery {
 		return queryExecution.execute(statement, resultType);
 	}

+	/**
+	 * Creates a {@link Statement} using the given {@link ParameterAccessor}
+	 *
+	 * @param accessor must not be {@literal null}.
+	 */
+	protected abstract Statement createQuery(CassandraParameterAccessor accessor);
+
 	/**
 	 * Returns the execution instance to use.
 	 *
@@ -138,47 +114,4 @@ public abstract class AbstractCassandraQuery implements RepositoryQuery {
 			return new SingleEntityExecution(getOperations());
 		}
 	}
-
-	/**
-	 * Creates a {@link Statement} using the given {@link ParameterAccessor}
-	 *
-	 * @param accessor must not be {@literal null}.
-	 */
-	protected abstract Statement createQuery(CassandraParameterAccessor accessor);
-
-	@RequiredArgsConstructor
-	private class CassandraReturnedType {
-
-		private final ReturnedType returnedType;
-		private final CustomConversions customConversions;
-
-		boolean isProjecting() {
-
-			if (!returnedType.isProjecting()) {
-				return false;
-			}
-
-			// Spring Data Cassandra allows List<Map<String, Object> and Map<String, Object> declarations
-			// on query methods so we don't want to let projection kick in
-			if (ClassUtils.isAssignable(Map.class, returnedType.getReturnedType())) {
-				return false;
-			}
-
-			// Type conversion using registered conversions is handled on template level
-			if (customConversions.hasCustomWriteTarget(returnedType.getReturnedType())) {
-				return false;
-			}
-
-			// Don't apply projection on Cassandra simple types
-			return !customConversions.isSimpleType(returnedType.getReturnedType());
-		}
-
-		Class<?> getDomainType() {
-			return returnedType.getDomainType();
-		}
-
-		Class<?> getReturnedType() {
-			return returnedType.getReturnedType();
-		}
-	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java
@@ -26,7 +26,6 @@ import org.springframework.data.cassandra.repository.query.ReactiveCassandraQuer
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingConverter;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingExecution;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.SingleEntityExecution;
-import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
@@ -40,14 +39,10 @@ import com.datastax.driver.core.Statement;
 * @author Mark Paluch
 * @since 2.0
 */
-public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery {
-
-	private final ReactiveCassandraQueryMethod method;
+public abstract class AbstractReactiveCassandraQuery extends CassandraRepositoryQuerySupport {

 	private final ReactiveCassandraOperations operations;

-	private final EntityInstantiators instantiators;
-
 	/**
 	 * Create a new {@link AbstractReactiveCassandraQuery} from the given {@link CassandraQueryMethod} and
 	 * {@link CassandraOperations}.
@@ -57,17 +52,15 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 	 */
 	public AbstractReactiveCassandraQuery(ReactiveCassandraQueryMethod method, ReactiveCassandraOperations operations) {

-		Assert.notNull(method, "ReactiveCassandraQueryMethod must not be null");
+		super(method);
+
 		Assert.notNull(operations, "ReactiveCassandraOperations must not be null");

-		this.method = method;
 		this.operations = operations;
-		this.instantiators = new EntityInstantiators();
 	}

-	/* (non-Javadoc) */
-	protected EntityInstantiators getEntityInstantiators() {
-		return this.instantiators;
+	protected ReactiveCassandraOperations getReactiveCassandraOperations() {
+		return this.operations;
 	}

 	/*
@@ -76,12 +69,7 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 	 */
 	@Override
 	public ReactiveCassandraQueryMethod getQueryMethod() {
-		return this.method;
-	}
-
-	/* (non-Javadoc) */
-	protected ReactiveCassandraOperations getReactiveCassandraOperations() {
-		return this.operations;
+		return (ReactiveCassandraQueryMethod) super.getQueryMethod();
 	}

 	/*
@@ -95,6 +83,13 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 				: execute(new ReactiveCassandraParameterAccessor(getQueryMethod(), parameters)));
 	}

+	/**
+	 * Creates a string query using the given {@link ParameterAccessor}
+	 *
+	 * @param accessor must not be {@literal null}.
+	 */
+	protected abstract Statement createQuery(CassandraParameterAccessor accessor);
+
 	@SuppressWarnings("unchecked")
 	private Object executeDeferred(Object[] parameters) {

@@ -125,13 +120,6 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 		return queryExecution.execute(statement, resultType);
 	}

-	/**
-	 * Creates a string query using the given {@link ParameterAccessor}
-	 *
-	 * @param accessor must not be {@literal null}.
-	 */
-	protected abstract Statement createQuery(CassandraParameterAccessor accessor);
-
 	/**
 	 * Returns the execution instance to use.
 	 *
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java
@@ -15,6 +15,7 @@
 */
 package org.springframework.data.cassandra.repository.query;

+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.lang.Nullable;
@@ -69,4 +70,13 @@ public interface CassandraParameterAccessor extends ParameterAccessor {
 	 * @since 1.5
 	 */
 	Object[] getValues();
+
+	/**
+	 * Returns the {@link QueryOptions} associated of the associated query method.
+	 *
+	 * @return the {@link QueryOptions} or {@literal null} if none.
+	 * @since 2.0
+	 */
+	@Nullable
+	QueryOptions getQueryOptions();
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java
@@ -18,11 +18,13 @@ package org.springframework.data.cassandra.repository.query;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Method;
+import java.util.Arrays;
 import java.util.List;

 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.cassandra.repository.query.CassandraParameters.CassandraParameter;
 import org.springframework.data.repository.query.Parameter;
@@ -39,6 +41,8 @@ import org.springframework.lang.Nullable;
 */
 public class CassandraParameters extends Parameters<CassandraParameters, CassandraParameter> {

+	private final @Nullable Integer queryOptionsIndex;
+
 	/**
 	 * Create a new {@link CassandraParameters} instance from the given {@link Method}
 	 *
@@ -46,10 +50,16 @@ public class CassandraParameters extends Parameters<CassandraParameters, Cassand
 	 */
 	public CassandraParameters(Method method) {
 		super(method);
+
+		List<Class<?>> parameterTypes = Arrays.asList(method.getParameterTypes());
+
+		this.queryOptionsIndex = parameterTypes.indexOf(QueryOptions.class);
 	}

-	private CassandraParameters(List<CassandraParameter> originals) {
+	private CassandraParameters(List<CassandraParameter> originals, @Nullable Integer queryOptionsIndex) {
 		super(originals);
+
+		this.queryOptionsIndex = queryOptionsIndex;
 	}

 	/* (non-Javadoc)
@@ -65,7 +75,17 @@ public class CassandraParameters extends Parameters<CassandraParameters, Cassand
 	 */
 	@Override
 	protected CassandraParameters createFrom(List<CassandraParameter> parameters) {
-		return new CassandraParameters(parameters);
+		return new CassandraParameters(parameters, queryOptionsIndex);
+	}
+
+	/**
+	 * Returns the index of the {@link QueryOptions} parameter to be applied to queries.
+	 *
+	 * @return
+	 * @since 2.0
+	 */
+	public int getQueryOptionsIndex() {
+		return queryOptionsIndex != null ? queryOptionsIndex : -1;
 	}

 	/**
@@ -78,7 +98,7 @@ public class CassandraParameters extends Parameters<CassandraParameters, Cassand
 		private final @Nullable CassandraType cassandraType;
 		private final Class<?> parameterType;

-		protected CassandraParameter(MethodParameter parameter) {
+		CassandraParameter(MethodParameter parameter) {

 			super(parameter);

@@ -93,6 +113,14 @@ public class CassandraParameters extends Parameters<CassandraParameters, Cassand
 			parameterType = potentiallyUnwrapParameterType(parameter);
 		}

+		/* (non-Javadoc)
+		 * @see org.springframework.data.repository.query.Parameter#isSpecialParameter()
+		 */
+		@Override
+		public boolean isSpecialParameter() {
+			return super.isSpecialParameter() || QueryOptions.class.isAssignableFrom(getType());
+		}
+
 		/**
 		 * Returns the {@link CassandraType} for the declared parameter if specified using
 		 * {@link org.springframework.data.cassandra.core.mapping.CassandraType}.
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java
@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.repository.query;
 import java.util.Arrays;
 import java.util.List;

+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.repository.query.ParameterAccessor;
@@ -51,15 +52,6 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 		this.values = Arrays.asList(values);
 	}

-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#findCassandraType(int)
-	 */
-	@Nullable
-	public CassandraType findCassandraType(int index) {
-		return getParameters().getParameter(index).getCassandraType();
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#getDataType(int)
@@ -75,11 +67,11 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc

 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.repository.query.ParametersParameterAccessor#getParameters()
+	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#findCassandraType(int)
 	 */
-	@Override
-	public CassandraParameters getParameters() {
-		return (CassandraParameters) super.getParameters();
+	@Nullable
+	public CassandraType findCassandraType(int index) {
+		return getParameters().getParameter(index).getCassandraType();
 	}

 	/*
@@ -91,6 +83,15 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 		return getParameters().getParameter(index).getType();
 	}

+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.ParametersParameterAccessor#getParameters()
+	 */
+	@Override
+	public CassandraParameters getParameters() {
+		return (CassandraParameters) super.getParameters();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.repository.query.CassandraParameterAccessor#getValues()
@@ -99,4 +100,27 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 	public Object[] getValues() {
 		return values.toArray();
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.repository.query.CassandraParameterAccessor#getQueryOptions()
+	 */
+	@Nullable
+	@Override
+	public QueryOptions getQueryOptions() {
+
+		int queryOptionsIndex = getParameters().getQueryOptionsIndex();
+
+		if (queryOptionsIndex == -1) {
+			return null;
+		}
+
+		Object value = getValue(queryOptionsIndex);
+
+		if (value == null) {
+			return null;
+		}
+
+		return (QueryOptions) value;
+	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryCreator.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryCreator.java
@@ -115,10 +115,6 @@ class CassandraQueryCreator extends AbstractQueryCreator<Query, CriteriaDefiniti
 	@Override
 	protected CriteriaDefinition and(Part part, CriteriaDefinition base, Iterator<Object> iterator) {

-		if (base == null) {
-			return getQueryBuilder().and(create(part, iterator));
-		}
-
 		getQueryBuilder().and(base);

 		return create(part, iterator);
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java
@@ -22,6 +22,7 @@ import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.cassandra.repository.Consistency;
 import org.springframework.data.cassandra.repository.Query;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionFactory;
@@ -34,6 +35,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;

+import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.ResultSet;

 /**
@@ -52,6 +54,8 @@ public class CassandraQueryMethod extends QueryMethod {

 	private final Optional<Query> query;

+	private final Optional<Consistency> consistency;
+
 	private @Nullable CassandraEntityMetadata<?> entityMetadata;

 	/**
@@ -74,6 +78,7 @@ public class CassandraQueryMethod extends QueryMethod {
 		this.method = method;
 		this.mappingContext = mappingContext;
 		this.query = Optional.ofNullable(AnnotatedElementUtils.findMergedAnnotation(method, Query.class));
+		this.consistency = Optional.ofNullable(AnnotatedElementUtils.findMergedAnnotation(method, Consistency.class));
 	}

 	/**
@@ -156,6 +161,26 @@ public class CassandraQueryMethod extends QueryMethod {
 		return query.map(Query::value).orElse(null);
 	}

+	/**
+	 * @return whether the method has an annotated {@link com.datastax.driver.core.ConsistencyLevel}.
+	 * @since 2.0
+	 */
+	public boolean hasConsistencyLevel() {
+		return consistency.isPresent();
+	}
+
+	/**
+	 * Returns the {@link ConsistencyLevel} in a {@link Query} annotation or throws {@link IllegalStateException} if the
+	 * annotation was not found.
+	 *
+	 * @return the {@link ConsistencyLevel}.
+	 * @throws IllegalStateException if the required annotation was not found.
+	 */
+	public ConsistencyLevel getRequiredAnnotatedConsistencyLevel() throws IllegalStateException {
+		return consistency.map(Consistency::value)
+				.orElseThrow(() -> new IllegalStateException("No @Consistency annotation found"));
+	}
+
 	/**
 	 * Returns the required query string declared in a {@link Query} annotation or throws {@link IllegalStateException} if
 	 * neither the annotation found nor the attribute was specified.
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraRepositoryQuerySupport.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraRepositoryQuerySupport.java
@@ -0,0 +1,116 @@
+/*
+ * 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.repository.query;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.cassandra.core.CassandraOperations;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Base class for Cassandra {@link RepositoryQuery} implementations providing common infrastructure such as
+ * {@link EntityInstantiators} and {@link QueryStatementCreator}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+public abstract class CassandraRepositoryQuerySupport implements RepositoryQuery {
+
+	protected final Logger log = LoggerFactory.getLogger(getClass());
+
+	private final CassandraQueryMethod queryMethod;
+
+	private final EntityInstantiators instantiators;
+
+	private final QueryStatementCreator queryStatementCreator;
+
+	/**
+	 * Create a new {@link AbstractCassandraQuery} from the given {@link CassandraQueryMethod} and
+	 * {@link CassandraOperations}.
+	 *
+	 * @param queryMethod must not be {@literal null}.
+	 * @param operations must not be {@literal null}.
+	 */
+	public CassandraRepositoryQuerySupport(CassandraQueryMethod queryMethod) {
+
+		Assert.notNull(queryMethod, "CassandraQueryMethod must not be null");
+
+		this.queryMethod = queryMethod;
+		this.instantiators = new EntityInstantiators();
+		this.queryStatementCreator = new QueryStatementCreator(queryMethod);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.RepositoryQuery#getQueryMethod()
+	 */
+	@Override
+	public CassandraQueryMethod getQueryMethod() {
+		return this.queryMethod;
+	}
+
+	protected EntityInstantiators getEntityInstantiators() {
+		return this.instantiators;
+	}
+
+	protected QueryStatementCreator getQueryStatementCreator() {
+		return queryStatementCreator;
+	}
+
+	@RequiredArgsConstructor
+	class CassandraReturnedType {
+
+		private final ReturnedType returnedType;
+		private final CustomConversions customConversions;
+
+		boolean isProjecting() {
+
+			if (!returnedType.isProjecting()) {
+				return false;
+			}
+
+			// Spring Data Cassandra allows List<Map<String, Object> and Map<String, Object> declarations
+			// on query methods so we don't want to let projection kick in
+			if (ClassUtils.isAssignable(Map.class, returnedType.getReturnedType())) {
+				return false;
+			}
+
+			// Type conversion using registered conversions is handled on template level
+			if (customConversions.hasCustomWriteTarget(returnedType.getReturnedType())) {
+				return false;
+			}
+
+			// Don't apply projection on Cassandra simple types
+			return !customConversions.isSimpleType(returnedType.getReturnedType());
+		}
+
+		Class<?> getDomainType() {
+			return returnedType.getDomainType();
+		}
+
+		Class<?> getReturnedType() {
+			return returnedType.getReturnedType();
+		}
+	}
+}
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java
@@ -20,6 +20,7 @@ import java.util.Optional;
 import java.util.Set;

 import org.springframework.data.cassandra.core.convert.CassandraConverter;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
@@ -88,14 +89,6 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 		return potentiallyConvert(index, delegate.getBindableValue(index));
 	}

-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#findCassandraType(int)
-	 */
-	@Override
-	public CassandraType findCassandraType(int index) {
-		return delegate.findCassandraType(index);
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#getDataType(int)
 	 */
@@ -104,6 +97,15 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 		return delegate.getDataType(index);
 	}

+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#findCassandraType(int)
+	 */
+	@Override
+	public CassandraType findCassandraType(int index) {
+		return delegate.findCassandraType(index);
+	}
+
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#getParameterType(int)
 	 */
@@ -112,6 +114,15 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 		return delegate.getParameterType(index);
 	}

+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.repository.query.CassandraParameterAccessor#getQueryOptions()
+	 */
+	@Nullable
+	@Override
+	public QueryOptions getQueryOptions() {
+		return delegate.getQueryOptions();
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.query.ParameterAccessor#hasBindableNullValue()
 	 */
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQuery.java
@@ -22,9 +22,7 @@ import org.springframework.data.cassandra.core.convert.UpdateMapper;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
-import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.repository.query.QueryCreationException;
 import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.parser.PartTree;
@@ -96,30 +94,7 @@ public class PartTreeCassandraQuery extends AbstractCassandraQuery {
 	 */
 	@Override
 	protected Statement createQuery(CassandraParameterAccessor parameterAccessor) {
-
-		CassandraQueryCreator queryCreator = new CassandraQueryCreator(getTree(), parameterAccessor, getMappingContext());
-
-		Query query = queryCreator.createQuery();
-
-		try {
-
-			if (getTree().isLimiting()) {
-				query = query.limit(getTree().getMaxResults());
-			}
-
-			if (getQueryMethod().getQueryAnnotation().map(org.springframework.data.cassandra.repository.Query::allowFiltering)
-					.orElse(false)) {
-
-				query = query.withAllowFiltering();
-			}
-
-			CassandraPersistentEntity<?> persistentEntity = getMappingContext()
-					.getRequiredPersistentEntity(getQueryMethod().getDomainClass());
-
-			return getStatementFactory().select(query, persistentEntity);
-
-		} catch (RuntimeException e) {
-			throw QueryCreationException.create(getQueryMethod(), e);
-		}
+		return getQueryStatementCreator().select(getStatementFactory(), getTree(), getMappingContext(),
+				parameterAccessor);
 	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java
@@ -0,0 +1,139 @@
+/*
+ * 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.repository.query;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.cassandra.core.StatementFactory;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
+import org.springframework.data.cassandra.core.cql.QueryOptionsUtil;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.repository.query.QueryCreationException;
+import org.springframework.data.repository.query.parser.PartTree;
+
+import com.datastax.driver.core.RegularStatement;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.Statement;
+
+/**
+ * Creates {@link Statement}s for {@link CassandraQueryMethod query methods} based on {@link PartTree} and
+ * {@link StringBasedQuery}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+@RequiredArgsConstructor
+class QueryStatementCreator {
+
+	private static final Logger LOG = LoggerFactory.getLogger(QueryStatementCreator.class);
+
+	private final CassandraQueryMethod queryMethod;
+
+	/**
+	 * Create a {@literal SELECT} {@link Statement} from a {@link PartTree} and apply query options.
+	 *
+	 * @param statementFactory must not be {@literal null}.
+	 * @param tree must not be {@literal null}.
+	 * @param mappingContext must not be {@literal null}.
+	 * @param parameterAccessor must not be {@literal null}.
+	 * @return the {@literal SELECT} {@link Statement}.
+	 */
+	Statement select(StatementFactory statementFactory, PartTree tree,
+			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext,
+			CassandraParameterAccessor parameterAccessor) {
+
+		CassandraQueryCreator queryCreator = new CassandraQueryCreator(tree, parameterAccessor, mappingContext);
+
+		Query query = queryCreator.createQuery();
+
+		try {
+
+			if (tree.isLimiting()) {
+				query = query.limit(tree.getMaxResults());
+			}
+
+			if (queryMethod.getQueryAnnotation().map(org.springframework.data.cassandra.repository.Query::allowFiltering)
+					.orElse(false)) {
+
+				query = query.withAllowFiltering();
+			}
+
+			Optional<QueryOptions> queryOptions = Optional.ofNullable(parameterAccessor.getQueryOptions());
+
+			if (queryOptions.isPresent()) {
+				query = Optional.ofNullable(parameterAccessor.getQueryOptions()).map(query::queryOptions).orElse(query);
+			} else if (queryMethod.hasConsistencyLevel()) {
+				query = query.queryOptions(
+						QueryOptions.builder().consistencyLevel(queryMethod.getRequiredAnnotatedConsistencyLevel()).build());
+			}
+
+			CassandraPersistentEntity<?> persistentEntity = mappingContext
+					.getRequiredPersistentEntity(queryMethod.getDomainClass());
+
+			RegularStatement statement = statementFactory.select(query, persistentEntity);
+
+			if (LOG.isDebugEnabled()) {
+				LOG.debug(String.format("Created query [%s].", statement));
+			}
+
+			return statement;
+
+		} catch (RuntimeException e) {
+			throw QueryCreationException.create(queryMethod, e);
+		}
+	}
+
+	/**
+	 * Create a {@link Statement} from a {@link StringBasedQuery} and apply query options.
+	 *
+	 * @param stringBasedQuery must not be {@literal null}.
+	 * @param parameterAccessor must not be {@literal null}.
+	 * @return the {@link Statement}.
+	 */
+	SimpleStatement select(StringBasedQuery stringBasedQuery, CassandraParameterAccessor parameterAccessor) {
+
+		try {
+
+			SimpleStatement boundQuery = stringBasedQuery.bindQuery(parameterAccessor, queryMethod);
+
+			Optional<QueryOptions> queryOptions = Optional.ofNullable(parameterAccessor.getQueryOptions());
+
+			SimpleStatement queryToUse = boundQuery;
+
+			if (queryOptions.isPresent()) {
+				queryToUse = Optional.ofNullable(parameterAccessor.getQueryOptions())
+						.map(it -> QueryOptionsUtil.addQueryOptions(boundQuery, it)).orElse(boundQuery);
+			} else if (queryMethod.hasConsistencyLevel()) {
+				queryToUse.setConsistencyLevel(queryMethod.getRequiredAnnotatedConsistencyLevel());
+			}
+
+			if (LOG.isDebugEnabled()) {
+				LOG.debug(String.format("Created query [%s].", queryToUse));
+			}
+
+			return queryToUse;
+		} catch (RuntimeException e) {
+			throw QueryCreationException.create(queryMethod, e);
+		}
+	}
+}
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java
@@ -21,9 +21,7 @@ import org.springframework.data.cassandra.core.convert.UpdateMapper;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
-import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.repository.query.QueryCreationException;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.parser.PartTree;

@@ -96,30 +94,6 @@ public class ReactivePartTreeCassandraQuery extends AbstractReactiveCassandraQue
 	 */
 	@Override
 	protected Statement createQuery(CassandraParameterAccessor parameterAccessor) {
-
-		CassandraQueryCreator queryCreator = new CassandraQueryCreator(getTree(), parameterAccessor, getMappingContext());
-
-		Query query = queryCreator.createQuery();
-
-		try {
-
-			if (getTree().isLimiting()) {
-				query = query.limit(getTree().getMaxResults());
-			}
-
-			if (getQueryMethod().getQueryAnnotation().map(org.springframework.data.cassandra.repository.Query::allowFiltering)
-					.orElse(false)) {
-
-				query = query.withAllowFiltering();
-			}
-
-			CassandraPersistentEntity<?> persistentEntity = getMappingContext()
-					.getRequiredPersistentEntity(getQueryMethod().getDomainClass());
-
-			return getStatementFactory().select(query, persistentEntity);
-
-		} catch (RuntimeException e) {
-			throw QueryCreationException.create(getQueryMethod(), e);
-		}
+		return getQueryStatementCreator().select(statementFactory, getTree(), getMappingContext(), parameterAccessor);
 	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java
@@ -15,11 +15,8 @@
 */
 package org.springframework.data.cassandra.repository.query;

-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.repository.query.EvaluationContextProvider;
-import org.springframework.data.repository.query.QueryCreationException;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.util.Assert;

@@ -38,8 +35,6 @@ import com.datastax.driver.core.SimpleStatement;
 */
 public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandraQuery {

-	private static final Logger LOG = LoggerFactory.getLogger(ReactiveStringBasedCassandraQuery.class);
-
 	private final StringBasedQuery stringBasedQuery;

 	/**
@@ -81,7 +76,6 @@ public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandra
 				new ExpressionEvaluatingParameterBinder(expressionParser, evaluationContextProvider));
 	}

-	/* (non-Javadoc) */
 	protected StringBasedQuery getStringBasedQuery() {
 		return this.stringBasedQuery;
 	}
@@ -91,17 +85,6 @@ public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandra
 	 */
 	@Override
 	public SimpleStatement createQuery(CassandraParameterAccessor parameterAccessor) {
-
-		try {
-			SimpleStatement boundQuery = getStringBasedQuery().bindQuery(parameterAccessor, getQueryMethod());
-
-			if (LOG.isDebugEnabled()) {
-				LOG.debug(String.format("Created query [%s].", boundQuery));
-			}
-
-			return boundQuery;
-		} catch (RuntimeException e) {
-			throw QueryCreationException.create(getQueryMethod(), e);
-		}
+		return getQueryStatementCreator().select(getStringBasedQuery(), parameterAccessor);
 	}
 }
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQuery.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQuery.java
@@ -15,11 +15,8 @@
 */
 package org.springframework.data.cassandra.repository.query;

-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.repository.query.EvaluationContextProvider;
-import org.springframework.data.repository.query.QueryCreationException;
 import org.springframework.expression.spel.standard.SpelExpressionParser;

 import com.datastax.driver.core.SimpleStatement;
@@ -37,8 +34,6 @@ import com.datastax.driver.core.SimpleStatement;
 */
 public class StringBasedCassandraQuery extends AbstractCassandraQuery {

-	private static final Logger LOG = LoggerFactory.getLogger(StringBasedCassandraQuery.class);
-
 	private final StringBasedQuery stringBasedQuery;

 	/**
@@ -77,7 +72,6 @@ public class StringBasedCassandraQuery extends AbstractCassandraQuery {
 				new ExpressionEvaluatingParameterBinder(expressionParser, evaluationContextProvider));
 	}

-	/* (non-Javadoc) */
 	protected StringBasedQuery getStringBasedQuery() {
 		return this.stringBasedQuery;
 	}
@@ -87,17 +81,6 @@ public class StringBasedCassandraQuery extends AbstractCassandraQuery {
 	 */
 	@Override
 	public SimpleStatement createQuery(CassandraParameterAccessor parameterAccessor) {
-
-		try {
-			SimpleStatement boundQuery = getStringBasedQuery().bindQuery(parameterAccessor, getQueryMethod());
-
-			if (LOG.isDebugEnabled()) {
-				LOG.debug(String.format("Created query [%s].", boundQuery));
-			}
-
-			return boundQuery;
-		} catch (RuntimeException e) {
-			throw QueryCreationException.create(getQueryMethod(), e);
-		}
+		return getQueryStatementCreator().select(getStringBasedQuery(), parameterAccessor);
 	}
 }
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessorUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessorUnitTests.java
@@ -20,10 +20,12 @@ import static org.assertj.core.api.Assertions.*;
 import java.lang.reflect.Method;
 import java.util.List;

+import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.cassandra.domain.AllPossibleTypes;
@@ -49,7 +51,7 @@ public class CassandraParametersParameterAccessorUnitTests {
 	RepositoryMetadata metadata = new DefaultRepositoryMetadata(PossibleRepository.class);
 	CassandraMappingContext context = new CassandraMappingContext();

-	@Test // DATACASS-296
+	@Test // DATACASS-296, DATACASS-146
 	public void returnsCassandraSimpleType() throws Exception {

 		Method method = PossibleRepository.class.getMethod("findByFirstname", String.class);
@@ -57,6 +59,7 @@ public class CassandraParametersParameterAccessorUnitTests {
 				"firstname");

 		assertThat(accessor.getDataType(0)).isEqualTo(DataType.varchar());
+		assertThat(accessor.getQueryOptions()).isNull();
 	}

 	@Test // DATACASS-296
@@ -97,6 +100,18 @@ public class CassandraParametersParameterAccessorUnitTests {
 		assertThat(accessor.getDataType(0)).isEqualTo(DataType.date());
 	}

+	@Test // DATACASS-146
+	public void shouldProvideQueryOptions() throws Exception {
+
+		QueryOptions options = QueryOptions.builder().retryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE).build();
+
+		Method method = PossibleRepository.class.getMethod("findByFirstname", QueryOptions.class, String.class);
+		CassandraParameterAccessor accessor = new CassandraParametersParameterAccessor(getCassandraQueryMethod(method),
+				options, "firstname");
+
+		assertThat(accessor.getQueryOptions()).isEqualTo(options);
+	}
+
 	private CassandraQueryMethod getCassandraQueryMethod(Method method) {
 		return new CassandraQueryMethod(method, metadata, projectionFactory, context);
 	}
@@ -110,5 +125,7 @@ public class CassandraParametersParameterAccessorUnitTests {
 		List<AllPossibleTypes> findByAnnotatedBpLocalDateTime(@CassandraType(type = Name.DATE) LocalDateTime dateTime);

 		List<AllPossibleTypes> findByAnnotatedObject(@CassandraType(type = Name.DATE) Object dateTime);
+
+		List<AllPossibleTypes> findByFirstname(QueryOptions queryOptions, String firstname);
 	}
 }
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
@@ -34,12 +34,14 @@ import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.domain.AddressType;
 import org.springframework.data.cassandra.domain.Group;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.AllowFiltering;
+import org.springframework.data.cassandra.repository.Consistency;
 import org.springframework.data.cassandra.repository.MapIdCassandraRepository;
 import org.springframework.data.cassandra.repository.Query;
 import org.springframework.data.projection.ProjectionFactory;
@@ -47,6 +49,7 @@ import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
 import org.springframework.util.ClassUtils;

+import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.UDTValue;
 import com.datastax.driver.core.UserType;
@@ -176,6 +179,26 @@ public class PartTreeCassandraQueryUnitTests {
 		assertThat(query.toString()).isEqualTo("SELECT * FROM person WHERE firstname='foo' ALLOW FILTERING;");
 	}

+	@Test // DATACASS-146
+	public void shouldApplyQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().fetchSize(777).build();
+		Statement statement = deriveQueryFromMethod(Repo.class, "findByFirstname",
+				new Class[] { QueryOptions.class, String.class }, queryOptions, "Walter");
+
+		assertThat(statement.toString()).isEqualTo("SELECT * FROM person WHERE firstname='Walter';");
+		assertThat(statement.getFetchSize()).isEqualTo(777);
+	}
+
+	@Test // DATACASS-146
+	public void shouldApplyConsistencyLevel() {
+
+		Statement statement = deriveQueryFromMethod(Repo.class, "findPersonBy", new Class[0]);
+
+		assertThat(statement.toString()).isEqualTo("SELECT * FROM person;");
+		assertThat(statement.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_ONE);
+	}
+
 	private String deriveQueryFromMethod(String method, Object... args) {

 		Class<?>[] types = new Class<?>[args.length];
@@ -210,9 +233,7 @@ public class PartTreeCassandraQueryUnitTests {
 					new DefaultRepositoryMetadata(repositoryInterface), factory, mappingContext);

 			return new PartTreeCassandraQuery(queryMethod, mockCassandraOperations);
-		} catch (NoSuchMethodException e) {
-			throw new IllegalArgumentException(e.getMessage(), e);
-		} catch (SecurityException e) {
+		} catch (NoSuchMethodException | SecurityException e) {
 			throw new IllegalArgumentException(e.getMessage(), e);
 		}
 	}
@@ -233,8 +254,9 @@ public class PartTreeCassandraQueryUnitTests {

 		Person findPersonByFirstnameAndLastname(String firstname, String lastname);

-		Person findByAge(Integer age);
+		Person findByFirstname(QueryOptions queryOptions, String firstname);

+		@Consistency(ConsistencyLevel.LOCAL_ONE)
 		Person findPersonBy();

 		Person findByMainAddress(AddressType address);
@@ -251,7 +273,6 @@ public class PartTreeCassandraQueryUnitTests {
 		PersonProjection findPersonProjectedByNickname(String nickname);

 		<T> T findDynamicallyProjectedBy(Class<T> type);
-
 	}

 	interface PersonProjection {
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java
@@ -15,13 +15,17 @@
 */
 package org.springframework.data.cassandra.repository.query;

-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.when;
-
-import java.lang.reflect.Method;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;

+import com.datastax.driver.core.ConsistencyLevel;
+import org.springframework.data.cassandra.repository.Consistency;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import rx.Single;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;

 import org.junit.Before;
 import org.junit.Rule;
@@ -30,9 +34,9 @@ import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
-
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.domain.Person;
@@ -41,8 +45,9 @@ import org.springframework.data.cassandra.repository.Query;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
+import org.springframework.util.ClassUtils;

-import rx.Single;
+import com.datastax.driver.core.Statement;

 /**
 * Unit tests for {@link ReactivePartTreeCassandraQuery}.
@@ -70,6 +75,7 @@ public class ReactivePartTreeCassandraQueryUnitTests {

 	@Test // DATACASS-335
 	public void shouldDeriveSimpleQuery() {
+
 		String query = deriveQueryFromMethod("findByLastname", "foo");

 		assertThat(query).isEqualTo("SELECT * FROM person WHERE lastname='foo';");
@@ -77,6 +83,7 @@ public class ReactivePartTreeCassandraQueryUnitTests {

 	@Test // DATACASS-335
 	public void shouldDeriveSimpleQueryWithoutNames() {
+
 		String query = deriveQueryFromMethod("findPersonBy");

 		assertThat(query).isEqualTo("SELECT * FROM person;");
@@ -84,6 +91,7 @@ public class ReactivePartTreeCassandraQueryUnitTests {

 	@Test // DATACASS-335
 	public void shouldDeriveAndQuery() {
+
 		String query = deriveQueryFromMethod("findByFirstnameAndLastname", "foo", "bar");

 		assertThat(query).isEqualTo("SELECT * FROM person WHERE firstname='foo' AND lastname='bar';");
@@ -99,39 +107,67 @@ public class ReactivePartTreeCassandraQueryUnitTests {

 	@Test // DATACASS-335
 	public void usesDynamicProjection() {
+
 		String query = deriveQueryFromMethod("findDynamicallyProjectedBy", PersonProjection.class);

 		assertThat(query).isEqualTo("SELECT * FROM person;");
 	}

+	@Test // DATACASS-146
+	public void shouldApplyQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().fetchSize(777).build();
+		Statement statement = deriveQueryFromMethod(Repo.class, "findByFirstname",
+				new Class[] { QueryOptions.class, String.class }, queryOptions, "Walter");
+
+		assertThat(statement.toString()).isEqualTo("SELECT * FROM person WHERE firstname='Walter';");
+		assertThat(statement.getFetchSize()).isEqualTo(777);
+	}
+
+	@Test // DATACASS-146
+	public void shouldApplyConsistencyLevel() {
+
+		Statement statement = deriveQueryFromMethod(Repo.class, "findPersonBy", new Class[0]);
+
+		assertThat(statement.toString()).isEqualTo("SELECT * FROM person;");
+		assertThat(statement.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_ONE);
+	}
+
 	private String deriveQueryFromMethod(String method, Object... args) {

 		Class<?>[] types = new Class<?>[args.length];

 		for (int i = 0; i < args.length; i++) {
-			types[i] = args[i].getClass();
+			types[i] = ClassUtils.getUserClass(args[i].getClass());
 		}

-		ReactivePartTreeCassandraQuery partTreeQuery = createQueryForMethod(method, types);
-
-		CassandraParameterAccessor accessor =
-				new CassandraParametersParameterAccessor(partTreeQuery.getQueryMethod(), args);
-
-		return partTreeQuery.createQuery(
-				new ConvertingParameterAccessor(mockCassandraOperations.getConverter(), accessor)).toString();
+		return deriveQueryFromMethod(Repo.class, method, types, args).toString();
 	}

-	private ReactivePartTreeCassandraQuery createQueryForMethod(String methodName, Class<?>... paramTypes) {
+	private Statement deriveQueryFromMethod(Class<?> repositoryInterface, String method, Class<?>[] types,
+			Object... args) {
+
+		ReactivePartTreeCassandraQuery partTreeQuery = createQueryForMethod(repositoryInterface, method, types);
+
+		CassandraParameterAccessor accessor = new CassandraParametersParameterAccessor(partTreeQuery.getQueryMethod(),
+				args);
+
+		return partTreeQuery.createQuery(new ConvertingParameterAccessor(mockCassandraOperations.getConverter(), accessor));
+	}
+
+	private ReactivePartTreeCassandraQuery createQueryForMethod(Class<?> repositoryInterface, String methodName,
+			Class<?>... paramTypes) {
+		Class<?>[] userTypes = Arrays.stream(paramTypes)//
+				.map(it -> it.getName().contains("Mockito") ? it.getSuperclass() : it)//
+				.toArray(size -> new Class<?>[size]);
 		try {
-			Method method = Repo.class.getMethod(methodName, paramTypes);
+			Method method = repositoryInterface.getMethod(methodName, userTypes);
 			ProjectionFactory factory = new SpelAwareProxyProjectionFactory();
 			ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method,
-					new DefaultRepositoryMetadata(Repo.class), factory, mappingContext);
+					new DefaultRepositoryMetadata(repositoryInterface), factory, mappingContext);

 			return new ReactivePartTreeCassandraQuery(queryMethod, mockCassandraOperations);
-		} catch (NoSuchMethodException e) {
-			throw new IllegalArgumentException(e.getMessage(), e);
-		} catch (SecurityException e) {
+		} catch (NoSuchMethodException | SecurityException e) {
 			throw new IllegalArgumentException(e.getMessage(), e);
 		}
 	}
@@ -146,8 +182,9 @@ public class ReactivePartTreeCassandraQueryUnitTests {

 		Flux<Person> findPersonByFirstnameAndLastname(String firstname, String lastname);

-		Flux<Person> findByAge(Integer age);
+		Flux<Person> findByFirstname(QueryOptions queryOptions, String firstname);

+		@Consistency(ConsistencyLevel.LOCAL_ONE)
 		Flux<Person> findPersonBy();

 		@Query(allowFiltering = true)
@@ -156,7 +193,6 @@ public class ReactivePartTreeCassandraQueryUnitTests {
 		Mono<PersonProjection> findPersonProjectedBy();

 		<T> Single<T> findDynamicallyProjectedBy(Class<T> type);
-
 	}

 	interface PersonProjection {
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java
@@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*;

 import java.lang.reflect.Method;

+import com.datastax.driver.core.ConsistencyLevel;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -27,9 +28,11 @@ import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.cassandra.ReactiveSession;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.domain.Person;
+import org.springframework.data.cassandra.repository.Consistency;
 import org.springframework.data.cassandra.repository.Query;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
@@ -88,6 +91,37 @@ public class ReactiveStringBasedCassandraQueryUnitTests {
 		assertThat(stringQuery.getObject(0)).isEqualTo("White");
 	}

+	@Test // DATACASS-146
+	public void shouldApplyQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().fetchSize(777).build();
+
+		ReactiveStringBasedCassandraQuery cassandraQuery = getQueryMethod("findByLastname", QueryOptions.class,
+				String.class);
+		CassandraParametersParameterAccessor accessor = new CassandraParametersParameterAccessor(
+				cassandraQuery.getQueryMethod(), queryOptions, "White");
+
+		SimpleStatement actual = cassandraQuery.createQuery(accessor);
+
+		assertThat(actual.toString()).isEqualTo("SELECT * FROM person WHERE lastname=?;");
+		assertThat(actual.getObject(0)).isEqualTo("White");
+		assertThat(actual.getFetchSize()).isEqualTo(777);
+	}
+
+	@Test // DATACASS-146
+	public void shouldApplyConsistencyLevel() {
+
+		ReactiveStringBasedCassandraQuery cassandraQuery = getQueryMethod("findByLastname", String.class);
+		CassandraParametersParameterAccessor accessor = new CassandraParametersParameterAccessor(
+				cassandraQuery.getQueryMethod(), "Matthews");
+
+		SimpleStatement actual = cassandraQuery.createQuery(accessor);
+
+		assertThat(actual.toString()).isEqualTo("SELECT * FROM person WHERE lastname=?;");
+		assertThat(actual.getObject(0)).isEqualTo("Matthews");
+		assertThat(actual.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_ONE);
+	}
+
 	private ReactiveStringBasedCassandraQuery getQueryMethod(String name, Class<?>... args) {

 		Method method = ReflectionUtils.findMethod(SampleRepository.class, name, args);
@@ -101,6 +135,10 @@ public class ReactiveStringBasedCassandraQueryUnitTests {
 	private interface SampleRepository extends Repository<Person, String> {

 		@Query("SELECT * FROM person WHERE lastname=?0;")
+		@Consistency(ConsistencyLevel.LOCAL_ONE)
 		Person findByLastname(String lastname);
+
+		@Query("SELECT * FROM person WHERE lastname=?0;")
+		Person findByLastname(QueryOptions queryOptions, String lastname);
 	}
 }
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQueryUnitTests.java
@@ -27,6 +27,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;

+import com.datastax.driver.core.ConsistencyLevel;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -35,10 +36,12 @@ import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.domain.AddressType;
 import org.springframework.data.cassandra.domain.Person;
+import org.springframework.data.cassandra.repository.Consistency;
 import org.springframework.data.cassandra.repository.Query;
 import org.springframework.data.cassandra.support.UserTypeBuilder;
 import org.springframework.data.projection.ProjectionFactory;
@@ -335,6 +338,36 @@ public class StringBasedCassandraQueryUnitTests {
 		assertThat(stringQuery.getObject(0).toString()).isEqualTo("udtValue");
 	}

+	@Test // DATACASS-146
+	public void shouldApplyQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().fetchSize(777).build();
+
+		StringBasedCassandraQuery cassandraQuery = getQueryMethod("findByLastname", QueryOptions.class, String.class);
+		CassandraParametersParameterAccessor accessor = new CassandraParametersParameterAccessor(
+				cassandraQuery.getQueryMethod(), queryOptions, "Matthews");
+
+		SimpleStatement actual = cassandraQuery.createQuery(accessor);
+
+		assertThat(actual.toString()).isEqualTo("SELECT * FROM person WHERE lastname = ?;");
+		assertThat(actual.getObject(0)).isEqualTo("Matthews");
+		assertThat(actual.getFetchSize()).isEqualTo(777);
+	}
+
+	@Test // DATACASS-146
+	public void shouldApplyConsistencyLevel() {
+
+		StringBasedCassandraQuery cassandraQuery = getQueryMethod("findByLastname", String.class);
+		CassandraParametersParameterAccessor accessor = new CassandraParametersParameterAccessor(
+				cassandraQuery.getQueryMethod(), "Matthews");
+
+		SimpleStatement actual = cassandraQuery.createQuery(accessor);
+
+		assertThat(actual.toString()).isEqualTo("SELECT * FROM person WHERE lastname = ?;");
+		assertThat(actual.getObject(0)).isEqualTo("Matthews");
+		assertThat(actual.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_ONE);
+	}
+
 	private StringBasedCassandraQuery getQueryMethod(String name, Class<?>... args) {

 		Method method = ReflectionUtils.findMethod(SampleRepository.class, name, args);
@@ -347,8 +380,12 @@ public class StringBasedCassandraQueryUnitTests {
 	private interface SampleRepository extends Repository<Person, String> {

 		@Query("SELECT * FROM person WHERE lastname = ?0;")
+		@Consistency(ConsistencyLevel.LOCAL_ONE)
 		Person findByLastname(String lastname);

+		@Query("SELECT * FROM person WHERE lastname = ?0;")
+		Person findByLastname(QueryOptions queryOptions, String lastname);
+
 		@Query("SELECT * FROM person WHERE lastname = ?0 or firstname = ?0;")
 		Person findByLastnameUsedTwice(String lastname);

--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java
@@ -20,10 +20,12 @@ import java.util.Iterator;
 import java.util.Optional;

 import org.springframework.data.cassandra.core.convert.CassandraConverter;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.lang.Nullable;

 import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
@@ -64,6 +66,12 @@ class StubParameterAccessor implements CassandraParameterAccessor {
 		return values[index].getClass();
 	}

+	@Nullable
+	@Override
+	public QueryOptions getQueryOptions() {
+		return null;
+	}
+
 	@Override
 	public Pageable getPageable() {
 		return null;
--- a/src/main/asciidoc/reference/cassandra-repositories.adoc
+++ b/src/main/asciidoc/reference/cassandra-repositories.adoc
@@ -147,23 +147,26 @@ the Apache Cassandra database. Defining such a query is just a matter of declari
 ----
 public interface PersonRepository extends CrudRepository<Person, String> {

-    List<Person> findByLastname(String lastname);                      <1>
+    List<Person> findByLastname(String lastname);                        <1>

-    List<Person> findByFirstname(String firstname, Sort sort);         <2>
+    List<Person> findByFirstname(String firstname, Sort sort);           <2>

-    Person findByShippingAddress(Address address);                     <3>
+    List<Person> findByFirstname(String firstname, QueryOptions opts);   <3>

-    Stream<Person> findAllBy();                                        <4>
+    Person findByShippingAddress(Address address);                       <4>
+
+    Stream<Person> findAllBy();                                          <5>
 }
 ----
 <1> The method shows a query for all people with the given `lastname`. The query will be derived from parsing
 the method name for constraints which can be concatenated with `And`. Thus the method name will result in
 a query expression of `SELECT * from person WHERE lastname = 'lastname'`.
-<2> Applies dynamic sorting to a query. Just add a `Sort` parameter to your method signature and Spring Data
+<2> Applies dynamic sorting to a query. Just add a `Sort` parameter to your method signature and Spring Data.
 will automatically apply ordering to the query accordingly.
-<3> Shows that you can query based on properties which are not a primitive type using registered `Converter`'s
+<3> Passing a `QueryOptions` object will apply the query options to the resulting query before it's execution.
+<4> Shows that you can query based on properties which are not a primitive type using registered `Converter`'s.
 in `CustomConversions`.
-<4> Uses a Java 8 `Stream` which reads and converts individual elements while iterating the stream.
+<5> Uses a Java 8 `Stream` which reads and converts individual elements while iterating the stream.
 ====

 NOTE: Querying non-primary key properties requires secondary indexes.
@@ -231,6 +234,31 @@ NOTE: Querying non-primary key properties requires secondary indexes.

 include::../{spring-data-commons-docs}/repository-projections.adoc[leveloffset=+2]

+=== Query options
+
+You can specify query options for query methods by passing a `QueryOptions` object
+to apply options to the query before the actual query execution.
+`QueryOptions` is treated as non-query parameter and isn't considered as query parameter value.
+
+For static declaration of a consistency level, use the `@Consistency` annotation on query methods.
+The declared consistency level is applied to the query each time it is executed.
+
+Query options are applicable to derived and string `@Query` repository methods.
+
+----
+public interface PersonRepository extends CrudRepository<Person, String> {
+
+    @Consistency(ConsistencyLevel.LOCAL_ONE)
+    List<Person> findByLastname(String lastname);
+
+    List<Person> findByFirstname(String firstname, QueryOptions options);
+}
+----
+
+NOTE: You can control fetch size, consistency level and retry policy defaults by configuring these parameters
+on the CQL API instances `CqlTemplate`, `AsyncCqlTemplate`, and `ReactiveCqlTemplate`. Defaults apply if the particular
+query option is not set.
+
 [[cassandra.repositories.misc]]
 == Miscellaneous

--- a/src/main/asciidoc/reference/cassandra.adoc
+++ b/src/main/asciidoc/reference/cassandra.adoc
@@ -664,6 +664,10 @@ All CQL issued by this class is logged at the `DEBUG` level under the category c
 name of the template instance (typically `CqlTemplate`, but it may be different if you are using a custom subclass
 of the `CqlTemplate` class).

+You can control fetch size, consistency level and retry policy defaults by configuring these parameters
+on the CQL API instances `CqlTemplate`, `AsyncCqlTemplate`, and `ReactiveCqlTemplate`. Defaults apply if the particular
+query option is not set.
+
 NOTE: `CqlTemplate` comes in different execution model flavors. The basic `CqlTemplate` uses a blocking execution model.
 You can use `AsyncCqlTemplate` for asynchronous execution and synchronization with ``ListenableFuture``s or
 <<cassandra.reactive.cql-template,`ReactiveCqlTemplate`>> for reactive execution.