#189 - Accept StatementFilterFunction in DatabaseClient.

We now accept StatementFilterFunction and ExecuteFunction via DatabaseClient to filter Statement execution. StatementFilterFunctions can be used to pre-process the statement or post-process Result objects.

databaseClient.execute(…)
		.filter((s, next) -> next.execute(s.returnGeneratedValues("my_id")))
		.filter((s, next) -> next.execute(s.fetchSize(25)))

databaseClient.execute(…)
		.filter(s -> s.returnGeneratedValues("my_id"))
		.filter(s -> s.fetchSize(25))

Original pull request: #308.

src/main/asciidoc/new-features.adoc

@@ -5,7 +5,8 @@
 == What's New in Spring Data R2DBC 1.1.0 RELEASE
 
 * Introduction of `R2dbcEntityTemplate` for entity-oriented operations.
-* Support interface projections with `DatabaseClient.as(…)`
+* Support interface projections with `DatabaseClient.as(…)`.
+* <<r2dbc.datbaseclient.filter,Support for `ExecuteFunction` and `StatementFilterFunction` via `DatabaseClient.filter(…)`>>.
 
 [[new-features.1-0-0-RELEASE]]
 == What's New in Spring Data R2DBC 1.0.0 RELEASE

src/main/asciidoc/reference/r2dbc-sql.adoc

@@ -134,7 +134,7 @@ In JDBC, the actual drivers translate `?` bind markers to database-native marker
 
 Spring Data R2DBC lets you use native bind markers or named bind markers with the `:name` syntax.
 
-Named parameter support leverages a `R2dbcDialect` instance  to expand named parameters to native bind markers at the time of query execution, which gives you a certain degree of query portability across various database vendors.
+Named parameter support leverages a `R2dbcDialect` instance to expand named parameters to native bind markers at the time of query execution, which gives you a certain degree of query portability across various database vendors.
 ****
 
 The query-preprocessor unrolls named `Collection` parameters into a series of bind markers to remove the need of dynamic query creation based on the number of arguments.
@@ -159,7 +159,7 @@ tuples.add(new Object[] {"John", 35});
 tuples.add(new Object[] {"Ann",  50});
 
 db.execute("SELECT id, name, state FROM table WHERE (name, age) IN (:tuples)")
-    .bind("tuples", tuples);
+    .bind("tuples", tuples)
 ----
 ====
 
@@ -171,6 +171,38 @@ The following example shows a simpler variant using `IN` predicates:
 [source,java]
 ----
 db.execute("SELECT id, name, state FROM table WHERE age IN (:ages)")
-    .bind("ages", Arrays.asList(35, 50));
+    .bind("ages", Arrays.asList(35, 50))
 ----
 ====
+
+[[r2dbc.datbaseclient.filter]]
+== Statement Filters
+
+You can register a `Statement` filter (`StatementFilterFunction`) through `DatabaseClient` to intercept and modify statements in their execution, as the following example shows:
+
+====
+[source,java]
+----
+db.execute("INSERT INTO table (name, state) VALUES(:name, :state)")
+    .filter((s, next) -> next.execute(s.returnGeneratedValues("id")))
+    .bind("name", …)
+    .bind("state", …)
+----
+====
+
+`DatabaseClient` exposes also simplified `filter(…)` overload accepting `UnaryOperator<Statement>`:
+
+====
+[source,java]
+----
+db.execute("INSERT INTO table (name, state) VALUES(:name, :state)")
+    .filter(s -> s.returnGeneratedValues("id"))
+    .bind("name", …)
+    .bind("state", …)
+
+db.execute("SELECT id, name, state FROM table")
+    .filter(s -> s.fetchSize(25))
+----
+====
+
+`StatementFilterFunction` allow filtering of the executed `Statement` and filtering of `Result` objects.

src/main/java/org/springframework/data/r2dbc/core/DatabaseClient.java

@@ -18,6 +18,7 @@ package org.springframework.data.r2dbc.core;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
+import io.r2dbc.spi.Statement;
 import reactor.core.publisher.Mono;
 
 import java.util.Arrays;
@@ -26,6 +27,7 @@ import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 import org.reactivestreams.Publisher;
 
@@ -37,6 +39,7 @@ import org.springframework.data.r2dbc.query.Criteria;
 import org.springframework.data.r2dbc.query.Update;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
  * A non-blocking, reactive client for performing database calls requests with Reactive Streams back pressure. Provides
@@ -142,6 +145,16 @@ public interface DatabaseClient {
 		 */
 		Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator);
 
