DATACASS-656 - Add StatementBuilder and AsyncResultStream utilities.

We now provide a utility class to build CQL statements using a fluent functional declaration.

AsyncResultStream can forward-stream over a AsyncResultStream by fetching subsequent pages and applying either a Consumer or Collector for non-blocking processing of asynchronously fetched results.

Original pull request: #167.

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

@@ -1,26 +0,0 @@
-/*
- * Copyright 2013-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.config;
-
-/**
- * Spring CQL extension of CassandraCqlClusterFactoryBean. This class exists only in the name of symmetry, based on the
- * other CassandraData*FactoryBean classes.
- *
- * @author Matthew T. Adams
- * @author Mark Paluch
- */
-public class CassandraCqlClusterFactoryBean extends CassandraClusterFactoryBean {}

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

@@ -0,0 +1,235 @@
+/*
+ * Copyright 2019 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.cql;
+
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.stream.Collector;
+
+import org.springframework.util.Assert;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.SettableListenableFuture;
+
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+
+/**
+ * Asynchronous supplied sequence of elements supporting sequential operations over a {@link AsyncResultSet a result
+ * set}. An asynchronous stream represents a pipeline of operations to process a {@link AsyncResultSet}.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class AsyncResultStream<T> {
+
+	private final AsyncResultSet resultSet;
+
+	private final RowMapper<T> mapper;
+
+	private AsyncResultStream(AsyncResultSet resultSet, RowMapper<T> mapper) {
+		this.resultSet = resultSet;
+		this.mapper = mapper;
+	}
+
+	/**
+	 * Creates a {@link AsyncResultStream} given {@link AsyncResultSet}.
+	 *
+	 * @param resultSet the result set to process.
+	 * @return a new {@link AsyncResultStream} instance.
+	 */
+	static AsyncResultStream<Row> from(AsyncResultSet resultSet) {
+
+		Assert.notNull(resultSet, "AsyncResultSet must not be null");
+
+		return new AsyncResultStream<>(resultSet, (row, rowNum) -> row);
+	}
+
+	/**
+	 * Returns a stream consisting of the results of applying the given function to the elements of this stream.
+	 * <p>
+	 * This is an intermediate operation.
+	 *
+	 * @param <R> The element type of the new stream
+	 * @param mapper a non-interfering, stateless {@link RowMapper}.
+	 */
+	<R> AsyncResultStream<R> map(RowMapper<R> mapper) {
+
+		Assert.notNull(mapper, "RowMapper must not be null");
+
+		return new AsyncResultStream<>(resultSet, mapper);
+	}
+
+	/**
+	 * Performs a mutable reduction operation on the elements of this stream using a {@link Collector} resulting in a
+	 * {@link ListenableFuture}.
+	 * <p>
+	 * This is a terminal operation.
+	 *
+	 * @param <R> the type of the result
+	 * @param <A> the intermediate accumulation type of the {@link Collector}
+	 * @param collector the {@link Collector} describing the reduction
+	 * @return the result of the reduction
+	 */
+	<R, A> ListenableFuture<R> collect(Collector<? super T, A, R> collector) {
+
+		Assert.notNull(collector, "Collector must not be null");
+
+		SettableListenableFuture<R> future = new SettableListenableFuture<>();
+		CollectState<A, R> collectState = new CollectState<>(collector);
+
+		collectState.collectAsync(future, this.resultSet);
+
+		return future;
+	}
+
+	/**
+	 * Performs an action for each element of this stream. This method returns a {@link ListenableFuture} that completes
+	 * without a value ({@code null}) once all elements have been processed.
+	 * <p>
+	 * This is a terminal operation.
+	 * <p>
+	 * If the action accesses shared state, it is responsible for providing the required synchronization.
+	 *
+	 * @param action a non-interfering action to perform on the elements.
+	 */
+	ListenableFuture<Void> forEach(Consumer<T> action) {
+
+		Assert.notNull(action, "Action must not be null");
+
+		SettableListenableFuture<Void> future = new SettableListenableFuture<>();
+		ForwardLoopState loopState = new ForwardLoopState(action);
+
+		loopState.forEachAsync(future, this.resultSet);
+
+		return future;
+	}
+
+	/**
+	 * State object for forward-looping using {@code forEach}.
+	 */
+	class ForwardLoopState {
+
+		private final AtomicInteger rowNumber = new AtomicInteger();
+		private final Consumer<T> consumer;
+
+		ForwardLoopState(Consumer<T> consumer) {
+			this.consumer = consumer;
+		}
+
+		void peekRow(Iterable<Row> rows) {
+			rows.forEach(row -> consumer.accept(mapper.mapRow(row, rowNumber.incrementAndGet())));
+		}
+
+		/**
+		 * Recursive async iteration.
+		 *
+		 * @param target
+		 * @param resultSet
+		 */
+		void forEachAsync(SettableListenableFuture<Void> target, AsyncResultSet resultSet) {
+
+			if (target.isCancelled()) {
+				return;
+			}
+
+			try {
+				peekRow(resultSet.currentPage());
+			} catch (RuntimeException e) {
+				target.setException(e);
+				return;
+			}
+
+			if (!resultSet.hasMorePages()) {
+				target.set(null);
+			} else {
+
+				CompletionStage<AsyncResultSet> nextPage = resultSet.fetchNextPage();
+
+				nextPage.whenComplete((nextResultSet, throwable) -> {
+
+					if (throwable != null) {
+						target.setException(throwable);
+					} else {
+						forEachAsync(target, nextResultSet);
+					}
+				});
+			}
+		}
+	}
+
+	/**
+	 * State object for collecting rows using {@code collect}.
+	 */
+	class CollectState<A, R> {
+
+		private final AtomicInteger rowNumber = new AtomicInteger();
+		private volatile A intermediate;
+		private final Collector<? super T, A, R> collector;
+
+		CollectState(Collector<? super T, A, R> collector) {
+			this.collector = collector;
+			this.intermediate = collector.supplier().get();
+		}
+
+		void collectPage(Iterable<Row> rows) {
+
+			for (Row row : rows) {
+				collector.accumulator().accept(intermediate, mapper.mapRow(row, rowNumber.incrementAndGet()));
+			}
+		}
+
+		R finish() {
+			return collector.finisher().apply(intermediate);
+		}
+
+		/**
+		 * Recursive collection.
+		 *
+		 * @param target
+		 * @param resultSet
+		 */
+		void collectAsync(SettableListenableFuture<R> target, AsyncResultSet resultSet) {
+
+			if (target.isCancelled()) {
+				return;
+			}
+
+			try {
+				collectPage(resultSet.currentPage());
+			} catch (RuntimeException e) {
+				target.setException(e);
+				return;
+			}
+
+			if (!resultSet.hasMorePages()) {
+				target.set(finish());
+			} else {
+
+				CompletionStage<AsyncResultSet> nextPage = resultSet.fetchNextPage();
+
+				nextPage.whenComplete((nextResultSet, throwable) -> {
+
+					if (throwable != null) {
+						target.setException(throwable);
+					} else {
+						collectAsync(target, nextResultSet);
+					}
+				});
+			}
+		}
+	}
+}

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

