Use prepare(String) for Prepared Statement preparation to prevent bind values from being cached.

We now also apply query options from the initial statement to the bound statement by copying these if query options are set.

Closes #1213

Author: mp911de

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

@@ -1081,7 +1081,7 @@ public ListenableFuture<PreparedStatement> createPreparedStatement(CqlSession se
 		 * @return
 		 */
 		protected CompletionStage<PreparedStatement> doPrepare(CqlSession session) {
-			return session.prepareAsync(statement);
+			return session.prepareAsync(statement.getQuery());
 		}
 
 		/*

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

@@ -1087,7 +1087,7 @@ public PreparedStatementHandler(Statement<?> statement) {
 		 */
 		@Override
 		public PreparedStatement createPreparedStatement(CqlSession session) throws DriverException {
-			return session.prepare(statement);
+			return session.prepare(statement.getQuery());
 		}
 
 		/*

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

@@ -16,11 +16,18 @@
 package org.springframework.data.cassandra.core;
 
 import java.util.Map;
+import java.util.Objects;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 
 import org.springframework.data.cassandra.core.cql.QueryExtractorDelegate;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
+import org.springframework.util.function.SingletonSupplier;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
@@ -30,6 +37,7 @@
 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;
+import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
 
 /**
  * Support class for Cassandra Template API implementation classes that want to make use of prepared statements.
@@ -40,30 +48,51 @@
 class PreparedStatementDelegate {
 
 	/**
-	 * Bind values held in {@link SimpleStatement} to the {@link PreparedStatement}.
+	 * Bind values held in {@link SimpleStatement} to the {@link PreparedStatement} and apply query options that are set
+	 * or do not match the default value.
 	 *
-	 * @param statement
+	 * @param source
 	 * @param ps
 	 * @return the bound statement.
 	 */
-	static BoundStatement bind(SimpleStatement statement, PreparedStatement ps) {
+	static BoundStatement bind(SimpleStatement source, PreparedStatement ps) {
+
+		BoundStatementBuilder builder = ps.boundStatementBuilder(source.getPositionalValues().toArray());
+
+		Mapper mapper = Mapper.INSTANCE;
 
-		BoundStatementBuilder boundStatementBuilder = ps.boundStatementBuilder(statement.getPositionalValues().toArray());
-		Map<CqlIdentifier, Object> namedValues = statement.getNamedValues();
+		mapper.from(source.getExecutionProfileName()).whenHasText().to(builder::setExecutionProfileName);
+		mapper.from(source.getExecutionProfile()).whenNonNull().to(builder::setExecutionProfile);
+		mapper.from(source.getRoutingKeyspace()).whenNonNull().to(builder::setRoutingKeyspace);
+		mapper.from(source.getRoutingKey()).whenNonNull().to(builder::setRoutingKey);
+		mapper.from(source.getRoutingToken()).whenNonNull().to(builder::setRoutingToken);
+		mapper.from(source.isIdempotent()).whenNonNull().to(builder::setIdempotence);
+		mapper.from(source.isTracing()).whenNonNull().to(builder::setTracing);
+		mapper.from(source.getQueryTimestamp()).whenNot(it -> it == Statement.NO_DEFAULT_TIMESTAMP)
+				.to(builder::setQueryTimestamp);
+		mapper.from(source.getPagingState()).whenNonNull().to(builder::setPagingState);
+		mapper.from(source.getPageSize()).whenNot(it -> it == 0L).to(builder::setPageSize);
+		mapper.from(source.getConsistencyLevel()).whenNonNull().to(builder::setConsistencyLevel);
+		mapper.from(source.getSerialConsistencyLevel()).whenNonNull().to(builder::setSerialConsistencyLevel);
+		mapper.from(source.getTimeout()).whenNonNull().to(builder::setTimeout);
+		mapper.from(source.getNode()).whenNonNull().to(builder::setNode);
+		mapper.from(source.getNowInSeconds()).whenNot(it -> it == Statement.NO_NOW_IN_SECONDS).to(builder::setNowInSeconds);
+
+		Map<CqlIdentifier, Object> namedValues = source.getNamedValues();
 
 		ColumnDefinitions variableDefinitions = ps.getVariableDefinitions();
+		CodecRegistry codecRegistry = builder.codecRegistry();
 		for (Map.Entry<CqlIdentifier, Object> entry : namedValues.entrySet()) {
 
 			if (entry.getValue() == null) {
-				boundStatementBuilder = boundStatementBuilder.setToNull(entry.getKey());
+				builder = builder.setToNull(entry.getKey());
 			} else {
 				DataType type = variableDefinitions.get(entry.getKey()).getType();
-				boundStatementBuilder = boundStatementBuilder.set(entry.getKey(), entry.getValue(),
-						boundStatementBuilder.codecRegistry().codecFor(type));
+				builder = builder.set(entry.getKey(), entry.getValue(), codecRegistry.codecFor(type));
 			}
 		}
 
-		return ps.bind(statement.getPositionalValues().toArray());
+		return builder.build();
 	}
 
 	/**
@@ -117,4 +146,146 @@ private static String getMessage(Statement<?> statement) {
 		return String.format("Cannot prepare statement %s. Statement must be a SimpleStatement.", statement);
 	}
 
+	enum Mapper {
+
+		INSTANCE;
+
+		/**
+		 * Return a new {@link Source} from the specified value supplier that can be used to perform the mapping.
+		 *
+		 * @param <T> the source type
+		 * @param supplier the value supplier
+		 * @return a {@link Source} that can be used to complete the mapping
+		 * @see #from(Object)
+		 */
+		public <T> Source<T> from(Supplier<T> supplier) {
+
+			Assert.notNull(supplier, "Supplier must not be null");
+			return getSource(supplier);
+		}
+
+		/**
+		 * Return a new {@link Source} from the specified value that can be used to perform the mapping.
+		 *
+		 * @param <T> the source type
+		 * @param value the value
+		 * @return a {@link Source} that can be used to complete the mapping
+		 */
+		public <T> Source<T> from(@Nullable T value) {
+			return from(() -> value);
+		}
+
+		private <T> Source<T> getSource(Supplier<T> supplier) {
+			return new Source<>(SingletonSupplier.of(supplier), t -> true);
+		}
+	}
+
+	/**
+	 * A source value/supplier that is in the process of being mapped.
+	 *
+	 * @param <T> the source type
+	 */
+	static class Source<T> {
+
+		private final Supplier<T> supplier;
+
+		private final Predicate<T> predicate;
+
+		private Source(Supplier<T> supplier, Predicate<T> predicate) {
+
+			Assert.notNull(predicate, "Predicate must not be null");
+
+			this.supplier = supplier;
+			this.predicate = predicate;
+		}
+
+		/**
+		 * Return a filtered version of the source that won't map non-null values or suppliers that throw a
+		 * {@link NullPointerException}.
+		 *
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenNonNull() {
+			return new Source<>(this.supplier, Objects::nonNull);
+		}
+
+		/**
+		 * Return a filtered version of the source that will only map values that are {@code true}.
+		 *
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenTrue() {
+			return when(Boolean.TRUE::equals);
+		}
+
+		/**
+		 * Return a filtered version of the source that will only map values that are {@code false}.
+		 *
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenFalse() {
+			return when(Boolean.FALSE::equals);
+		}
+
+		/**
+		 * Return a filtered version of the source that will only map values that have a {@code toString()} containing
+		 * actual text.
+		 *
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenHasText() {
+			return when((value) -> StringUtils.hasText(Objects.toString(value, null)));
+		}
+
+		/**
+		 * Return a filtered version of the source that will only map values equal to the specified {@code object}.
+		 *
+		 * @param object the object to match
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenEqualTo(Object object) {
+			return when(object::equals);
+		}
+
+		/**
+		 * Return a filtered version of the source that won't map values that match the given predicate.
+		 *
+		 * @param predicate the predicate used to filter values
+		 * @return a new filtered source instance
+		 */
+		public Source<T> whenNot(Predicate<T> predicate) {
+
+			Assert.notNull(predicate, "Predicate must not be null");
+			return when(predicate.negate());
+		}
+
+		/**
+		 * Return a filtered version of the source that won't map values that don't match the given predicate.
+		 *
+		 * @param predicate the predicate used to filter values
+		 * @return a new filtered source instance
+		 */
+		public Source<T> when(Predicate<T> predicate) {
+
+			Assert.notNull(predicate, "Predicate must not be null");
+			return new Source<>(this.supplier, (this.predicate != null) ? this.predicate.and(predicate) : predicate);
+		}
+
+		/**
+		 * Complete the mapping by passing any non-filtered value to the specified consumer.
+		 *
+		 * @param consumer the consumer that should accept the value if it's not been filtered
+		 */
+		public void to(Consumer<T> consumer) {
+
+			Assert.notNull(consumer, "Consumer must not be null");
+
+			T value = this.supplier.get();
+			if (this.predicate.test(value)) {
+				consumer.accept(value);
+			}
+		}
+
+	}
+
 }

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

@@ -1060,7 +1060,7 @@ public PreparedStatementHandler(Statement<?> statement) {
 		 */
 		@Override
 		public Mono<PreparedStatement> createPreparedStatement(ReactiveSession session) throws DriverException {
-			return session.prepare(statement);
+			return session.prepare(statement.getQuery());
 		}
 
 		/*

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

@@ -181,4 +181,5 @@ private static int getTtlSeconds(Duration ttl) {
 	private static boolean hasTtl(Duration ttl) {
 		return !ttl.isZero() && !ttl.isNegative();
 	}
+
 }