Spring/spring-data-r2dbc
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

#73 - Introduce PreparedOperation.

Browse Source 
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.
This commit is contained in:
Mark Paluch
2019-03-18 14:52:30 +01:00
parent 72ffccbfbc
commit ef2d885b1e
12 changed files with 991 additions and 585 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.java
View File

@@ -1,32 +0,0 @@
package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Statement;
/**
* Extension to {@link BindableOperation} for operations that allow parameter substitution for a single {@code id}
* column that accepts either a single value or multiple values, depending on the underlying operation.
*
* @author Mark Paluch
* @see Statement#bind
* @see Statement#bindNull
*/
public interface BindIdOperation extends BindableOperation {
/**
* Bind the given {@code value} to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param value the actual value. Must not be {@literal null}.
* @see Statement#bind
*/
void bindId(Statement statement, Object value);
/**
* Bind the given {@code values} to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param values the actual values.
* @see Statement#bind
*/
void bindIds(Statement statement, Iterable<? extends Object> values);
}

5
src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
View File

@@ -26,6 +26,7 @@ import java.util.function.Consumer;
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 @@ public interface DatabaseClient {
* 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 @@ public interface DatabaseClient {
*
* @param sqlSupplier must not be {@literal null}.
* @return a new {@link GenericExecuteSpec}.
* @see PreparedOperation
*/
GenericExecuteSpec sql(Supplier<String> sqlSupplier);
}

65
src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
View File

@@ -37,7 +37,6 @@ import java.util.LinkedHashMap;
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.apache.commons.logging.LogFactory;
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.domain.SettableValue;
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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,18 +1018,17 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
OutboundRow outboundRow = dataAccessStrategy.getOutboundRow(toInsert);
Set<String> columns = new LinkedHashSet<>();
PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
it -> {
outboundRow.forEach((k, v) -> {
outboundRow.forEach((k, v) -> {
if (v.hasValue()) {
it.bind(k, v);
}
});
});
if (v.hasValue()) {
columns.add(k);
}
});
BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, columns);
String sql = bindableInsert.toQuery();
String sql = operation.toQuery();
Function<Connection, Statement> insertFunction = it -> {
@@ -1017,15 +1036,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
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());

356
src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
View File

