#105 - Move named parameter resolution to ReactiveDataAccessStrategy.

Named parameter resolution is now provided as part of ReactiveDataAccessStrategy. This allows us to hide implementation internals (bind markers). DatabaseClient allows configuration whether to use named parameter expansion.

Detailed configuration of named parameter support is now moved to DefaultReactiveDataAccessStrategy.

Original pull request: #105.

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.r2dbc.core;
 
+import org.springframework.data.util.Streamable;
 import org.springframework.lang.Nullable;
 
 /**
@@ -58,4 +59,11 @@ public interface BindParameterSource {
 	default Class<?> getType(String paramName) {
 		return Object.class;
 	}
+
+	/**
+	 * Returns parameter names of the underlying parameter source.
+	 *
+	 * @return parameter names of the underlying parameter source.
+	 */
+	Streamable<String> getParameterNames();
 }

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

@@ -1,73 +0,0 @@
-/*
- * 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.r2dbc.core;
-
-import io.r2dbc.spi.Statement;
-
-import org.springframework.data.r2dbc.dialect.BindTarget;
-import org.springframework.data.r2dbc.mapping.SettableValue;
-
-/**
- * Extension to {@link QueryOperation} for operations that allow parameter substitution by binding parameter values.
- * {@link BindableOperation} is typically created with a {@link Set} of column names or parameter names that accept bind
- * parameters by calling {@link #bind(Statement, String, Object)}.
- *
- * @author Mark Paluch
- * @see Statement#bind
- * @see Statement#bindNull TODO: Refactor to {@link PreparedOperation}.
- */
-public interface BindableOperation extends QueryOperation {
-
-	/**
-	 * Bind the given {@code value} to the {@link Statement} using the underlying binding strategy.
-	 *
-	 * @param bindTarget the bindTarget to bind the value to.
-	 * @param identifier named identifier that is considered by the underlying binding strategy.
-	 * @param value the actual value. Must not be {@literal null}. Use {@link #bindNull(Statement, Class)} for
-	 *          {@literal null} values.
-	 * @see Statement#bind
-	 */
-	void bind(BindTarget bindTarget, String identifier, Object value);
-
-	/**
-	 * Bind a {@literal null} value to the {@link Statement} using the underlying binding strategy.
-	 *
-	 * @param bindTarget the bindTarget to bind the value to.
-	 * @param identifier named identifier that is considered by the underlying binding strategy.
-	 * @param valueType value type, must not be {@literal null}.
-	 * @see Statement#bindNull
-	 */
-	void bindNull(BindTarget bindTarget, String identifier, Class<?> valueType);
-
-	/**
-	 * Bind a {@link SettableValue} to the {@link Statement} using the underlying binding strategy. Binds either the
-	 * {@link SettableValue#getValue()} or {@literal null}, depending on whether the value is {@literal null}.
-	 *
-	 * @param bindTarget the bindTarget to bind the value to.
-	 * @param value the settable value
-	 * @see Statement#bind
-	 * @see Statement#bindNull
-	 */
-	default void bind(BindTarget bindTarget, String identifier, SettableValue value) {
-
-		if (value.getValue() == null) {
-			bindNull(bindTarget, identifier, value.getType());
-		} else {
-			bind(bindTarget, identifier, value.getValue());
-		}
-	}
-
-}

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

@@ -157,14 +157,14 @@ public interface DatabaseClient {
 		Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy);
 
 		/**
-		 * Configures {@link NamedParameterExpander}.
+		 * Configures whether to use named parameter expansion. Defaults to {@literal true}.
 		 *
-		 * @param expander must not be {@literal null}.
+		 * @param enabled {@literal true} to use named parameter expansion. {@literal false} to disable named parameter
+		 *          expansion.
 		 * @return {@code this} {@link Builder}.
-		 * @see NamedParameterExpander#enabled()
-		 * @see NamedParameterExpander#disabled()
+		 * @see NamedParameterExpander
 		 */