+		/**
+		 * Configures a {@link ExecuteFunction} to execute {@link Statement} objects.
+		 *
+		 * @param executeFunction must not be {@literal null}.
+		 * @return {@code this} {@link Builder}.
+		 * @since 1.1
+		 * @see Statement#execute()
+		 */
+		Builder executeFunction(ExecuteFunction executeFunction);
+
 		/**
 		 * Configures a {@link ReactiveDataAccessStrategy}.
 		 *
@@ -186,7 +199,7 @@ public interface DatabaseClient {
 	/**
 	 * Contract for specifying a SQL call along with options leading to the exchange.
 	 */
-	interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec> {
+	interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec>, StatementFilterSpec<GenericExecuteSpec> {
 
 		/**
 		 * Define the target type the result should be mapped to. <br />
@@ -231,7 +244,7 @@ public interface DatabaseClient {
 	/**
 	 * Contract for specifying a SQL call along with options leading to the exchange.
 	 */
-	interface TypedExecuteSpec<T> extends BindSpec<TypedExecuteSpec<T>> {
+	interface TypedExecuteSpec<T> extends BindSpec<TypedExecuteSpec<T>>, StatementFilterSpec<TypedExecuteSpec<T>> {
 
 		/**
 		 * Define the target type the result should be mapped to. <br />
@@ -866,4 +879,31 @@ public interface DatabaseClient {
 		 */
 		S bindNull(String name, Class<?> type);
 	}
+
+	/**
+	 * Contract for applying a {@link StatementFilterFunction}.
+	 *
+	 * @since 1.1
+	 */
+	interface StatementFilterSpec<S extends StatementFilterSpec<S>> {
+
+		/**
+		 * Add the given filter to the end of the filter chain.
+		 *
+		 * @param filter the filter to be added to the chain.
+		 */
+		default S filter(UnaryOperator<Statement> filter) {
+
+			Assert.notNull(filter, "Statement FilterFunction must not be null!");
+
+			return filter((statement, next) -> next.execute(filter.apply(statement)));
+		}
+
+		/**
+		 * Add the given filter to the end of the filter chain.
+		 *
+		 * @param filter the filter to be added to the chain.
+		 */
+		S filter(StatementFilterFunction filter);
+	}
 }

src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java

@@ -78,6 +78,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 	private final R2dbcExceptionTranslator exceptionTranslator;
 
+	private final ExecuteFunction executeFunction;
+
 	private final ReactiveDataAccessStrategy dataAccessStrategy;
 
 	private final boolean namedParameters;
@@ -87,11 +89,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	private final ProjectionFactory projectionFactory;
 
 	DefaultDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy dataAccessStrategy, boolean namedParameters, ProjectionFactory projectionFactory,
-			DefaultDatabaseClientBuilder builder) {
+			ExecuteFunction executeFunction, ReactiveDataAccessStrategy dataAccessStrategy, boolean namedParameters,
+			ProjectionFactory projectionFactory, DefaultDatabaseClientBuilder builder) {
 
 		this.connector = connector;
 		this.exceptionTranslator = exceptionTranslator;
+		this.executeFunction = executeFunction;
 		this.dataAccessStrategy = dataAccessStrategy;
 		this.namedParameters = namedParameters;
 		this.projectionFactory = projectionFactory;
@@ -264,25 +267,26 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	 * Customization hook.
 	 */
 	protected <T> DefaultTypedExecuteSpec<T> createTypedExecuteSpec(Map<Integer, SettableValue> byIndex,
-			Map<String, SettableValue> byName, Supplier<String> sqlSupplier, Class<T> typeToRead) {
-		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead);
+			Map<String, SettableValue> byName, Supplier<String> sqlSupplier, StatementFilterFunction filterFunction,
+			Class<T> typeToRead) {
+		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, filterFunction, typeToRead);
 	}
 
 	/**
 	 * Customization hook.
 	 */
 	protected <T> DefaultTypedExecuteSpec<T> createTypedExecuteSpec(Map<Integer, SettableValue> byIndex,
-			Map<String, SettableValue> byName, Supplier<String> sqlSupplier,
+			Map<String, SettableValue> byName, Supplier<String> sqlSupplier, StatementFilterFunction filterFunction,
 			BiFunction<Row, RowMetadata, T> mappingFunction) {
-		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, mappingFunction);
+		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, filterFunction, mappingFunction);
 	}
 
 	/**
 	 * Customization hook.
 	 */
 	protected ExecuteSpecSupport createGenericExecuteSpec(Map<Integer, SettableValue> byIndex,
-			Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
-		return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier);
+			Map<String, SettableValue> byName, Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
+		return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier, filterFunction);
 	}
 
 	/**
@@ -327,19 +331,22 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		final Map<Integer, SettableValue> byIndex;
 		final Map<String, SettableValue> byName;
 		final Supplier<String> sqlSupplier;
+		final StatementFilterFunction filterFunction;
 
 		ExecuteSpecSupport(Supplier<String> sqlSupplier) {
 
 			this.byIndex = Collections.emptyMap();
 			this.byName = Collections.emptyMap();
 			this.sqlSupplier = sqlSupplier;
+			this.filterFunction = StatementFilterFunctions.empty();
 		}
 
 		ExecuteSpecSupport(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier) {
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
 			this.byIndex = byIndex;
 			this.byName = byName;
 			this.sqlSupplier = sqlSupplier;
+			this.filterFunction = filterFunction;
 		}
 
 		<T> FetchSpec<T> exchange(Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
@@ -404,7 +411,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				return statement;
 			};
 
-			Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, executeFunction);
+			Function<Connection, Flux<Result>> resultFunction = toFunction(sql, filterFunction, executeFunction);
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
@@ -426,7 +433,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				byIndex.put(index, SettableValue.fromOrEmpty(value, value.getClass()));
 			}
 
-			return createInstance(byIndex, this.byName, this.sqlSupplier);
+			return createInstance(byIndex, this.byName, this.sqlSupplier, this.filterFunction);
 		}
 
 		public ExecuteSpecSupport bindNull(int index, Class<?> type) {
@@ -436,7 +443,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
 			byIndex.put(index, SettableValue.empty(type));
 
-			return createInstance(byIndex, this.byName, this.sqlSupplier);
+			return createInstance(byIndex, this.byName, this.sqlSupplier, this.filterFunction);
 		}
 
 		public ExecuteSpecSupport bind(String name, Object value) {
@@ -455,7 +462,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				byName.put(name, SettableValue.fromOrEmpty(value, value.getClass()));
 			}
 
-			return createInstance(this.byIndex, byName, this.sqlSupplier);
+			return createInstance(this.byIndex, byName, this.sqlSupplier, this.filterFunction);
 		}
 
 		public ExecuteSpecSupport bindNull(String name, Class<?> type) {
@@ -466,7 +473,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
 			byName.put(name, SettableValue.empty(type));
 
-			return createInstance(this.byIndex, byName, this.sqlSupplier);
+			return createInstance(this.byIndex, byName, this.sqlSupplier, this.filterFunction);
+		}
+
+		public ExecuteSpecSupport filter(StatementFilterFunction filter) {
+
+			Assert.notNull(filter, "Statement FilterFunction must not be null!");
+
+			return createInstance(this.byIndex, byName, this.sqlSupplier, this.filterFunction.andThen(filter));
 		}
 
 		private void assertNotPreparedOperation() {
@@ -476,8 +490,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier) {
-			return new ExecuteSpecSupport(byIndex, byName, sqlSupplier);
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
+			return new ExecuteSpecSupport(byIndex, byName, sqlSupplier, filterFunction);
 		}
 	}
 
@@ -487,8 +501,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	protected class DefaultGenericExecuteSpec extends ExecuteSpecSupport implements GenericExecuteSpec {
 
 		DefaultGenericExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier) {
-			super(byIndex, byName, sqlSupplier);
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
+			super(byIndex, byName, sqlSupplier, filterFunction);
 		}
 
 		DefaultGenericExecuteSpec(Supplier<String> sqlSupplier) {
@@ -500,7 +514,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.notNull(resultType, "Result type must not be null!");
 
-			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType);
+			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, this.filterFunction, resultType);
 		}
 
 		@Override
@@ -549,10 +563,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return (DefaultGenericExecuteSpec) super.bindNull(name, type);
 		}
 
+		@Override
+		public DefaultGenericExecuteSpec filter(StatementFilterFunction filter) {
+			return (DefaultGenericExecuteSpec) super.filter(filter);
+		}
+
 		@Override
 		protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier) {
-			return createGenericExecuteSpec(byIndex, byName, sqlSupplier);
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
+			return createGenericExecuteSpec(byIndex, byName, sqlSupplier, filterFunction);
 		}
 	}
 
@@ -566,9 +585,9 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		private final BiFunction<Row, RowMetadata, T> mappingFunction;
 
 		DefaultTypedExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier, Class<T> typeToRead) {
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction, Class<T> typeToRead) {
 
-			super(byIndex, byName, sqlSupplier);
+			super(byIndex, byName, sqlSupplier, filterFunction);
 
 			this.typeToRead = typeToRead;
 
@@ -581,9 +600,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		DefaultTypedExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
-				Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
+				Supplier<String> sqlSupplier, StatementFilterFunction filterFunction,
+				BiFunction<Row, RowMetadata, T> mappingFunction) {
 
-			super(byIndex, byName, sqlSupplier);
+			super(byIndex, byName, sqlSupplier, filterFunction);
 
 			this.typeToRead = null;
 			this.mappingFunction = mappingFunction;
@@ -594,7 +614,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.notNull(resultType, "Result type must not be null!");
 
-			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType);
+			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, this.filterFunction, resultType);
 		}
 
 		@Override
@@ -643,10 +663,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return (DefaultTypedExecuteSpec<T>) super.bindNull(name, type);
 		}
 
+		@Override
+		public DefaultTypedExecuteSpec<T> filter(StatementFilterFunction filter) {
+			return (DefaultTypedExecuteSpec<T>) super.filter(filter);
+		}
+
 		@Override
 		protected DefaultTypedExecuteSpec<T> createInstance(Map<Integer, SettableValue> byIndex,
-				Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
-			return createTypedExecuteSpec(byIndex, byName, sqlSupplier, this.typeToRead);
+				Map<String, SettableValue> byName, Supplier<String> sqlSupplier, StatementFilterFunction filterFunction) {
+			return createTypedExecuteSpec(byIndex, byName, sqlSupplier, filterFunction, this.typeToRead);
 		}
 	}
 
@@ -735,7 +760,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			String sql = getRequiredSql(preparedOperation);
 			Function<Connection, Statement> selectFunction = wrapPreparedOperation(sql, preparedOperation);
-			Function<Connection, Flux<Result>> resultFunction = DefaultDatabaseClient.toExecuteFunction(sql, selectFunction);
+			Function<Connection, Flux<Result>> resultFunction = toFunction(sql, StatementFilterFunctions.empty(),
+					selectFunction);
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
@@ -1432,7 +1458,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		String sql = getRequiredSql(operation);
 		Function<Connection, Statement> insertFunction = wrapPreparedOperation(sql, operation)
 				.andThen(statement -> statement.returnGeneratedValues());
-		Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, insertFunction);
+		Function<Connection, Flux<Result>> resultFunction = toFunction(sql, StatementFilterFunctions.empty(),
+				insertFunction);
 
 		return new DefaultSqlResult<>(this, //
 				sql, //
@@ -1445,7 +1472,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		String sql = getRequiredSql(operation);
 		Function<Connection, Statement> executeFunction = wrapPreparedOperation(sql, operation);
-		Function<Connection, Flux<Result>> resultFunction = toExecuteFunction(sql, executeFunction);
+		Function<Connection, Flux<Result>> resultFunction = toFunction(sql, StatementFilterFunctions.empty(),
+				executeFunction);
 
 		return new DefaultSqlResult<>(this, //
 				sql, //
@@ -1476,12 +1504,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		};
 	}
 
-	private static Function<Connection, Flux<Result>> toExecuteFunction(String sql,
-			Function<Connection, Statement> executeFunction) {
+	private Function<Connection, Flux<Result>> toFunction(String sql, StatementFilterFunction filterFunction,
+			Function<Connection, Statement> statementFactory) {
 
 		return it -> {
 
-			Flux<Result> from = Flux.defer(() -> executeFunction.apply(it).execute()).cast(Result.class);
+			Flux<Result> from = Flux.defer(() -> {
+				Statement statement = statementFactory.apply(it);
+				return filterFunction.filter(statement, executeFunction);
+			}).cast(Result.class);
 			return from.checkpoint("SQL \"" + sql + "\" [DatabaseClient]");
 		};
 	}

src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientBuilder.java

@@ -17,6 +17,7 @@
 package org.springframework.data.r2dbc.core;
 
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Statement;
 
 import java.util.function.Consumer;
 
@@ -40,6 +41,8 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 	private @Nullable R2dbcExceptionTranslator exceptionTranslator;
 
+	private ExecuteFunction executeFunction = Statement::execute;
+
 	private ReactiveDataAccessStrategy accessStrategy;
 
 	private boolean namedParameters = true;
@@ -54,6 +57,7 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 		this.connectionFactory = other.connectionFactory;
 		this.exceptionTranslator = other.exceptionTranslator;
+		this.executeFunction = other.executeFunction;
 		this.accessStrategy = other.accessStrategy;
 		this.namedParameters = other.namedParameters;
 		this.projectionFactory = other.projectionFactory;
@@ -85,6 +89,19 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 		return this;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#executeFunction(org.springframework.data.r2dbc.core.ExecuteFunction)
+	 */
+	@Override
+	public Builder executeFunction(ExecuteFunction executeFunction) {
+
+		Assert.notNull(executeFunction, "ExecuteFunction must not be null!");
+
+		this.executeFunction = executeFunction;
+		return this;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#dataAccessStrategy(org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy)
@@ -143,8 +160,8 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 			accessStrategy = new DefaultReactiveDataAccessStrategy(dialect);
 		}
 
