From b43b11936f14789e9252d6980b60a68fd00ec7ea Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Fri, 4 Jan 2019 14:37:37 +0100
Subject: [PATCH] #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/asciidoc/new-features.adoc           |  13 +-
 .../reference/r2dbc-repositories.adoc         |   6 +-
 src/main/asciidoc/reference/r2dbc.adoc        |  69 ++-
 .../r2dbc/function/BindParameterSource.java   |  61 +++
 .../data/r2dbc/function/DatabaseClient.java   |  17 +-
 .../r2dbc/function/DefaultDatabaseClient.java |  26 +-
 .../DefaultDatabaseClientBuilder.java         |  53 +-
 .../DefaultReactiveDataAccessStrategy.java    |  10 +
 .../DefaultTransactionalDatabaseClient.java   |   5 +-
 ...ultTransactionalDatabaseClientBuilder.java |  18 +-
 .../function/MapBindParameterSource.java      | 114 ++++
 .../function/NamedParameterExpander.java      | 160 ++++++
 .../r2dbc/function/NamedParameterUtils.java   | 485 ++++++++++++++++++
 .../data/r2dbc/function/ParsedSql.java        | 143 ++++++
 .../function/ReactiveDataAccessStrategy.java  |   8 +
 .../function/TransactionalDatabaseClient.java |  35 +-
 ...bstractDatabaseClientIntegrationTests.java |  10 +-
 ...ctionalDatabaseClientIntegrationTests.java |  33 +-
 .../NamedParameterUtilsUnitTests.java         | 295 +++++++++++
 ...ostgresDatabaseClientIntegrationTests.java |   5 -
 ...ctionalDatabaseClientIntegrationTests.java |   5 -
 ...lServerDatabaseClientIntegrationTests.java |   5 -
 ...stgresR2dbcRepositoryIntegrationTests.java |   2 +-
 ...ServerR2dbcRepositoryIntegrationTests.java |   2 +-
 24 files changed, 1494 insertions(+), 86 deletions(-)
 create mode 100644 src/main/java/org/springframework/data/r2dbc/function/BindParameterSource.java
 create mode 100644 src/main/java/org/springframework/data/r2dbc/function/MapBindParameterSource.java
 create mode 100644 src/main/java/org/springframework/data/r2dbc/function/NamedParameterExpander.java
 create mode 100644 src/main/java/org/springframework/data/r2dbc/function/NamedParameterUtils.java
 create mode 100644 src/main/java/org/springframework/data/r2dbc/function/ParsedSql.java
 create mode 100644 src/test/java/org/springframework/data/r2dbc/function/NamedParameterUtilsUnitTests.java

diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc
index a475570..b47c2ee 100644
--- a/src/main/asciidoc/new-features.adoc
+++ b/src/main/asciidoc/new-features.adoc
@@ -1,10 +1,15 @@
 [[new-features]]
 = New & Noteworthy
 
+[[new-features.1-0-0-M2]]
+== What's New in Spring Data R2DBC 1.0.0 M2
+
+* Support for named parameters.
+
 [[new-features.1-0-0-M1]]
 == What's New in Spring Data R2DBC 1.0.0 M1
 
