DATACASS-510 - Introduce support for prepared statements using CassandraTemplate.

We now support prepared statement usage through CassandraTemplate and its asynchronous and reactive variants. All statements created or received by CassandraTemplate will be prepared. CassandraTemplate is an infrastructure class for repositories so prepared statements will affect repositories, too.

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

@@ -29,6 +29,7 @@ import org.springframework.data.cassandra.core.query.Update;
 import org.springframework.data.domain.Slice;
 import org.springframework.util.concurrent.ListenableFuture;
 
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
 import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
@@ -102,6 +103,17 @@ public interface AsyncCassandraOperations {
 	// Methods dealing with com.datastax.oss.driver.api.core.cql.Statement
 	// -------------------------------------------------------------------------
 
+	/**
+	 * Execute the a Cassandra {@link Statement}. Any errors that result from executing this command will be converted
+	 * into Spring's DAO exception hierarchy.
+	 *
+	 * @param statement a Cassandra {@link Statement}, must not be {@literal null}.
+	 * @return the {@link AsyncResultSet}.
+	 * @throws DataAccessException if there is any problem executing the query.
+	 * @since 3.2
+	 */
+	ListenableFuture<AsyncResultSet> execute(Statement<?> statement) throws DataAccessException;
+
 	/**
 	 * Execute a {@code SELECT} query and convert the resulting items to a {@link List} of entities.
 	 *

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

@@ -22,6 +22,9 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -29,18 +32,12 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.data.cassandra.SessionFactory;
 import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
-import org.springframework.data.cassandra.core.cql.AsyncCqlOperations;
-import org.springframework.data.cassandra.core.cql.AsyncCqlTemplate;
-import org.springframework.data.cassandra.core.cql.AsyncSessionCallback;
-import org.springframework.data.cassandra.core.cql.CassandraAccessor;
-import org.springframework.data.cassandra.core.cql.CqlExceptionTranslator;
-import org.springframework.data.cassandra.core.cql.CqlProvider;
-import org.springframework.data.cassandra.core.cql.QueryOptions;
-import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.data.cassandra.core.cql.*;
 import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
 import org.springframework.data.cassandra.core.cql.util.CassandraFutureAdapter;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
@@ -59,6 +56,7 @@ import org.springframework.data.domain.Slice;
 import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.util.Streamable;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.annotation.AsyncResult;
 import org.springframework.util.Assert;
@@ -69,6 +67,9 @@ import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
+import com.datastax.oss.driver.api.core.cql.BoundStatement;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
@@ -89,6 +90,13 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
  * Can be used within a service implementation via direct instantiation with a {@link CqlSession} reference, or get
  * prepared in an application context and given to services as bean reference.
  * <p>
+ * This class supports the use of prepared statements when enabling {@link #setUsePreparedStatements(boolean)}. All
+ * statements created by methods of this class (such as {@link #select(Query, Class)} or
+ * {@link #update(Query, org.springframework.data.cassandra.core.query.Update, Class)} will be executed as prepared
+ * statements. Also, statements accepted by methods (such as {@link #select(String, Class)} or
+ * {@link #select(Statement, Class) and others}) will be prepared prior to execution. Note that {@link Statement}
+ * objects passed to methods must be {@link SimpleStatement} so that these can be prepared.
+ * <p>
  * Note: The {@link CqlSession} should always be configured as a bean in the application context, in the first case
  * given to the service directly, in the second case to the prepared template.
  *
@@ -100,6 +108,8 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
 public class AsyncCassandraTemplate
 		implements AsyncCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
+	private final Logger logger = LoggerFactory.getLogger(getClass());
+
 	private final AsyncCqlOperations cqlOperations;
 
 	private final CassandraConverter converter;
@@ -116,6 +126,8 @@ public class AsyncCassandraTemplate
 
 	private @Nullable EntityCallbacks entityCallbacks;
 
+	private boolean usePreparedStatements = true;
+
 	/**
 	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link CqlSession} and a default
 	 * {@link MappingCassandraConverter}.
@@ -136,7 +148,7 @@ public class AsyncCassandraTemplate
 	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
 	 *          {@literal null}.
 	 * @see CassandraConverter
-	 * @see Session
+	 * @see CqlSession
 	 */
 	public AsyncCassandraTemplate(CqlSession session, CassandraConverter converter) {
 		this(new DefaultSessionFactory(session), converter);
@@ -150,7 +162,7 @@ public class AsyncCassandraTemplate
 	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
 	 *          {@literal null}.
 	 * @see CassandraConverter
-	 * @see Session
+	 * @see CqlSession
 	 */
 	public AsyncCassandraTemplate(SessionFactory sessionFactory, CassandraConverter converter) {
 		this(new AsyncCqlTemplate(sessionFactory), converter);
@@ -164,7 +176,7 @@ public class AsyncCassandraTemplate
 	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
 	 *          {@literal null}.
 	 * @see CassandraConverter
-	 * @see Session
+	 * @see CqlSession
 	 */
 	public AsyncCassandraTemplate(AsyncCqlTemplate asyncCqlTemplate, CassandraConverter converter) {
 
@@ -226,6 +238,32 @@ public class AsyncCassandraTemplate
 		return this.converter;
 	}
 
+	/**
+	 * Returns whether this instance is configured to use {@link PreparedStatement prepared statements}. If enabled
+	 * (default), then all persistence methods (such as {@link #select}, {@link #update}, and others) will make use of
+	 * prepared statements. Note that methods accepting a {@link Statement} must be called with {@link SimpleStatement}
+	 * instances to participate in statement preparation.
+	 *
+	 * @return {@literal true} if prepared statements usage is enabled; {@literal false} otherwise.
+	 * @since 3.2
+	 */
+	public boolean isUsePreparedStatements() {
+		return usePreparedStatements;
+	}
+
+	/**
+	 * Enable/disable {@link PreparedStatement prepared statements} usage. If enabled (default), then all persistence
+	 * methods (such as {@link #select}, {@link #update}, and others) will make use of prepared statements. Note that
+	 * methods accepting a {@link Statement} must be called with {@link SimpleStatement} instances to participate in
+	 * statement preparation.
+	 *
+	 * @param usePreparedStatements whether to use prepared statements.
+	 * @since 3.2
+	 */
+	public void setUsePreparedStatements(boolean usePreparedStatements) {
+		this.usePreparedStatements = usePreparedStatements;
+	}
+
 	/**
 	 * Returns the {@link EntityOperations} used to perform data access operations on an entity inside a Cassandra data
 	 * source.
@@ -314,6 +352,17 @@ public class AsyncCassandraTemplate
 	// Methods dealing with com.datastax.oss.driver.api.core.cql.Statement
 	// -------------------------------------------------------------------------
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(com.datastax.oss.driver.api.core.cql.Statement, java.lang.Class)
+	 */
+	@Override
+	public ListenableFuture<AsyncResultSet> execute(Statement<?> statement) throws DataAccessException {
+
+		Assert.notNull(statement, "Statement must not be null");
+
+		return doQueryForResultSet(statement);
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(com.datastax.oss.driver.api.core.cql.Statement, java.lang.Class)
 	 */
@@ -325,7 +374,7 @@ public class AsyncCassandraTemplate
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
-		return getAsyncCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
+		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -341,7 +390,7 @@ public class AsyncCassandraTemplate
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
-		return getAsyncCqlOperations().query(statement, row -> {
+		return doQuery(statement, row -> {
 			entityConsumer.accept(mapper.apply(row));
 		});
 	}
@@ -364,7 +413,7 @@ public class AsyncCassandraTemplate
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		ListenableFuture<AsyncResultSet> resultSet = getAsyncCqlOperations().queryForResultSet(statement);
+		ListenableFuture<AsyncResultSet> resultSet = doQueryForResultSet(statement);
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
@@ -439,8 +488,8 @@ public class AsyncCassandraTemplate
 		Assert.notNull(update, "Update must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		return getAsyncCqlOperations()
-				.execute(getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass)).build());
+		return doExecute(getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass)).build(),
+				AsyncResultSet::wasApplied);
 	}
 
 	/* (non-Javadoc)
@@ -463,7 +512,7 @@ public class AsyncCassandraTemplate
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName));
 
-		ListenableFuture<Boolean> future = getAsyncCqlOperations().execute(delete);
+		ListenableFuture<Boolean> future = doExecute(delete, AsyncResultSet::wasApplied);
 
 		future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)), e -> {});
 
@@ -504,7 +553,13 @@ public class AsyncCassandraTemplate
 
 		SimpleStatement statement = countStatement.build();
 
-		ListenableFuture<Long> result = getAsyncCqlOperations().queryForObject(statement, Long.class);
+		ListenableFuture<Long> result = doExecute(statement, it -> {
+
+			SingleColumnRowMapper<Long> mapper = SingleColumnRowMapper.newInstance(Long.class);
+
+			Row row = DataAccessUtils.requiredSingleResult(Streamable.of(it.currentPage()).toList());
+			return mapper.mapRow(row, 0);
+		});
 
 		return new MappingListenableFutureAdapter<>(result, it -> it != null ? it : 0L);
 	}
@@ -523,8 +578,7 @@ public class AsyncCassandraTemplate
 		StatementBuilder<com.datastax.oss.driver.api.querybuilder.select.Select> select = getStatementFactory()
 				.selectOneById(id, entity, entity.getTableName());
 
-		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(select.build()),
-				resultSet -> resultSet.one() != null);
+		return doExecute(select.build(), resultSet -> resultSet.one() != null);
 	}
 
 	/* (non-Javadoc)
@@ -539,8 +593,7 @@ public class AsyncCassandraTemplate
 		StatementBuilder<com.datastax.oss.driver.api.querybuilder.select.Select> select = getStatementFactory()
 				.select(query.limit(1), getRequiredPersistentEntity(entityClass), getTableName(entityClass));
 
-		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(select.build()),
-				resultSet -> resultSet.one() != null);
+		return doExecute(select.build(), resultSet -> resultSet.one() != null);
 	}
 
 	/* (non-Javadoc)
@@ -557,8 +610,7 @@ public class AsyncCassandraTemplate
 		StatementBuilder<Select> select = getStatementFactory().selectOneById(id, entity, tableName);
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, tableName);
 
-		return new MappingListenableFutureAdapter<>(
-				getAsyncCqlOperations().query(select.build(), (row, rowNum) -> mapper.apply(row)),
+		return new MappingListenableFutureAdapter<>(doQuery(select.build(), (row, rowNum) -> mapper.apply(row)),
 				it -> it.isEmpty() ? null : it.get(0));
 	}
 
@@ -617,8 +669,7 @@ public class AsyncCassandraTemplate
 
 	@SuppressWarnings("unused")
 	private <T> ListenableFuture<EntityWriteResult<T>> doInsert(SimpleStatement insert, T entity,
-			AdaptibleEntity<T> source,
-			CqlIdentifier tableName) {
+			AdaptibleEntity<T> source, CqlIdentifier tableName) {
 
 		return executeSave(entity, tableName, insert);
 	}
@@ -743,7 +794,7 @@ public class AsyncCassandraTemplate
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName));
 
-		ListenableFuture<Boolean> future = getAsyncCqlOperations().execute(delete);
+		ListenableFuture<Boolean> future = doExecute(delete, AsyncResultSet::wasApplied);
 		future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)), e -> {});
 
 		return future;
@@ -763,7 +814,7 @@ public class AsyncCassandraTemplate
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entityClass, tableName));
 
-		ListenableFuture<Boolean> future = getAsyncCqlOperations().execute(statement);
+		ListenableFuture<Boolean> future = doExecute(statement, AsyncResultSet::wasApplied);
 		future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entityClass, tableName)), e -> {});
 
 		return new MappingListenableFutureAdapter<>(future, aBoolean -> null);
@@ -785,7 +836,7 @@ public class AsyncCassandraTemplate
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
 		T entityToSave = maybeCallBeforeSave(entity, tableName, statement);
 
-		ListenableFuture<AsyncResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
+		ListenableFuture<AsyncResultSet> result = doQueryForResultSet(statement);
 
 		return new MappingListenableFutureAdapter<>(result, resultSet -> {
 
@@ -806,7 +857,7 @@ public class AsyncCassandraTemplate
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
 
-		ListenableFuture<AsyncResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
+		ListenableFuture<AsyncResultSet> result = doQueryForResultSet(statement);
 
 		return new MappingListenableFutureAdapter<>(result, resultSet -> {
 
@@ -821,6 +872,44 @@ public class AsyncCassandraTemplate
 		});
 	}
 
+	private <T> ListenableFuture<List<T>> doQuery(Statement<?> statement, RowMapper<T> rowMapper) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getAsyncCqlOperations().query(statementHandler, statementHandler, rowMapper);
+		}
+
+		return getAsyncCqlOperations().query(statement, rowMapper);
+	}
+
+	private ListenableFuture<Void> doQuery(Statement<?> statement, RowCallbackHandler callbackHandler) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getAsyncCqlOperations().query(statementHandler, statementHandler, callbackHandler);
+		}
+
+		return getAsyncCqlOperations().query(statement, callbackHandler);
+	}
+
+	private ListenableFuture<AsyncResultSet> doQueryForResultSet(Statement<?> statement) {
+		return doExecute(statement, Function.identity());
+	}
+
+	private <T> ListenableFuture<T> doExecute(Statement<?> statement, Function<AsyncResultSet, T> mappingFunction) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getAsyncCqlOperations().query(statementHandler, statementHandler,
+					(AsyncResultSetExtractor<T>) resultSet -> new AsyncResult<>(mappingFunction.apply(resultSet)));
+		}
+
+		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(statement), mappingFunction);
+	}
+
 	private static List<Row> getFirstPage(AsyncResultSet resultSet) {
 		return StreamSupport.stream(resultSet.currentPage().spliterator(), false).collect(Collectors.toList());
 	}
@@ -893,7 +982,7 @@ public class AsyncCassandraTemplate
 	protected <T> T maybeCallBeforeConvert(T object, CqlIdentifier tableName) {
 
 		if (null != entityCallbacks) {
-			return (T) entityCallbacks.callback(BeforeConvertCallback.class, object, tableName);
+			return entityCallbacks.callback(BeforeConvertCallback.class, object, tableName);
 		}
 
 		return object;
@@ -902,7 +991,7 @@ public class AsyncCassandraTemplate
 	protected <T> T maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement<?> statement) {
 
 		if (null != entityCallbacks) {
-			return (T) entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement);
+			return entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement);
 		}
 
 		return object;
@@ -927,24 +1016,37 @@ public class AsyncCassandraTemplate
 		}
 	}
 
-	class AsyncStatementCallback implements AsyncSessionCallback<AsyncResultSet>, CqlProvider {
+	/**
+	 * Utility class to prepare a {@link SimpleStatement} and bind values associated with the statement to a
+	 * {@link BoundStatement}.
+	 *
+	 * @since 3.2
+	 */
+	private class PreparedStatementHandler
+			implements AsyncPreparedStatementCreator, PreparedStatementBinder, CqlProvider {
 
-		SimpleStatement statement;
+		private final SimpleStatement statement;
 
-		AsyncStatementCallback(SimpleStatement statement) {
-			this.statement = statement;
+		public PreparedStatementHandler(Statement<?> statement) {
+			this.statement = PreparedStatementDelegate.getStatementForPrepare(statement);
 		}
 
 		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.AsyncSessionCallback#doInSession(com.datastax.oss.driver.api.core.CqlSession)
+		 * @see org.springframework.data.cassandra.core.cql.AsyncPreparedStatementCreator#createPreparedStatement(com.datastax.oss.driver.api.core.CqlSession)
 		 */
 		@Override
-		public ListenableFuture<AsyncResultSet> doInSession(CqlSession session)
-				throws DriverException, DataAccessException {
-			return new CassandraFutureAdapter<>(session.executeAsync(this.statement),
-					e -> e instanceof DriverException ? exceptionTranslator.translate("AsyncStatementCallback", getCql(), e)
-							: exceptionTranslator.translateExceptionIfPossible(e));
+		public ListenableFuture<PreparedStatement> createPreparedStatement(CqlSession session) throws DriverException {
+			return new CassandraFutureAdapter<>(session.prepareAsync(statement), exceptionTranslator);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.PreparedStatementBinder#bindValues(com.datastax.oss.driver.api.core.cql.PreparedStatement)
+		 */
+		@Override
+		public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+			return PreparedStatementDelegate.bind(statement, ps);
 		}
 
 		/*
@@ -953,7 +1055,8 @@ public class AsyncCassandraTemplate
 		 */
 		@Override
 		public String getCql() {
-			return this.statement.getQuery();
+			return statement.getQuery();
 		}
+
 	}
 }

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

@@ -31,6 +31,7 @@ import org.springframework.data.domain.Slice;
 import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
@@ -57,13 +58,6 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	 */
 	CassandraBatchOperations batchOps();
 
-	/**
-	 * Returns the underlying {@link CassandraConverter}.
-	 *
-	 * @return the underlying {@link CassandraConverter}.
-	 */
-	CassandraConverter getConverter();
-
 	/**
 	 * Expose the underlying {@link CqlOperations} to allow CQL operations.
 	 *
@@ -72,6 +66,13 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	 */
 	CqlOperations getCqlOperations();
 
+	/**
+	 * Returns the underlying {@link CassandraConverter}.
+	 *
+	 * @return the underlying {@link CassandraConverter}.
+	 */
+	CassandraConverter getConverter();
+
 	/**
 	 * The table name used for the specified class by this template.
 	 *
@@ -123,6 +124,17 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	// Methods dealing with com.datastax.oss.driver.api.core.cql.Statement
 	// -------------------------------------------------------------------------
 
+	/**
+	 * Execute the a Cassandra {@link Statement}. Any errors that result from executing this command will be converted
+	 * into Spring's DAO exception hierarchy.
+	 *
+	 * @param statement a Cassandra {@link Statement}, must not be {@literal null}.
+	 * @return the {@link ResultSet}.
+	 * @throws DataAccessException if there is any problem executing the query.
+	 * @since 3.2
+	 */
+	ResultSet execute(Statement<?> statement) throws DataAccessException;
+
 	/**
 	 * Execute a {@code SELECT} query and convert the resulting items to a {@link List} of entities.
 	 *

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

@@ -20,6 +20,9 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -27,6 +30,7 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.data.cassandra.SessionFactory;
 import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
@@ -37,13 +41,15 @@ import org.springframework.data.cassandra.core.cql.CassandraAccessor;
 import org.springframework.data.cassandra.core.cql.CqlOperations;
 import org.springframework.data.cassandra.core.cql.CqlProvider;
 import org.springframework.data.cassandra.core.cql.CqlTemplate;
+import org.springframework.data.cassandra.core.cql.PreparedStatementBinder;
+import org.springframework.data.cassandra.core.cql.PreparedStatementCreator;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
-import org.springframework.data.cassandra.core.cql.SessionCallback;
+import org.springframework.data.cassandra.core.cql.RowMapper;
+import org.springframework.data.cassandra.core.cql.SingleColumnRowMapper;
 import org.springframework.data.cassandra.core.cql.WriteOptions;
 import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
-import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
@@ -57,7 +63,6 @@ import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.mapping.callback.EntityCallbacks;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
@@ -67,6 +72,8 @@ import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.cql.BoundStatement;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
@@ -88,6 +95,13 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
  * Can be used within a service implementation via direct instantiation with a {@link CqlSession} reference, or get
  * prepared in an application context and given to services as bean reference.
  * <p>
+ * This class supports the use of prepared statements when enabling {@link #setUsePreparedStatements(boolean)}. All
+ * statements created by methods of this class (such as {@link #select(Query, Class)} or
+ * {@link #update(Query, org.springframework.data.cassandra.core.query.Update, Class)} will be executed as prepared
+ * statements. Also, statements accepted by methods (such as {@link #select(String, Class)} or
+ * {@link #select(Statement, Class) and others}) will be prepared prior to execution. Note that {@link Statement}
+ * objects passed to methods must be {@link SimpleStatement} so that these can be prepared.
+ * <p>
  * Note: The {@link CqlSession} should always be configured as a bean in the application context, in the first case
  * given to the service directly, in the second case to the prepared template.
  *
@@ -99,22 +113,24 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
  */
 public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
-	private @Nullable ApplicationEventPublisher eventPublisher;
-
-	private @Nullable EntityCallbacks entityCallbacks;
-
-	private final CassandraConverter converter;
+	private final Logger logger = LoggerFactory.getLogger(getClass());
 
 	private final CqlOperations cqlOperations;
 
-	private final EntityOperations entityOperations;
+	private final CassandraConverter converter;
 
-	private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
+	private final EntityOperations entityOperations;
 
 	private final SpelAwareProxyProjectionFactory projectionFactory;
 
 	private final StatementFactory statementFactory;
 
+	private @Nullable ApplicationEventPublisher eventPublisher;
+
+	private @Nullable EntityCallbacks entityCallbacks;
+
+	private boolean usePreparedStatements = true;
+
 	/**
 	 * Creates an instance of {@link CassandraTemplate} initialized with the given {@link CqlSession} and a default
 	 * {@link MappingCassandraConverter}.
@@ -173,7 +189,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		this.converter = converter;
 		this.cqlOperations = cqlOperations;
 		this.entityOperations = new EntityOperations(converter.getMappingContext());
-		this.mappingContext = converter.getMappingContext();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
 		this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
 	}
@@ -217,6 +232,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		this.entityCallbacks = entityCallbacks;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraOperations#CqlOperations()
+	 */
+	@Override
+	public CqlOperations getCqlOperations() {
+		return this.cqlOperations;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraOperations#getConverter()
 	 */
@@ -225,12 +248,30 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return this.converter;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.core.CassandraOperations#CqlOperations()
+	/**
+	 * Returns whether this instance is configured to use {@link PreparedStatement prepared statements}. If enabled
+	 * (default), then all persistence methods (such as {@link #select}, {@link #update}, and others) will make use of
+	 * prepared statements. Note that methods accepting a {@link Statement} must be called with {@link SimpleStatement}
+	 * instances to participate in statement preparation.
+	 *
+	 * @return {@literal true} if prepared statements usage is enabled; {@literal false} otherwise.
+	 * @since 3.2
 	 */
-	@Override
-	public CqlOperations getCqlOperations() {
-		return this.cqlOperations;
+	public boolean isUsePreparedStatements() {
+		return usePreparedStatements;
+	}
+
+	/**
+	 * Enable/disable {@link PreparedStatement prepared statements} usage. If enabled (default), then all persistence
+	 * methods (such as {@link #select}, {@link #update}, and others) will make use of prepared statements. Note that
+	 * methods accepting a {@link Statement} must be called with {@link SimpleStatement} instances to participate in
+	 * statement preparation.
+	 *
+	 * @param usePreparedStatements whether to use prepared statements.
+	 * @since 3.2
+	 */
+	public void setUsePreparedStatements(boolean usePreparedStatements) {
+		this.usePreparedStatements = usePreparedStatements;
 	}
 
 	/**
@@ -323,6 +364,17 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 	// Methods dealing with com.datastax.oss.driver.api.core.cql.Statement
 	// -------------------------------------------------------------------------
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraOperations#execute(com.datastax.oss.driver.api.core.cql.Statement)
+	 */
+	@Override
+	public ResultSet execute(Statement<?> statement) {
+
+		Assert.notNull(statement, "Statement must not be null");
+
+		return doQueryForResultSet(statement);
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraOperations#select(com.datastax.oss.driver.api.core.cql.Statement, java.lang.Class)
 	 */
@@ -334,7 +386,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
-		return getCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
+		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -356,7 +408,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		ResultSet resultSet = getCqlOperations().queryForResultSet(statement);
+		ResultSet resultSet = doQueryForResultSet(statement);
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
@@ -374,7 +426,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(entityClass, "Entity type must not be null");
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
-		return getCqlOperations().queryForStream(statement, (row, rowNum) -> mapper.apply(row));
+		return doQueryForStream(statement, (row, rowNum) -> mapper.apply(row));
 	}
 
 	// -------------------------------------------------------------------------
@@ -405,7 +457,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
 
-		return getCqlOperations().query(select.build(), (row, rowNum) -> mapper.apply(row));
+		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -451,7 +503,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 				tableName);
 
 		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
-		return getCqlOperations().queryForStream(select.build(), (row, rowNum) -> mapper.apply(row));
+		return doQueryForStream(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -468,7 +520,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		StatementBuilder<Update> updateStatement = getStatementFactory().update(query, update,
 				getRequiredPersistentEntity(entityClass));
 
-		return getCqlOperations().execute(updateStatement.build());
+		return doExecute(updateStatement.build()).wasApplied();
 	}
 
 	@Nullable
@@ -478,7 +530,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		StatementBuilder<Update> updateStatement = getStatementFactory().update(query, update,
 				getRequiredPersistentEntity(entityClass), tableName);
 
-		return getCqlOperations().execute(new StatementCallback(updateStatement.build()));
+		return doExecute(updateStatement.build());
 	}
 
 	/* (non-Javadoc)
@@ -504,7 +556,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entityClass, tableName));
 
-		WriteResult writeResult = getCqlOperations().execute(new StatementCallback(statement));
+		WriteResult writeResult = doExecute(statement);
 
 		maybeEmitEvent(new AfterDeleteEvent<>(statement, entityClass, tableName));
 
@@ -543,10 +595,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		StatementBuilder<Select> countStatement = getStatementFactory().count(query,
 				getRequiredPersistentEntity(entityClass), tableName);
 
-		SimpleStatement statement = countStatement.build();
-		Long count = getCqlOperations().queryForObject(statement, Long.class);
-
-		return count != null ? count : 0L;
+		return doQueryForObject(countStatement.build(), Long.class);
 	}
 
 	/* (non-Javadoc)
@@ -561,7 +610,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
 		StatementBuilder<Select> select = getStatementFactory().selectOneById(id, entity, entity.getTableName());
 
-		return getCqlOperations().queryForResultSet(select.build()).one() != null;
+		return doQueryForResultSet(select.build()).one() != null;
 	}
 
 	/* (non-Javadoc)
@@ -581,7 +630,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		StatementBuilder<Select> select = getStatementFactory().select(query.limit(1),
 				getRequiredPersistentEntity(entityClass), tableName);
 
-		return getCqlOperations().queryForResultSet(select.build()).one() != null;
+		return doQueryForResultSet(select.build()).one() != null;
 	}
 
 	/* (non-Javadoc)
@@ -597,7 +646,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		CqlIdentifier tableName = entity.getTableName();
 		StatementBuilder<Select> select = getStatementFactory().selectOneById(id, entity, tableName);
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, tableName);
-		List<T> result = getCqlOperations().query(select.build(), (row, rowNum) -> mapper.apply(row));
+		List<T> result = doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 
 		return result.isEmpty() ? null : result.get(0);
 	}
@@ -777,7 +826,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entityClass, tableName));
 
-		boolean result = getCqlOperations().execute(statement);
+		boolean result = doExecute(statement).wasApplied();
 
 		maybeEmitEvent(new AfterDeleteEvent<>(statement, entityClass, tableName));
 
@@ -798,7 +847,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entityClass, tableName));
 
-		getCqlOperations().execute(statement);
+		doExecute(statement);
 
 		maybeEmitEvent(new AfterDeleteEvent<>(statement, entityClass, tableName));
 	}
@@ -853,7 +902,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
 		T entityToSave = maybeCallBeforeSave(entity, tableName, statement);
 
-		WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
+		WriteResult result = doExecute(statement);
 		resultConsumer.accept(result);
 
 		maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName));
@@ -866,7 +915,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
 
-		WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
+		WriteResult result = doExecute(statement);
 
 		resultConsumer.accept(result);
 
@@ -875,6 +924,51 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return result;
 	}
 
+	private <T> List<T> doQuery(Statement<?> statement, RowMapper<T> rowMapper) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getCqlOperations().query(statementHandler, statementHandler, rowMapper);
+		}
+
+		return getCqlOperations().query(statement, rowMapper);
+	}
+
+	private <T> T doQueryForObject(Statement<?> statement, Class<T> resultType) {
+		return DataAccessUtils.requiredSingleResult(doQuery(statement, SingleColumnRowMapper.newInstance(resultType)));
+	}
+
+	private <T> Stream<T> doQueryForStream(Statement<?> statement, RowMapper<T> rowMapper) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getCqlOperations().queryForStream(statementHandler, statementHandler, rowMapper);
+		}
+
+		return getCqlOperations().queryForStream(statement, rowMapper);
+	}
+
+	private WriteResult doExecute(SimpleStatement statement) {
+		return doExecute(statement, WriteResult::of);
+	}
+
+	private ResultSet doQueryForResultSet(Statement<?> statement) {
+		return doExecute(statement, Function.identity());
+	}
+
+	private <T> T doExecute(Statement<?> statement, Function<ResultSet, T> mappingFunction) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getCqlOperations().query(statementHandler, statementHandler, mappingFunction::apply);
+		}
+
+		return mappingFunction.apply(getCqlOperations().queryForResultSet(statement));
+	}
+
 	private int getConfiguredPageSize(CqlSession session) {
 		return session.getContext().getConfig().getDefaultProfile().getInt(DefaultDriverOption.REQUEST_PAGE_SIZE, 5000);
 	}
@@ -957,21 +1051,37 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return object;
 	}
 
-	static class StatementCallback implements SessionCallback<WriteResult>, CqlProvider {
+	/**
+	 * Utility class to prepare a {@link SimpleStatement} and bind values associated with the statement to a
+	 * {@link BoundStatement}.
+	 *
+	 * @since 3.2
+	 */
+	public static class PreparedStatementHandler
+			implements PreparedStatementCreator, PreparedStatementBinder, CqlProvider {
 
 		private final SimpleStatement statement;
 
-		StatementCallback(SimpleStatement statement) {
-			this.statement = statement;
+		public PreparedStatementHandler(Statement<?> statement) {
+			this.statement = PreparedStatementDelegate.getStatementForPrepare(statement);
 		}
 
 		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.SessionCallback#doInSession(org.springframework.data.cassandra.Session)
+		 * @see org.springframework.data.cassandra.core.cql.PreparedStatementCreator#createPreparedStatement(com.datastax.oss.driver.api.core.CqlSession)
 		 */
 		@Override
-		public WriteResult doInSession(CqlSession session) throws DriverException, DataAccessException {
-			return WriteResult.of(session.execute(this.statement));
+		public PreparedStatement createPreparedStatement(CqlSession session) throws DriverException {
+			return session.prepare(statement);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.PreparedStatementBinder#bindValues(com.datastax.oss.driver.api.core.cql.PreparedStatement)
+		 */
+		@Override
+		public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+			return PreparedStatementDelegate.bind(statement, ps);
 		}
 
 		/*
@@ -980,7 +1090,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		 */
 		@Override
 		public String getCql() {
-			return this.statement.getQuery();
+			return statement.getQuery();
 		}
 	}
 }

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+
+import org.springframework.data.cassandra.core.cql.QueryExtractorDelegate;
+import org.springframework.util.StringUtils;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.BoundStatement;
+import com.datastax.oss.driver.api.core.cql.BoundStatementBuilder;
+import com.datastax.oss.driver.api.core.cql.ColumnDefinitions;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.cql.Statement;
+import com.datastax.oss.driver.api.core.type.DataType;
+
+/**
+ * Support class for Cassandra Template API implementation classes that want to make use of prepared statements.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+class PreparedStatementDelegate {
+
+	/**
+	 * Bind values held in {@link SimpleStatement} to the {@link PreparedStatement}.
+	 *
+	 * @param statement
+	 * @param ps
+	 * @return the bound statement.
+	 */
+	static BoundStatement bind(SimpleStatement statement, PreparedStatement ps) {
+
+		BoundStatementBuilder boundStatementBuilder = ps.boundStatementBuilder(statement.getPositionalValues().toArray());
+		Map<CqlIdentifier, Object> namedValues = statement.getNamedValues();
+
+		ColumnDefinitions variableDefinitions = ps.getVariableDefinitions();
+		for (Map.Entry<CqlIdentifier, Object> entry : namedValues.entrySet()) {
+
+			if (entry.getValue() == null) {
+				boundStatementBuilder = boundStatementBuilder.setToNull(entry.getKey());
+			} else {
+				DataType type = variableDefinitions.get(entry.getKey()).getType();
+				boundStatementBuilder = boundStatementBuilder.set(entry.getKey(), entry.getValue(),
+						boundStatementBuilder.codecRegistry().codecFor(type));
+			}
+		}
+
+		return ps.bind(statement.getPositionalValues().toArray());
+	}
+
+	/**
+	 * Ensure the given {@link Statement} is a {@link SimpleStatement}. Throw a {@link IllegalArgumentException}
+	 * otherwise.
+	 *
+	 * @param statement
+	 * @return the {@link SimpleStatement}.
+	 */
+	static SimpleStatement getStatementForPrepare(Statement<?> statement) {
+
+		if (statement instanceof SimpleStatement) {
+			return (SimpleStatement) statement;
+		}
+
+		throw new IllegalArgumentException(getMessage(statement));
+	}
+
+	/**
+	 * Check whether to use prepared statements. When {@code usePreparedStatements} is {@literal true}, then verifying
+	 * additionally that the given {@link Statement} is a {@link SimpleStatement}, otherwise log the mismatch and fallback
+	 * to non-prepared usage.
+	 *
+	 * @param usePreparedStatements
+	 * @param statement
+	 * @param logger
+	 * @return
+	 */
+	static boolean canPrepare(boolean usePreparedStatements, Statement<?> statement, Logger logger) {
+
+		if (usePreparedStatements) {
+
+			if (statement instanceof SimpleStatement) {
+				return true;
+			}
+
+			logger.warn(getMessage(statement));
+		}
+
+		return false;
+	}
+
+	private static String getMessage(Statement<?> statement) {
+
+		String cql = QueryExtractorDelegate.getCql(statement);
+
+		if (StringUtils.hasText(cql)) {
+			return String.format("Cannot prepare statement %s (%s). Statement must be a SimpleStatement.", cql, statement);
+		}
+
+		return String.format("Cannot prepare statement %s. Statement must be a SimpleStatement.", statement);
+	}
+
+}

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

@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.cassandra.ReactiveResultSet;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
@@ -58,13 +59,6 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	 */
 	ReactiveCassandraBatchOperations batchOps();
 
-	/**
-	 * Returns the underlying {@link CassandraConverter}.
-	 *
-	 * @return the underlying {@link CassandraConverter}.
-	 */
-	CassandraConverter getConverter();
-
 	/**
 	 * Expose the underlying {@link ReactiveCqlOperations} to allow CQL operations.
 	 *
@@ -73,6 +67,13 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	 */
 	ReactiveCqlOperations getReactiveCqlOperations();
 
+	/**
+	 * Returns the underlying {@link CassandraConverter}.
+	 *
+	 * @return the underlying {@link CassandraConverter}.
+	 */
+	CassandraConverter getConverter();
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with static CQL
 	// -------------------------------------------------------------------------
@@ -137,6 +138,17 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	// Methods dealing with org.springframework.data.cassandra.core.query.Query
 	// -------------------------------------------------------------------------
 
+	/**
+	 * Execute the a Cassandra {@link Statement}. Any errors that result from executing this command will be converted
+	 * into Spring's DAO exception hierarchy.
+	 *
+	 * @param statement a Cassandra {@link Statement}, must not be {@literal null}.
+	 * @return the {@link ReactiveResultSet}.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 * @since 3.2
+	 */
+	Mono<ReactiveResultSet> execute(Statement<?> statement) throws DataAccessException;
+
 	/**
 	 * Execute a {@code SELECT} query and convert the resulting items to a stream of entities.
 	 *

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

@@ -23,7 +23,8 @@ import java.util.Collections;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 
-import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
@@ -32,20 +33,14 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.data.cassandra.ReactiveResultSet;
 import org.springframework.data.cassandra.ReactiveSession;
 import org.springframework.data.cassandra.ReactiveSessionFactory;
 import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
-import org.springframework.data.cassandra.core.cql.CassandraAccessor;
-import org.springframework.data.cassandra.core.cql.CqlProvider;
-import org.springframework.data.cassandra.core.cql.QueryOptions;
-import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
-import org.springframework.data.cassandra.core.cql.ReactiveCqlTemplate;
-import org.springframework.data.cassandra.core.cql.ReactiveSessionCallback;
-import org.springframework.data.cassandra.core.cql.RowMapper;
-import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.data.cassandra.core.cql.*;
 import org.springframework.data.cassandra.core.cql.session.DefaultReactiveSessionFactory;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
@@ -73,6 +68,8 @@ import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.cql.BoundStatement;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
@@ -93,6 +90,13 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
  * Can be used within a service implementation via direct instantiation with a {@link ReactiveSessionFactory} reference,
  * or get prepared in an application context and given to services as bean reference.
  * <p>
+ * This class supports the use of prepared statements when enabling {@link #setUsePreparedStatements(boolean)}. All
+ * statements created by methods of this class (such as {@link #select(Query, Class)} or
+ * {@link #update(Query, org.springframework.data.cassandra.core.query.Update, Class)} will be executed as prepared
+ * statements. Also, statements accepted by methods (such as {@link #select(String, Class)} or
+ * {@link #select(Statement, Class) and others}) will be prepared prior to execution. Note that {@link Statement}
+ * objects passed to methods must be {@link SimpleStatement} so that these can be prepared.
+ * <p>
  * Note: The {@link ReactiveSessionFactory} should always be configured as a bean in the application context, in the
  * first case given to the service directly, in the second case to the prepared template.
  *
@@ -105,20 +109,24 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
 public class ReactiveCassandraTemplate
 		implements ReactiveCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
-	private @Nullable ApplicationEventPublisher eventPublisher;
+	private final Logger logger = LoggerFactory.getLogger(getClass());
 
-	private @Nullable ReactiveEntityCallbacks entityCallbacks;
+	private final ReactiveCqlOperations cqlOperations;
 
 	private final CassandraConverter converter;
 
 	private final EntityOperations entityOperations;
 
-	private final ReactiveCqlOperations cqlOperations;
-
 	private final SpelAwareProxyProjectionFactory projectionFactory;
 
 	private final StatementFactory statementFactory;
 
+	private @Nullable ApplicationEventPublisher eventPublisher;
+
+	private @Nullable ReactiveEntityCallbacks entityCallbacks;
+
+	private boolean usePreparedStatements = true;
+
 	/**
 	 * Creates an instance of {@link ReactiveCassandraTemplate} initialized with the given {@link ReactiveSession} and a
 	 * default {@link MappingCassandraConverter}.
@@ -221,6 +229,14 @@ public class ReactiveCassandraTemplate
 		this.entityCallbacks = entityCallbacks;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getReactiveCqlOperations()
+	 */
+	@Override
+	public ReactiveCqlOperations getReactiveCqlOperations() {
+		return this.cqlOperations;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter()
 	 */
@@ -229,6 +245,32 @@ public class ReactiveCassandraTemplate
 		return this.converter;
 	}
 
+	/**
+	 * Returns whether this instance is configured to use {@link PreparedStatement prepared statements}. If enabled
+	 * (default), then all persistence methods (such as {@link #select}, {@link #update}, and others) will make use of
+	 * prepared statements. Note that methods accepting a {@link Statement} must be called with {@link SimpleStatement}
+	 * instances to participate in statement preparation.
+	 *
+	 * @return {@literal true} if prepared statements usage is enabled; {@literal false} otherwise.
+	 * @since 3.2
+	 */
+	public boolean isUsePreparedStatements() {
+		return usePreparedStatements;
+	}
+
+	/**
+	 * Enable/disable {@link PreparedStatement prepared statements} usage. If enabled (default), then all persistence
+	 * methods (such as {@link #select}, {@link #update}, and others) will make use of prepared statements. Note that
+	 * methods accepting a {@link Statement} must be called with {@link SimpleStatement} instances to participate in
+	 * statement preparation.
+	 *
+	 * @param usePreparedStatements whether to use prepared statements.
+	 * @since 3.2
+	 */
+	public void setUsePreparedStatements(boolean usePreparedStatements) {
+		this.usePreparedStatements = usePreparedStatements;
+	}
+
 	/**
 	 * Returns the {@link EntityOperations} used to perform data access operations on an entity inside a Cassandra data
 	 * source.
@@ -253,14 +295,6 @@ public class ReactiveCassandraTemplate
 		return this.projectionFactory;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getReactiveCqlOperations()
-	 */
-	@Override
-	public ReactiveCqlOperations getReactiveCqlOperations() {
-		return this.cqlOperations;
-	}
-
 	private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
 		return getEntityOperations().getRequiredPersistentEntity(entityType);
 	}
@@ -307,6 +341,17 @@ public class ReactiveCassandraTemplate
 	// Methods dealing with com.datastax.oss.driver.api.core.cql.Statement
 	// -------------------------------------------------------------------------
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#execute(com.datastax.oss.driver.api.core.cql.Statement)
+	 */
+	@Override
+	public Mono<ReactiveResultSet> execute(Statement<?> statement) throws DataAccessException {
+
+		Assert.notNull(statement, "Statement must not be null");
+
+		return doExecute(statement, Function.identity());
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(com.datastax.oss.driver.api.core.cql.Statement, java.lang.Class)
 	 */
@@ -318,7 +363,7 @@ public class ReactiveCassandraTemplate
 
 		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
-		return getReactiveCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
+		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -338,7 +383,7 @@ public class ReactiveCassandraTemplate
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Mono<ReactiveResultSet> resultSetMono = getReactiveCqlOperations().queryForResultSet(statement);
+		Mono<ReactiveResultSet> resultSetMono = doExecute(statement, Function.identity());
 		Mono<Integer> effectiveFetchSizeMono = getEffectiveFetchSize(statement);
 		RowMapper<T> rowMapper = (row, i) -> getConverter().read(entityClass, row);
 
@@ -382,7 +427,7 @@ public class ReactiveCassandraTemplate
 
 		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
 
-		return getReactiveCqlOperations().query(select.build(), (row, rowNum) -> mapper.apply(row));
+		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
 
 	/* (non-Javadoc)
@@ -431,7 +476,7 @@ public class ReactiveCassandraTemplate
 		StatementBuilder<Update> statement = getStatementFactory().update(query, update,
 				getRequiredPersistentEntity(entityClass), tableName);
 
-		return getReactiveCqlOperations().execute(new StatementCallback(statement.build())).next();
+		return doExecuteAndFlatMap(statement.build(), ReactiveCassandraTemplate::toWriteResult);
 	}
 
 	/* (non-Javadoc)
@@ -453,8 +498,8 @@ public class ReactiveCassandraTemplate
 
 		SimpleStatement delete = builder.build();
 
-		Mono<WriteResult> writeResult = getReactiveCqlOperations().execute(new StatementCallback(delete))
-				.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName))).next();
+		Mono<WriteResult> writeResult = doExecuteAndFlatMap(delete, ReactiveCassandraTemplate::toWriteResult)
+				.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)));
 
 		return writeResult.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)));
 	}
@@ -491,7 +536,14 @@ public class ReactiveCassandraTemplate
 		StatementBuilder<Select> count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass),
 				tableName);
 
-		return getReactiveCqlOperations().queryForObject(count.build(), Long.class).switchIfEmpty(Mono.just(0L));
+		SingleColumnRowMapper<Long> mapper = SingleColumnRowMapper.newInstance(Long.class);
+
+		Mono<Long> mono = doExecuteAndFlatMap(count.build(), rs -> rs.rows() //
+				.map(it -> mapper.mapRow(it, 0)) //
+				.buffer() //
+				.map(DataAccessUtils::requiredSingleResult).next());
+
+		return mono.switchIfEmpty(Mono.just(0L));
 	}
 
 	/* (non-Javadoc)
@@ -506,7 +558,7 @@ public class ReactiveCassandraTemplate
 		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
 		StatementBuilder<Select> builder = getStatementFactory().selectOneById(id, entity, entity.getTableName());
 
-		return getReactiveCqlOperations().queryForRows(builder.build()).hasElements();
+		return doQuery(builder.build(), (row, rowNum) -> row).hasElements();
 	}
 
 	/* (non-Javadoc)
@@ -526,7 +578,7 @@ public class ReactiveCassandraTemplate
 		StatementBuilder<Select> builder = getStatementFactory().select(query.limit(1),
 				getRequiredPersistentEntity(entityClass), tableName);
 
-		return getReactiveCqlOperations().queryForRows(builder.build()).hasElements();
+		return doQuery(builder.build(), (row, rowNum) -> row).hasElements();
 	}
 
 	/* (non-Javadoc)
@@ -734,7 +786,7 @@ public class ReactiveCassandraTemplate
 		StatementBuilder<Delete> builder = getStatementFactory().deleteById(id, entity, tableName);
 		SimpleStatement delete = builder.build();
 
-		Mono<Boolean> result = getReactiveCqlOperations().execute(delete)
+		Mono<Boolean> result = doExecute(delete, ReactiveResultSet::wasApplied)
 				.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)));
 
 		return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)));
@@ -752,7 +804,7 @@ public class ReactiveCassandraTemplate
 		Truncate truncate = QueryBuilder.truncate(tableName);
 		SimpleStatement statement = truncate.build();
 
-		Mono<Boolean> result = getReactiveCqlOperations().execute(statement)
+		Mono<Boolean> result = doExecute(statement, ReactiveResultSet::wasApplied)
 				.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(statement, entityClass, tableName)));
 
 		return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entityClass, tableName))).then();
@@ -810,13 +862,12 @@ public class ReactiveCassandraTemplate
 			maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
 
 			return maybeCallBeforeSave(entity, tableName, statement).flatMapMany(entityToSave -> {
-				Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+				Mono<WriteResult> execute = doExecuteAndFlatMap(statement, ReactiveCassandraTemplate::toWriteResult);
 
 				return execute.map(it -> EntityWriteResult.of(it, entityToSave)).handle(handler) //
 						.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName)));
 			}).next();
 		});
-
 	}
 
 	private Mono<WriteResult> executeDelete(Object entity, CqlIdentifier tableName, SimpleStatement statement,
@@ -824,12 +875,46 @@ public class ReactiveCassandraTemplate
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
 
-		Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+		Mono<WriteResult> execute = doExecuteAndFlatMap(statement, ReactiveCassandraTemplate::toWriteResult);
 
 		return execute.map(it -> EntityWriteResult.of(it, entity)).handle(handler) //
 				.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement))) //
-				.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName))) //
-				.next();
+				.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName)));
+	}
+
+	private <T> Flux<T> doQuery(Statement<?> statement, RowMapper<T> rowMapper) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getReactiveCqlOperations().query(statementHandler, statementHandler, rowMapper);
+		}
+
+		return getReactiveCqlOperations().query(statement, rowMapper);
+	}
+
+	private <T> Mono<T> doExecute(Statement<?> statement, Function<ReactiveResultSet, T> mappingFunction) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getReactiveCqlOperations()
+					.query(statementHandler, statementHandler, rs -> Mono.just(mappingFunction.apply(rs))).next();
+		}
+
+		return getReactiveCqlOperations().queryForResultSet(statement).map(mappingFunction);
+	}
+
+	private <T> Mono<T> doExecuteAndFlatMap(Statement<?> statement,
+			Function<ReactiveResultSet, Mono<T>> mappingFunction) {
+
+		if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
+
+			PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
+			return getReactiveCqlOperations().query(statementHandler, statementHandler, mappingFunction::apply).next();
+		}
+
+		return getReactiveCqlOperations().queryForResultSet(statement).flatMap(mappingFunction);
 	}
 
 	private int getConfiguredPageSize(DriverContext context) {
@@ -875,6 +960,11 @@ public class ReactiveCassandraTemplate
 		};
 	}
 
+	static Mono<WriteResult> toWriteResult(ReactiveResultSet resultSet) {
+		return resultSet.rows().collectList()
+				.map(rows -> new WriteResult(resultSet.getAllExecutionInfo(), resultSet.wasApplied(), rows));
+	}
+
 	private Class<?> resolveTypeToRead(Class<?> entityType, Class<?> targetType) {
 		return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
 	}
@@ -913,21 +1003,37 @@ public class ReactiveCassandraTemplate
 		return Mono.just(object);
 	}
 
-	static class StatementCallback implements ReactiveSessionCallback<WriteResult>, CqlProvider {
+	/**
+	 * Utility class to prepare a {@link SimpleStatement} and bind values associated with the statement to a
+	 * {@link BoundStatement}.
+	 *
+	 * @since 3.2
+	 */
+	private static class PreparedStatementHandler
+			implements ReactivePreparedStatementCreator, PreparedStatementBinder, CqlProvider {
 
 		private final SimpleStatement statement;
 
-		StatementCallback(SimpleStatement statement) {
-			this.statement = statement;
+		public PreparedStatementHandler(Statement<?> statement) {
+			this.statement = PreparedStatementDelegate.getStatementForPrepare(statement);
 		}
 
 		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.ReactiveSessionCallback#doInSession(org.springframework.data.cassandra.ReactiveSession)
+		 * @see org.springframework.data.cassandra.core.cql.ReactivePreparedStatementCreator#doInSession(org.springframework.data.cassandra.ReactiveSession)
 		 */
 		@Override
-		public Publisher<WriteResult> doInSession(ReactiveSession session) throws DriverException, DataAccessException {
-			return session.execute(this.statement).flatMap(StatementCallback::toWriteResult);
+		public Mono<PreparedStatement> createPreparedStatement(ReactiveSession session) throws DriverException {
+			return session.prepare(statement);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.PreparedStatementBinder#bindValues(com.datastax.oss.driver.api.core.cql.PreparedStatement)
+		 */
+		@Override
+		public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+			return PreparedStatementDelegate.bind(statement, ps);
 		}
 
 		/*
@@ -936,12 +1042,7 @@ public class ReactiveCassandraTemplate
 		 */
 		@Override
 		public String getCql() {
-			return this.statement.getQuery();
-		}
-
-		private static Mono<WriteResult> toWriteResult(ReactiveResultSet resultSet) {
-			return resultSet.rows().collectList()
-					.map(rows -> new WriteResult(resultSet.getAllExecutionInfo(), resultSet.wasApplied(), rows));
+			return statement.getQuery();
 		}
 	}
 }

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

@@ -15,12 +15,12 @@
  */
 package org.springframework.data.cassandra.core.cql;
 
+import org.springframework.util.concurrent.ListenableFuture;
+
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 
-import org.springframework.util.concurrent.ListenableFuture;
-
 /**
  * One of the two central callback interfaces used by the {@link AsyncCqlTemplate} class. This interface prepares a CQL
  * statement returning a {@link org.springframework.util.concurrent.ListenableFuture} given a {@link CqlSession},
@@ -30,13 +30,14 @@ import org.springframework.util.concurrent.ListenableFuture;
  * concern themselves with {@link DriverException}s that may be thrown from operations they attempt. The
  * {@link AsyncCqlTemplate} class will catch and handle {@link DriverException}s appropriately.
  * <p>
- * A {@link AsyncPreparedStatementCreator} should also implement the {@link CqlProvider} interface if it is able to
- * provide the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case
- * of exceptions.
+ * Classes implementing this interface should also implement the {@link CqlProvider} interface if it is able to provide
+ * the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case of
+ * exceptions.
  *
  * @author Mark Paluch
  * @since 2.0
  * @see AsyncCqlTemplate#execute(AsyncPreparedStatementCreator, PreparedStatementCallback)
+ * @see CqlProvider
  */
 @FunctionalInterface
 public interface AsyncPreparedStatementCreator {

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

@@ -26,11 +26,16 @@ import com.datastax.oss.driver.api.core.cql.PreparedStatement;
  * <p>
  * Implementations <i>do not</i> need to concern themselves with {@link DriverException}s that may be thrown from
  * operations they attempt. The {@link CqlTemplate} class will catch and handle {@link DriverException}s appropriately.
+ * <p>
+ * Classes implementing this interface should also implement the {@link CqlProvider} interface if it is able to provide
+ * the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case of
+ * exceptions.
  *
  * @author David Webb
  * @author Mark Paluch
  * @see CqlTemplate#execute(PreparedStatementCreator, PreparedStatementCallback)
  * @see CqlTemplate#query(PreparedStatementCreator, RowCallbackHandler)
+ * @see CqlProvider
  */
 @FunctionalInterface
 public interface PreparedStatementCreator {

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

@@ -15,12 +15,13 @@
  */
 package org.springframework.data.cassandra.core.cql;
 
-import com.datastax.oss.driver.api.core.DriverException;
-import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.cassandra.ReactiveSession;
 
+import com.datastax.oss.driver.api.core.DriverException;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+
 /**
  * One of the two central callback interfaces used by the {@link ReactiveCqlTemplate} class. This interface creates a
  * {@link PreparedStatement} given a {@link ReactiveSession}, provided by the {@link ReactiveCqlTemplate} class.
@@ -29,12 +30,14 @@ import org.springframework.data.cassandra.ReactiveSession;
  * concern themselves with {@link DriverException}s that may be thrown from operations they attempt. The
  * {@link ReactiveCqlTemplate} class will catch and handle {@link DriverException}s appropriately.
  * <p>
- * A {@link ReactivePreparedStatementCreator} should also implement the {@link CqlProvider} interface if it is able to
- * provide the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case
- * of exceptions.
+ * Classes implementing this interface should also implement the {@link CqlProvider} interface if it is able to provide
+ * the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case of
+ * exceptions.
  *
  * @author Mark Paluch
  * @since 2.0
+ * @see ReactiveCqlTemplate#execute(ReactivePreparedStatementCreator, ReactivePreparedStatementCallback)
+ * @see CqlProvider
  */
 @FunctionalInterface
 public interface ReactivePreparedStatementCreator {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java

@@ -149,7 +149,7 @@ public abstract class AbstractCassandraQuery extends CassandraRepositoryQuerySup
 		} else if (isExistsQuery()) {
 			return new ExistsExecution(getOperations());
 		} else if (isModifyingQuery()) {
-			return ((statement, type) -> getOperations().getCqlOperations().queryForResultSet(statement).wasApplied());
+			return ((statement, type) -> getOperations().execute(statement).wasApplied());
 		} else {
 			return new SingleEntityExecution(getOperations(), isLimiting());
 		}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java

@@ -146,8 +146,9 @@ public abstract class AbstractReactiveCassandraQuery extends CassandraRepository
 		} else if (isExistsQuery()) {
 			return new ExistsExecution(getReactiveCassandraOperations());
 		} else if (isModifyingQuery()) {
-			return (statement, type) -> getReactiveCassandraOperations().getReactiveCqlOperations()
-					.queryForResultSet(statement).map(ReactiveResultSet::wasApplied);
+
+			return (statement, type) -> getReactiveCassandraOperations().execute(statement)
+					.map(ReactiveResultSet::wasApplied);
 		} else {
 			return new SingleEntityExecution(getReactiveCassandraOperations(), isLimiting());
 		}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-import java.util.Iterator;
 import java.util.List;
 
 import org.springframework.core.convert.converter.Converter;
@@ -34,7 +33,6 @@ import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 
-import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.Statement;
 
@@ -193,25 +191,22 @@ interface CassandraQueryExecution {
 		@Override
 		public Object execute(Statement<?> statement, Class<?> type) {
 
-			ResultSet resultSet = this.operations.getCqlOperations().queryForResultSet(statement);
+			List<Row> resultSet = this.operations.select(statement, Row.class);
 
-			Iterator<Row> iterator = resultSet.iterator();
-
-			if (iterator.hasNext()) {
-
-				Row row = iterator.next();
-
-				if (!iterator.hasNext() && ProjectionUtil.qualifiesAsCountProjection(row)) {
-
-					Object object = row.getObject(0);
-
-					return ((Number) object).longValue() > 0;
-				}
-
-				return true;
+			if (resultSet.isEmpty()) {
+				return false;
 			}
 
-			return false;
+			Row row = resultSet.get(0);
+
+			if (resultSet.size() == 1 && ProjectionUtil.qualifiesAsCountProjection(row)) {
+
+				Object object = row.getObject(0);
+
+				return ((Number) object).longValue() > 0;
+			}
+
+			return true;
 		}
 	}
 
@@ -233,7 +228,7 @@ interface CassandraQueryExecution {
 		 */
 		@Override
 		public Object execute(Statement<?> statement, Class<?> type) {
-			return operations.getCqlOperations().queryForResultSet(statement);
+			return operations.execute(statement);
 		}
 	}
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java

@@ -142,17 +142,17 @@ interface ReactiveCassandraQueryExecution {
 		@Override
 		public Publisher<? extends Object> execute(Statement<?> statement, Class<?> type) {
 
-			return operations.select(statement, type).buffer(2).map(objects -> {
+			return operations.select(statement, type).buffer(2).handle((objects, sink) -> {
 
 				if (objects.isEmpty()) {
-					return null;
+					return;
 				}
 
 				if (objects.size() == 1 || limiting) {
-					return objects.get(0);
+					sink.next(objects.get(0));
 				}
 
-				throw new IncorrectResultSizeDataAccessException(1, objects.size());
+				sink.error(new IncorrectResultSizeDataAccessException(1, objects.size()));
 			});
 		}
 	}
@@ -178,7 +178,7 @@ interface ReactiveCassandraQueryExecution {
 		@Override
 		public Publisher<? extends Object> execute(Statement<?> statement, Class<?> type) {
 
-			Mono<List<Row>> rows = this.operations.getReactiveCqlOperations().queryForRows(statement).buffer(2).next();
+			Mono<List<Row>> rows = this.operations.select(statement, Row.class).buffer(2).next();
 
 			return rows.map(it -> {
 

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

@@ -57,6 +57,7 @@ class AsyncCassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		MappingCassandraConverter converter = new MappingCassandraConverter();
 		CassandraTemplate cassandraTemplate = new CassandraTemplate(session, converter);
 		template = new AsyncCassandraTemplate(new AsyncCqlTemplate(session), converter);
+		prepareTemplate(template);
 
 		SchemaTestUtils.potentiallyCreateTableFor(User.class, cassandraTemplate);
 		SchemaTestUtils.potentiallyCreateTableFor(UserToken.class, cassandraTemplate);
@@ -64,6 +65,15 @@ class AsyncCassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		SchemaTestUtils.truncate(UserToken.class, cassandraTemplate);
 	}
 
+	/**
+	 * Post-process the {@link AsyncCassandraTemplate} before running the tests.
+	 *
+	 * @param template
+	 */
+	void prepareTemplate(AsyncCassandraTemplate template) {
+		template.setUsePreparedStatements(false);
+	}
+
 	@Test // DATACASS-343
 	void shouldSelectByQueryWithSorting() {
 

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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;
+
+/**
+ * Integration tests for {@link AsyncCassandraTemplate} with
+ * {@link AsyncCassandraTemplate#setUsePreparedStatements(boolean) prepared statements enabled}.
+ *
+ * @author Mark Paluch
+ */
+class AsyncCassandraTemplatePreparedStatementsIntegrationTests extends AsyncCassandraTemplateIntegrationTests {
+
+	@Override
+	void prepareTemplate(AsyncCassandraTemplate template) {
+		template.setUsePreparedStatements(true);
+	}
+}

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

@@ -87,6 +87,7 @@ public class AsyncCassandraTemplateUnitTests {
 	void setUp() {
 
 		template = new AsyncCassandraTemplate(session);
+		template.setUsePreparedStatements(false);
 
 		when(session.executeAsync(any(Statement.class))).thenReturn(new TestResultSetFuture(resultSet));
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);

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

@@ -97,6 +97,8 @@ class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrat
 
 		template = new CassandraTemplate(new CqlTemplate(session), converter);
 
+		prepareTemplate(template);
+
 		SchemaTestUtils.potentiallyCreateTableFor(User.class, template);
 		SchemaTestUtils.potentiallyCreateTableFor(UserToken.class, template);
 		SchemaTestUtils.potentiallyCreateTableFor(BookReference.class, template);
@@ -120,6 +122,15 @@ class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrat
 		SchemaTestUtils.truncate(WithMappedUdtList.class, template);
 	}
 
+	/**
+	 * Post-process the {@link CassandraTemplate} before running the tests.
+	 *
+	 * @param template
+	 */
+	void prepareTemplate(CassandraTemplate template) {
+
+	}
+
 	@Test // DATACASS-343
 	void shouldSelectByQueryWithAllowFiltering() {
 

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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;
+
+/**
+ * Integration tests for {@link CassandraTemplate} with {@link CassandraTemplate#setUsePreparedStatements(boolean)
+ * prepared statements enabled}.
+ *
+ * @author Mark Paluch
+ */
+class CassandraTemplatePreparedStatementIntegrationTests extends CassandraTemplateIntegrationTests {
+
+	@Override
+	void prepareTemplate(CassandraTemplate template) {
+		template.setUsePreparedStatements(true);
+	}
+}

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

@@ -82,6 +82,7 @@ class CassandraTemplateUnitTests {
 	void setUp() {
 
 		template = new CassandraTemplate(session);
+		template.setUsePreparedStatements(false);
 
 		when(session.execute(any(Statement.class))).thenReturn(resultSet);
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);

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

@@ -59,6 +59,7 @@ class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceCreating
 		DefaultBridgedReactiveSession session = new DefaultBridgedReactiveSession(this.session);
 
 		template = new ReactiveCassandraTemplate(new ReactiveCqlTemplate(session), converter);
+		prepareTemplate(template);
 
 		SchemaTestUtils.potentiallyCreateTableFor(User.class, cassandraTemplate);
 		SchemaTestUtils.potentiallyCreateTableFor(UserToken.class, cassandraTemplate);
@@ -66,6 +67,15 @@ class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceCreating
 		SchemaTestUtils.truncate(UserToken.class, cassandraTemplate);
 	}
 
+	/**
+	 * Post-process the {@link ReactiveCassandraTemplate} before running the tests.
+	 *
+	 * @param template
+	 */
+	void prepareTemplate(ReactiveCassandraTemplate template) {
+		template.setUsePreparedStatements(false);
+	}
+
 	@Test // DATACASS-335
 	void insertShouldInsertEntity() {
 

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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;
+
+/**
+ * Integration tests for {@link ReactiveCassandraTemplate} with
+ * {@link ReactiveCassandraTemplate#setUsePreparedStatements(boolean) prepared statements enabled}.
+ *
+ * @author Mark Paluch
+ */
+class ReactiveCassandraTemplatePreparedStatementIntegrationTests extends ReactiveCassandraTemplateIntegrationTests {
+
+	@Override
+	void prepareTemplate(ReactiveCassandraTemplate template) {
+		template.setUsePreparedStatements(true);
+	}
+}

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

@@ -84,6 +84,7 @@ class ReactiveCassandraTemplateUnitTests {
 	void setUp() {
 
 		template = new ReactiveCassandraTemplate(session);
+		template.setUsePreparedStatements(false);
 
 		when(session.execute(any(Statement.class))).thenReturn(Mono.just(reactiveResultSet));
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/example/CassandraTemplateExamples.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https:://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.example;
+
+import static org.springframework.data.cassandra.core.query.Criteria.*;
+import static org.springframework.data.cassandra.core.query.Query.*;
+
+import org.springframework.data.cassandra.core.CassandraTemplate;
+
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+
+/**
+ * @author Mark Paluch
+ */
+// @formatter:off
+public class CassandraTemplateExamples {
+
+	private CassandraTemplate template = null;
+
+	void examples() {
+		// tag::preparedStatement[]
+		template.setUsePreparedStatements(true);
+
+		Actor actorByQuery = template.selectOne(query(where("id").is(42)), Actor.class);
+
+		Actor actorByStatement = template.selectOne(
+				SimpleStatement.newInstance("SELECT id, name FROM actor WHERE id = ?", 42),
+				Actor.class);
+		// end::preparedStatement[]
+	}
+
+	static class Actor {
+
+	}
+
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/example/CqlTemplateExamples.java

@@ -74,6 +74,13 @@ public class CqlTemplateExamples {
 					}
 				});
 		// end::listOfRowMapper[]
+
+		// tag::preparedStatement[]
+		List<String> lastNames = cqlTemplate.query(
+				session -> session.prepare("SELECT last_name FROM t_actor WHERE id = ?"),
+				ps -> ps.bind(1212L),
+				(row, rowNum) -> row.getString(0));
+		// end::preparedStatement[]
 	}
 
 	// tag::findAllActors[]

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/isolated/RepositoryQueryMethodParameterTypesIntegrationTests.java

@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.cassandra.CassandraInvalidQueryException;
 import org.springframework.data.cassandra.config.SchemaAction;
+import org.springframework.data.cassandra.core.CassandraAdminTemplate;
 import org.springframework.data.cassandra.core.convert.CassandraCustomConversions;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -75,6 +76,13 @@ class RepositoryQueryMethodParameterTypesIntegrationTests
 		public SchemaAction getSchemaAction() {
 			return SchemaAction.RECREATE_DROP_UNUSED;
 		}
+
+		@Override
+		public CassandraAdminTemplate cassandraTemplate() {
+			CassandraAdminTemplate template = super.cassandraTemplate();
+			template.setUsePreparedStatements(false);
+			return template;
+		}
 	}
 
 	@Autowired AllPossibleTypesRepository allPossibleTypesRepository;

spring-data-cassandra/src/test/kotlin/org/springframework/data/cassandra/repository/CoroutineRepositoryUnitTests.kt

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.cassandra.repository
 
-import com.datastax.oss.driver.api.core.cql.Statement
 import io.mockk.every
 import io.mockk.mockk
 import kotlinx.coroutines.runBlocking
@@ -55,7 +54,7 @@ class CoroutineRepositoryUnitTests {
 	fun `should discard result of suspended query method without result`() {
 
 		every { resultSet.wasApplied() } returns true
-		every { cqlOperations.queryForResultSet(any<Statement<*>>()) } returns Mono.just(resultSet)
+		every { operations.execute(any()) } returns Mono.just(resultSet)
 
 		val repository = repositoryFactory.getRepository(PersonRepository::class.java)
 

src/main/asciidoc/new-features.adoc

@@ -3,6 +3,11 @@
 
 This chapter summarizes changes and new features for each release.
 
+[[new-features.3-2-0]]
+== What's new in Spring Data for Apache Cassandra 3.2
+
+* <<cassandra.template.prepared-statements,Support for prepared statements>> using `CassandraTemplate` and repositories (enabled by default).
+
 [[new-features.3-1-0]]
 == What's new in Spring Data for Apache Cassandra 3.1
 

src/main/asciidoc/reference/cassandra-repositories.adoc

@@ -3,6 +3,7 @@
 
 This chapter covers the details of the Spring Data Repository support for Apache Cassandra.
 Cassandra's repository support builds on the core repository support explained in "`<<repositories>>`".
+Cassandra repositories use `CassandraTemplate` and its wired `CqlTemplate` as infrastructure beans.
 You should understand the basic concepts explained there before proceeding.
 
 [[cassandra-repo-usage]]

src/main/asciidoc/reference/cassandra.adoc

@@ -520,7 +520,7 @@ The `CqlTemplate` class executes CQL queries and update statements, performs ite
 It also catches CQL exceptions and translates them to the generic, more informative, exception hierarchy defined in the `org.springframework.dao` package.
 
 When you use the `CqlTemplate` for your code, you need only implement callback interfaces, which have a clearly defined contract.
-Given a `Connection`, the `PreparedStatementCreator` callback interface creates a prepared statement with the provided CQL and any necessary parameter arguments.
+Given a `Connection`, the `PreparedStatementCreator` callback interface creates a <<cassandra.template.prepared-statements.cql,prepared statement>> with the provided CQL and any necessary parameter arguments.
 The `RowCallbackHandler` interface extracts values from each row of a `ResultSet`.
 
 The `CqlTemplate` can be used within a DAO implementation through direct instantiation with a `SessionFactory` reference or be configured in the Spring container and given to DAOs as a bean reference. `CqlTemplate` is a foundational building block for <<cassandra-template,`CassandraTemplate`>>.
@@ -1118,3 +1118,78 @@ The terminating methods (`first()`, `one()`, `all()`, and `stream()`) handle swi
 WARNING: The new fluent template API methods (that is, `query(..)`, `insert(..)`, `update(..)`, and `delete(..)`) use effectively thread-safe supporting objects to compose the CQL statement.
 However, it comes at the added cost of additional young-gen JVM heap overhead, since the design is based on final fields for the various CQL statement components and construction on mutation.
 You should be careful when possibly inserting or deleting a large number of objects (such as inside of a loop, for instance).
+
+[[cassandra.template.prepared-statements]]
+== Prepared Statements
+
+CQL statements that are executed multiple times can be prepared and stored in a `PreparedStatement` object to improve query performance.
+Both, the driver and Cassandra maintain a mapping of `PreparedStatement` queries to their metadata.
+You can use prepared statements through the following abstractions:
+
+* `CqlTemplate` through the choice of API
+* `CassandraTemplate` by enabling prepared statements
+* Cassandra repositories as they are built on `CassandraTemplate`
+
+[[cassandra.template.prepared-statements.cql]]
+=== Using `CqlTemplate`
+
+The `CqlTemplate` class (and its asynchronous and reactive variants) offers various methods accepting static CQL, `Statement` objects and `PreparedStatementCreator`.
+Methods accepting static CQL without additional arguments typically run the CQL statement as-is without further processing.
+Methods accepting static CQL in combination with an arguments array (such as `execute(String cql, Object... args)` and `queryForRows(String cql, Object... args)`) use prepared statements.
+Internally, these methods create a `PreparedStatementCreator` and `PreparedStatementBinder` objects to prepare the statement and later on to bind values to the statement to run it.
+Spring Data Cassandra generally uses index-based parameter bindings for prepared statements.
+
+Since Cassandra Driver version 4, prepared statements are cached on the driver level which removes the need to keep track of prepared statements in the application.
+
+The following example shows how to issue a query with a parametrized prepared statement:
+
+====
+[source,java,indent=0]
+----
+include::../{example-root}/CqlTemplateExamples.java[tags=lastName]
+----
+====
+
+In cases where you require more control over statement preparation and parameter binding (for example, using named binding parameters), you can fully control prepared statement creation and parameter binding by calling query methods with `PreparedStatementCreator` and `PreparedStatementBinder` arguments:
+
+====
+[source,java,indent=0]
+----
+include::../{example-root}/CqlTemplateExamples.java[tags=preparedStatement]
+----
+====
+
+Spring Data Cassandra ships with classes supporting that pattern in the `cql` package:
+
+* `SimplePreparedStatementCreator` - utility class to create a prepared statement.
+* `ArgumentPreparedStatementBinder` - utility class to bind arguments to a prepared statement.
+
+[[cassandra.template.prepared-statements.cassandra-template]]
+=== Using `CassandraTemplate`
+
+The `CassandraTemplate` class is built on top of `CqlTemplate` to provide a higher level of abstraction.
+The use of prepared statements can be controlled directly on `CassandraTemplate` (and its asynchronous and reactive variants) by calling `setUsePreparedStatements(false)` respective `setUsePreparedStatements(true)`.
+Note that the use of prepared statements by `CassandraTemplate` is enabled by default.
+
+The following example shows the use of methods that generate and that accept CQL:
+
+====
+[source,java,indent=0]
+----
+include::../{example-root}/CassandraTemplateExamples.java[tags=preparedStatement]
+----
+====
+
+Calling entity-bound methods such as `select(Query, Class<T>)` or `update(Query, Update, Class<T>)` build CQL statements themselves to perform the intended operations.
+Some `CassandraTemplate` methods (such as `select(Statement<?>, Class<T>)`) also accepts CQL `Statement` objects as part of their API.
+
+It's possible to participate in prepared statements when calling methods accepting a `Statement` with a `SimpleStatement` object.
+The template API extracts the query string and parameters (positional and named parameters) and uses these to prepare, bind, and run the statement.
+Non-``SimpleStatement`` objects cannot be used with prepared statements.
+
+[[cassandra.template.prepared-statements.caching]]
+=== Caching Prepared Statements
+
+Since Cassandra driver 4.0, prepared statements are cached by the `CqlSession` cache so it is okay to prepare the same string twice.
+Previous versions required caching of prepared statements outside of the driver.
+See also the https://docs.datastax.com/en/developer/java-driver/latest/manual/core/statements/prepared/[Driver documentation on Prepared Statements] for further reference.

src/main/asciidoc/reference/reactive-cassandra-repositories.adoc

@@ -4,6 +4,8 @@
 This chapter outlines the specialties handled by the reactive repository support for Apache Cassandra.
 It builds on the core repository infrastructure explained in <<cassandra.repositories>>, so you should have a good understanding of the basic concepts explained there.
 
+Cassandra repositories use `ReactiveCassandraTemplate` and its wired `ReactiveCqlTemplate` as infrastructure beans.
+
 Reactive usage is broken up into two phases: Composition and Execution.
 
 Calling repository methods lets you compose a reactive sequence by obtaining `Publisher` instances and applying operators.

src/main/asciidoc/reference/reactive-cassandra.adoc

@@ -150,7 +150,7 @@ The `ReactiveCqlTemplate` class runs CQL queries and update statements and perfo
 It also catches CQL exceptions and translates them into the generic, more informative, exception hierarchy defined in the `org.springframework.dao` package.
 
 When you use the `ReactiveCqlTemplate` in your code, you need only implement callback interfaces, which have a clearly defined contract.
-Given a `Connection`, the `ReactivePreparedStatementCreator` callback interface creates a prepared statement with the provided CQL and any necessary parameter arguments.
+Given a `Connection`, the `ReactivePreparedStatementCreator` callback interface creates a <<cassandra.template.prepared-statements.cql,prepared statement>> with the provided CQL and any necessary parameter arguments.
 The `RowCallbackHandler`
 interface extracts values from each row of a `ReactiveResultSet`.