-		return new DefaultDatabaseClient(this.connectionFactory, exceptionTranslator, accessStrategy, namedParameters,
-				projectionFactory, new DefaultDatabaseClientBuilder(this));
+		return new DefaultDatabaseClient(this.connectionFactory, exceptionTranslator, executeFunction, accessStrategy,
+				namedParameters, projectionFactory, new DefaultDatabaseClientBuilder(this));
 	}
 
 	/*

src/main/java/org/springframework/data/r2dbc/core/ExecuteFunction.java

@@ -0,0 +1,46 @@
+/*
+ * 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.r2dbc.core;
+
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Statement;
+
+import java.util.function.BiFunction;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * Represents a function that executes a {@link io.r2dbc.spi.Statement} for a (delayed) {@link io.r2dbc.spi.Result}
+ * stream.
+ * <p>
+ * Note that discarded {@link Result} objects must be consumed according to the R2DBC spec via either
+ * {@link Result#getRowsUpdated()} or {@link Result#map(BiFunction)}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ * @see Statement#execute()
+ */
+@FunctionalInterface
+public interface ExecuteFunction {
+
+	/**
+	 * Execute the given {@link Statement} for a stream of {@link Result}s.
+	 *
+	 * @param statement the request to execute.
+	 * @return the delayed result stream.
+	 */
+	Publisher<? extends Result> execute(Statement statement);
+}