-* Initial R2DBC support through `DatabaseClient`
-* Initial Transaction support through `TransactionalDatabaseClient`
-* Initial R2DBC Repository Support through `R2dbcRepository`
-* Initial Dialect support for Postgres and Microsoft SQL Server
+* Initial R2DBC support through `DatabaseClient`.
+* Initial Transaction support through `TransactionalDatabaseClient`.
+* Initial R2DBC Repository Support through `R2dbcRepository`.
+* Initial Dialect support for Postgres and Microsoft SQL Server.
diff --git a/src/main/asciidoc/reference/r2dbc-repositories.adoc b/src/main/asciidoc/reference/r2dbc-repositories.adoc
index 7a46cf0..1c62b29 100644
--- a/src/main/asciidoc/reference/r2dbc-repositories.adoc
+++ b/src/main/asciidoc/reference/r2dbc-repositories.adoc
@@ -104,7 +104,7 @@ Defining such a query is a matter of declaring a method on the repository interf
 ----
 public interface PersonRepository extends ReactiveCrudRepository<Person, Long> {
 
-    @Query("SELECT * FROM person WHERE lastname = $1")
+    @Query("SELECT * FROM person WHERE lastname = :lastname")
     Flux<Person> findByLastname(String lastname);                      <1>
 
     @Query("SELECT firstname, lastname FROM person WHERE lastname = $1")
@@ -114,10 +114,10 @@ public interface PersonRepository extends ReactiveCrudRepository<Person, Long> {
 ----
 <1> The `findByLastname` method shows a query for all people with the given last name.
 The query is provided as R2DBC repositories do not support query derivation.
+<2> A query for a single `Person` entity projecting only `firstname` and `lastname` columns.
 The annotated query uses native bind markers, which are Postgres bind markers in this example.
-<4> A query for a single `Person` entity projecting only `firstname` and `lastname` columns.
 ====
 
 NOTE: R2DBC repositories do not support query derivation.
 
-NOTE: R2DBC repositories require native parameter bind markers that are bound by index.
+NOTE: R2DBC repositories bind parameters to placeholders by index.
diff --git a/src/main/asciidoc/reference/r2dbc.adoc b/src/main/asciidoc/reference/r2dbc.adoc
index b5d47ce..8fb038d 100644
--- a/src/main/asciidoc/reference/r2dbc.adoc
+++ b/src/main/asciidoc/reference/r2dbc.adoc
@@ -446,20 +446,61 @@ Parameter binding supports various binding strategies:
 * By Index using zero-based parameter indexes.
 * By Name using the placeholder name.
 
-The following example shows parameter binding for a PostgreSQL query:
+The following example shows parameter binding for a query:
 
 [source,java]
 ----
 db.execute()
-    .sql("INSERT INTO person (id, name, age) VALUES($1, $2, $3)")
-    .bind(0, "joe")
-    .bind(1, "Joe")
-    .bind(2, 34);
+    .sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)")
+    .bind("id", "joe")
+    .bind("name", "Joe")
+    .bind("age", 34);
 ----
 
-NOTE: If you are familiar with JDBC, then you're also familiar with `?` (question mark) bind markers.
+.R2DBC Native Bind Markers
+****
+R2DBC uses database-native bind markers that depend on the actual database.
+If you are familiar with JDBC, then you're also familiar with `?` (question mark) bind markers.
 JDBC drivers translate question mark bind markers to database-native markers as part of statement execution.
-Make sure to use the appropriate bind markers that are supported by your database as R2DBC requires database-native parameter bind markers.
+
+Postgres uses indexed markers (`$1`, `$2`), SQL Server uses named bind markers prefixed with `@` as its native bind marker syntax.
+Spring Data R2DBC leverages `Dialect` implementations 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.
+You can still use native bind markers if you prefer to do so.
+****
+
+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.
+Nested object arrays are expanded to allow usage of e.g. select lists.
+
+Consider the following query:
+
+[source,sql]
+----
+SELECT id, name, state FROM table WHERE (name, age) IN (('John', 35), ('Ann', 50))
+----
+
+This query can be parametrized and executed as:
+
+[source,java]
+----
+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);
+----
+
+NOTE: Usage of select lists is vendor-dependent.
+
+A simpler variant using `IN` predicates:
+
+[source,java]
+----
+db.execute()
+    .sql("SELECT id, name, state FROM table WHERE age IN (:ages)")
+    .bind("ages", Arrays.asList(35, 50));
+----
 
 [[r2dbc.datbaseclient.transactions]]
 === Transactions
