#23 - Add support for named parameters.

DatabaseClient now supports named parameters prefixed with a colon such as :name in addition to database-native bind markers. Named parameters thus are supported in annotated repository query methods which also increases portability of queries across database vendors.

Named parameter support unrolls collection arguments to reduce the need for argument-specific SQL statements:

db.execute()
    .sql("SELECT id, name, state FROM table WHERE age IN (:ages)")
    .bind("ages", Arrays.asList(35, 50));

Results in a query: SELECT id, name, state FROM table WHERE age IN (35, 50)

Collection arguments containing nested object arrays can be used to use select lists:

List<Object[]> tuples = new ArrayList<>();
tuples.add(new Object[] {"John", 35});
tuples.add(new Object[] {"Ann",  50});

db.execute()
    .sql("SELECT id, name, state FROM table WHERE (name, age) IN (:tuples)")
    .bind("tuples", tuples);

translates to: SELECT id, name, state FROM table WHERE (name, age) IN (('John', 35), ('Ann', 50))

Original pull request: #47.

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

@@ -0,0 +1,61 @@
+/*
+ * 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
+ *
+ *      http://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.function;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Interface that defines common functionality for objects that can offer parameter values for named bind parameters,
+ * serving as argument for {@link NamedParameterExpander} operations.
+ * <p>
+ * This interface allows for the specification of the type in addition to parameter values. All parameter values and
+ * types are identified by specifying the name of the parameter.
+ * <p>
+ * Intended to wrap various implementations like a {@link java.util.Map} with a consistent interface.
+ *
+ * @author Mark Paluch
+ * @see MapBindParameterSource
+ */
+public interface BindParameterSource {
+
+	/**
+	 * Determine whether there is a value for the specified named parameter.
+	 *
+	 * @param paramName the name of the parameter.
+	 * @return {@literal true} if there is a value defined; {@literal false} otherwise.
+	 */
+	boolean hasValue(String paramName);
+
+	/**
+	 * Return the parameter value for the requested named parameter.
+	 *
+	 * @param paramName the name of the parameter.
+	 * @return the value of the specified parameter, can be {@literal null}.
+	 * @throws IllegalArgumentException if there is no value for the requested parameter.
+	 */
+	@Nullable
+	Object getValue(String paramName) throws IllegalArgumentException;
+
+	/**
+	 * Determine the type for the specified named parameter.
+	 *
+	 * @param paramName the name of the parameter.
+	 * @return the type of the specified parameter, or {@link Object#getClass()} if not known.
+	 */
+	default Class<?> getType(String paramName) {
+		return Object.class;
+	}
+}

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

@@ -109,6 +109,16 @@ public interface DatabaseClient {
 		 */
 		Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy);
 
+		/**
+		 * Configures {@link NamedParameterExpander}.
+		 *
+		 * @param namedParameters must not be {@literal null}.
+		 * @return {@code this} {@link Builder}.
+		 * @see NamedParameterExpander#enabled()
+		 * @see NamedParameterExpander#disabled()
+		 */
+		Builder namedParameters(NamedParameterExpander namedParameters);
+
 		/**
 		 * Configures a {@link Consumer} to configure this builder.
 		 *
@@ -124,7 +134,12 @@ public interface DatabaseClient {
 	}
 
 	/**
-	 * Contract for specifying a SQL call along with options leading to the exchange.
+	 * Contract for specifying a SQL call along with options leading to the exchange. The SQL string can contain either
+	 * native parameter bind markers (e.g. {@literal $1, $2} for Postgres, {@literal @P0, @P1} for SQL Server) or named
+	 * parameters (e.g. {@literal :foo, :bar}) when {@link NamedParameterExpander} is enabled.
+	 *
+	 * @see NamedParameterExpander
+	 * @see DatabaseClient.Builder#namedParameters(NamedParameterExpander)
 	 */
 	interface SqlSpec {
 

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

@@ -66,9 +66,6 @@ import org.springframework.util.Assert;
  */
 class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
-	/**
-	 * Logger available to subclasses
-	 */
 	private final Log logger = LogFactory.getLog(getClass());
 
 	private final ConnectionFactory connector;
@@ -77,14 +74,18 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 	private final ReactiveDataAccessStrategy dataAccessStrategy;
 
+	private final NamedParameterExpander namedParameters;
+
 	private final DefaultDatabaseClientBuilder builder;
 
 	DefaultDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy dataAccessStrategy, DefaultDatabaseClientBuilder builder) {
+			ReactiveDataAccessStrategy dataAccessStrategy, NamedParameterExpander namedParameters,
+			DefaultDatabaseClientBuilder builder) {
 
 		this.connector = connector;
 		this.exceptionTranslator = exceptionTranslator;
 		this.dataAccessStrategy = dataAccessStrategy;
+		this.namedParameters = namedParameters;
 		this.builder = builder;
 	}
 