@@ -0,0 +1,267 @@
+/*
+ * Copyright 2019 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.cql.util;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.UnaryOperator;
+
+import org.springframework.util.Assert;
+
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.cql.SimpleStatementBuilder;
+import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
+import com.datastax.oss.driver.api.querybuilder.BuildableQuery;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+
+/**
+ * Functional builder for Cassandra {@link BuildableQuery statements}. Statements are built by applying
+ * {@link UnaryOperator builder functions} that get applied when {@link #build() building the actual
+ * {@link SimpleStatement statement}. The {@code StatmentBuilder} provides a mutable container for statement creation
+ * allowing a functional declaration of actions that are necessary to build a statement. This class helps building CQL
+ * statements as a {@link BuildableQuery} classes are typically immutable and require return value tracking across
+ * methods that want to apply modifications to a statment.
+ * <p>
+ * Building a statement consists of three phases:
+ * <ol>
+ * <li>Creation of the {@link StatementBuilder} with a {@link BuildableQuery query stub}</li>
+ * <li>Functional declaration applying {@link UnaryOperator builder functions}, {@link BindFunction bind functions} and
+ * {@link Consumer on build signals}</li>
+ * <li>Building the statement using {@link #build()}</li>
+ * </ol>
+ * The initial {@link BuildableQuery query stub} is used as base object for all built queries. Builder functions are
+ * applied each time a statement is built allowing to build multiple statement instances while evolving the actual
+ * statement.
+ * <p>
+ * The builder can be used for structural evolution and value evolution of statements. Values are bound through
+ * {@link BindFunction binding functions} that accept the statement and a {@link TermFactory}. Values can be bound
+ * inline or through bind markers when {@link #build(ParameterHandling, CodecRegistry) building} the statement. All
+ * functions are applied in the order of their declaration.
+ * <p>
+ * All methods returning {@link StatementBuilder} point to the same instance. This class is intended for internal use.
+ *
+ * @author Mark Paluch
+ * @param <S> Statement type
+ * @since 3.0
+ */
+public class StatementBuilder<S extends BuildableQuery> {
+
+	private S statement;
+
+	private List<BuilderRunnable<S>> queryActions = new ArrayList<>();
+	private List<Consumer<SimpleStatementBuilder>> onBuild = new ArrayList<>();
+
+	private StatementBuilder(S statement) {
+		this.statement = statement;
+	}
+
+	/**
+	 * Create a new {@link StatementBuilder} with the given {@link BuildableQuery query stub}. The stub is used as base
+	 * for the built query so each query inherits properties of this stub.
+	 *
+	 * @param stub the query stub to use.
+	 * @param <S> query type.
+	 * @return the {@link StatementBuilder} for the {@link BuildableQuery query stub}.
+	 */
+	public static <S extends BuildableQuery> StatementBuilder<S> of(S stub) {
+
+		Assert.notNull(stub, "Query stub must not be null");
+
+		return new StatementBuilder<>(stub);
+	}
+
+	/**
+	 * Apply a {@link BindFunction} to the statement. Bind functions are applied on {@link #build()}.
+	 *
+	 * @param action the bind function to be applied to the statement.
+	 * @return {@code this} {@link StatementBuilder}.
+	 */
+	public StatementBuilder<S> bind(BindFunction<S> action) {
+
+		Assert.notNull(action, "BindFunction must not be null");
+
+		queryActions.add(action::bind);
+		return this;
+	}
+
+	/**
+	 * Apply a {@link UnaryOperator builder function} to the statement. Builder functions are applied on {@link #build()}.
+	 *
+	 * @param action the builder function to be applied to the statement.
+	 * @return {@code this} {@link StatementBuilder}.
+	 */
+	public StatementBuilder<S> apply(UnaryOperator<S> action) {
+
+		Assert.notNull(action, "BindFunction must not be null");
+
+		queryActions.add((source, termFactory) -> action.apply(source));
+		return this;
+	}
+
+	/**
+	 * Add behavior when the statement is built. The {@link Consumer} gets invoked with a {@link SimpleStatementBuilder}
+	 * allowing association of the final statement with additional settings. The {@link Consumer} is applied on
+	 * {@link #build()}.
+	 *
+	 * @param action the {@link Consumer} function that gets notified on {@link #build()}.
+	 * @return {@code this} {@link StatementBuilder}.
+	 */
+	public StatementBuilder<S> onBuild(Consumer<SimpleStatementBuilder> action) {
+
+		Assert.notNull(action, "Consumer must not be null");
+
+		onBuild.add(action);
+		return this;
+	}
+
+	/**
+	 * Build a {@link SimpleStatement statement} by applying builder and bind functions using the default
+	 * {@link CodecRegistry} and {@link ParameterHandling#INLINE} parameter rendering.
+	 *
+	 * @return the built {@link SimpleStatement}.
+	 */
+	public SimpleStatement build() {
+		return build(ParameterHandling.INLINE, CodecRegistry.DEFAULT);
+	}
+
+	/**
+	 * Build a {@link SimpleStatement statement} by applying builder and bind functions using the given
+	 * {@link ParameterHandling}.
+	 *
+	 * @param parameterHandling
+	 * @return the built {@link SimpleStatement}.
+	 */
+	public SimpleStatement build(ParameterHandling parameterHandling) {
+		return build(parameterHandling, CodecRegistry.DEFAULT);
+	}
+
+	/**
+	 * Build a {@link SimpleStatement statement} by applying builder and bind functions using the given
+	 * {@link CodecRegistry} and {@link ParameterHandling}.
+	 *
+	 * @param parameterHandling
+	 * @param codecRegistry
+	 * @return the built {@link SimpleStatement}.
+	 */
+	public SimpleStatement build(ParameterHandling parameterHandling, CodecRegistry codecRegistry) {
+
+		Assert.notNull(parameterHandling, "ParameterHandling must not be null");
+		Assert.notNull(codecRegistry, "CodecRegistry must not be null");
+
+		S statement = this.statement;
+
+		if (parameterHandling == ParameterHandling.INLINE) {
+
+			TermFactory termFactory = value -> QueryBuilder.literal(value, codecRegistry);
+
+			for (BuilderRunnable<S> runnable : queryActions) {
+				statement = runnable.run(statement, termFactory);
+			}
+
+			return onBuild(statement.builder()).build();
+		}
+
+		if (parameterHandling == ParameterHandling.BY_INDEX) {
+
+			List<Object> values = new ArrayList<>();
+			TermFactory termFactory = value -> {
+				values.add(value);
+				return QueryBuilder.bindMarker();
+			};
+
+			for (BuilderRunnable<S> runnable : queryActions) {
+				statement = runnable.run(statement, termFactory);
+			}
+
+			return onBuild(statement.builder().addPositionalValues(values)).build();
+		}
+
+		if (parameterHandling == ParameterHandling.BY_NAME) {
+
+			Map<String, Object> values = new LinkedHashMap<>();
+			TermFactory termFactory = value -> {
+				String name = "p" + values.size();
+				values.put(name, value);
+				return QueryBuilder.bindMarker(name);
+			};
+
+			for (BuilderRunnable<S> runnable : queryActions) {
+				statement = runnable.run(statement, termFactory);
+			}
+
+			SimpleStatementBuilder builder = statement.builder();
+			values.forEach(builder::addNamedValue);
+
+			return onBuild(builder).build();
+		}
+
+		throw new UnsupportedOperationException(String.format("ParameterHandling %s not supported", parameterHandling));
+	}
+
+	private SimpleStatementBuilder onBuild(SimpleStatementBuilder statementBuilder) {
+
+		onBuild.forEach(it -> it.accept(statementBuilder));
+
+		return statementBuilder;
+	}
+
+	/**
+	 * Binding function. This function gets called with the current statement and {@link TermFactory}.
+	 *
+	 * @param <S>
+	 */
+	@FunctionalInterface
+	public interface BindFunction<S> {
+
+		/**
+		 * Apply a binding operation on the {@link BuildableQuery statement} and return the modified statement instance.
+		 *
+		 * @param statement the initial statement instance.
+		 * @param factory factory to create {@link com.datastax.oss.driver.api.querybuilder.term.Term} objects.
+		 * @return the modified statement instance.
+		 */
+		S bind(S statement, TermFactory factory);
+	}
+
+	@FunctionalInterface
+	interface BuilderRunnable<S> {
+		S run(S source, TermFactory termFactory);
+	}
+
+	/**
+	 * Enumeration to represent how parameters are rendered.
+	 */
+	public enum ParameterHandling {
+
+		/**
+		 * CQL inline rendering as literals.
+		 */
+		INLINE,
+
+		/**
+		 * Index-based bind markers.
+		 */
+		BY_INDEX,
+
+		/**
+		 * Named bind markers.
+		 */
+		BY_NAME;
+	}
+}

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2019 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.cql.util;
+
+import org.springframework.lang.Nullable;
+
+import com.datastax.oss.driver.api.querybuilder.term.Term;
+
+/**
+ * Factory for {@link Term} objects encapsulating a binding {@code value}. Classes implementing this factory interface
+ * may return inline terms to render values as part of the query string, or bind markers to supply parameters on
+ * statement creation/parameter binding.
+ * <p>
+ * A {@ling TermFactory} is typically used with {@link StatementBuilder}.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@FunctionalInterface
+public interface TermFactory {
+
+	/**
+	 * Create a {@link Term} for the given {@code value}.
+	 *
+	 * @param value the value to bind, can be {@literal null}.
+	 * @return the {@link Term} for the given {@code value}.
+	 */
+	Term create(@Nullable Object value);
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/util/package-info.java