src/main/java/org/springframework/data/r2dbc/core/StatementFilterFunction.java

@@ -0,0 +1,65 @@
+/*
+ * 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.r2dbc.core;
+
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Statement;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.util.Assert;
+
+/**
+ * Represents a function that filters an {@link ExecuteFunction execute function}.
+ * <p>
+ * The filter is executed when a {@link org.reactivestreams.Subscriber} subscribes to the {@link Publisher} returned by
+ * the {@link DatabaseClient}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ * @see ExecuteFunction
+ */
+@FunctionalInterface
+public interface StatementFilterFunction {
+
+	/**
+	 * Apply this filter to the given {@link Statement} and {@link ExecuteFunction}.
+	 * <p>
+	 * The given {@link ExecuteFunction} represents the next entity in the chain, to be invoked via
+	 * {@link ExecuteFunction#execute(Statement)} invoked} in order to proceed with the exchange, or not invoked to
+	 * shortcut the chain.
+	 *
+	 * @param statement the current {@link Statement}.
+	 * @param next the next exchange function in the chain.
+	 * @return the filtered {@link Result}s.
+	 */
+	Publisher<? extends Result> filter(Statement statement, ExecuteFunction next);
+
+	/**
+	 * Return a composed filter function that first applies this filter, and then applies the given {@code "after"}
+	 * filter.
+	 *
+	 * @param afterFilter the filter to apply after this filter.
+	 * @return the composed filter.
+	 */
+	default StatementFilterFunction andThen(StatementFilterFunction afterFilter) {
+
+		Assert.notNull(afterFilter, "StatementFilterFunction must not be null");
+
+		return (request, next) -> filter(request, afterRequest -> afterFilter.filter(afterRequest, next));
+	}
+
+}

