#73 - Introduce PreparedOperation.

We now encapsulate prepared operations from the StatementFactory within PreparedOperation that renders SQL and provides binding values.

StatementFactory supports SELECT/INSERT/UPDATE/DELETE statement creation considering Dialect-specific rendering.
StatementFactory replaces String-based statement methods in ReactiveDataAccessStrategy.

PreparedOperation<Update> operation = accessStrategy.getStatements().update(entity.getTableName(), binder -> {
	binder.bind("name", "updated value");
	binder.filterBy("id", SettableValue.from(42));
});

databaseClient.execute().sql(operation).then();

Original pull request: #82.

Author: mp911de

src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.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/DatabaseClient.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.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -336,9 +337,17 @@ <T> FetchSpec<T> exchange(String sql, BiFunction<Row, RowMetadata, T> mappingFun
 					logger.debug("Executing SQL statement [" + sql + "]");
 				}
 
+				if (sqlSupplier instanceof PreparedOperation<?>) {
+					return ((PreparedOperation<?>) sqlSupplier).bind(it.createStatement(sql));
+				}
+
 				BindableOperation operation = namedParameters.expand(sql, dataAccessStrategy.getBindMarkersFactory(),
 						new MapBindParameterSource(byName));
 
+				if (logger.isTraceEnabled()) {
+					logger.trace("Expanded SQL [" + operation.toQuery() + "]");
+				}
+
 				Statement statement = it.createStatement(operation.toQuery());
 
 				byName.forEach((name, o) -> {
@@ -366,6 +375,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);
@@ -376,6 +386,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));
 
@@ -384,6 +396,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));
@@ -396,6 +410,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);
@@ -404,6 +419,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);
@@ -881,20 +902,19 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 				throw new IllegalStateException("Insert fields is empty!");
 			}
 
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet());
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						byName.forEach(it::bind);
+					});
 
-			String sql = bindableInsert.toQuery();
+			String sql = operation.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));
-
-				return statement;
+				return operation.bind(it.createStatement(sql));
 			};
 
 			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
@@ -998,34 +1018,25 @@ 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);
-				}
-			});
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						outboundRow.forEach((k, v) -> {
 
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, columns);
+							if (v.hasValue()) {
+								it.bind(k, v);
+							}
+						});
+					});
 
-			String sql = bindableInsert.toQuery();
+			String sql = operation.toQuery();
 
 			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;
+				return operation.bind(it.createStatement(sql));
 			};
 
 			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());