-		Builder namedParameters(NamedParameterExpander expander);
+		Builder namedParameters(boolean enabled);
 
 		/**
 		 * Configures a {@link Consumer} to configure this builder.

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

@@ -79,13 +79,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 	private final ReactiveDataAccessStrategy dataAccessStrategy;
 
-	private final NamedParameterExpander namedParameters;
+	private final boolean namedParameters;
 
 	private final DefaultDatabaseClientBuilder builder;
 
 	DefaultDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy dataAccessStrategy, NamedParameterExpander namedParameters,
-			DefaultDatabaseClientBuilder builder) {
+			ReactiveDataAccessStrategy dataAccessStrategy, boolean namedParameters, DefaultDatabaseClientBuilder builder) {
 
 		this.connector = connector;
 		this.exceptionTranslator = exceptionTranslator;
@@ -284,6 +283,18 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		return new DefaultGenericExecuteSpec(sqlSupplier);
 	}
 
+	private static void bindByName(Statement statement, Map<String, SettableValue> byName) {
+
+		byName.forEach((name, o) -> {
+
+			if (o.getValue() != null) {
+				statement.bind(name, o.getValue());
+			} else {
+				statement.bindNull(name, o.getType());
+			}
+		});
+	}
+
 	private static void bindByIndex(Statement statement, Map<Integer, SettableValue> byIndex) {
 
 		byIndex.forEach((i, o) -> {
@@ -353,27 +364,28 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					return statement;
 				}
 
-				BindableOperation operation = namedParameters.expand(sql, dataAccessStrategy.getBindMarkersFactory(),
-						new MapBindParameterSource(this.byName));
+				if (namedParameters) {
 
-				String expanded = getRequiredSql(operation);
-				if (logger.isTraceEnabled()) {
-					logger.trace("Expanded SQL [" + expanded + "]");
+					PreparedOperation<?> operation = dataAccessStrategy.processNamedParameters(sql, this.byName);
+
+					String expanded = getRequiredSql(operation);
+					if (logger.isTraceEnabled()) {
+						logger.trace("Expanded SQL [" + expanded + "]");
+					}
+
+					Statement statement = it.createStatement(expanded);
+					BindTarget bindTarget = new StatementWrapper(statement);
+
+					operation.bindTo(bindTarget);
+					bindByIndex(statement, this.byIndex);
+
+					return statement;
 				}
 
-				Statement statement = it.createStatement(expanded);
-				BindTarget bindTarget = new StatementWrapper(statement);
-
-				this.byName.forEach((name, o) -> {
-
-					if (o.getValue() != null) {
-						operation.bind(bindTarget, name, o.getValue());
-					} else {
-						operation.bindNull(bindTarget, name, o.getType());
-					}
-				});
+				Statement statement = it.createStatement(sql);
 
 				bindByIndex(statement, this.byIndex);
+				bindByName(statement, this.byName);
 
 				return statement;
 			};

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

@@ -36,9 +36,12 @@ import org.springframework.util.Assert;
 class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 	private @Nullable ConnectionFactory connectionFactory;
+
 	private @Nullable R2dbcExceptionTranslator exceptionTranslator;
+
 	private ReactiveDataAccessStrategy accessStrategy;
-	private NamedParameterExpander namedParameters;
+
+	private boolean namedParameters = true;
 
 	DefaultDatabaseClientBuilder() {}
 
@@ -93,14 +96,12 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#namedParameters(org.springframework.data.r2dbc.function.NamedParameterExpander)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#namedParameters(boolean)
 	 */
 	@Override
-	public Builder namedParameters(NamedParameterExpander expander) {
+	public Builder namedParameters(boolean enabled) {
 
-		Assert.notNull(expander, "NamedParameterExpander must not be null!");
-
-		this.namedParameters = expander;
+		this.namedParameters = enabled;
 		return this;
 	}
 
@@ -125,19 +126,12 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 			accessStrategy = new DefaultReactiveDataAccessStrategy(dialect);
 		}
 