src/main/java/org/springframework/data/r2dbc/core/StatementFilterFunctions.java

@@ -0,0 +1,46 @@
+/*
+ * 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.r2dbc.core;
+
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Statement;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * Collection of default {@link StatementFilterFunction}s.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+enum StatementFilterFunctions implements StatementFilterFunction {
+
+	EMPTY_FILTER;
+
+	@Override
+	public Publisher<? extends Result> filter(Statement statement, ExecuteFunction next) {
+		return next.execute(statement);
+	}
+
+	/**
+	 * Return an empty {@link StatementFilterFunction} that delegates to {@link ExecuteFunction}.
+	 *
+	 * @return an empty {@link StatementFilterFunction} that delegates to {@link ExecuteFunction}.
+	 */
+	public static StatementFilterFunction empty() {
+		return EMPTY_FILTER;
+	}
+}

src/test/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientUnitTests.java

@@ -37,6 +37,7 @@ import java.util.Arrays;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.reactivestreams.Publisher;
@@ -468,7 +469,120 @@ public class DefaultDatabaseClientUnitTests {
 
 				}) //
 				.verifyComplete();
+	}
 
+	@Test // gh-189
+	public void shouldApplyExecuteFunction() {
+
+		Statement statement = mock(Statement.class);
+		when(connection.createStatement(anyString())).thenReturn(statement);
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
+		MockResult result = MockResult.builder().rowMetadata(metadata)
+				.row(MockRow.builder().identified(0, Object.class, "Walter").build()).build();
+
+		DatabaseClient databaseClient = DatabaseClient.builder() //
+				.connectionFactory(connectionFactory) //
+				.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)) //
+				.executeFunction(it -> Mono.just(result)).build();
+
+		databaseClient.execute("SELECT") //
+				.fetch().all() //
+				.as(StepVerifier::create) //
+				.expectNextCount(1).verifyComplete();
+
+		verify(statement, never()).execute();
+	}
+
+	@Test // gh-189
+	public void shouldApplyStatementFilterFunctions() {
+
+		Statement statement = mock(Statement.class);
+		when(connection.createStatement(anyString())).thenReturn(statement);
+		when(statement.returnGeneratedValues(anyString())).thenReturn(statement);
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
+		MockResult result = MockResult.builder().rowMetadata(metadata).build();
+
+		doReturn(Flux.just(result)).when(statement).execute();
+
+		DatabaseClient databaseClient = DatabaseClient.builder() //
+				.connectionFactory(connectionFactory) //
+				.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)) //
+				.build();
+
+		databaseClient.execute("SELECT") //
+				.filter((s, next) -> next.execute(s.returnGeneratedValues("foo"))) //
+				.filter((s, next) -> next.execute(s.returnGeneratedValues("bar"))) //
+				.fetch().all() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		InOrder inOrder = inOrder(statement);
+		inOrder.verify(statement).returnGeneratedValues("foo");
+		inOrder.verify(statement).returnGeneratedValues("bar");
+		inOrder.verify(statement).execute();
+	}
+
+	@Test // gh-189
+	public void shouldApplyStatementFilterFunctionsToTypedExecute() {
+
+		Statement statement = mock(Statement.class);
+		when(connection.createStatement(anyString())).thenReturn(statement);
+		when(statement.returnGeneratedValues(anyString())).thenReturn(statement);
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
+		MockResult result = MockResult.builder().rowMetadata(metadata).build();
+
+		doReturn(Flux.just(result)).when(statement).execute();
+
+		DatabaseClient databaseClient = DatabaseClient.builder() //
+				.connectionFactory(connectionFactory) //
+				.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)) //
+				.build();
+
+		databaseClient.execute("SELECT") //
+				.filter((s, next) -> next.execute(s.returnGeneratedValues("foo"))) //
+				.as(Person.class) //
+				.fetch().all() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(statement).returnGeneratedValues("foo");
+	}
+
+	@Test // gh-189
+	public void shouldApplySimpleStatementFilterFunctions() {
+
+		Statement statement = mock(Statement.class);
+		when(connection.createStatement(anyString())).thenReturn(statement);
+		when(statement.returnGeneratedValues(anyString())).thenReturn(statement);
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
+		MockResult result = MockResult.builder().rowMetadata(metadata).build();
+
+		doReturn(Flux.just(result)).when(statement).execute();
+
+		DatabaseClient databaseClient = DatabaseClient.builder() //
+				.connectionFactory(connectionFactory) //
+				.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)) //
+				.build();
+
+		databaseClient.execute("SELECT") //
+				.filter(s -> s.returnGeneratedValues("foo")) //
+				.filter(s -> s.returnGeneratedValues("bar")) //
+				.fetch().all() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		InOrder inOrder = inOrder(statement);
+		inOrder.verify(statement).returnGeneratedValues("foo");
+		inOrder.verify(statement).returnGeneratedValues("bar");
+		inOrder.verify(statement).execute();
 	}
 
 	static class Person {