#73 - Polishing.

Original pull request: #82.

Author: schauder

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/DefaultDatabaseClient.java

@@ -331,40 +331,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 + "]");
-				}
-
-				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) -> {
-
-					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, //
@@ -907,20 +882,10 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 						byName.forEach(it::bind);
 					});
 
-			String sql = operation.toQuery();
-			Function<Connection, Statement> insertFunction = it -> {
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				return operation.bind(it.createStatement(sql));
-			};
-
-			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);
@@ -1028,21 +993,10 @@ private <MR> FetchSpec<MR> exchange(Object toInsert, BiFunction<Row, RowMetadata
 						});
 					});
 
-			String sql = operation.toQuery();
-
-			Function<Connection, Statement> insertFunction = it -> {
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				return operation.bind(it.createStatement(sql));
-			};
-
-			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) //