-		NamedParameterExpander namedParameters = this.namedParameters;
-
-		if (namedParameters == null) {
-			namedParameters = NamedParameterExpander.enabled();
-		}
-
 		return doBuild(this.connectionFactory, exceptionTranslator, accessStrategy, namedParameters,
 				new DefaultDatabaseClientBuilder(this));
 	}
 
 	protected DatabaseClient doBuild(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy accessStrategy, NamedParameterExpander namedParameters,
-			DefaultDatabaseClientBuilder builder) {
+			ReactiveDataAccessStrategy accessStrategy, boolean namedParameters, DefaultDatabaseClientBuilder builder) {
 		return new DefaultDatabaseClient(connector, exceptionTranslator, accessStrategy, namedParameters, builder);
 	}
 

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

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiFunction;
 
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
@@ -31,7 +32,6 @@ import org.springframework.data.r2dbc.convert.EntityRowMapper;
 import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
-import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.SettableValue;
@@ -57,6 +57,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	private final UpdateMapper updateMapper;
 	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 	private final StatementMapper statementMapper;
+	private final NamedParameterExpander expander;
 
 	/**
 	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link R2dbcDialect} and optional
@@ -81,7 +82,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		this(dialect, createConverter(dialect, converters));
 	}
 
-	private static R2dbcConverter createConverter(R2dbcDialect dialect, Collection<?> converters) {
+	public static R2dbcConverter createConverter(R2dbcDialect dialect, Collection<?> converters) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(converters, "Converters must not be null");
@@ -101,11 +102,24 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * @param dialect the {@link R2dbcDialect} to use.
 	 * @param converter must not be {@literal null}.
 	 */