@@ -478,14 +519,14 @@ TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(
 
 Flux<Void> completion = databaseClient.inTransaction(db -> {
 
-    return db.execute().sql("INSERT INTO person (id, name, age) VALUES($1, $2, $3)") //
-            .bind(0, "joe") //
-            .bind(1, "Joe") //
-            .bind(2, 34) //
+    return db.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)")
+            .bind("id", "joe")
+            .bind("name", "Joe")
+            .bind("age", 34)
             .fetch().rowsUpdated()
-            .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES($1, $2)")
-                    .bind(0, "joe")
-                    .bind(1, "Joe")
+            .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)")
+                    .bind("id", "joe")
+                    .bind("name", "Joe")
                     .fetch().rowsUpdated())
             .then();
 });
diff --git a/src/main/java/org/springframework/data/r2dbc/function/BindParameterSource.java b/src/main/java/org/springframework/data/r2dbc/function/BindParameterSource.java
new file mode 100644
index 0000000..c89eae8
--- /dev/null
+++ b/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;
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
index 5893c3c..242deda 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
+++ b/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 {
 
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
index 34c8e82..2b56479 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
+++ b/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;
 			};
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientBuilder.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientBuilder.java
index 5d1beab..b45df52 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientBuilder.java
+++ b/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");
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
index 72059aa..6a56cea 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
+++ b/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);
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClient.java
index 1edd16d..99f9312 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClient.java
+++ b/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
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClientBuilder.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClientBuilder.java
index c25e2c8..b9ab26a 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultTransactionalDatabaseClientBuilder.java
+++ b/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);
 	}
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/function/MapBindParameterSource.java b/src/main/java/org/springframework/data/r2dbc/function/MapBindParameterSource.java
new file mode 100644
index 0000000..09de5d8
--- /dev/null
+++ b/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();
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/NamedParameterExpander.java b/src/main/java/org/springframework/data/r2dbc/function/NamedParameterExpander.java
new file mode 100644
index 0000000..372e936
--- /dev/null
+++ b/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;
+				}
+			};
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/NamedParameterUtils.java b/src/main/java/org/springframework/data/r2dbc/function/NamedParameterUtils.java
new file mode 100644
index 0000000..3b15a90
--- /dev/null
+++ b/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;
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/ParsedSql.java b/src/main/java/org/springframework/data/r2dbc/function/ParsedSql.java
new file mode 100644
index 0000000..706931e
--- /dev/null
+++ b/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;
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
index 8670961..f715e6e 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
+++ b/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.
diff --git a/src/main/java/org/springframework/data/r2dbc/function/TransactionalDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/TransactionalDatabaseClient.java
index b596d8d..266fbb9 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/TransactionalDatabaseClient.java
+++ b/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);
 
diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
index 9a5f313..8700e4a 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
@@ -90,7 +90,9 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 	/**
 	 * Get a parameterized {@code INSERT INTO legoset} statement setting id, name, and manual values.
 	 */