@@ -325,8 +326,21 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					logger.debug("Executing SQL statement [" + sql + "]");
 				}
 
-				Statement<?> statement = it.createStatement(sql);
-				doBind(statement, byName, byIndex);
+				BindableOperation operation = namedParameters.expand(sql, dataAccessStrategy.getBindMarkersFactory(),
+						new MapBindParameterSource(byName));
+
+				Statement<?> statement = it.createStatement(operation.toQuery());
+
+				byName.forEach((name, o) -> {
+
+					if (o.getValue() != null) {
+						operation.bind(statement, name, o.getValue());
+					} else {
+						operation.bindNull(statement, name, o.getType());
+					}
+				});
+
+				doBind(statement, Collections.emptyMap(), byIndex);
 
 				return statement;
 			};

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

@@ -38,6 +38,7 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 	private @Nullable ConnectionFactory connectionFactory;
 	private @Nullable R2dbcExceptionTranslator exceptionTranslator;
 	private ReactiveDataAccessStrategy accessStrategy;
+	private NamedParameterExpander namedParameters;
 
 	DefaultDatabaseClientBuilder() {}
 
@@ -48,8 +49,13 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 		this.connectionFactory = other.connectionFactory;
 		this.exceptionTranslator = other.exceptionTranslator;
 		this.accessStrategy = other.accessStrategy;