-	@SuppressWarnings("unchecked")
 	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect, R2dbcConverter converter) {
+		this(dialect, converter, new NamedParameterExpander());
+	}
+
+	/**
+	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link R2dbcDialect} and {@link R2dbcConverter}.
+	 *
+	 * @param dialect the {@link R2dbcDialect} to use.
+	 * @param converter must not be {@literal null}.
+	 * @param expander must not be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect, R2dbcConverter converter,
+			NamedParameterExpander expander) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(converter, "RelationalConverter must not be null");
+		Assert.notNull(expander, "NamedParameterExpander must not be null");
 
 		this.converter = converter;
 		this.updateMapper = new UpdateMapper(converter);
@@ -116,6 +130,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		RenderContextFactory factory = new RenderContextFactory(dialect);
 		this.statementMapper = new DefaultStatementMapper(dialect, factory.createRenderContext(), this.updateMapper,
 				this.mappingContext);
+		this.expander = expander;
 	}
 
 	/*
@@ -213,6 +228,15 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return new EntityRowMapper<>(typeToRead, this.converter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy#processNamedParameters(java.lang.String, java.util.Map)
+	 */
+	@Override
+	public PreparedOperation<?> processNamedParameters(String query, Map<String, SettableValue> bindings) {
+		return this.expander.expand(query, this.dialect.getBindMarkersFactory(), new MapBindParameterSource(bindings));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getTableName(java.lang.Class)
@@ -231,15 +255,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return this.statementMapper;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getBindMarkersFactory()
-	 */
-	@Override
-	public BindMarkersFactory getBindMarkersFactory() {
-		return this.dialect.getBindMarkersFactory();
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getConverter()

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

@@ -39,7 +39,7 @@ import org.springframework.transaction.NoTransactionException;
 class DefaultTransactionalDatabaseClient extends DefaultDatabaseClient implements TransactionalDatabaseClient {
 
 	DefaultTransactionalDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy dataAccessStrategy, NamedParameterExpander namedParameters,
+			ReactiveDataAccessStrategy dataAccessStrategy, boolean namedParameters,
 			DefaultDatabaseClientBuilder builder) {
 		super(connector, exceptionTranslator, dataAccessStrategy, namedParameters, builder);
 	}

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

@@ -70,11 +70,11 @@ class DefaultTransactionalDatabaseClientBuilder extends DefaultDatabaseClientBui
 	}
 
 	/* (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder#dataAccessStrategy(org.springframework.data.r2dbc.function.NamedParameterSupport)
+	 * @see org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder#dataAccessStrategy(boolean)
 	 */
 	@Override
-	public TransactionalDatabaseClient.Builder namedParameters(NamedParameterExpander expander) {
-		super.namedParameters(expander);
+	public TransactionalDatabaseClient.Builder namedParameters(boolean enabled) {
+		super.namedParameters(enabled);
 		return this;
 	}
 
@@ -97,7 +97,7 @@ class DefaultTransactionalDatabaseClientBuilder extends DefaultDatabaseClientBui
 
 	@Override
 	protected DatabaseClient doBuild(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy accessStrategy, NamedParameterExpander namedParameters,
+			ReactiveDataAccessStrategy accessStrategy, boolean namedParameters,
 			DefaultDatabaseClientBuilder builder) {
 		return new DefaultTransactionalDatabaseClient(connector, exceptionTranslator, accessStrategy, namedParameters,
 				builder);

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

@@ -19,6 +19,7 @@ import java.util.LinkedHashMap;
 import java.util.Map;
 
 import org.springframework.data.r2dbc.mapping.SettableValue;
+import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
 
 /**
@@ -111,4 +112,13 @@ class MapBindParameterSource implements BindParameterSource {
 
 		return this.values.get(paramName).getValue();
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.SqlParameterSource#getParameterNames()
+	 */
+	@Override
+	public Streamable<String> getParameterNames() {
+		return Streamable.of(this.values.keySet());
+	}
 }

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

@@ -22,7 +22,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
-import org.springframework.data.r2dbc.dialect.BindTarget;
 
 /**
  * SQL translation support allowing the use of named parameters rather than native placeholders.
@@ -56,25 +55,12 @@ public class NamedParameterExpander {
 		}
 	};
 
-	private NamedParameterExpander() {}
-
 	/**
-	 * Creates a disabled instance of {@link NamedParameterExpander}.
-	 *
-	 * @return a disabled instance of {@link NamedParameterExpander}.
-	 */
-	public static NamedParameterExpander disabled() {
-		return Disabled.INSTANCE;
-	}
-
-	/**
-	 * Creates a new enabled instance of {@link NamedParameterExpander}.
+	 * Create a new enabled instance of {@link NamedParameterExpander}.
 	 *
 	 * @return a new enabled instance of {@link NamedParameterExpander}.
 	 */
-	public static NamedParameterExpander enabled() {
-		return new NamedParameterExpander();
-	}
+	public NamedParameterExpander() {}
 
 	/**
 	 * Specify the maximum number of entries for the SQL cache. Default is 256.
@@ -98,7 +84,7 @@ public class NamedParameterExpander {
 	 * @param sql the original SQL statement
 	 * @return a representation of the parsed SQL statement
 	 */
-	protected ParsedSql getParsedSql(String sql) {
+	private ParsedSql getParsedSql(String sql) {
 
 		if (getCacheLimit() <= 0) {
 			return NamedParameterUtils.parseSqlStatement(sql);
@@ -116,51 +102,36 @@ public class NamedParameterExpander {
 		}
 	}
 
-	BindableOperation expand(String sql, BindMarkersFactory bindMarkersFactory, BindParameterSource paramSource) {
+	/**
+	 * Parse the SQL statement and locate any placeholders or named parameters. Named parameters are substituted for a
+	 * native placeholder, and any select list is expanded to the required number of placeholders. Select lists may
+	 * contain an array of objects, and in that case the placeholders will be grouped and enclosed with parentheses. This
+	 * allows for the use of "expression lists" in the SQL statement like: <br />
+	 * <br />
+	 * {@code select id, name, state from table where (name, age) in (('John', 35), ('Ann', 50))}
+	 * <p>
+	 * The parameter values passed in are used to determine the number of placeholders to be used for a select list.
+	 * Select lists should be limited to 100 or fewer elements. A larger number of elements is not guaranteed to be
+	 * supported by the database and is strictly vendor-dependent.
+	 *
+	 * @param sql sql the original SQL statement
+	 * @param bindMarkersFactory the bind marker factory.
+	 * @param paramSource the source for named parameters.
+	 * @return the expanded sql that accepts bind parameters and allows for execution without further translation wrapped
+	 *         as {@link PreparedOperation}.
+	 */
+	public PreparedOperation<String> expand(String sql, BindMarkersFactory bindMarkersFactory,
+			BindParameterSource paramSource) {
 
 		ParsedSql parsedSql = getParsedSql(sql);
 
-		BindableOperation expanded = NamedParameterUtils.substituteNamedParameters(parsedSql, bindMarkersFactory,
+		PreparedOperation<String> expanded = NamedParameterUtils.substituteNamedParameters(parsedSql, bindMarkersFactory,
 				paramSource);
 
-		if (logger.isDebugEnabled()) {
-			logger.debug(String.format("Expanding SQL statement [%s] to [%s]", sql, expanded.toQuery()));
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug(String.format("Expanding SQL statement [%s] to [%s]", sql, expanded.toQuery()));
 		}
 
 		return expanded;
 	}
-
-	/**
-	 * Disabled named parameter support.
-	 */
-	static class Disabled extends NamedParameterExpander {
-
-		private static final Disabled INSTANCE = new Disabled();
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.NamedParameterSupport#expand(java.lang.String, org.springframework.data.r2dbc.dialect.BindMarkersFactory, org.springframework.data.r2dbc.function.SqlParameterSource)
-		 */
-		@Override
-		BindableOperation expand(String sql, BindMarkersFactory bindMarkersFactory, BindParameterSource paramSource) {
-
-			return new BindableOperation() {
-
-				@Override
-				public void bind(BindTarget target, String identifier, Object value) {
-					target.bind(identifier, value);
-				}
-
-				@Override
-				public void bindNull(BindTarget target, String identifier, Class<?> valueType) {
-					target.bindNull(identifier, valueType);
-				}
-
-				@Override
-				public String toQuery() {
-					return sql;
-				}
-			};
-		}
-	}
 }

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

@@ -258,15 +258,15 @@ abstract class NamedParameterUtils {
 	 * @return the expanded query that accepts bind parameters and allows for execution without further translation.
 	 * @see #parseSqlStatement
 	 */
-	public static BindableOperation substituteNamedParameters(ParsedSql parsedSql, BindMarkersFactory bindMarkersFactory,
-			BindParameterSource paramSource) {
+	public static PreparedOperation<String> substituteNamedParameters(ParsedSql parsedSql,
+			BindMarkersFactory bindMarkersFactory, BindParameterSource paramSource) {
 
 		BindMarkerHolder markerHolder = new BindMarkerHolder(bindMarkersFactory.create());
 
 		String originalSql = parsedSql.getOriginalSql();
 		List<String> paramNames = parsedSql.getParameterNames();
 		if (paramNames.isEmpty()) {
-			return new ExpandedQuery(originalSql, markerHolder);
+			return new ExpandedQuery(originalSql, markerHolder, paramSource);
 		}
 
 		StringBuilder actualSql = new StringBuilder(originalSql.length());
@@ -313,7 +313,7 @@ abstract class NamedParameterUtils {
 		}
 		actualSql.append(originalSql, lastIndex, originalSql.length());
 
-		return new ExpandedQuery(actualSql.toString(), markerHolder);
+		return new ExpandedQuery(actualSql.toString(), markerHolder, paramSource);
 	}
 
 	/**
@@ -338,7 +338,7 @@ abstract class NamedParameterUtils {
 	 * @param paramSource the source for named parameters.
 	 * @return the expanded query that accepts bind parameters and allows for execution without further translation.
 	 */
-	public static BindableOperation substituteNamedParameters(String sql, BindMarkersFactory bindMarkersFactory,
+	public static PreparedOperation<String> substituteNamedParameters(String sql, BindMarkersFactory bindMarkersFactory,
 			BindParameterSource paramSource) {
 		ParsedSql parsedSql = parseSqlStatement(sql);
 		return substituteNamedParameters(parsedSql, bindMarkersFactory, paramSource);
@@ -368,8 +368,8 @@ abstract class NamedParameterUtils {
 
 		String addMarker(String name) {
 
-			BindMarker bindMarker = bindMarkers.next(name);
-			markers.computeIfAbsent(name, ignore -> new ArrayList<>()).add(bindMarker);
+			BindMarker bindMarker = this.bindMarkers.next(name);
+			this.markers.computeIfAbsent(name, ignore -> new ArrayList<>()).add(bindMarker);
 			return bindMarker.getPlaceholder();
 		}
 	}
@@ -378,22 +378,20 @@ abstract class NamedParameterUtils {
 	 * Expanded query that allows binding of parameters using parameter names that were used to expand the query. Binding
 	 * unrolls {@link Collection}s and nested arrays.
 	 */
-	private static class ExpandedQuery implements BindableOperation {
+	private static class ExpandedQuery implements PreparedOperation<String> {
 
 		private final String expandedSql;
 
 		private final Map<String, List<BindMarker>> markers;
 
-		ExpandedQuery(String expandedSql, BindMarkerHolder bindMarkerHolder) {
+		private final BindParameterSource parameterSource;
+
+		ExpandedQuery(String expandedSql, BindMarkerHolder bindMarkerHolder, BindParameterSource parameterSource) {
 			this.expandedSql = expandedSql;
 			this.markers = bindMarkerHolder.markers;
+			this.parameterSource = parameterSource;
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(BindTarget, java.lang.String, java.lang.Object)
-		 */
-		@Override
 		@SuppressWarnings("unchecked")
 		public void bind(BindTarget target, String identifier, Object value) {
 
@@ -443,11 +441,6 @@ abstract class NamedParameterUtils {
 			markers.next().bind(target, valueToBind);
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(BindTarget, java.lang.String, java.lang.Class)
-		 */
-		@Override
 		public void bindNull(BindTarget target, String identifier, Class<?> valueType) {
 
 			List<BindMarker> bindMarkers = getBindMarkers(identifier);
@@ -467,7 +460,27 @@ abstract class NamedParameterUtils {
 		}
 
 		private List<BindMarker> getBindMarkers(String identifier) {
-			return markers.get(identifier);
+			return this.markers.get(identifier);
+		}
+
+		@Override
+		public String getSource() {
+			return this.expandedSql;
+		}
+
+		@Override
+		public void bindTo(BindTarget target) {
+
+			for (String namedParameter : this.parameterSource.getParameterNames()) {
+
+				Object value = this.parameterSource.getValue(namedParameter);
+
+				if (value == null) {
+					bindNull(target, namedParameter, this.parameterSource.getType(namedParameter));
+				} else {
+					bind(target, namedParameter, value);
+				}
+			}
 		}
 
 		/*
@@ -476,7 +489,7 @@ abstract class NamedParameterUtils {
 		 */
 		@Override
 		public String toQuery() {
-			return expandedSql;
+			return this.expandedSql;
 		}
 	}
 }

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

@@ -19,10 +19,10 @@ import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiFunction;
 
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
-import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 
@@ -32,7 +32,7 @@ import org.springframework.data.r2dbc.mapping.SettableValue;
  * primary keys.
  *
  * @author Mark Paluch
- * @see BindableOperation
+ * @see PreparedOperation
  */
 public interface ReactiveDataAccessStrategy {
 
@@ -65,6 +65,15 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	String getTableName(Class<?> type);
 
+	/**
+	 * Expand named parameters and return a {@link PreparedOperations} wrapping named bindings.
+	 * 
+	 * @param query the query to expand.
+	 * @param bindings named parameter bindings.
+	 * @return the {@link PreparedOperation} encapsulating expanded SQL and bindings.
+	 */
+	PreparedOperation<?> processNamedParameters(String query, Map<String, SettableValue> bindings);
+
 	/**
 	 * Returns the {@link org.springframework.data.r2dbc.dialect.R2dbcDialect}-specific {@link StatementMapper}.
 	 *
@@ -72,13 +81,6 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	StatementMapper getStatementMapper();
 
-	/**
-	 * Returns the configured {@link BindMarkersFactory} to create native parameter placeholder markers.
-	 *
-	 * @return the configured {@link BindMarkersFactory}.
-	 */
-	BindMarkersFactory getBindMarkersFactory();
-
 	/**
 	 * Returns the {@link R2dbcConverter}.
 	 *

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

@@ -194,14 +194,14 @@ public interface TransactionalDatabaseClient extends DatabaseClient {
 		Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy);
 
 		/**
-		 * Configures {@link NamedParameterExpander}.
+		 * Configures whether to use named parameter expansion. Defaults to {@literal true}.
 		 *
-		 * @param expander must not be {@literal null}.
-		 * @return {@code this} {@link Builder}.
-		 * @see NamedParameterExpander#enabled()
-		 * @see NamedParameterExpander#disabled()
+		 * @param enabled {@literal true} to use named parameter expansion. {@literal false} to disable named parameter
+		 *          expansion.
+		 * @return {@code this} {@link DatabaseClient.Builder}.
+		 * @see NamedParameterExpander
 		 */
-		Builder namedParameters(NamedParameterExpander expander);
+		Builder namedParameters(boolean enabled);
 
 		/**
 		 * Configures a {@link Consumer} to configure this builder.

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

@@ -22,10 +22,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 
 import org.junit.Test;
-import org.springframework.data.r2dbc.core.BindableOperation;
-import org.springframework.data.r2dbc.core.MapBindParameterSource;
-import org.springframework.data.r2dbc.core.NamedParameterUtils;
-import org.springframework.data.r2dbc.core.ParsedSql;
+
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.BindTarget;
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
@@ -67,12 +64,12 @@ public class NamedParameterUtilsUnitTests {
 		MapBindParameterSource namedParams = new MapBindParameterSource(new HashMap<>());
 		namedParams.addValue("a", "a").addValue("b", "b").addValue("c", "c");
 
-		BindableOperation operation = NamedParameterUtils.substituteNamedParameters("xxx :a :b :c",
+		PreparedOperation<?> operation = NamedParameterUtils.substituteNamedParameters("xxx :a :b :c",
 				PostgresDialect.INSTANCE.getBindMarkersFactory(), namedParams);
 
 		assertThat(operation.toQuery()).isEqualTo("xxx $1 $2 $3");
 
-		BindableOperation operation2 = NamedParameterUtils.substituteNamedParameters("xxx :a :b :c",
+		PreparedOperation<?> operation2 = NamedParameterUtils.substituteNamedParameters("xxx :a :b :c",
 				SqlServerDialect.INSTANCE.getBindMarkersFactory(), namedParams);
 
 		assertThat(operation2.toQuery()).isEqualTo("xxx @P0_a @P1_b @P2_c");
@@ -85,12 +82,12 @@ public class NamedParameterUtilsUnitTests {
 		namedParams.addValue("a",
 				Arrays.asList(new Object[] { "Walter", "Heisenberg" }, new Object[] { "Walt Jr.", "Flynn" }));
 
-		BindableOperation operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
+		PreparedOperation<?> operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
 
 		assertThat(operation.toQuery()).isEqualTo("xxx ($1, $2), ($3, $4)");
 	}
 
-	@Test // gh-23
+	@Test // gh-23, gh-105
 	public void shouldBindObjectArray() {
 
 		MapBindParameterSource namedParams = new MapBindParameterSource(new HashMap<>());
@@ -99,8 +96,8 @@ public class NamedParameterUtilsUnitTests {
 
 		BindTarget bindTarget = mock(BindTarget.class);
 
-		BindableOperation operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
-		operation.bind(bindTarget, "a", namedParams.getValue("a"));
+		PreparedOperation<?> operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
+		operation.bindTo(bindTarget);
 
 		verify(bindTarget).bind(0, "Walter");
 		verify(bindTarget).bind(1, "Heisenberg");
@@ -133,7 +130,7 @@ public class NamedParameterUtilsUnitTests {
 		String sql = "select 'first name' from artists where id = :id and birth_date=:birthDate::timestamp";
 
 		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
-		BindableOperation operation = NamedParameterUtils.substituteNamedParameters(parsedSql, BIND_MARKERS,
+		PreparedOperation<?> operation = NamedParameterUtils.substituteNamedParameters(parsedSql, BIND_MARKERS,
 				new MapBindParameterSource());
 
 		assertThat(operation.toQuery()).isEqualTo(expectedSql);