Introduce PreparedOperation

pom.xml

@@ -7,7 +7,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-r2dbc</artifactId>
-	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<version>1.0.0.gh-73-SNAPSHOT</version>
 
 	<name>Spring Data R2DBC</name>
 	<description>Spring Data module for R2DBC.</description>

src/main/java/org/springframework/data/r2dbc/dialect/IndexedBindMarker.java

@@ -20,11 +20,11 @@
 /**
  * A single indexed bind marker.
  */
-class IndexedBindMarker implements BindMarker {
+public class IndexedBindMarker implements BindMarker {
 
 	private final String placeholder;
 
-	private int index;
+	private final int index;
 
 	IndexedBindMarker(String placeholder, int index) {
 		this.placeholder = placeholder;
@@ -57,4 +57,10 @@ public void bind(Statement statement, Object value) {
 	public void bindNull(Statement statement, Class<?> valueType) {
 		statement.bindNull(this.index, valueType);
 	}
+
+
+	public int getIndex() {
+		return index;
+	}
+
 }

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import io.r2dbc.spi.Statement;
+import lombok.RequiredArgsConstructor;
+
+import java.util.List;
+
+import org.springframework.data.r2dbc.domain.SettableValue;
+
+/**
+ * @author Jens Schauder
+ */
+public class Bindings {
+
+	private final List<SingleBinding> bindings;
+
+	public Bindings(List<SingleBinding> bindings) {
+		this.bindings = bindings;
+	}
+
+	public void apply(Statement statement) {
+		bindings.forEach(sb -> sb.bindTo(statement));
+	}
+
+	@RequiredArgsConstructor
+	public static abstract class SingleBinding<T> {
+
+		final T identifier;
+		final SettableValue value;
+
+		public abstract void bindTo(Statement statement);
+
+		public abstract boolean isIndexed();
+
+		public final boolean isNamed() {
+			return !isIndexed();
+		}
+	}
+
+
+	public static class IndexedSingleBinding extends SingleBinding<Integer> {
+
+		public IndexedSingleBinding(Integer identifier, SettableValue value) {
+			super(identifier, value);
+		}
+
+		@Override
+		public void bindTo(Statement statement) {
+
+			if (value.isEmpty()) {
+				statement.bindNull((int) identifier, value.getType());
+			} else {
+				statement.bind((int) identifier, value.getValue());
+			}
+		}
+
+		@Override
+		public boolean isIndexed() {
+			return true;
+		}
+	}
+
+	public static class NamedExpandedSingleBinding extends SingleBinding<String> {
+
+		private final BindableOperation operation;
+
+		public NamedExpandedSingleBinding(String identifier, SettableValue value, BindableOperation operation) {
+
+			super(identifier, value);
+
+			this.operation = operation;
+		}
+
+		@Override
+		public void bindTo(Statement statement) {
+
+			if (value != null) {
+				operation.bind(statement, identifier, value);
+			} else {
+				operation.bindNull(statement, identifier, value.getType());
+			}
+		}
+
+		@Override
+		public boolean isIndexed() {
+			return false;
+		}
+	}
+}

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

@@ -26,6 +26,7 @@
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
@@ -137,6 +138,9 @@ interface Builder {
 	 * 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.
+	 * <p>
+	 * Accepts {@link PreparedOperation} as SQL and binding {@link Supplier}.
+	 * </p>
 	 *
 	 * @see NamedParameterExpander
 	 * @see DatabaseClient.Builder#namedParameters(NamedParameterExpander)
@@ -156,6 +160,7 @@ interface SqlSpec {
 		 *
 		 * @param sqlSupplier must not be {@literal null}.
 		 * @return a new {@link GenericExecuteSpec}.
+		 * @see PreparedOperation
 		 */
 		GenericExecuteSpec sql(Supplier<String> sqlSupplier);
 	}

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

@@ -37,7 +37,6 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -49,6 +48,7 @@
 import org.reactivestreams.Publisher;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.UncategorizedR2dbcException;
@@ -57,6 +57,7 @@
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
+import org.springframework.data.relational.core.sql.Insert;
 import org.springframework.jdbc.core.SqlProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -331,32 +332,15 @@ protected String getSql() {
 
 		<T> FetchSpec<T> exchange(String sql, BiFunction<Row, RowMetadata, T> mappingFunction) {
 
-			Function<Connection, Statement> executeFunction = it -> {
+				PreparedOperation pop;
 
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				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());
-					}
-				});
-
-				bindByIndex(statement, byIndex);
-
-				return statement;
-			};
+			if (sqlSupplier instanceof PreparedOperation<?>) {
+				pop = ((PreparedOperation<?>) sqlSupplier);
+			} else {
+				pop = new ExpandedPreparedOperation(sql, namedParameters, dataAccessStrategy, byName, byIndex);
+			}
 
-			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(executeFunction.apply(it).execute());
+			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(pop.createBoundStatement(it).execute());
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
@@ -367,6 +351,7 @@ <T> FetchSpec<T> exchange(String sql, BiFunction<Row, RowMetadata, T> mappingFun
 
 		public ExecuteSpecSupport bind(int index, Object value) {
 
+			assertNotPreparedOperation();
 			Assert.notNull(value, () -> String.format("Value at index %d must not be null. Use bindNull(…) instead.", index));
 
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
@@ -377,6 +362,8 @@ public ExecuteSpecSupport bind(int index, Object value) {
 
 		public ExecuteSpecSupport bindNull(int index, Class<?> type) {
 
+			assertNotPreparedOperation();
+
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
 			byIndex.put(index, SettableValue.empty(type));
 
@@ -385,6 +372,8 @@ public ExecuteSpecSupport bindNull(int index, Class<?> type) {
 
 		public ExecuteSpecSupport bind(String name, Object value) {
 
+			assertNotPreparedOperation();
+
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 			Assert.notNull(value,
 					() -> String.format("Value for parameter %s must not be null. Use bindNull(…) instead.", name));
@@ -397,6 +386,7 @@ public ExecuteSpecSupport bind(String name, Object value) {
 
 		public ExecuteSpecSupport bindNull(String name, Class<?> type) {
 
+			assertNotPreparedOperation();
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
@@ -405,6 +395,12 @@ public ExecuteSpecSupport bindNull(String name, Class<?> type) {
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
 
+		private void assertNotPreparedOperation() {
+			if (sqlSupplier instanceof PreparedOperation<?>) {
+				throw new InvalidDataAccessApiUsageException("Cannot add bindings to a PreparedOperation");
+			}
+		}
+
 		protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
 				Supplier<String> sqlSupplier) {
 			return new ExecuteSpecSupport(byIndex, byName, sqlSupplier);
@@ -882,26 +878,15 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 				throw new IllegalStateException("Insert fields is empty!");
 			}
 
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet());
-
-			String sql = bindableInsert.toQuery();
-			Function<Connection, Statement> insertFunction = it -> {
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				Statement statement = it.createStatement(sql).returnGeneratedValues();
-
-				byName.forEach((k, v) -> bindableInsert.bind(statement, k, v));
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						byName.forEach(it::bind);
+					});
 
-				return statement;
-			};
-
-			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
+			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(operation.createBoundStatement(it).execute());
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
-					sql, //
+					operation.toQuery(), //
 					resultFunction, //
 					it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), //
 					mappingFunction);
@@ -999,40 +984,20 @@ private <MR> FetchSpec<MR> exchange(Object toInsert, BiFunction<Row, RowMetadata
 
 			OutboundRow outboundRow = dataAccessStrategy.getOutboundRow(toInsert);
 
-			Set<String> columns = new LinkedHashSet<>();
-
-			outboundRow.forEach((k, v) -> {
-
-				if (v.hasValue()) {
-					columns.add(k);
-				}
-			});
-
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, columns);
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						outboundRow.forEach((k, v) -> {
 
-			String sql = bindableInsert.toQuery();
+							if (v.hasValue()) {
+								it.bind(k, v);
+							}
+						});
+					});
 
-			Function<Connection, Statement> insertFunction = it -> {
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				Statement statement = it.createStatement(sql).returnGeneratedValues();
-
-				outboundRow.forEach((k, v) -> {
-					if (v.hasValue()) {
-						bindableInsert.bind(statement, k, v);
-					}
-				});
-
-				return statement;
-			};
-
-			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
+			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(operation.createBoundStatement(it).execute());
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
-					sql, //
+					operation.toQuery(), //
 					resultFunction, //
 					it -> resultFunction //
 							.apply(it) //