+		this.namedParameters = other.namedParameters;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#connectionFactory(io.r2dbc.spi.ConnectionFactory)
+	 */
 	@Override
 	public Builder connectionFactory(ConnectionFactory factory) {
 
@@ -59,6 +65,10 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 		return this;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#exceptionTranslator(org.springframework.data.r2dbc.support.R2dbcExceptionTranslator)
+	 */
 	@Override
 	public Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator) {
 
@@ -68,6 +78,10 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 		return this;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#dataAccessStrategy(org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy)
+	 */
 	@Override
 	public Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy) {
 
@@ -77,6 +91,23 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 		return this;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#namedParameters(org.springframework.data.r2dbc.function.NamedParameterExpander)
+	 */
+	@Override
+	public Builder namedParameters(NamedParameterExpander namedParameters) {
+
+		Assert.notNull(namedParameters, "NamedParameterExpander must not be null!");
+
+		this.namedParameters = namedParameters;
+		return this;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#build()
+	 */
 	@Override
 	public DatabaseClient build() {
 
@@ -97,19 +128,35 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 			accessStrategy = new DefaultReactiveDataAccessStrategy(dialect);
 		}
 
-		return doBuild(this.connectionFactory, exceptionTranslator, accessStrategy, new DefaultDatabaseClientBuilder(this));
+		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, DefaultDatabaseClientBuilder builder) {
-		return new DefaultDatabaseClient(connector, exceptionTranslator, accessStrategy, builder);
+			ReactiveDataAccessStrategy accessStrategy, NamedParameterExpander namedParameters,
+			DefaultDatabaseClientBuilder builder) {
+		return new DefaultDatabaseClient(connector, exceptionTranslator, accessStrategy, namedParameters, builder);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#clone()
+	 */
 	@Override
 	public DatabaseClient.Builder clone() {
 		return new DefaultDatabaseClientBuilder(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DatabaseClient.Builder#apply(java.util.function.Consumer)
+	 */
 	@Override
 	public DatabaseClient.Builder apply(Consumer<DatabaseClient.Builder> builderConsumer) {
 		Assert.notNull(builderConsumer, "BuilderConsumer must not be null");

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

@@ -41,6 +41,7 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.ArrayColumns;
 import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.dialect.LimitClause;
 import org.springframework.data.r2dbc.dialect.LimitClause.Position;
@@ -238,6 +239,15 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return getRequiredPersistentEntity(type).getTableName();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getBindMarkersFactory()
+	 */
+	@Override
+	public BindMarkersFactory getBindMarkersFactory() {
+		return dialect.getBindMarkersFactory();
+	}
+
 	private RelationalPersistentEntity<?> getRequiredPersistentEntity(Class<?> typeToRead) {
 		return mappingContext.getRequiredPersistentEntity(typeToRead);
 	}

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

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

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

@@ -69,6 +69,15 @@ class DefaultTransactionalDatabaseClientBuilder extends DefaultDatabaseClientBui
 		return this;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder#dataAccessStrategy(org.springframework.data.r2dbc.function.NamedParameterSupport)
+	 */
+	@Override
+	public TransactionalDatabaseClient.Builder namedParameters(NamedParameterExpander namedParameters) {
+		super.namedParameters(namedParameters);
+		return this;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder#apply(java.util.function.Consumer)
 	 */
@@ -86,12 +95,11 @@ class DefaultTransactionalDatabaseClientBuilder extends DefaultDatabaseClientBui
 		return (TransactionalDatabaseClient) super.build();
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder#doBuild(io.r2dbc.spi.ConnectionFactory, org.springframework.data.r2dbc.support.R2dbcExceptionTranslator, org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy, org.springframework.data.r2dbc.function.DefaultDatabaseClientBuilder)
-	 */
 	@Override
 	protected DatabaseClient doBuild(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
-			ReactiveDataAccessStrategy accessStrategy, DefaultDatabaseClientBuilder builder) {
-		return new DefaultTransactionalDatabaseClient(connector, exceptionTranslator, accessStrategy, builder);
+			ReactiveDataAccessStrategy accessStrategy, NamedParameterExpander namedParameters,
+			DefaultDatabaseClientBuilder builder) {
+		return new DefaultTransactionalDatabaseClient(connector, exceptionTranslator, accessStrategy, namedParameters,
+				builder);
 	}
 }

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

@@ -0,0 +1,114 @@
+/*
+ * 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
+ *
+ *      http://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.function;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.data.r2dbc.function.convert.SettableValue;
+import org.springframework.util.Assert;
+
+/**
+ * {@link BindParameterSource} implementation that holds a given {@link Map} of parameters encapsulated as
+ * {@link SettableValue}.
+ * <p>
+ * This class is intended for passing in a simple Map of parameter values to the methods of the
+ * {@link NamedParameterExpander} class.
+ *
+ * @author Mark Paluch
+ */
+class MapBindParameterSource implements BindParameterSource {
+
+	private final Map<String, SettableValue> values;
+
+	/**
+	 * Creates a new empty {@link MapBindParameterSource}.
+	 */
+	MapBindParameterSource() {
+		this(new LinkedHashMap<>());
+	}
+
+	/**
+	 * Creates a new {@link MapBindParameterSource} given {@link Map} of {@link SettableValue}.
+	 *
+	 * @param values the parameter mapping.
+	 */
+	MapBindParameterSource(Map<String, SettableValue> values) {
+
+		Assert.notNull(values, "Values must not be null");
+
+		this.values = values;
+	}
+
+	/**
+	 * Add a key-value pair to the {@link MapBindParameterSource}. The value must not be {@literal null}.
+	 *
+	 * @param paramName must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @return {@code this} {@link MapBindParameterSource}
+	 */
+	MapBindParameterSource addValue(String paramName, Object value) {
+
+		Assert.notNull(paramName, "Parameter name must not be null!");
+		Assert.notNull(value, "Value must not be null!");
+
+		this.values.put(paramName, new SettableValue(paramName, value, value.getClass()));
+		return this;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.SqlParameterSource#hasValue(java.lang.String)
+	 */
+	@Override
+	public boolean hasValue(String paramName) {
+
+		Assert.notNull(paramName, "Parameter name must not be null!");
+
+		return values.containsKey(paramName);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.SqlParameterSource#getType(java.lang.String)
+	 */
+	@Override
+	public Class<?> getType(String paramName) {
+
+		Assert.notNull(paramName, "Parameter name must not be null!");
+
+		SettableValue settableValue = this.values.get(paramName);
+		if (settableValue != null) {
+			return settableValue.getType();
+		}
+
+		return Object.class;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.SqlParameterSource#getValue(java.lang.String)
+	 */
+	@Override
+	public Object getValue(String paramName) throws IllegalArgumentException {
+
+		if (!hasValue(paramName)) {
+			throw new IllegalArgumentException("No value registered for key '" + paramName + "'");
+		}
+
+		return this.values.get(paramName).getValue();
+	}
+}

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

@@ -0,0 +1,160 @@
+/*
+ * 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
+ *
+ *      http://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.function;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+
+/**
+ * SQL translation support allowing the use of named parameters rather than native placeholders.
+ * <p>
+ * This class expands SQL from named parameters to native style placeholders at execution time. It also allows for
+ * expanding a {@link java.util.List} of values to the appropriate number of placeholders.
+ * <p>
+ * <b>NOTE: An instance of this class is thread-safe once configured.</b>
+ *
+ * @author Mark Paluch
+ */
+public class NamedParameterExpander {
+
+	/** Default maximum number of entries for the SQL cache: 256. */
+	public static final int DEFAULT_CACHE_LIMIT = 256;
+
+	private volatile int cacheLimit = DEFAULT_CACHE_LIMIT;
+
+	private final Log logger = LogFactory.getLog(getClass());
+
+	/** Cache of original SQL String to ParsedSql representation. */
+	@SuppressWarnings("serial") private final Map<String, ParsedSql> parsedSqlCache = new LinkedHashMap<String, ParsedSql>(
+			DEFAULT_CACHE_LIMIT, 0.75f, true) {
+		@Override
+		protected boolean removeEldestEntry(Map.Entry<String, ParsedSql> eldest) {
+			return size() > getCacheLimit();
+		}
+	};
+
+	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}.
+	 *
+	 * @return a new enabled instance of {@link NamedParameterExpander}.
+	 */
+	public static NamedParameterExpander enabled() {
+		return new NamedParameterExpander();
+	}
+
+	/**
+	 * Specify the maximum number of entries for the SQL cache. Default is 256.
+	 */
+	public void setCacheLimit(int cacheLimit) {
+		this.cacheLimit = cacheLimit;
+	}
+
+	/**
+	 * Return the maximum number of entries for the SQL cache.
+	 */
+	public int getCacheLimit() {
+		return this.cacheLimit;
+	}
+
+	/**
+	 * Obtain a parsed representation of the given SQL statement.
+	 * <p>
+	 * The default implementation uses an LRU cache with an upper limit of 256 entries.
+	 *
+	 * @param sql the original SQL statement
+	 * @return a representation of the parsed SQL statement
+	 */
+	protected ParsedSql getParsedSql(String sql) {
+
+		if (getCacheLimit() <= 0) {
+			return NamedParameterUtils.parseSqlStatement(sql);
+		}
+
+		synchronized (this.parsedSqlCache) {
+			ParsedSql parsedSql = this.parsedSqlCache.get(sql);
+			if (parsedSql == null) {
+				parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+				this.parsedSqlCache.put(sql, parsedSql);
+			}
+			return parsedSql;
+		}
+	}
+
+	BindableOperation expand(String sql, BindMarkersFactory bindMarkersFactory, BindParameterSource paramSource) {
+
+		ParsedSql parsedSql = getParsedSql(sql);
+
+		BindableOperation expanded = NamedParameterUtils.substituteNamedParameters(parsedSql, bindMarkersFactory,
+				paramSource);
+
+		if (logger.isDebugEnabled()) {
+			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(Statement<?> statement, String identifier, Object value) {
+					statement.bind(identifier, value);
+				}
+
+				@Override
+				public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
+					statement.bindNull(identifier, valueType);
+				}
+
+				@Override
+				public String toQuery() {
+					return sql;
+				}
+			};
+		}
+	}
+}

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

@@ -0,0 +1,485 @@
+/*
+ * Copyright 2002-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
+ *
+ *      http://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.function;
+
+import io.r2dbc.spi.Statement;
+import lombok.Value;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Helper methods for named parameter parsing.
+ * <p>
+ * Only intended for internal use within Spring's Data's R2DBC framework. Partially extracted from Spring's JDBC named
+ * parameter support.
+ * <p>
+ * This is a subset of Spring Frameworks's {@code org.springframework.jdbc.core.namedparam.NamedParameterUtils}.
+ *
+ * @author Thomas Risberg
+ * @author Juergen Hoeller
+ * @author Mark Paluch
+ */
+abstract class NamedParameterUtils {
+
+	/**
+	 * Set of characters that qualify as comment or quotes starting characters.
+	 */
+	private static final String[] START_SKIP = new String[] { "'", "\"", "--", "/*" };
+
+	/**
+	 * Set of characters that at are the corresponding comment or quotes ending characters.
+	 */
+	private static final String[] STOP_SKIP = new String[] { "'", "\"", "\n", "*/" };
+
+	/**
+	 * Set of characters that qualify as parameter separators, indicating that a parameter name in a SQL String has ended.
+	 */
+	private static final String PARAMETER_SEPARATORS = "\"':&,;()|=+-*%/\\<>^";
+
+	/**
+	 * An index with separator flags per character code. Technically only needed between 34 and 124 at this point.
+	 */
+	private static final boolean[] separatorIndex = new boolean[128];
+
+	static {
+		for (char c : PARAMETER_SEPARATORS.toCharArray()) {
+			separatorIndex[c] = true;
+		}
+	}
+
+	// -------------------------------------------------------------------------
+	// Core methods used by NamedParameterSupport.
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Parse the SQL statement and locate any placeholders or named parameters. Named parameters are substituted for a
+	 * placeholder.
+	 *
+	 * @param sql the SQL statement
+	 * @return the parsed statement, represented as {@link ParsedSql} instance.
+	 */
+	public static ParsedSql parseSqlStatement(String sql) {
+
+		Assert.notNull(sql, "SQL must not be null");
+
+		Set<String> namedParameters = new HashSet<>();
+		String sqlToUse = sql;
+		List<ParameterHolder> parameterList = new ArrayList<>();
+
+		char[] statement = sql.toCharArray();
+		int namedParameterCount = 0;
+		int unnamedParameterCount = 0;
+		int totalParameterCount = 0;
+
+		int escapes = 0;
+		int i = 0;
+		while (i < statement.length) {
+			int skipToPosition = i;
+			while (i < statement.length) {
+				skipToPosition = skipCommentsAndQuotes(statement, i);
+				if (i == skipToPosition) {
+					break;
+				} else {
+					i = skipToPosition;
+				}
+			}
+			if (i >= statement.length) {
+				break;
+			}
+			char c = statement[i];
+			if (c == ':' || c == '&') {
+				int j = i + 1;
+				if (c == ':' && j < statement.length && statement[j] == ':') {
+					// Postgres-style "::" casting operator should be skipped
+					i = i + 2;
+					continue;
+				}
+				String parameter = null;
+				if (c == ':' && j < statement.length && statement[j] == '{') {
+					// :{x} style parameter
+					while (statement[j] != '}') {
+						j++;
+						if (j >= statement.length) {
+							throw new InvalidDataAccessApiUsageException(
+									"Non-terminated named parameter declaration " + "at position " + i + " in statement: " + sql);
+						}
+						if (statement[j] == ':' || statement[j] == '{') {
+							throw new InvalidDataAccessApiUsageException("Parameter name contains invalid character '" + statement[j]
+									+ "' at position " + i + " in statement: " + sql);
+						}
+					}
+					if (j - i > 2) {
+						parameter = sql.substring(i + 2, j);
+						namedParameterCount = addNewNamedParameter(namedParameters, namedParameterCount, parameter);
+						totalParameterCount = addNamedParameter(parameterList, totalParameterCount, escapes, i, j + 1, parameter);
+					}
+					j++;
+				} else {
+					while (j < statement.length && !isParameterSeparator(statement[j])) {
+						j++;
+					}
+					if (j - i > 1) {
+						parameter = sql.substring(i + 1, j);
+						namedParameterCount = addNewNamedParameter(namedParameters, namedParameterCount, parameter);
+						totalParameterCount = addNamedParameter(parameterList, totalParameterCount, escapes, i, j, parameter);
+					}
+				}
+				i = j - 1;
+			} else {
+				if (c == '\\') {
+					int j = i + 1;
+					if (j < statement.length && statement[j] == ':') {
+						// escaped ":" should be skipped
+						sqlToUse = sqlToUse.substring(0, i - escapes) + sqlToUse.substring(i - escapes + 1);
+						escapes++;
+						i = i + 2;
+						continue;
+					}
+				}
+				if (c == '?') {
+					int j = i + 1;
+					if (j < statement.length && (statement[j] == '?' || statement[j] == '|' || statement[j] == '&')) {
+						// Postgres-style "??", "?|", "?&" operator should be skipped
+						i = i + 2;
+						continue;
+					}
+					unnamedParameterCount++;
+					totalParameterCount++;
+				}
+			}
+			i++;
+		}
+		ParsedSql parsedSql = new ParsedSql(sqlToUse);
+		for (ParameterHolder ph : parameterList) {
+			parsedSql.addNamedParameter(ph.getParameterName(), ph.getStartIndex(), ph.getEndIndex());
+		}
+		parsedSql.setNamedParameterCount(namedParameterCount);
+		parsedSql.setUnnamedParameterCount(unnamedParameterCount);
+		parsedSql.setTotalParameterCount(totalParameterCount);
+		return parsedSql;
+	}
+
+	private static int addNamedParameter(List<ParameterHolder> parameterList, int totalParameterCount, int escapes, int i,
+			int j, String parameter) {
+
+		parameterList.add(new ParameterHolder(parameter, i - escapes, j - escapes));
+		totalParameterCount++;
+		return totalParameterCount;
+	}
+
+	private static int addNewNamedParameter(Set<String> namedParameters, int namedParameterCount, String parameter) {
+		if (!namedParameters.contains(parameter)) {
+			namedParameters.add(parameter);
+			namedParameterCount++;
+		}
+		return namedParameterCount;
+	}
+
+	/**
+	 * Skip over comments and quoted names present in an SQL statement.
+	 *
+	 * @param statement character array containing SQL statement.
+	 * @param position current position of statement.
+	 * @return next position to process after any comments or quotes are skipped.
+	 */
+	private static int skipCommentsAndQuotes(char[] statement, int position) {
+
+		for (int i = 0; i < START_SKIP.length; i++) {
+			if (statement[position] == START_SKIP[i].charAt(0)) {
+				boolean match = true;
+				for (int j = 1; j < START_SKIP[i].length(); j++) {
+					if (statement[position + j] != START_SKIP[i].charAt(j)) {
+						match = false;
+						break;
+					}
+				}
+				if (match) {
+					int offset = START_SKIP[i].length();
+					for (int m = position + offset; m < statement.length; m++) {
+						if (statement[m] == STOP_SKIP[i].charAt(0)) {
+							boolean endMatch = true;
+							int endPos = m;
+							for (int n = 1; n < STOP_SKIP[i].length(); n++) {
+								if (m + n >= statement.length) {
+									// last comment not closed properly
+									return statement.length;
+								}
+								if (statement[m + n] != STOP_SKIP[i].charAt(n)) {
+									endMatch = false;
+									break;
+								}
+								endPos = m + n;
+							}
+							if (endMatch) {
+								// found character sequence ending comment or quote
+								return endPos + 1;
+							}
+						}
+					}
+					// character sequence ending comment or quote not found
+					return statement.length;
+				}
+			}
+		}
+		return position;
+	}
+
+	/**
+	 * 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 parsedSql the parsed representation of the SQL statement.
+	 * @param bindMarkersFactory the bind marker factory.
+	 * @param paramSource the source for named parameters.
+	 * @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) {
+
+		BindMarkerHolder markerHolder = new BindMarkerHolder(bindMarkersFactory.create());
+
+		String originalSql = parsedSql.getOriginalSql();
+		List<String> paramNames = parsedSql.getParameterNames();
+		if (paramNames.isEmpty()) {
+			return new ExpandedQuery(originalSql, markerHolder);
+		}
+
+		StringBuilder actualSql = new StringBuilder(originalSql.length());
+		int lastIndex = 0;
+		for (int i = 0; i < paramNames.size(); i++) {
+			String paramName = paramNames.get(i);
+			int[] indexes = parsedSql.getParameterIndexes(i);
+			int startIndex = indexes[0];
+			int endIndex = indexes[1];
+			actualSql.append(originalSql, lastIndex, startIndex);
+			if (paramSource.hasValue(paramName)) {
+				Object value = paramSource.getValue(paramName);
+				if (value instanceof Collection) {
+					Iterator<?> entryIter = ((Collection<?>) value).iterator();
+					int k = 0;
+					while (entryIter.hasNext()) {
+						if (k > 0) {
+							actualSql.append(", ");
+						}
+						k++;
+						Object entryItem = entryIter.next();
+						if (entryItem instanceof Object[]) {
+							Object[] expressionList = (Object[]) entryItem;
+							actualSql.append('(');
+							for (int m = 0; m < expressionList.length; m++) {
+								if (m > 0) {
+									actualSql.append(", ");
+								}
+								actualSql.append(markerHolder.addMarker(paramName));
+							}
+							actualSql.append(')');
+						} else {
+							actualSql.append(markerHolder.addMarker(paramName));
+						}
+
+					}
+				} else {
+					actualSql.append(markerHolder.addMarker(paramName));
+				}
+			} else {
+				actualSql.append(markerHolder.addMarker(paramName));
+			}
+			lastIndex = endIndex;
+		}
+		actualSql.append(originalSql, lastIndex, originalSql.length());
+
+		return new ExpandedQuery(actualSql.toString(), markerHolder);
+	}
+
+	/**
+	 * Determine whether a parameter name ends at the current position, that is, whether the given character qualifies as
+	 * a separator.
+	 */
+	private static boolean isParameterSeparator(char c) {
+		return (c < 128 && separatorIndex[c]) || Character.isWhitespace(c);
+	}
+
+	// -------------------------------------------------------------------------
+	// Convenience methods operating on a plain SQL String
+	// -------------------------------------------------------------------------
+
+	/**
+	 * 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.
+	 * <p>
+	 *
+	 * @param sql the SQL statement.
+	 * @param bindMarkersFactory the bind marker factory.
+	 * @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,
+			BindParameterSource paramSource) {
+		ParsedSql parsedSql = parseSqlStatement(sql);
+		return substituteNamedParameters(parsedSql, bindMarkersFactory, paramSource);
+	}
+
+	@Value
+	private static class ParameterHolder {
+
+		String parameterName;
+
+		int startIndex;
+
+		int endIndex;
+	}
+
+	/**
+	 * Holder for bind marker progress.
+	 */
+	private static class BindMarkerHolder {
+
+		private final BindMarkers bindMarkers;
+		private final Map<String, List<BindMarker>> markers = new TreeMap<>();
+
+		BindMarkerHolder(BindMarkers bindMarkers) {
+			this.bindMarkers = bindMarkers;
+		}
+
+		String addMarker(String name) {
+
+			BindMarker bindMarker = bindMarkers.next(name);
+			markers.computeIfAbsent(name, ignore -> new ArrayList<>()).add(bindMarker);
+			return bindMarker.getPlaceholder();
+		}
+	}
+
+	/**
+	 * 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 final String expandedSql;
+
+		private final Map<String, List<BindMarker>> markers;
+
+		ExpandedQuery(String expandedSql, BindMarkerHolder bindMarkerHolder) {
+			this.expandedSql = expandedSql;
+			this.markers = bindMarkerHolder.markers;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
+		 */
+		@Override
+		@SuppressWarnings("unchecked")
+		public void bind(Statement<?> statement, String identifier, Object value) {
+
+			List<BindMarker> bindMarkers = getBindMarkers(identifier);
+
+			if (bindMarkers.size() == 1) {
+				bindMarkers.get(0).bind(statement, value);
+			} else {
+
+				Assert.isInstanceOf(Collection.class, value,
+						() -> String.format("Value [%s] must be an Collection with a size of [%d]", value, bindMarkers.size()));
+
+				Collection<Object> collection = (Collection<Object>) value;
+
+				Iterator<Object> iterator = collection.iterator();
+				Iterator<BindMarker> markers = bindMarkers.iterator();
+
+				while (iterator.hasNext()) {
+
+					Object valueToBind = iterator.next();
+
+					if (valueToBind instanceof Object[]) {
+						Object[] objects = (Object[]) valueToBind;
+						for (Object object : objects) {
+							bind(statement, markers, object);
+						}
+					} else {
+						bind(statement, markers, valueToBind);
+					}
+				}
+			}
+		}
+
+		private void bind(Statement<?> statement, Iterator<BindMarker> markers, Object valueToBind) {
+
+			Assert.isTrue(markers.hasNext(),
+					() -> String.format(
+							"No bind marker for value [%s] in SQL [%s]. Check that the query was expanded using the same arguments.",
+							valueToBind, toQuery()));
+
+			markers.next().bind(statement, valueToBind);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
+		 */
+		@Override
+		public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
+
+			List<BindMarker> bindMarkers = getBindMarkers(identifier);
+
+			if (bindMarkers.size() == 1) {
+				bindMarkers.get(0).bindNull(statement, valueType);
+				return;
+			}
+
+			throw new UnsupportedOperationException("bindNull(…) can bind only singular values");
+		}
+
+		private List<BindMarker> getBindMarkers(String identifier) {
+
+			List<BindMarker> bindMarkers = markers.get(identifier);
+
+			Assert.notNull(bindMarkers, () -> String.format("Parameter name [%s] is unknown. Known parameters names are: %s",
+					identifier, markers.keySet()));
+			return bindMarkers;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
+		 */
+		@Override
+		public String toQuery() {
+			return expandedSql;
+		}
+	}
+}

src/main/java/org/springframework/data/r2dbc/function/ParsedSql.java

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2002-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
+ *
+ *      http://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.function;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Holds information about a parsed SQL statement.
+ * <p>
+ * This is a copy of Spring Frameworks's {@code org.springframework.jdbc.core.namedparam.ParsedSql}.
+ *
+ * @author Thomas Risberg
+ * @author Juergen Hoeller
+ */
+class ParsedSql {
+
+	private String originalSql;
+
+	private List<String> parameterNames = new ArrayList<>();
+
+	private List<int[]> parameterIndexes = new ArrayList<>();
+
+	private int namedParameterCount;
+
+	private int unnamedParameterCount;
+
+	private int totalParameterCount;
+
+	/**
+	 * Create a new instance of the {@link ParsedSql} class.
+	 *
+	 * @param originalSql the SQL statement that is being (or is to be) parsed
+	 */
+	ParsedSql(String originalSql) {
+		this.originalSql = originalSql;
+	}
+
+	/**
+	 * Return the SQL statement that is being parsed.
+	 */
+	String getOriginalSql() {
+		return this.originalSql;
+	}
+
+	/**
+	 * Add a named parameter parsed from this SQL statement.
+	 *
+	 * @param parameterName the name of the parameter
+	 * @param startIndex the start index in the original SQL String
+	 * @param endIndex the end index in the original SQL String
+	 */
+	void addNamedParameter(String parameterName, int startIndex, int endIndex) {
+		this.parameterNames.add(parameterName);
+		this.parameterIndexes.add(new int[] { startIndex, endIndex });
+	}
+
+	/**
+	 * Return all of the parameters (bind variables) in the parsed SQL statement. Repeated occurrences of the same
+	 * parameter name are included here.
+	 */
+	List<String> getParameterNames() {
+		return this.parameterNames;
+	}
+
+	/**
+	 * Return the parameter indexes for the specified parameter.
+	 *
+	 * @param parameterPosition the position of the parameter (as index in the parameter names List)
+	 * @return the start index and end index, combined into a int array of length 2
+	 */
+	int[] getParameterIndexes(int parameterPosition) {
+		return this.parameterIndexes.get(parameterPosition);
+	}
+
+	/**
+	 * Set the count of named parameters in the SQL statement. Each parameter name counts once; repeated occurrences do
+	 * not count here.
+	 */
+	void setNamedParameterCount(int namedParameterCount) {
+		this.namedParameterCount = namedParameterCount;
+	}
+
+	/**
+	 * Return the count of named parameters in the SQL statement. Each parameter name counts once; repeated occurrences do
+	 * not count here.
+	 */
+	int getNamedParameterCount() {
+		return this.namedParameterCount;
+	}
+
+	/**
+	 * Set the count of all of the unnamed parameters in the SQL statement.
+	 */
+	void setUnnamedParameterCount(int unnamedParameterCount) {
+		this.unnamedParameterCount = unnamedParameterCount;
+	}
+
+	/**
+	 * Return the count of all of the unnamed parameters in the SQL statement.
+	 */
+	int getUnnamedParameterCount() {
+		return this.unnamedParameterCount;
+	}
+
+	/**
+	 * Set the total count of all of the parameters in the SQL statement. Repeated occurrences of the same parameter name
+	 * do count here.
+	 */
+	void setTotalParameterCount(int totalParameterCount) {
+		this.totalParameterCount = totalParameterCount;
+	}
+
+	/**
+	 * Return the total count of all of the parameters in the SQL statement. Repeated occurrences of the same parameter
+	 * name do count here.
+	 */
+	int getTotalParameterCount() {
+		return this.totalParameterCount;
+	}
+
+	/**
+	 * Exposes the original SQL String.
+	 */
+	@Override
+	public String toString() {
+		return this.originalSql;
+	}
+
+}

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

@@ -26,6 +26,7 @@ import java.util.function.BiFunction;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.function.convert.SettableValue;
 
 /**
@@ -76,6 +77,13 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	String getTableName(Class<?> type);
 
+	/**
+	 * Returns the configured {@link BindMarkersFactory} to create native parameter placeholder markers.
+	 *
+	 * @return the configured {@link BindMarkersFactory}.
+	 */
+	BindMarkersFactory getBindMarkersFactory();
+
 	// -------------------------------------------------------------------------
 	// Methods creating SQL operations.
 	// Subject to be moved into a SQL creation DSL.

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

@@ -40,10 +40,10 @@ import org.springframework.util.Assert;
  * <pre class="code">
  * Flux<Integer> transactionalFlux = databaseClient.inTransaction(db -> {
  *
- * 	return db.execute().sql("INSERT INTO person (id, firstname, lastname) VALUES($1, $2, $3)") //
- * 			.bind(0, 1) //
- * 			.bind(1, "Walter") //
- * 			.bind(2, "White") //
+ * 	return db.execute().sql("INSERT INTO person (id, firstname, lastname) VALUES(:id, :firstname, :lastname)") //
+ * 			.bind("id", 1) //
+ * 			.bind("firstname", "Walter") //
+ * 			.bind("lastname", "White") //
  * 			.fetch().rowsUpdated();
  * });
  * </pre>
@@ -54,10 +54,11 @@ import org.springframework.util.Assert;
  *
  * <pre class="code">
  * Mono<Void> mono = databaseClient.beginTransaction()
- * 		.then(databaseClient.execute().sql("INSERT INTO person (id, firstname, lastname) VALUES($1, $2, $3)") //
- * 				.bind(0, 1) //
- * 				.bind(1, "Walter") //
- * 				.bind(2, "White") //
+ * 		.then(databaseClient.execute()
+ * 				.sql("INSERT INTO person (id, firstname, lastname) VALUES(:id, :firstname, :lastname)") //
+ * 				.bind("id", 1) //
+ * 				.bind("firstname", "Walter") //
+ * 				.bind("lastname", "White") //
  * 				.fetch().rowsUpdated())
  * 		.then(databaseClient.commitTransaction());
  *
@@ -168,7 +169,7 @@ public interface TransactionalDatabaseClient extends DatabaseClient {
 		 * Configures the {@link ConnectionFactory R2DBC connector}.
 		 *
 		 * @param factory must not be {@literal null}.
-		 * @return {@code this} {@link DatabaseClient.Builder}.
+		 * @return {@code this} {@link Builder}.
 		 */
 		Builder connectionFactory(ConnectionFactory factory);
 
@@ -176,7 +177,7 @@ public interface TransactionalDatabaseClient extends DatabaseClient {
 		 * Configures a {@link R2dbcExceptionTranslator}.
 		 *
 		 * @param exceptionTranslator must not be {@literal null}.
-		 * @return {@code this} {@link DatabaseClient.Builder}.
+		 * @return {@code this} {@link Builder}.
 		 */
 		Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator);
 
@@ -184,15 +185,25 @@ public interface TransactionalDatabaseClient extends DatabaseClient {
 		 * Configures a {@link ReactiveDataAccessStrategy}.
 		 *
 		 * @param accessStrategy must not be {@literal null}.
-		 * @return {@code this} {@link DatabaseClient.Builder}.
+		 * @return {@code this} {@link Builder}.
 		 */
 		Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy);
 
+		/**
+		 * Configures {@link NamedParameterExpander}.
+		 *
+		 * @param namedParameters must not be {@literal null}.
+		 * @return {@code this} {@link Builder}.
+		 * @see NamedParameterExpander#enabled()
+		 * @see NamedParameterExpander#disabled()
+		 */
+		Builder namedParameters(NamedParameterExpander namedParameters);
+
 		/**
 		 * Configures a {@link Consumer} to configure this builder.
 		 *
 		 * @param builderConsumer must not be {@literal null}.
-		 * @return {@code this} {@link DatabaseClient.Builder}.
+		 * @return {@code this} {@link Builder}.
 		 */
 		Builder apply(Consumer<DatabaseClient.Builder> builderConsumer);