@@ -0,0 +1,7 @@
+/**
+ * Utility classes for basic CQL interaction.
+ */
+@NonNullApi
+package org.springframework.data.cassandra.core.cql.util;
+
+import org.springframework.lang.NonNullApi;

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

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2019 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.cql;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import org.junit.Test;
+
+import org.springframework.util.concurrent.ListenableFuture;
+
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+
+/**
+ * Unit tests for {@link AsyncResultStream}.
+ *
+ * @author Mark Paluch
+ */
+public class AsyncResultStreamUnitTests {
+
+	AsyncResultSet first = mock(AsyncResultSet.class);
+	AsyncResultSet last = mock(AsyncResultSet.class);
+	Row row1 = mock(Row.class);
+	Row row2 = mock(Row.class);
+
+	@Test // DATACASS-656
+	public void shouldIterateFirstPage() {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+
+		List<Row> rows = new ArrayList<>();
+
+		AsyncResultStream.from(first).forEach(rows::add);
+
+		assertThat(rows).containsOnly(row1);
+	}
+
+	@Test // DATACASS-656
+	public void shouldIterateMappedFirstPage() {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+
+		List<String> rows = new ArrayList<>();
+
+		AsyncResultStream.from(first).map((row, rowNum) -> "row-" + rowNum).forEach(rows::add);
+
+		assertThat(rows).containsOnly("row-1");
+	}
+
+	@Test // DATACASS-656
+	public void shouldIterateMappedPages() {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+		when(last.currentPage()).thenReturn(Collections.singletonList(row2));
+		when(first.fetchNextPage()).thenReturn(CompletableFuture.completedFuture(last));
+		when(first.hasMorePages()).thenReturn(true);
+
+		List<String> rows = new ArrayList<>();
+
+		AsyncResultStream.from(first).map((row, rowNum) -> "row-" + rowNum).forEach(rows::add);
+
+		assertThat(rows).containsOnly("row-1", "row-2");
+	}
+
+	@Test // DATACASS-656
+	public void shouldPropagateExceptionOnIterate() {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+
+		CompletableFuture<AsyncResultSet> failed = new CompletableFuture<>();
+		failed.completeExceptionally(new RuntimeException("boo"));
+		when(first.fetchNextPage()).thenReturn(failed);
+		when(first.hasMorePages()).thenReturn(true);
+
+		List<String> rows = new ArrayList<>();
+
+		ListenableFuture<Void> completion = AsyncResultStream.from(first).map((row, rowNum) -> "row-" + rowNum)
+				.forEach(rows::add);
+
+		assertThatThrownBy(completion::get).hasRootCauseInstanceOf(RuntimeException.class);
+	}
+
+	@Test // DATACASS-656
+	public void shouldCollectFirstPage() throws ExecutionException, InterruptedException {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+
+		ListenableFuture<List<Row>> collect = AsyncResultStream.from(first).collect(Collectors.toList());
+
+		assertThat(collect.get()).containsOnly(row1);
+	}
+
+	@Test // DATACASS-656
+	public void shouldCollectMappedPages() throws ExecutionException, InterruptedException {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+		when(last.currentPage()).thenReturn(Collections.singletonList(row2));
+		when(first.fetchNextPage()).thenReturn(CompletableFuture.completedFuture(last));
+		when(first.hasMorePages()).thenReturn(true);
+
+		ListenableFuture<List<String>> rows = AsyncResultStream.from(first).map((row, rowNum) -> "row-" + rowNum)
+				.collect(Collectors.toList());
+
+		assertThat(rows.get()).containsOnly("row-1", "row-2");
+	}
+
+	@Test // DATACASS-656
+	public void shouldPropagateExceptionOnCollect() {
+
+		when(first.currentPage()).thenReturn(Collections.singletonList(row1));
+
+		CompletableFuture<AsyncResultSet> failed = new CompletableFuture<>();
+		failed.completeExceptionally(new RuntimeException("boo"));
+		when(first.fetchNextPage()).thenReturn(failed);
+		when(first.hasMorePages()).thenReturn(true);
+
+		ListenableFuture<List<Row>> collect = AsyncResultStream.from(first).collect(Collectors.toList());
+
+		assertThatThrownBy(collect::get).hasRootCauseInstanceOf(RuntimeException.class);
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/util/StatementBuilderUnitTests.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2019 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.cql.util;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.Test;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.metadata.schema.ClusteringOrder;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+import com.datastax.oss.driver.api.querybuilder.relation.Relation;
+
+/**
+ * Unit tests for {@link StatementBuilder}.
+ *
+ * @author Mark Paluch
+ */
+public class StatementBuilderUnitTests {
+
+	@Test // DATACASS-656
+	public void shouldCreateSimpleStatement() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all()).build();
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person");
+	}
+
+	@Test // DATACASS-656
+	public void shouldApplyBuilderFunction() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.apply(select -> select.orderBy("foo", ClusteringOrder.ASC)).build();
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person ORDER BY foo ASC");
+	}
+
+	@Test // DATACASS-656
+	public void shouldApplyBindFunction() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.bind((select, factory) -> select.where(Relation.column("foo").isEqualTo(factory.create("bar")))).build();
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person WHERE foo='bar'");
+	}
+
+	@Test // DATACASS-656
+	public void shouldBindByIndex() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.bind((select, factory) -> select.where(Relation.column("foo").isEqualTo(factory.create("bar"))))
+				.build(StatementBuilder.ParameterHandling.BY_INDEX);
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person WHERE foo=?");
+		assertThat(statement.getPositionalValues()).containsOnly("bar");
+	}
+
+	@Test // DATACASS-656
+	public void shouldBindByName() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.bind((select, factory) -> select.where(Relation.column("foo").isEqualTo(factory.create("bar"))))
+				.build(StatementBuilder.ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person WHERE foo=:p0");
+		assertThat(statement.getNamedValues()).containsEntry(CqlIdentifier.fromCql("p0"), "bar");
+	}
+
+	@Test // DATACASS-656
+	public void shouldApplyFunctionsInOrder() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.bind((select, factory) -> select.where(Relation.column("foo").isEqualTo(factory.create("bar"))))
+				.apply(select -> select.orderBy("one", ClusteringOrder.ASC))
+				.bind((select, factory) -> select.where(Relation.column("bar").isEqualTo(factory.create("baz"))))
+				.apply(select -> select.orderBy("two", ClusteringOrder.ASC)).build();
+
+		assertThat(statement.getQuery())
+				.isEqualTo("SELECT * FROM person WHERE foo='bar' AND bar='baz' ORDER BY one ASC,two ASC");
+	}
+
+	@Test // DATACASS-656
+	public void shouldNotifyOnBuild() {
+
+		SimpleStatement statement = StatementBuilder.of(QueryBuilder.selectFrom("person").all())
+				.onBuild(statementBuilder -> statementBuilder.addPositionalValue("foo"))
+				.build(StatementBuilder.ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person");
+		assertThat(statement.getPositionalValues()).hasSize(1);
+	}
+}