@@ -17,14 +17,11 @@ package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.OptionalLong;
import java.util.Set;
import java.util.function.BiFunction;
@@ -37,8 +34,6 @@ import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Order;
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.domain.OutboundRow;
@@ -53,13 +48,17 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentEnti
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.sql.Expression;
import org.springframework.data.relational.core.sql.OrderByField;
import org.springframework.data.relational.core.sql.Select;
import org.springframework.data.relational.core.sql.SelectBuilder.SelectFromAndOrderBy;
import org.springframework.data.relational.core.sql.StatementBuilder;
import org.springframework.data.relational.core.sql.Table;
import org.springframework.data.relational.core.sql.render.NamingStrategies;
import org.springframework.data.relational.core.sql.render.RenderContext;
import org.springframework.data.relational.core.sql.render.RenderNamingStrategy;
import org.springframework.data.relational.core.sql.render.SelectRenderContext;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
* Default {@link ReactiveDataAccessStrategy} implementation.
@@ -71,6 +70,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
private final Dialect dialect;
private final R2dbcConverter converter;
private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
private final StatementFactory statements;
/**
* Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect}.
@@ -118,6 +118,30 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
this.mappingContext = (MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty>) this.converter
.getMappingContext();
this.dialect = dialect;
RenderContext renderContext = new RenderContext() {
@Override
public RenderNamingStrategy getNamingStrategy() {
return NamingStrategies.asIs();
}
@Override
public SelectRenderContext getSelect() {
return new SelectRenderContext() {
@Override
public Function<Select, ? extends CharSequence> afterSelectList() {
return it -> "";
}
@Override
public Function<Select, ? extends CharSequence> afterOrderBy(boolean hasOrderBy) {
return it -> "";
}
};
}
};
this.statements = new DefaultStatementFactory(this.dialect, renderContext);
}
/*
@@ -218,7 +242,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getRowMapper(java.lang.Class)
*/
@SuppressWarnings("unchecked")
@Override
public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
return new EntityRowMapper<>(typeToRead, converter);
@@ -233,6 +256,15 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return getRequiredPersistentEntity(type).getTableName();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getStatements()
*/
@Override
public StatementFactory getStatements() {
return this.statements;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getBindMarkersFactory()
@@ -251,15 +283,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return mappingContext.getPersistentEntity(typeToRead);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#insertAndReturnGeneratedKeys(java.lang.String, java.util.Set)
*/
@Override
public BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns) {
return new DefaultBindableInsert(dialect.getBindMarkersFactory().create(), table, columns);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#select(java.lang.String, java.util.Set, org.springframework.data.domain.Sort, org.springframework.data.domain.Pageable)
@@ -290,6 +313,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
offset = OptionalLong.of(page.getOffset());
}
// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
return StatementRenderUtil.render(selectBuilder.build(), limit, offset, this.dialect);
}
@@ -310,304 +334,4 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return fields;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#updateById(java.lang.String, java.util.Set, java.lang.String)
*/
@Override
public BindIdOperation updateById(String table, Set<String> columns, String idColumn) {
return new DefaultBindableUpdate(dialect.getBindMarkersFactory().create(), table, columns, idColumn);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteById(java.lang.String, java.lang.String)
*/
@Override
public BindIdOperation deleteById(String table, String idColumn) {
return new DefaultBindIdOperation(dialect.getBindMarkersFactory().create(),
marker -> String.format("DELETE FROM %s WHERE %s = %s", table, idColumn, marker.getPlaceholder()));
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteByIdIn(java.lang.String, java.lang.String)
*/
@Override
public BindIdOperation deleteByIdIn(String table, String idColumn) {
String query = String.format("DELETE FROM %s", table);
return new DefaultBindIdIn(dialect.getBindMarkersFactory().create(), query, idColumn);
}
/**
* Default {@link BindableOperation} implementation for a {@code INSERT} operation.
*/
static class DefaultBindableInsert implements BindableOperation {
private final Map<String, BindMarker> markers = new LinkedHashMap<>();
private final String query;
DefaultBindableInsert(BindMarkers bindMarkers, String table, Collection<String> columns) {
StringBuilder builder = new StringBuilder();
List<String> placeholders = new ArrayList<>(columns.size());
for (String column : columns) {
BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
placeholders.add(marker.getPlaceholder());
}
String columnsString = StringUtils.collectionToDelimitedString(columns, ", ");
String placeholdersString = StringUtils.collectionToDelimitedString(placeholders, ", ");
builder.append("INSERT INTO ").append(table).append(" (").append(columnsString).append(")").append(" VALUES(")
.append(placeholdersString).append(")");
this.query = builder.toString();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement statement, String identifier, Object value) {
markers.get(identifier).bind(statement, value);
}
/*
* (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) {
markers.get(identifier).bindNull(statement, valueType);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code UPDATE} operation using a single key.
*/
static class DefaultBindableUpdate implements BindIdOperation {
private final Map<String, BindMarker> markers = new LinkedHashMap<>();
private final BindMarker idMarker;
private final String query;
DefaultBindableUpdate(BindMarkers bindMarkers, String tableName, Set<String> columns, String idColumnName) {
this.idMarker = bindMarkers.next();
StringBuilder setClause = new StringBuilder();
for (String column : columns) {
BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
if (setClause.length() != 0) {
setClause.append(", ");
}
setClause.append(column).append(" = ").append(marker.getPlaceholder());
}
this.query = String.format("UPDATE %s SET %s WHERE %s = %s", tableName, setClause, idColumnName,
idMarker.getPlaceholder());
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement statement, String identifier, Object value) {
markers.get(identifier).bind(statement, value);
}
/*
* (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) {
markers.get(identifier).bindNull(statement, valueType);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement statement, Object value) {
idMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement statement, Iterable<?> values) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code SELECT} or {@code DELETE} operation using a single key
* in the {@code WHERE} predicate.
*/
static class DefaultBindIdOperation implements BindIdOperation {
private final BindMarker idMarker;
private final String query;
DefaultBindIdOperation(BindMarkers bindMarkers, Function<BindMarker, String> queryFunction) {
this.idMarker = bindMarkers.next();
this.query = queryFunction.apply(this.idMarker);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement statement, String identifier, Object value) {
throw new UnsupportedOperationException();
}
/*
* (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) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement statement, Object value) {
idMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement statement, Iterable<?> values) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code SELECT … WHERE id IN (…)} or
* {@code DELETE … WHERE id IN (…)}.
*/
static class DefaultBindIdIn implements BindIdOperation {
private final List<String> markers = new ArrayList<>();
private final BindMarkers bindMarkers;
private final String baseQuery;
private final String idColumnName;
DefaultBindIdIn(BindMarkers bindMarkers, String baseQuery, String idColumnName) {
this.bindMarkers = bindMarkers;
this.baseQuery = baseQuery;
this.idColumnName = idColumnName;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement statement, String identifier, Object value) {
throw new UnsupportedOperationException();
}
/*
* (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) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement statement, Object value) {
BindMarker bindMarker = bindMarkers.next();
markers.add(bindMarker.getPlaceholder());
bindMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement statement, Iterable<?> values) {
for (Object value : values) {
bindId(statement, value);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
if (this.markers.isEmpty()) {
throw new UnsupportedOperationException();
}
String in = StringUtils.collectionToDelimitedString(this.markers, ", ");
return String.format("%s WHERE %s IN (%s)", this.baseQuery, this.idColumnName, in);
}
}
}

474
src/main/java/org/springframework/data/r2dbc/function/DefaultStatementFactory.java Normal file
View File

@@ -0,0 +1,474 @@
/*
* 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.Getter;
import lombok.RequiredArgsConstructor;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
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.Dialect;
import org.springframework.data.r2dbc.domain.SettableValue;
import org.springframework.data.relational.core.sql.AssignValue;
import org.springframework.data.relational.core.sql.Assignment;
import org.springframework.data.relational.core.sql.Column;
import org.springframework.data.relational.core.sql.Condition;
import org.springframework.data.relational.core.sql.Delete;
import org.springframework.data.relational.core.sql.DeleteBuilder;
import org.springframework.data.relational.core.sql.Expression;
import org.springframework.data.relational.core.sql.Insert;
import org.springframework.data.relational.core.sql.SQL;
import org.springframework.data.relational.core.sql.Select;
import org.springframework.data.relational.core.sql.SelectBuilder;
import org.springframework.data.relational.core.sql.StatementBuilder;
import org.springframework.data.relational.core.sql.Table;
import org.springframework.data.relational.core.sql.Update;
import org.springframework.data.relational.core.sql.UpdateBuilder;
import org.springframework.data.relational.core.sql.render.RenderContext;
import org.springframework.data.relational.core.sql.render.SqlRenderer;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Default {@link StatementFactory} implementation.
*
* @author Mark Paluch
*/
@RequiredArgsConstructor
class DefaultStatementFactory implements StatementFactory {
private final Dialect dialect;
private final RenderContext renderContext;
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory#select(java.lang.String, java.util.Collection, java.util.function.Consumer)
*/
@Override
public PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
Consumer<StatementBinderBuilder> binderConsumer) {
Assert.hasText(tableName, "Table must not be empty");
Assert.notEmpty(columnNames, "Columns must not be empty");
Assert.notNull(binderConsumer, "Binder Consumer must not be null");
DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
@Override
public void bind(String identifier, SettableValue settable) {
throw new InvalidDataAccessApiUsageException("Binding for SELECT not supported. Use filterBy(…)");
}
};
binderConsumer.accept(binderBuilder);
return withDialect((dialect, renderContext) -> {
Table table = Table.create(tableName);
List<Column> columns = table.columns(columnNames);
SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
Binding binding = binderBuilder.build(table, bindMarkers);
Select select;
if (binding.hasCondition()) {
select = selectBuilder.where(binding.getCondition()).build();
} else {
select = selectBuilder.build();
}
return new DefaultPreparedOperation<>(select, renderContext, binding);
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory#insert(java.lang.String, java.util.Collection, java.util.function.Consumer)
*/
@Override
public PreparedOperation<Insert> insert(String tableName, Collection<String> generatedKeysNames,
Consumer<StatementBinderBuilder> binderConsumer) {
Assert.hasText(tableName, "Table must not be empty");
Assert.notNull(generatedKeysNames, "Generated key names must not be null");
Assert.notNull(binderConsumer, "Binder Consumer must not be null");
DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
@Override
public void filterBy(String identifier, SettableValue settable) {
throw new InvalidDataAccessApiUsageException("Filter-Binding for INSERT not supported. Use bind(…)");
}
};
binderConsumer.accept(binderBuilder);
return withDialect((dialect, renderContext) -> {
BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
Table table = Table.create(tableName);
Map<BindMarker, SettableValue> expressionBindings = new LinkedHashMap<>();
List<Expression> expressions = new ArrayList<>();
binderBuilder.forEachBinding((column, settableValue) -> {
BindMarker bindMarker = bindMarkers.next(column);
expressions.add(SQL.bindMarker(bindMarker.getPlaceholder()));
expressionBindings.put(bindMarker, settableValue);
});
if (expressions.isEmpty()) {
throw new IllegalStateException("INSERT contains no value expressions");
}
Binding binding = binderBuilder.build(table, bindMarkers).withBindings(expressionBindings);
Insert insert = StatementBuilder.insert().into(table).columns(table.columns(binderBuilder.bindings.keySet()))
.values(expressions).build();
return new DefaultPreparedOperation<Insert>(insert, renderContext, binding) {
@Override
public Statement bind(Statement to) {
return super.bind(to).returnGeneratedValues(generatedKeysNames.toArray(new String[0]));
}
};
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory#update(java.lang.String, java.util.function.Consumer)
*/
@Override
public PreparedOperation<Update> update(String tableName, Consumer<StatementBinderBuilder> binderConsumer) {
Assert.hasText(tableName, "Table must not be empty");
Assert.notNull(binderConsumer, "Binder Consumer must not be null");
DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder();
binderConsumer.accept(binderBuilder);
return withDialect((dialect, renderContext) -> {
BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
Table table = Table.create(tableName);
Map<BindMarker, SettableValue> assignmentBindings = new LinkedHashMap<>();
List<Assignment> assignments = new ArrayList<>();
binderBuilder.forEachBinding((column, settableValue) -> {
BindMarker bindMarker = bindMarkers.next(column);
AssignValue assignment = table.column(column).set(SQL.bindMarker(bindMarker.getPlaceholder()));
assignments.add(assignment);
assignmentBindings.put(bindMarker, settableValue);
});
if (assignments.isEmpty()) {
throw new IllegalStateException("UPDATE contains no assignments");
}
UpdateBuilder.UpdateWhere updateBuilder = StatementBuilder.update(table).set(assignments);
Binding binding = binderBuilder.build(table, bindMarkers).withBindings(assignmentBindings);
Update update;
if (binding.hasCondition()) {
update = updateBuilder.where(binding.getCondition()).build();
} else {
update = updateBuilder.build();
}
return new DefaultPreparedOperation<>(update, renderContext, binding);
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory#delete(java.lang.String, java.util.function.Consumer)
*/
@Override
public PreparedOperation<Delete> delete(String tableName, Consumer<StatementBinderBuilder> binderConsumer) {
Assert.hasText(tableName, "Table must not be empty");
Assert.notNull(binderConsumer, "Binder Consumer must not be null");
DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
@Override
public void bind(String identifier, SettableValue settable) {
throw new InvalidDataAccessApiUsageException("Binding for DELETE not supported. Use filterBy(…)");
}
};
binderConsumer.accept(binderBuilder);
return withDialect((dialect, renderContext) -> {
Table table = Table.create(tableName);
DeleteBuilder.DeleteWhere deleteBuilder = StatementBuilder.delete().from(table);
BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
Binding binding = binderBuilder.build(table, bindMarkers);
Delete delete;
if (binding.hasCondition()) {
delete = deleteBuilder.where(binding.getCondition()).build();
} else {
delete = deleteBuilder.build();
}
return new DefaultPreparedOperation<>(delete, renderContext, binding);
});
}
private <T> T withDialect(BiFunction<Dialect, RenderContext, T> action) {
Assert.notNull(action, "Action must not be null");
return action.apply(this.dialect, this.renderContext);
}
/**
* Default {@link StatementBinderBuilder} implementation.
*/
static class DefaultBinderBuilder implements StatementBinderBuilder {
final Map<String, SettableValue> filters = new LinkedHashMap<>();
final Map<String, SettableValue> bindings = new LinkedHashMap<>();
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory.StatementBinderBuilder#filterBy(java.lang.String, org.springframework.data.r2dbc.domain.SettableValue)
*/
@Override
public void filterBy(String identifier, SettableValue settable) {
Assert.hasText(identifier, "FilterBy identifier must not be empty");
Assert.notNull(settable, "SettableValue for Filter must not be null");
this.filters.put(identifier, settable);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.StatementFactory.StatementBinderBuilder#bind(java.lang.String, org.springframework.data.r2dbc.domain.SettableValue)
*/
@Override
public void bind(String identifier, SettableValue settable) {
Assert.hasText(identifier, "Bind value identifier must not be empty");
Assert.notNull(settable, "SettableValue must not be null");
this.bindings.put(identifier, settable);
}
/**
* Call {@link BiConsumer} for each filter binding.
*
* @param consumer the consumer to notify.
*/
void forEachFilter(BiConsumer<String, SettableValue> consumer) {
filters.forEach(consumer);
}
/**
* Call {@link BiConsumer} for each value binding.
*
* @param consumer the consumer to notify.
*/
void forEachBinding(BiConsumer<String, SettableValue> consumer) {
bindings.forEach(consumer);
}
Binding build(Table table, BindMarkers bindMarkers) {
Map<BindMarker, Object> values = new LinkedHashMap<>();
Map<BindMarker, SettableValue> nulls = new LinkedHashMap<>();
AtomicReference<Condition> conditionRef = new AtomicReference<>();
forEachFilter((k, v) -> {
Condition condition = toCondition(bindMarkers, table.column(k), v, values, nulls);
Condition current = conditionRef.get();
if (current == null) {
current = condition;
} else {
current = current.and(condition);
}
conditionRef.set(current);
});
return new Binding(values, nulls, conditionRef.get());
}
private static Condition toCondition(BindMarkers bindMarkers, Column column, SettableValue value,
Map<BindMarker, Object> values, Map<BindMarker, SettableValue> nulls) {
if (value.hasValue()) {
Object bindValue = value.getValue();
if (bindValue instanceof Iterable) {
Iterable<?> iterable = (Iterable<?>) bindValue;
List<Expression> expressions = new ArrayList<>();
for (Object o : iterable) {
BindMarker marker = bindMarkers.next(column.getName());
if (o == null) {
nulls.put(marker, value);
} else {
values.put(marker, o);
}
expressions.add(SQL.bindMarker(marker.getPlaceholder()));
}
return column.in(expressions.toArray(new Expression[0]));
}
BindMarker marker = bindMarkers.next(column.getName());
values.put(marker, value.getValue());
return column.isEqualTo(SQL.bindMarker(marker.getPlaceholder()));
}
return column.isNull();
}
}
/**
* Value object holding value and {@code NULL} bindings.
*
* @see SettableValue
*/
@RequiredArgsConstructor
@Getter
static class Binding {
private final Map<BindMarker, Object> values;
private final Map<BindMarker, SettableValue> nulls;
private final @Nullable Condition condition;
boolean hasCondition() {
return condition != null;
}
/**
* Append bindings.
*
* @param assignmentBindings
* @return
*/
Binding withBindings(Map<BindMarker, SettableValue> assignmentBindings) {
assignmentBindings.forEach(((bindMarker, settableValue) -> {
if (settableValue.isEmpty()) {
nulls.put(bindMarker, settableValue);
} else {
values.put(bindMarker, settableValue.getValue());
}
}));
return this;
}
/**
* Apply bindings to a {@link Statement}.
*
* @param to
*/
void apply(Statement to) {
values.forEach((marker, value) -> marker.bind(to, value));
nulls.forEach((marker, value) -> marker.bindNull(to, value.getType()));
}
}
/**
* Default implementation of {@link PreparedOperation}.
*
* @param <T>
*/
@RequiredArgsConstructor
static class DefaultPreparedOperation<T> implements PreparedOperation<T> {
private final T source;
private final RenderContext renderContext;
private final Binding binding;
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.PreparedOperation#getSource()
*/
@Override
public T getSource() {
return this.source;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
SqlRenderer sqlRenderer = SqlRenderer.create(renderContext);
if (this.source instanceof Select) {
return sqlRenderer.render((Select) this.source);
}
if (this.source instanceof Insert) {
return sqlRenderer.render((Insert) this.source);
}
if (this.source instanceof Update) {
return sqlRenderer.render((Update) this.source);
}
if (this.source instanceof Delete) {
return sqlRenderer.render((Delete) this.source);
}
throw new IllegalStateException("Cannot render " + this.getSource());
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.PreparedOperation#bind(io.r2dbc.spi.Statement)
*/
@Override
public Statement bind(Statement to) {
binding.apply(to);
return to;
}
}
}

48
src/main/java/org/springframework/data/r2dbc/function/PreparedOperation.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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 java.util.function.Supplier;
/**
* Extension to {@link QueryOperation} for a prepared SQL query {@link Supplier} with bound parameters. Contains
* parameter bindings that can be {@link #bind(Statement)} bound to a {@link Statement}.
* <p>
* Can be executed with {@link DatabaseClient}.
* </p>
*
* @param <T> underlying operation source.
* @author Mark Paluch
* @see DatabaseClient
* @see DatabaseClient.SqlSpec#sql(Supplier)
*/
public interface PreparedOperation<T> extends QueryOperation {
/**
* @return the query source, such as a statement/criteria object.
*/
T getSource();
/**
* Bind query parameters to a {@link Statement}.
*
* @param to the target statement to bind parameters to.
* @return the bound statement.
*/
Statement bind(Statement to);
}

43
src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
View File

@@ -17,7 +17,6 @@ package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import java.util.List;
import java.util.Set;
@@ -72,6 +71,8 @@ public interface ReactiveDataAccessStrategy {
*/
String getTableName(Class<?> type);
StatementFactory getStatements();
/**
* Returns the configured {@link BindMarkersFactory} to create native parameter placeholder markers.
*
@@ -91,15 +92,6 @@ public interface ReactiveDataAccessStrategy {
// Subject to be moved into a SQL creation DSL.
// -------------------------------------------------------------------------
/**
* Create an {@code INSERT} operation for the given {@code table} to insert {@code columns}.
*
* @param table the table to insert data to.
* @param columns column names that will be bound.
* @return the {@link BindableOperation} representing the {@code INSERT} statement.
*/
BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns);
/**
* Create a {@code SELECT … ORDER BY … LIMIT …} operation for the given {@code table} using {@code columns} to
* project.
@@ -111,35 +103,4 @@ public interface ReactiveDataAccessStrategy {
* @return
*/
String select(String table, Set<String> columns, Sort sort, Pageable page);
/**
* Create a {@code UPDATE … SET … WHERE id = ?} operation for the given {@code table} updating {@code columns} and
* {@code idColumn}.
*
* @param table the table to insert data to.
* @param columns columns to update.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation updateById(String table, Set<String> columns, String idColumn);
/**
* Create a {@code DELETE … WHERE id = ?} operation for the given {@code table} and {@code idColumn}.
*
* @param table the table to insert data to.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation deleteById(String table, String idColumn);
/**
* Create a {@code DELETE … WHERE id IN (?)} operation for the given {@code table} and {@code idColumn}. The actual
* {@link BindableOperation#toQuery() query} string depends on {@link BindIdOperation#bindIds(Statement, Iterable)
* bound parameters}.
*
* @param table the table to insert data to.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation deleteByIdIn(String table, String idColumn);
}

102
src/main/java/org/springframework/data/r2dbc/function/StatementFactory.java Normal file
View File

@@ -0,0 +1,102 @@
/*
* 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 java.util.Collection;
import java.util.function.Consumer;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.domain.SettableValue;
import org.springframework.data.relational.core.sql.Delete;
import org.springframework.data.relational.core.sql.Insert;
import org.springframework.data.relational.core.sql.Select;
import org.springframework.data.relational.core.sql.Update;
/**
* Interface declaring statement methods that are commonly used for {@code SELECT/INSERT/UPDATE/DELETE} operations.
* These methods consider {@link Dialect} specifics and accept bind parameters with values.
*
* @author Mark Paluch
* @see PreparedOperation
*/
public interface StatementFactory {
/**
* Creates a {@link Select} statement.
*
* @param tableName must not be {@literal null} or empty.
* @param columnNames the columns to project, must not be {@literal null} or empty.
* @param binderConsumer customizer for bindings. Supports only
* {@link StatementBinderBuilder#filterBy(String, SettableValue)} bindings.
* @return the {@link PreparedOperation} to select the given columns.
*/
PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
Consumer<StatementBinderBuilder> binderConsumer);
/**
* Creates a {@link Insert} statement.
*
* @param tableName must not be {@literal null} or empty.
* @param generatedKeysNames must not be {@literal null}.
* @param binderConsumer customizer for bindings. Supports only
* {@link StatementBinderBuilder#bind(String, SettableValue)} bindings.
* @return the {@link PreparedOperation} to update values in {@code tableName} assigning bound values.
*/
PreparedOperation<Insert> insert(String tableName, Collection<String> generatedKeysNames,
Consumer<StatementBinderBuilder> binderConsumer);
/**
* Creates a {@link Update} statement.
*
* @param tableName must not be {@literal null} or empty.
* @param binderConsumer customizer for bindings.
* @return the {@link PreparedOperation} to update values in {@code tableName} assigning bound values.
*/
PreparedOperation<Update> update(String tableName, Consumer<StatementBinderBuilder> binderConsumer);
/**
* Creates a {@link Delete} statement.
*
* @param tableName must not be {@literal null} or empty.
* @param binderConsumer customizer for bindings. Supports only
* {@link StatementBinderBuilder#filterBy(String, SettableValue)} bindings.
* @return the {@link PreparedOperation} to delete rows from {@code tableName}.
*/
PreparedOperation<Delete> delete(String tableName, Consumer<StatementBinderBuilder> binderConsumer);
/**
* Binder to specify parameter bindings by name. Bindings match to equals comparisons.
*/
interface StatementBinderBuilder {
/**
* Bind the given Id {@code value} to this builder using the underlying binding strategy to express a filter
* condition. {@link Collection} type values translate to {@code IN} matching.
*
* @param identifier named identifier that is considered by the underlying binding strategy.
* @param settable must not be {@literal null}. Use {@link SettableValue#empty(Class)} for {@code NULL} values.
*/
void filterBy(String identifier, SettableValue settable);
/**
* Bind the given {@code value} to this builder using the underlying binding strategy.
*
* @param identifier named identifier that is considered by the underlying binding strategy.
* @param settable must not be {@literal null}. Use {@link SettableValue#empty(Class)} for {@code NULL} values.
*/
void bind(String identifier, SettableValue settable);
}
}

100
src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
View File

@@ -20,29 +20,25 @@ import lombok.RequiredArgsConstructor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.reactivestreams.Publisher;
import org.springframework.data.r2dbc.dialect.BindMarker;
import org.springframework.data.r2dbc.dialect.BindMarkers;
import org.springframework.data.r2dbc.domain.SettableValue;
import org.springframework.data.r2dbc.function.BindIdOperation;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
import org.springframework.data.r2dbc.function.PreparedOperation;
import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.function.StatementFactory;
import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
import org.springframework.data.relational.core.sql.Conditions;
import org.springframework.data.relational.core.sql.Expression;
import org.springframework.data.relational.core.sql.Delete;
import org.springframework.data.relational.core.sql.Functions;
import org.springframework.data.relational.core.sql.SQL;
import org.springframework.data.relational.core.sql.Select;
import org.springframework.data.relational.core.sql.StatementBuilder;
import org.springframework.data.relational.core.sql.Table;
import org.springframework.data.relational.core.sql.Update;
import org.springframework.data.relational.core.sql.render.SqlRenderer;
import org.springframework.data.relational.repository.query.RelationalEntityInformation;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
@@ -83,15 +79,13 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Map<String, SettableValue> columns = accessStrategy.getOutboundRow(objectToSave);
columns.remove(getIdColumnName()); // do not update the Id column.
String idColumnName = getIdColumnName();
BindIdOperation update = accessStrategy.updateById(entity.getTableName(), columns.keySet(), idColumnName);
GenericExecuteSpec exec = databaseClient.execute().sql(update);
PreparedOperation<Update> operation = accessStrategy.getStatements().update(entity.getTableName(), binder -> {
columns.forEach(binder::bind);
binder.filterBy(idColumnName, SettableValue.from(id));
});
BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(exec);
columns.forEach((k, v) -> update.bind(wrapper, k, v));
update.bindId(wrapper, id);
return wrapper.getBoundOperation().as(entity.getJavaType()) //
return databaseClient.execute().sql(operation).as(entity.getJavaType()) //
.then() //
.thenReturn(objectToSave);
}
@@ -129,18 +123,14 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
String idColumnName = getIdColumnName();
BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
BindMarker bindMarker = bindMarkers.next("id");
StatementFactory statements;
Table table = Table.create(entity.getTableName());
Select select = StatementBuilder //
.select(table.columns(columns)) //
.from(table) //
.where(Conditions.isEqual(table.column(idColumnName), SQL.bindMarker(bindMarker.getPlaceholder()))) //
.build();
PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(), columns,
binder -> {
binder.filterBy(idColumnName, SettableValue.from(id));
});
return databaseClient.execute().sql(SqlRenderer.toString(select)) //
.bind(0, id) //
return databaseClient.execute().sql(operation) //
.as(entity.getJavaType()) //
.fetch() //
.one();
@@ -164,18 +154,12 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
String idColumnName = getIdColumnName();
BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
BindMarker bindMarker = bindMarkers.next("id");
PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(),
Collections.singleton(idColumnName), binder -> {
binder.filterBy(idColumnName, SettableValue.from(id));
});
Table table = Table.create(entity.getTableName());
Select select = StatementBuilder //
.select(table.column(idColumnName)) //
.from(table) //
.where(Conditions.isEqual(table.column(idColumnName), SQL.bindMarker(bindMarker.getPlaceholder()))) //
.build();
return databaseClient.execute().sql(SqlRenderer.toString(select)) //
.bind(0, id) //
return databaseClient.execute().sql(operation) //
.map((r, md) -> r) //
.first() //
.hasElement();
@@ -225,25 +209,12 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
String idColumnName = getIdColumnName();
BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(), columns,
binder -> {
binder.filterBy(idColumnName, SettableValue.from(ids));
});
List<Expression> markers = new ArrayList<>();
for (int i = 0; i < ids.size(); i++) {
markers.add(SQL.bindMarker(bindMarkers.next("id").getPlaceholder()));
}
Table table = Table.create(entity.getTableName());
Select select = StatementBuilder.select(table.columns(columns)).from(table)
.where(Conditions.in(table.column(idColumnName), markers)).build();
GenericExecuteSpec executeSpec = databaseClient.execute().sql(SqlRenderer.toString(select));
for (int i = 0; i < ids.size(); i++) {
executeSpec = executeSpec.bind(i, ids.get(i));
}
return executeSpec.as(entity.getJavaType()).fetch().all();
return databaseClient.execute().sql(operation).as(entity.getJavaType()).fetch().all();
});
}
@@ -274,12 +245,11 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Assert.notNull(id, "Id must not be null!");
BindIdOperation delete = accessStrategy.deleteById(entity.getTableName(), getIdColumnName());
BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(databaseClient.execute().sql(delete));
PreparedOperation<Delete> delete = accessStrategy.getStatements().delete(entity.getTableName(), binder -> {
binder.filterBy(getIdColumnName(), SettableValue.from(id));
});
delete.bindId(wrapper, id);
return wrapper.getBoundOperation() //
return databaseClient.execute().sql(delete) //
.fetch() //
.rowsUpdated() //
.then();
@@ -299,13 +269,11 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
return Flux.empty();
}
String idColumnName = getIdColumnName();
BindIdOperation delete = accessStrategy.deleteByIdIn(entity.getTableName(), idColumnName);
PreparedOperation<Delete> delete = accessStrategy.getStatements().delete(entity.getTableName(), binder -> {
binder.filterBy(getIdColumnName(), SettableValue.from(ids));
});
BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(databaseClient.execute().sql(delete));
delete.bindIds(wrapper, ids);
return wrapper.getBoundOperation().as(entity.getJavaType()).fetch().rowsUpdated();
return databaseClient.execute().sql(delete).then();
}).then();
}

1
src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
View File

@@ -43,6 +43,7 @@ public abstract class StatementRenderUtil {
String sql = SqlRenderer.toString(select);
// TODO: Replace with proper {@link Dialect} rendering for limit/offset.
// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
if (limit.isPresent()) {
LimitClause limitClause = dialect.limit();