-	protected abstract String getInsertIntoLegosetStatement();
+	protected String getInsertIntoLegosetStatement() {
+		return "INSERT INTO legoset (id, name, manual) VALUES(:id, :name, :manual)";
+	}
 
 	@Test // gh-2
 	public void executeInsert() {
@@ -98,9 +100,9 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
 
 		databaseClient.execute().sql(getInsertIntoLegosetStatement()) //
-				.bind(0, 42055) //
-				.bind(1, "SCHAUFELRADBAGGER") //
-				.bindNull(2, Integer.class) //
+				.bind("id", 42055) //
+				.bind("name", "SCHAUFELRADBAGGER") //
+				.bindNull("manual", Integer.class) //
 				.fetch().rowsUpdated() //
 				.as(StepVerifier::create) //
 				.expectNext(1) //
diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
index 52d267d..fc406a2 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
@@ -15,25 +15,27 @@
  */
 package org.springframework.data.r2dbc.function;
 
+import static org.assertj.core.api.Assertions.*;
+
 import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+
+import javax.sql.DataSource;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.transaction.NoTransactionException;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Hooks;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import javax.sql.DataSource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
-
-import static org.assertj.core.api.Assertions.*;
 
 /**
  * Abstract base class for integration tests for {@link TransactionalDatabaseClient}.
@@ -56,8 +58,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
 		jdbc = createJdbcTemplate(createDataSource());
 		try {
 			jdbc.execute("DROP TABLE legoset");
-		} catch (DataAccessException e) {
-		}
+		} catch (DataAccessException e) {}
 		jdbc.execute(getCreateTableStatement());
 		jdbc.execute("DELETE FROM legoset");
 	}
@@ -91,7 +92,9 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
 	/**
 	 * Get a parameterized {@code INSERT INTO legoset} statement setting id, name, and manual values.
 	 */
-	protected abstract String getInsertIntoLegosetStatement();
+	protected String getInsertIntoLegosetStatement() {
+		return "INSERT INTO legoset (id, name, manual) VALUES(:id, :name, :manual)";
+	}
 
 	/**
 	 * Get a statement that returns the current transactionId.
diff --git a/src/test/java/org/springframework/data/r2dbc/function/NamedParameterUtilsUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/NamedParameterUtilsUnitTests.java
new file mode 100644
index 0000000..4ab1acc
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/function/NamedParameterUtilsUnitTests.java
@@ -0,0 +1,295 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.Arrays;
+import java.util.HashMap;
+
+import org.junit.Test;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.dialect.SqlServerDialect;
+
+/**
+ * Unit tests for {@link NamedParameterUtils}.
+ *
+ * @author Mark Paluch
+ */
+public class NamedParameterUtilsUnitTests {
+
+	private final BindMarkersFactory BIND_MARKERS = PostgresDialect.INSTANCE.getBindMarkersFactory();
+
+	@Test // gh-23
+	public void shouldParseSql() {
+
+		String sql = "xxx :a yyyy :b :c :a zzzzz";
+		ParsedSql psql = NamedParameterUtils.parseSqlStatement(sql);
+		assertThat(psql.getParameterNames()).containsSequence("a", "b", "c", "a");
+		assertThat(psql.getTotalParameterCount()).isEqualTo(4);
+		assertThat(psql.getNamedParameterCount()).isEqualTo(3);
+
+		String sql2 = "xxx &a yyyy ? zzzzz";
+		ParsedSql psql2 = NamedParameterUtils.parseSqlStatement(sql2);
+		assertThat(psql2.getParameterNames().get(0)).isEqualTo("a");
+		assertThat(psql2.getTotalParameterCount()).isEqualTo(2);
+		assertThat(psql2.getNamedParameterCount()).isEqualTo(1);
+
+		String sql3 = "xxx &ä+:ö" + '\t' + ":ü%10 yyyy ? zzzzz";
+		ParsedSql psql3 = NamedParameterUtils.parseSqlStatement(sql3);
+		assertThat(psql3.getParameterNames()).containsSequence("ä", "ö", "ü");
+	}
+
+	@Test // gh-23
+	public void substituteNamedParameters() {
+
+		MapBindParameterSource namedParams = new MapBindParameterSource(new HashMap<>());
+		namedParams.addValue("a", "a").addValue("b", "b").addValue("c", "c");
+
+		BindableOperation 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",
+				SqlServerDialect.INSTANCE.getBindMarkersFactory(), namedParams);
+
+		assertThat(operation2.toQuery()).isEqualTo("xxx @P0_a @P1_b @P2_c");
+	}
+
+	@Test // gh-23
+	public void substituteObjectArray() {
+
+		MapBindParameterSource namedParams = new MapBindParameterSource(new HashMap<>());
+		namedParams.addValue("a",
+				Arrays.asList(new Object[] { "Walter", "Heisenberg" }, new Object[] { "Walt Jr.", "Flynn" }));
+
+		BindableOperation operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
+
+		assertThat(operation.toQuery()).isEqualTo("xxx ($1, $2), ($3, $4)");
+	}
+
+	@Test // gh-23
+	public void shouldBindObjectArray() {
+
+		MapBindParameterSource namedParams = new MapBindParameterSource(new HashMap<>());
+		namedParams.addValue("a",
+				Arrays.asList(new Object[] { "Walter", "Heisenberg" }, new Object[] { "Walt Jr.", "Flynn" }));
+
+		Statement<?> mockStatement = mock(Statement.class);
+
+		BindableOperation operation = NamedParameterUtils.substituteNamedParameters("xxx :a", BIND_MARKERS, namedParams);
+		operation.bind(mockStatement, "a", namedParams.getValue("a"));
+
+		verify(mockStatement).bind(0, "Walter");
+		verify(mockStatement).bind(1, "Heisenberg");
+		verify(mockStatement).bind(2, "Walt Jr.");
+		verify(mockStatement).bind(3, "Flynn");
+	}
+
+	@Test // gh-23
+	public void parseSqlContainingComments() {
+
+		String sql1 = "/*+ HINT */ xxx /* comment ? */ :a yyyy :b :c :a zzzzz -- :xx XX\n";
+
+		ParsedSql psql1 = NamedParameterUtils.parseSqlStatement(sql1);
+		assertThat(expand(psql1)).isEqualTo("/*+ HINT */ xxx /* comment ? */ $1 yyyy $2 $3 $4 zzzzz -- :xx XX\n");
+
+		MapBindParameterSource paramMap = new MapBindParameterSource(new HashMap<>());
+		paramMap.addValue("a", "a");
+		paramMap.addValue("b", "b");
+		paramMap.addValue("c", "c");
+
+		String sql2 = "/*+ HINT */ xxx /* comment ? */ :a yyyy :b :c :a zzzzz -- :xx XX";
+		ParsedSql psql2 = NamedParameterUtils.parseSqlStatement(sql2);
+		assertThat(expand(psql2)).isEqualTo("/*+ HINT */ xxx /* comment ? */ $1 yyyy $2 $3 $4 zzzzz -- :xx XX");
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithPostgresCasting() {
+
+		String expectedSql = "select 'first name' from artists where id = $1 and birth_date=$2::timestamp";
+		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,
+				new MapBindParameterSource());
+
+		assertThat(operation.toQuery()).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithPostgresContainedOperator() {
+
+		String expectedSql = "select 'first name' from artists where info->'stat'->'albums' = ?? $1 and '[\"1\",\"2\",\"3\"]'::jsonb ?? '4'";
+		String sql = "select 'first name' from artists where info->'stat'->'albums' = ?? :album and '[\"1\",\"2\",\"3\"]'::jsonb ?? '4'";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+
+		assertThat(parsedSql.getTotalParameterCount()).isEqualTo(1);
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithPostgresAnyArrayStringsExistsOperator() {
+
+		String expectedSql = "select '[\"3\", \"11\"]'::jsonb ?| '{1,3,11,12,17}'::text[]";
+		String sql = "select '[\"3\", \"11\"]'::jsonb ?| '{1,3,11,12,17}'::text[]";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+
+		assertThat(parsedSql.getTotalParameterCount()).isEqualTo(0);
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithPostgresAllArrayStringsExistsOperator() {
+
+		String expectedSql = "select '[\"3\", \"11\"]'::jsonb ?& '{1,3,11,12,17}'::text[] AND $1 = 'Back in Black'";
+		String sql = "select '[\"3\", \"11\"]'::jsonb ?& '{1,3,11,12,17}'::text[] AND :album = 'Back in Black'";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+		assertThat(parsedSql.getTotalParameterCount()).isEqualTo(1);
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithEscapedColon() {
+
+		String expectedSql = "select '0\\:0' as a, foo from bar where baz < DATE($1 23:59:59) and baz = $2";
+		String sql = "select '0\\:0' as a, foo from bar where baz < DATE(:p1 23\\:59\\:59) and baz = :p2";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+		assertThat(parsedSql.getParameterNames()).hasSize(2);
+		assertThat(parsedSql.getParameterNames().get(0)).isEqualTo("p1");
+		assertThat(parsedSql.getParameterNames().get(1)).isEqualTo("p2");
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithBracketDelimitedParameterNames() {
+
+		String expectedSql = "select foo from bar where baz = b$1$2z";
+		String sql = "select foo from bar where baz = b:{p1}:{p2}z";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+		assertThat(parsedSql.getParameterNames()).hasSize(2);
+		assertThat(parsedSql.getParameterNames().get(0)).isEqualTo("p1");
+		assertThat(parsedSql.getParameterNames().get(1)).isEqualTo("p2");
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithEmptyBracketsOrBracketsInQuotes() {
+
+		String expectedSql = "select foo from bar where baz = b:{}z";
+		String sql = "select foo from bar where baz = b:{}z";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+
+		assertThat(parsedSql.getParameterNames()).isEmpty();
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+
+		String expectedSql2 = "select foo from bar where baz = 'b:{p1}z'";
+		String sql2 = "select foo from bar where baz = 'b:{p1}z'";
+
+		ParsedSql parsedSql2 = NamedParameterUtils.parseSqlStatement(sql2);
+		assertThat(parsedSql2.getParameterNames()).isEmpty();
+		assertThat(expand(parsedSql2)).isEqualTo(expectedSql2);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithSingleLetterInBrackets() {
+		String expectedSql = "select foo from bar where baz = b$1z";
+		String sql = "select foo from bar where baz = b:{p}z";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(sql);
+		assertThat(parsedSql.getParameterNames()).hasSize(1);
+		assertThat(parsedSql.getParameterNames().get(0)).isEqualTo("p");
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithLogicalAnd() {
+
+		String expectedSql = "xxx & yyyy";
+
+		ParsedSql parsedSql = NamedParameterUtils.parseSqlStatement(expectedSql);
+
+		assertThat(expand(parsedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void substituteNamedParametersWithLogicalAnd() {
+
+		String expectedSql = "xxx & yyyy";
+
+		assertThat(expand(expectedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void variableAssignmentOperator() {
+
+		String expectedSql = "x := 1";
+
+		assertThat(expand(expectedSql)).isEqualTo(expectedSql);
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithQuotedSingleQuote() {
+
+		String sql = "SELECT ':foo'':doo', :xxx FROM DUAL";
+
+		ParsedSql psql = NamedParameterUtils.parseSqlStatement(sql);
+
+		assertThat(psql.getTotalParameterCount()).isEqualTo(1);
+		assertThat(psql.getParameterNames().get(0)).isEqualTo("xxx");
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithQuotesAndCommentBefore() {
+
+		String sql = "SELECT /*:doo*/':foo', :xxx FROM DUAL";
+
+		ParsedSql psql = NamedParameterUtils.parseSqlStatement(sql);
+
+		assertThat(psql.getTotalParameterCount()).isEqualTo(1);
+		assertThat(psql.getParameterNames().get(0)).isEqualTo("xxx");
+	}
+
+	@Test // gh-23
+	public void parseSqlStatementWithQuotesAndCommentAfter() {
+
+		String sql2 = "SELECT ':foo'/*:doo*/, :xxx FROM DUAL";
+
+		ParsedSql psql2 = NamedParameterUtils.parseSqlStatement(sql2);
+
+		assertThat(psql2.getTotalParameterCount()).isEqualTo(1);
+		assertThat(psql2.getParameterNames().get(0)).isEqualTo("xxx");
+	}
+
+	private String expand(ParsedSql sql) {
+		return NamedParameterUtils.substituteNamedParameters(sql, BIND_MARKERS, new MapBindParameterSource()).toQuery();
+	}
+
+	private String expand(String sql) {
+		return NamedParameterUtils.substituteNamedParameters(sql, BIND_MARKERS, new MapBindParameterSource()).toQuery();
+	}
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/function/PostgresDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/PostgresDatabaseClientIntegrationTests.java
index d5ec20b..a259800 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/PostgresDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/PostgresDatabaseClientIntegrationTests.java
@@ -48,11 +48,6 @@ public class PostgresDatabaseClientIntegrationTests extends AbstractDatabaseClie
 		return PostgresTestSupport.CREATE_TABLE_LEGOSET;
 	}
 
-	@Override
-	protected String getInsertIntoLegosetStatement() {
-		return PostgresTestSupport.INSERT_INTO_LEGOSET;
-	}
-
 	@Ignore("Adding RETURNING * lets Postgres report 0 affected rows.")
 	@Override
 	public void insert() {}
diff --git a/src/test/java/org/springframework/data/r2dbc/function/PostgresTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/PostgresTransactionalDatabaseClientIntegrationTests.java
index cfd8854..039df34 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/PostgresTransactionalDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/PostgresTransactionalDatabaseClientIntegrationTests.java
@@ -33,11 +33,6 @@ public class PostgresTransactionalDatabaseClientIntegrationTests
 		return PostgresTestSupport.CREATE_TABLE_LEGOSET;
 	}
 
-	@Override
-	protected String getInsertIntoLegosetStatement() {
-		return PostgresTestSupport.INSERT_INTO_LEGOSET;
-	}
-
 	@Override
 	protected String getCurrentTransactionIdStatement() {
 		return "SELECT txid_current();";
diff --git a/src/test/java/org/springframework/data/r2dbc/function/SqlServerDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/SqlServerDatabaseClientIntegrationTests.java
index 9e77bee..296fff0 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/SqlServerDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/SqlServerDatabaseClientIntegrationTests.java
@@ -46,9 +46,4 @@ public class SqlServerDatabaseClientIntegrationTests extends AbstractDatabaseCli
 	protected String getCreateTableStatement() {
 		return SqlServerTestSupport.CREATE_TABLE_LEGOSET;
 	}
-
-	@Override
-	protected String getInsertIntoLegosetStatement() {
-		return SqlServerTestSupport.INSERT_INTO_LEGOSET;
-	}
 }
diff --git a/src/test/java/org/springframework/data/r2dbc/repository/PostgresR2dbcRepositoryIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/repository/PostgresR2dbcRepositoryIntegrationTests.java
index 290160e..37ec7fc 100644
--- a/src/test/java/org/springframework/data/r2dbc/repository/PostgresR2dbcRepositoryIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/repository/PostgresR2dbcRepositoryIntegrationTests.java
@@ -88,7 +88,7 @@ public class PostgresR2dbcRepositoryIntegrationTests extends AbstractR2dbcReposi
 		Flux<Named> findAsProjection();
 
 		@Override
-		@Query("SELECT * FROM legoset WHERE manual = $1")
+		@Query("SELECT * FROM legoset WHERE manual = :manual")
 		Mono<LegoSet> findByManual(int manual);
 	}
 }
diff --git a/src/test/java/org/springframework/data/r2dbc/repository/SqlServerR2dbcRepositoryIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/repository/SqlServerR2dbcRepositoryIntegrationTests.java
index 8bea7d5..7a53754 100644
--- a/src/test/java/org/springframework/data/r2dbc/repository/SqlServerR2dbcRepositoryIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/repository/SqlServerR2dbcRepositoryIntegrationTests.java
@@ -93,7 +93,7 @@ public class SqlServerR2dbcRepositoryIntegrationTests extends AbstractR2dbcRepos
 		Flux<Named> findAsProjection();
 
 		@Override
-		@Query("SELECT * FROM legoset WHERE manual = @P0")
+		@Query("SELECT * FROM legoset WHERE manual = :manual")
 		Mono<LegoSet> findByManual(int manual);
 	}
 }