diff --git a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java index c35cffbc..912887b0 100644 --- a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java +++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java @@ -204,6 +204,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor { return ConnectionFactoryUtils.getConnection(obtainConnectionFactory()); } + /** + * Obtain the {@link ReactiveDataAccessStrategy}. + * + * @return a the ReactiveDataAccessStrategy. + */ + protected ReactiveDataAccessStrategy getDataAccessStrategy() { + return dataAccessStrategy; + } + /** * Release the {@link Connection}. * diff --git a/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityOperations.java b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityOperations.java new file mode 100644 index 00000000..2cb72138 --- /dev/null +++ b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityOperations.java @@ -0,0 +1,144 @@ +/* + * Copyright 2020 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.core; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.dao.DataAccessException; +import org.springframework.dao.TransientDataAccessResourceException; +import org.springframework.data.r2dbc.query.Query; +import org.springframework.data.r2dbc.query.Update; + +/** + * Interface specifying a basic set of reactive R2DBC operations using entities. Implemented by + * {@link R2dbcEntityTemplate}. Not often used directly, but a useful option to enhance testability, as it can easily be + * mocked or stubbed. + * + * @author Mark Paluch + * @since 1.1 + * @see DatabaseClient + */ +public interface R2dbcEntityOperations extends FluentR2dbcOperations { + + /** + * Expose the underlying {@link DatabaseClient} to allow SQL operations. + * + * @return the underlying {@link DatabaseClient}. + * @see DatabaseClient + */ + DatabaseClient getDatabaseClient(); + + // ------------------------------------------------------------------------- + // Methods dealing with org.springframework.data.r2dbc.query.Query + // ------------------------------------------------------------------------- + + /** + * Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users + * to further refine the selection Query using a {@link Query} predicate to determine how many entities of the given + * {@link Class type} match the Query. + * + * @param query user-defined count {@link Query} to execute; must not be {@literal null}. + * @param entityClass {@link Class type} of the entity; must not be {@literal null}. + * @return the number of existing entities. + * @throws DataAccessException if any problem occurs while executing the query. + */ + Mono count(Query query, Class entityClass) throws DataAccessException; + + /** + * Determine whether the result for {@code entityClass} {@link Query} yields at least one row. + * + * @param query user-defined exists {@link Query} to execute; must not be {@literal null}. + * @param entityClass {@link Class type} of the entity; must not be {@literal null}. + * @return {@literal true} if the object exists. + * @throws DataAccessException if any problem occurs while executing the query. + * @since 2.1 + */ + Mono exists(Query query, Class entityClass) throws DataAccessException; + + /** + * Execute a {@code SELECT} query and convert the resulting items to a stream of entities. + * + * @param query must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the result objects returned by the action. + * @throws DataAccessException if there is any problem issuing the execution. + */ + Flux select(Query query, Class entityClass) throws DataAccessException; + + /** + * Execute a {@code SELECT} query and convert the resulting item to an entity. + * + * @param query must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the result object returned by the action or {@link Mono#empty()}. + * @throws DataAccessException if there is any problem issuing the execution. + */ + Mono selectOne(Query query, Class entityClass) throws DataAccessException; + + /** + * Update the queried entities and return {@literal true} if the update was applied. + * + * @param query must not be {@literal null}. + * @param update must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the number of affected rows. + * @throws DataAccessException if there is any problem executing the query. + */ + Mono update(Query query, Update update, Class entityClass) throws DataAccessException; + + /** + * Remove entities (rows)/columns from the table by {@link Query}. + * + * @param query must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the number of affected rows. + * @throws DataAccessException if there is any problem issuing the execution. + */ + Mono delete(Query query, Class entityClass) throws DataAccessException; + + // ------------------------------------------------------------------------- + // Methods dealing with entities + // ------------------------------------------------------------------------- + + /** + * Insert the given entity and emit the entity if the insert was applied. + * + * @param entity The entity to insert, must not be {@literal null}. + * @return the inserted entity. + * @throws DataAccessException if there is any problem issuing the execution. + */ + Mono insert(T entity) throws DataAccessException; + + /** + * Update the given entity and emit the entity if the update was applied. + * + * @param entity The entity to update, must not be {@literal null}. + * @return the updated entity. + * @throws DataAccessException if there is any problem issuing the execution. + * @throws TransientDataAccessResourceException if the update did not affect any rows. + */ + Mono update(T entity) throws DataAccessException; + + /** + * Delete the given entity and emit the entity if the delete was applied. + * + * @param entity must not be {@literal null}. + * @return the deleted entity. + * @throws DataAccessException if there is any problem issuing the execution. + */ + Mono delete(T entity) throws DataAccessException; +} diff --git a/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java new file mode 100644 index 00000000..115bc161 --- /dev/null +++ b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java @@ -0,0 +1,491 @@ +/* + * Copyright 2020 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.core; + +import io.r2dbc.spi.Row; +import io.r2dbc.spi.RowMetadata; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.beans.FeatureDescriptor; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.function.BiFunction; +import java.util.stream.Collectors; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.TransientDataAccessResourceException; +import org.springframework.data.mapping.IdentifierAccessor; +import org.springframework.data.mapping.MappingException; +import org.springframework.data.mapping.context.MappingContext; +import org.springframework.data.projection.ProjectionInformation; +import org.springframework.data.projection.SpelAwareProxyProjectionFactory; +import org.springframework.data.r2dbc.mapping.R2dbcMappingContext; +import org.springframework.data.r2dbc.query.Criteria; +import org.springframework.data.r2dbc.query.Query; +import org.springframework.data.r2dbc.query.Update; +import org.springframework.data.relational.core.mapping.RelationalPersistentEntity; +import org.springframework.data.relational.core.mapping.RelationalPersistentProperty; +import org.springframework.data.relational.core.sql.Functions; +import org.springframework.data.util.ProxyUtils; +import org.springframework.util.Assert; + +/** + * Implementation of {@link R2dbcEntityOperations}. It simplifies the use of Reactive R2DBC usage through entities and + * helps to avoid common errors. This class uses {@link DatabaseClient} to execute SQL queries or updates, initiating + * iteration over {@link io.r2dbc.spi.Result}. + *

+ * Can be used within a service implementation via direct instantiation with a {@link DatabaseClient} reference, or get + * prepared in an application context and given to services as bean reference. + * + * @author Mark Paluch + * @since 1.1 + */ +public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware { + + private final DatabaseClient databaseClient; + + private final ReactiveDataAccessStrategy dataAccessStrategy; + + private final MappingContext, ? extends RelationalPersistentProperty> mappingContext; + + private final SpelAwareProxyProjectionFactory projectionFactory; + + /** + * Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient}. + * + * @param databaseClient + */ + public R2dbcEntityTemplate(DatabaseClient databaseClient) { + + Assert.notNull(databaseClient, "DatabaseClient must not be null"); + + this.databaseClient = databaseClient; + this.dataAccessStrategy = getDataAccessStrategy(databaseClient); + this.mappingContext = getMappingContext(this.dataAccessStrategy); + this.projectionFactory = new SpelAwareProxyProjectionFactory(); + } + + /** + * Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient} and {@link ReactiveDataAccessStrategy}. + * + * @param databaseClient + */ + public R2dbcEntityTemplate(DatabaseClient databaseClient, ReactiveDataAccessStrategy strategy) { + + Assert.notNull(databaseClient, "DatabaseClient must not be null"); + Assert.notNull(strategy, "ReactiveDataAccessStrategy must not be null"); + + this.databaseClient = databaseClient; + this.dataAccessStrategy = strategy; + this.mappingContext = strategy.getConverter().getMappingContext(); + this.projectionFactory = new SpelAwareProxyProjectionFactory(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#getDatabaseClient() + */ + @Override + public DatabaseClient getDatabaseClient() { + return this.databaseClient; + } + + /* + * (non-Javadoc) + * @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory) + */ + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.projectionFactory.setBeanFactory(beanFactory); + } + + // ------------------------------------------------------------------------- + // Methods dealing with org.springframework.data.r2dbc.core.FluentR2dbcOperations + // ------------------------------------------------------------------------- + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation#select(java.lang.Class) + */ + @Override + public ReactiveSelect select(Class domainType) { + return new ReactiveSelectOperationSupport(this).select(domainType); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation#insert(java.lang.Class) + */ + @Override + public ReactiveInsert insert(Class domainType) { + return new ReactiveInsertOperationSupport(this).insert(domainType); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation#update(java.lang.Class) + */ + @Override + public ReactiveUpdate update(Class domainType) { + return new ReactiveUpdateOperationSupport(this).update(domainType); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation#delete(java.lang.Class) + */ + @Override + public ReactiveDelete delete(Class domainType) { + return new ReactiveDeleteOperationSupport(this).delete(domainType); + } + + // ------------------------------------------------------------------------- + // Methods dealing with org.springframework.data.r2dbc.query.Query + // ------------------------------------------------------------------------- + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#count(org.springframework.data.r2dbc.query.Query, java.lang.Class) + */ + @Override + public Mono count(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "entity class must not be null"); + + return doCount(query, entityClass, getTableName(entityClass)); + } + + Mono doCount(Query query, Class entityClass, String tableName) { + + RelationalPersistentEntity entity = getRequiredEntity(entityClass); + StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass); + + StatementMapper.SelectSpec selectSpec = statementMapper // + .createSelect(tableName) // + .doWithTable((table, spec) -> { + return spec.withProjection(Functions.count(table.column(entity.getRequiredIdProperty().getColumnName()))); + }); + + Optional criteria = query.getCriteria(); + if (criteria.isPresent()) { + selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec); + } + + PreparedOperation operation = statementMapper.getMappedObject(selectSpec); + + return this.databaseClient.execute(operation) // + .map((r, md) -> r.get(0, Long.class)) // + .first() // + .defaultIfEmpty(0L); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#exists(org.springframework.data.r2dbc.query.Query, java.lang.Class) + */ + @Override + public Mono exists(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "entity class must not be null"); + + return doExists(query, entityClass, getTableName(entityClass)); + } + + Mono doExists(Query query, Class entityClass, String tableName) { + + RelationalPersistentEntity entity = getRequiredEntity(entityClass); + StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass); + + String columnName = entity.hasIdProperty() ? entity.getRequiredIdProperty().getColumnName() : "*"; + + StatementMapper.SelectSpec selectSpec = statementMapper // + .createSelect(tableName) // + .withProjection(columnName); + + Optional criteria = query.getCriteria(); + if (criteria.isPresent()) { + selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec); + } + + PreparedOperation operation = statementMapper.getMappedObject(selectSpec); + + return this.databaseClient.execute(operation) // + .map((r, md) -> r) // + .first() // + .hasElement(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#select(org.springframework.data.r2dbc.query.Query, java.lang.Class) + */ + @Override + public Flux select(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "entity class must not be null"); + + return doSelect(query, entityClass, getTableName(entityClass), entityClass).all(); + } + + RowsFetchSpec doSelect(Query query, Class entityClass, String tableName, Class returnType) { + + RelationalPersistentEntity entity = getRequiredEntity(entityClass); + StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass); + + StatementMapper.SelectSpec selectSpec = statementMapper // + .createSelect(tableName) // + .withProjection(getSelectProjection(query, returnType)); + + if (query.getLimit() > 0) { + selectSpec = selectSpec.limit(query.getLimit()); + } + + if (query.getOffset() > 0) { + selectSpec = selectSpec.offset(query.getOffset()); + } + + if (query.isSorted()) { + selectSpec = selectSpec.withSort(query.getSort()); + } + + Optional criteria = query.getCriteria(); + if (criteria.isPresent()) { + selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec); + } + + PreparedOperation operation = statementMapper.getMappedObject(selectSpec); + + BiFunction rowMapper; + if (returnType.isInterface()) { + rowMapper = dataAccessStrategy.getRowMapper(entityClass) + .andThen(o -> projectionFactory.createProjection(returnType, o)); + } else { + rowMapper = dataAccessStrategy.getRowMapper(returnType); + } + + return this.databaseClient.execute(operation).map(rowMapper); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#selectOne(org.springframework.data.r2dbc.query.Query, java.lang.Class) + */ + @Override + public Mono selectOne(Query query, Class entityClass) throws DataAccessException { + return doSelect(query.limit(2), entityClass, getTableName(entityClass), entityClass).one(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(org.springframework.data.r2dbc.query.Query, org.springframework.data.r2dbc.query.Update, java.lang.Class) + */ + @Override + public Mono update(Query query, Update update, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(update, "Update must not be null"); + Assert.notNull(entityClass, "entity class must not be null"); + + return doUpdate(query, update, entityClass, getTableName(entityClass)); + } + + Mono doUpdate(Query query, Update update, Class entityClass, String tableName) { + + StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass); + + StatementMapper.UpdateSpec selectSpec = statementMapper // + .createUpdate(tableName, update); + + Optional criteria = query.getCriteria(); + if (criteria.isPresent()) { + selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec); + } + + PreparedOperation operation = statementMapper.getMappedObject(selectSpec); + return this.databaseClient.execute(operation).fetch().rowsUpdated(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(org.springframework.data.r2dbc.query.Query, java.lang.Class) + */ + @Override + public Mono delete(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "entity class must not be null"); + + return doDelete(query, entityClass, getTableName(entityClass)); + } + + Mono doDelete(Query query, Class entityClass, String tableName) { + + StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass); + + StatementMapper.DeleteSpec selectSpec = statementMapper // + .createDelete(tableName); + + Optional criteria = query.getCriteria(); + if (criteria.isPresent()) { + selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec); + } + + PreparedOperation operation = statementMapper.getMappedObject(selectSpec); + return this.databaseClient.execute(operation).fetch().rowsUpdated().defaultIfEmpty(0); + } + + // ------------------------------------------------------------------------- + // Methods dealing with entities + // ------------------------------------------------------------------------- + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#insert(java.lang.Object) + */ + @Override + public Mono insert(T entity) throws DataAccessException { + + Assert.notNull(entity, "Entity must not be null"); + + return doInsert(entity, getRequiredEntity(entity).getTableName()); + } + + Mono doInsert(T entity, String tableName) { + + RelationalPersistentEntity persistentEntity = getRequiredEntity(entity); + + return this.databaseClient.insert() // + .into(persistentEntity.getType()) // + .table(tableName).using(entity) // + .map(this.dataAccessStrategy.getConverter().populateIdIfNecessary(entity)) // + .first() // + .defaultIfEmpty(entity); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(java.lang.Object) + */ + @Override + public Mono update(T entity) throws DataAccessException { + + Assert.notNull(entity, "Entity must not be null"); + + RelationalPersistentEntity persistentEntity = getRequiredEntity(entity); + + return this.databaseClient.update() // + .table(persistentEntity.getType()) // + .table(persistentEntity.getTableName()).using(entity) // + .fetch().rowsUpdated().handle((rowsUpdated, sink) -> { + + if (rowsUpdated == 0) { + sink.error(new TransientDataAccessResourceException( + String.format("Failed to update table [%s]. Row with Id [%s] does not exist.", + persistentEntity.getTableName(), persistentEntity.getIdentifierAccessor(entity).getIdentifier()))); + } else { + sink.next(entity); + } + }); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(java.lang.Object) + */ + @Override + public Mono delete(T entity) throws DataAccessException { + + Assert.notNull(entity, "Entity must not be null"); + + RelationalPersistentEntity persistentEntity = getRequiredEntity(entity); + + return delete(getByIdQuery(entity, persistentEntity), persistentEntity.getType()).thenReturn(entity); + } + + private Query getByIdQuery(T entity, RelationalPersistentEntity persistentEntity) { + if (!persistentEntity.hasIdProperty()) { + throw new MappingException("No id property found for object of type " + persistentEntity.getType() + "!"); + } + + IdentifierAccessor identifierAccessor = persistentEntity.getIdentifierAccessor(entity); + Object id = identifierAccessor.getRequiredIdentifier(); + + return Query.query(Criteria.where(persistentEntity.getRequiredIdProperty().getName()).is(id)); + } + + String getTableName(Class entityClass) { + return getRequiredEntity(entityClass).getTableName(); + } + + private RelationalPersistentEntity getRequiredEntity(Class entityClass) { + return this.mappingContext.getRequiredPersistentEntity(entityClass); + } + + private RelationalPersistentEntity getRequiredEntity(T entity) { + Class entityType = ProxyUtils.getUserClass(entity); + return (RelationalPersistentEntity) getRequiredEntity(entityType); + } + + private List getSelectProjection(Query query, Class returnType) { + + if (query.getColumns().isEmpty()) { + + if (returnType.isInterface()) { + + ProjectionInformation projectionInformation = projectionFactory.getProjectionInformation(returnType); + + if (projectionInformation.isClosed()) { + return projectionInformation.getInputProperties().stream().map(FeatureDescriptor::getName) + .collect(Collectors.toList()); + } + } + + return Collections.singletonList("*"); + } + + return query.getColumns(); + } + + private static ReactiveDataAccessStrategy getDataAccessStrategy(DatabaseClient databaseClient) { + + if (databaseClient instanceof DefaultDatabaseClient) { + + DefaultDatabaseClient client = (DefaultDatabaseClient) databaseClient; + return client.getDataAccessStrategy(); + } + + throw new IllegalStateException("Cannot obtain ReactiveDataAccessStrategy"); + } + + private static MappingContext, ? extends RelationalPersistentProperty> getMappingContext( + ReactiveDataAccessStrategy strategy) { + + if (strategy instanceof DefaultReactiveDataAccessStrategy) { + DefaultReactiveDataAccessStrategy strategy1 = (DefaultReactiveDataAccessStrategy) strategy; + return strategy1.getMappingContext(); + } + return new R2dbcMappingContext(); + } + +} diff --git a/src/main/java/org/springframework/data/r2dbc/query/Query.java b/src/main/java/org/springframework/data/r2dbc/query/Query.java new file mode 100644 index 00000000..5b06edb0 --- /dev/null +++ b/src/main/java/org/springframework/data/r2dbc/query/Query.java @@ -0,0 +1,234 @@ +/* + * Copyright 2020 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.query; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Sort; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +/** + * Query object representing {@link Criteria}, columns, {@link Sort}, and limit/offset for a SQL query. {@link Query} is + * created with a fluent API creating immutable objects. + * + * @author Mark Paluch + * @since 1.1 + * @see Criteria + * @see Sort + * @see Pageable + */ +public class Query { + + private final @Nullable Criteria criteria; + + // TODO: select list should be List + private final List columns; + private final Sort sort; + private final int limit; + private final long offset; + + /** + * Static factory method to create a {@link Query} using the provided {@link Criteria}. + * + * @param criteria must not be {@literal null}. + * @return a new {@link Query} for the given {@link Criteria}. + */ + public static Query query(Criteria criteria) { + return new Query(criteria); + } + + /** + * Creates a new {@link Query} using the given {@link Criteria}. + * + * @param criteria must not be {@literal null}. + */ + private Query(@Nullable Criteria criteria) { + this.criteria = criteria; + this.sort = Sort.unsorted(); + this.columns = Collections.emptyList(); + this.limit = -1; + this.offset = -1; + } + + private Query(Criteria criteria, List columns, Sort sort, int limit, long offset) { + this.criteria = criteria; + this.columns = columns; + this.sort = sort; + this.limit = limit; + this.offset = offset; + } + + /** + * Create a new empty {@link Query}. + * + * @return + */ + public static Query empty() { + return new Query(null); + } + + /** + * Add columns to the query. + * + * @param columns + * @return a new {@link Query} object containing the former settings with {@code columns} applied. + */ + public Query columns(String... columns) { + + Assert.notNull(columns, "Columns must not be null"); + + return columns(Arrays.asList(columns)); + } + + /** + * Add columns to the query. + * + * @param columns + * @return a new {@link Query} object containing the former settings with {@code columns} applied. + */ + public Query columns(Collection columns) { + + Assert.notNull(columns, "Columns must not be null"); + + List newColumns = new ArrayList<>(this.columns); + newColumns.addAll(columns); + return new Query(this.criteria, newColumns, this.sort, this.limit, offset); + } + + /** + * Set number of rows to skip before returning results. + * + * @param offset + * @return a new {@link Query} object containing the former settings with {@code offset} applied. + */ + public Query offset(long offset) { + return new Query(this.criteria, this.columns, this.sort, this.limit, offset); + } + + /** + * Limit the number of returned documents to {@code limit}. + * + * @param limit + * @return a new {@link Query} object containing the former settings with {@code limit} applied. + */ + public Query limit(int limit) { + return new Query(this.criteria, this.columns, this.sort, limit, this.offset); + } + + /** + * Set the given pagination information on the {@link Query} instance. Will transparently set {@code offset} and + * {@code limit} as well as applying the {@link Sort} instance defined with the {@link Pageable}. + * + * @param pageable + * @return a new {@link Query} object containing the former settings with {@link Pageable} applied. + */ + public Query with(Pageable pageable) { + + if (pageable.isUnpaged()) { + return this; + } + + assertNoCaseSort(pageable.getSort()); + + return new Query(this.criteria, this.columns, this.sort.and(sort), pageable.getPageSize(), pageable.getOffset()); + } + + /** + * Add a {@link Sort} to the {@link Query} instance. + * + * @param sort + * @return a new {@link Query} object containing the former settings with {@link Sort} applied. + */ + public Query sort(Sort sort) { + + Assert.notNull(sort, "Sort must not be null!"); + + if (sort.isUnsorted()) { + return this; + } + + assertNoCaseSort(sort); + + return new Query(this.criteria, this.columns, this.sort.and(sort), this.limit, this.offset); + } + + /** + * Return the {@link Criteria} to be applied. + * + * @return + */ + public Optional getCriteria() { + return Optional.ofNullable(this.criteria); + } + + /** + * Return the columns that this query should project. + * + * @return + */ + public List getColumns() { + return columns; + } + + /** + * Return {@literal true} if the {@link Query} has a sort parameter. + * + * @return {@literal true} if sorted. + * @see Sort#isSorted() + */ + public boolean isSorted() { + return sort.isSorted(); + } + + public Sort getSort() { + return sort; + } + + /** + * Return the number of rows to skip. + * + * @return + */ + public long getOffset() { + return this.offset; + } + + /** + * Return the maximum number of rows to be return. + * + * @return + */ + public int getLimit() { + return this.limit; + } + + private static void assertNoCaseSort(Sort sort) { + + for (Sort.Order order : sort) { + if (order.isIgnoreCase()) { + throw new IllegalArgumentException(String.format("Given sort contained an Order for %s with ignore case;" + + " R2DBC does not support sorting ignoring case currently", order.getProperty())); + } + } + } +} diff --git a/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java b/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java index 24103f8d..d22e093e 100644 --- a/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java +++ b/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Map; import java.util.function.UnaryOperator; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.data.domain.Sort; import org.springframework.data.mapping.PersistentPropertyPath; import org.springframework.data.mapping.PropertyPath; @@ -39,12 +40,14 @@ import org.springframework.data.r2dbc.query.Criteria.Combinator; import org.springframework.data.r2dbc.query.Criteria.Comparator; import org.springframework.data.relational.core.mapping.RelationalPersistentEntity; import org.springframework.data.relational.core.mapping.RelationalPersistentProperty; +import org.springframework.data.relational.core.sql.Aliased; import org.springframework.data.relational.core.sql.Column; import org.springframework.data.relational.core.sql.Condition; import org.springframework.data.relational.core.sql.Expression; import org.springframework.data.relational.core.sql.IdentifierProcessing; import org.springframework.data.relational.core.sql.SQL; import org.springframework.data.relational.core.sql.SqlIdentifier; +import org.springframework.data.relational.core.sql.SimpleFunction; import org.springframework.data.relational.core.sql.Table; import org.springframework.data.util.ClassTypeInformation; import org.springframework.data.util.TypeInformation; @@ -119,6 +122,51 @@ public class QueryMapper { return Sort.by(mappedOrder); } + /** + * Map the {@link Expression} object to apply field name mapping using {@link Class the type to read}. + * + * @param expression must not be {@literal null}. + * @param entity related {@link RelationalPersistentEntity}, can be {@literal null}. + * @return the mapped {@link Expression}. + * @since 1.1 + */ + public Expression getMappedObject(Expression expression, @Nullable RelationalPersistentEntity entity) { + + if (entity == null) { + return expression; + } + + if (expression instanceof Column) { + + Column column = (Column) expression; + Field field = createPropertyField(entity, column.getName()); + + return column instanceof Aliased + ? Column.aliased(field.getMappedColumnName(), column.getTable(), ((Aliased) column).getAlias()) + : Column.create(field.getMappedColumnName(), column.getTable()); + } + + if (expression instanceof SimpleFunction) { + + // Revisit after https://jira.spring.io/browse/DATAJDBC-478 + SimpleFunction function = (SimpleFunction) expression; + DirectFieldAccessor accessor = new DirectFieldAccessor(function); + + List arguments = (List) accessor.getPropertyValue("expressions"); + List mappedArguments = new ArrayList<>(arguments.size()); + + for (Expression argument : arguments) { + mappedArguments.add(getMappedObject(argument, entity)); + } + + SimpleFunction mappedFunction = SimpleFunction.create(function.getFunctionName(), mappedArguments); + + return function instanceof Aliased ? mappedFunction.as(((Aliased) function).getAlias()) : mappedFunction; + } + + throw new IllegalArgumentException(String.format("Cannot map %s", expression)); + } + /** * Map a {@link Criteria} object into {@link Condition} and consider value/{@code NULL} {@link Bindings}. * @@ -287,6 +335,10 @@ public class QueryMapper { } } + Field createPropertyField(@Nullable RelationalPersistentEntity entity, String key) { + return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext); + } + Field createPropertyField(@Nullable RelationalPersistentEntity entity, String key, MappingContext, RelationalPersistentProperty> mappingContext) { return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext); diff --git a/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java new file mode 100644 index 00000000..dd5031be --- /dev/null +++ b/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java @@ -0,0 +1,208 @@ +/* + * Copyright 2020 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.core; + +import static org.assertj.core.api.Assertions.*; + +import io.r2dbc.spi.test.MockColumnMetadata; +import io.r2dbc.spi.test.MockResult; +import io.r2dbc.spi.test.MockRow; +import io.r2dbc.spi.test.MockRowMetadata; +import reactor.test.StepVerifier; + +import java.util.Collections; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.data.annotation.Id; +import org.springframework.data.domain.Sort; +import org.springframework.data.r2dbc.dialect.PostgresDialect; +import org.springframework.data.r2dbc.mapping.SettableValue; +import org.springframework.data.r2dbc.query.Criteria; +import org.springframework.data.r2dbc.query.Query; +import org.springframework.data.r2dbc.query.Update; +import org.springframework.data.r2dbc.testing.StatementRecorder; +import org.springframework.data.relational.core.mapping.Column; + +/** + * Unit tests for {@link R2dbcEntityTemplate}. + * + * @author Mark Paluch + */ +public class R2dbcEntityTemplateUnitTests { + + DatabaseClient client; + R2dbcEntityTemplate entityTemplate; + StatementRecorder recorder; + + @Before + public void before() { + + recorder = StatementRecorder.newInstance(); + client = DatabaseClient.builder().connectionFactory(recorder) + .dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)).build(); + entityTemplate = new R2dbcEntityTemplate(client); + } + + @Test // gh-220 + public void shouldCountBy() { + + MockRowMetadata metadata = MockRowMetadata.builder() + .columnMetadata(MockColumnMetadata.builder().name("name").build()).build(); + MockResult result = MockResult.builder().rowMetadata(metadata) + .row(MockRow.builder().identified(0, Long.class, 1L).build()).build(); + + recorder.addStubbing(s -> s.startsWith("SELECT"), result); + + entityTemplate.count(Query.query(Criteria.where("name").is("Walter")), Person.class) // + .as(StepVerifier::create) // + .expectNext(1L) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT")); + + assertThat(statement.getSql()).isEqualTo("SELECT COUNT(person.id) FROM person WHERE person.THE_NAME = $1"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldExistsByCriteria() { + + MockRowMetadata metadata = MockRowMetadata.builder() + .columnMetadata(MockColumnMetadata.builder().name("name").build()).build(); + MockResult result = MockResult.builder().rowMetadata(metadata) + .row(MockRow.builder().identified(0, Long.class, 1L).build()).build(); + + recorder.addStubbing(s -> s.startsWith("SELECT"), result); + + entityTemplate.exists(Query.query(Criteria.where("name").is("Walter")), Person.class) // + .as(StepVerifier::create) // + .expectNext(true) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT")); + + assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldSelectByCriteria() { + + recorder.addStubbing(s -> s.startsWith("SELECT"), Collections.emptyList()); + + entityTemplate.select(Query.query(Criteria.where("name").is("Walter")).sort(Sort.by("name")), Person.class) // + .as(StepVerifier::create) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT")); + + assertThat(statement.getSql()) + .isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 ORDER BY THE_NAME ASC"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldSelectOne() { + + recorder.addStubbing(s -> s.startsWith("SELECT"), Collections.emptyList()); + + entityTemplate.selectOne(Query.query(Criteria.where("name").is("Walter")).sort(Sort.by("name")), Person.class) // + .as(StepVerifier::create) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT")); + + assertThat(statement.getSql()) + .isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 ORDER BY THE_NAME ASC LIMIT 2"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldUpdateByQuery() { + + MockRowMetadata metadata = MockRowMetadata.builder() + .columnMetadata(MockColumnMetadata.builder().name("name").build()).build(); + MockResult result = MockResult.builder().rowMetadata(metadata).rowsUpdated(1).build(); + + recorder.addStubbing(s -> s.startsWith("UPDATE"), result); + + entityTemplate + .update(Query.query(Criteria.where("name").is("Walter")), Update.update("name", "Heisenberg"), Person.class) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("UPDATE")); + + assertThat(statement.getSql()).isEqualTo("UPDATE person SET THE_NAME = $1 WHERE person.THE_NAME = $2"); + assertThat(statement.getBindings()).hasSize(2).containsEntry(0, SettableValue.from("Heisenberg")).containsEntry(1, + SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldDeleteByQuery() { + + MockRowMetadata metadata = MockRowMetadata.builder() + .columnMetadata(MockColumnMetadata.builder().name("name").build()).build(); + MockResult result = MockResult.builder().rowMetadata(metadata).rowsUpdated(1).build(); + + recorder.addStubbing(s -> s.startsWith("DELETE"), result); + + entityTemplate.delete(Query.query(Criteria.where("name").is("Walter")), Person.class) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("DELETE")); + + assertThat(statement.getSql()).isEqualTo("DELETE FROM person WHERE person.THE_NAME = $1"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + @Test // gh-220 + public void shouldDeleteEntity() { + + Person person = new Person(); + person.id = "Walter"; + recorder.addStubbing(s -> s.startsWith("DELETE"), Collections.emptyList()); + + entityTemplate.delete(person) // + .as(StepVerifier::create) // + .expectNext(person).verifyComplete(); + + StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("DELETE")); + + assertThat(statement.getSql()).isEqualTo("DELETE FROM person WHERE person.id = $1"); + assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter")); + } + + static class Person { + + @Id String id; + + @Column("THE_NAME") String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } +} diff --git a/src/test/java/org/springframework/data/r2dbc/testing/StatementRecorder.java b/src/test/java/org/springframework/data/r2dbc/testing/StatementRecorder.java new file mode 100644 index 00000000..d8dbd088 --- /dev/null +++ b/src/test/java/org/springframework/data/r2dbc/testing/StatementRecorder.java @@ -0,0 +1,312 @@ +/* + * Copyright 2020 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.testing; + +import io.r2dbc.spi.Batch; +import io.r2dbc.spi.Connection; +import io.r2dbc.spi.ConnectionFactory; +import io.r2dbc.spi.ConnectionFactoryMetadata; +import io.r2dbc.spi.ConnectionMetadata; +import io.r2dbc.spi.IsolationLevel; +import io.r2dbc.spi.Result; +import io.r2dbc.spi.Statement; +import io.r2dbc.spi.ValidationDepth; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.function.Supplier; +import java.util.regex.Pattern; + +import org.reactivestreams.Publisher; + +import org.springframework.data.r2dbc.mapping.SettableValue; + +/** + * Recorder utility for R2DBC {@link Statement}s. Allows stubbing and introspection. + * + * @author Mark Paluch + */ +public class StatementRecorder implements ConnectionFactory { + + private final Map, Supplier>> stubbings = new LinkedHashMap<>(); + private final List createdStatements = new ArrayList<>(); + private final List executedStatements = new ArrayList<>(); + + private StatementRecorder() {} + + /** + * Create a new {@link StatementRecorder}. + * + * @return + */ + public static StatementRecorder newInstance() { + return new StatementRecorder(); + } + + /** + * Create a new {@link StatementRecorder} accepting a {@link Consumer configurer}. + * + * @param configurer + * @return + */ + public static StatementRecorder newInstance(Consumer configurer) { + + StatementRecorder statementRecorder = new StatementRecorder(); + + configurer.accept(statementRecorder); + + return statementRecorder; + } + + /** + * Add a stubbing rule given the {@link Predicate SQL Predicate} and a {@link Result} that is emitted by the executed + * statement. Typical usage: + * + * 

+	 * recorder.addStubbing(sql -> sql.startsWith("SELECT"), result);
+	 *
+ * + * @param sqlPredicate + * @param result + */ + public void addStubbing(Predicate sqlPredicate, Result result) { + this.stubbings.put(sqlPredicate, () -> Collections.singletonList(result)); + } + + /** + * Add a stubbing rule given the {@link Predicate SQL Predicate} and a list of {@link Result results} that are emitted + * by the executed statement. Typical usage: + * + * 
+	 * recorder.addStubbing(sql -> sql.startsWith("SELECT"), results);
+	 *
+ * + * @param sqlPredicate + * @param result + */ + public void addStubbing(Predicate sqlPredicate, List results) { + this.stubbings.put(sqlPredicate, () -> results); + } + + /** + * Retrieve a statement by {@code sql}. + * + * @param sql + * @return + */ + public RecordedStatement getCreatedStatement(String sql) { + return getCreatedStatement(it -> compareSql(sql, it)); + } + + private static boolean compareSql(String pattern, String actual) { + return actual.equals(pattern) || Pattern.compile(pattern).matcher(actual).find(); + } + + /** + * Retrieve a statement by a {@link Predicate SQL predicate}. + * + * @param sql + * @return + */ + public RecordedStatement getCreatedStatement(Predicate predicate) { + + return createdStatements.stream().filter(recordedStatement -> { + return predicate.test(recordedStatement.getSql()) || predicate.test(recordedStatement.getSql().toLowerCase()) + || predicate.test(recordedStatement.getSql().toUpperCase()); + }).findFirst().orElseThrow(() -> new NoSuchElementException("No statement found")); + } + + public List getCreatedStatements() { + return createdStatements; + } + + public List getExecutedStatements() { + return executedStatements; + } + + @Override + public Publisher create() { + return Mono.just(new RecorderConnection()); + } + + @Override + public ConnectionFactoryMetadata getMetadata() { + return () -> "StatementRecorder"; + } + + class RecorderConnection implements Connection { + @Override + public Publisher beginTransaction() { + return createStatement("BEGIN").execute().then(); + } + + @Override + public Publisher close() { + return createStatement("CLOSE").execute().then(); + } + + @Override + public Publisher commitTransaction() { + return createStatement("COMMIT").execute().then(); + } + + @Override + public Batch createBatch() { + throw new UnsupportedOperationException("createBatch not yet supported"); + } + + @Override + public Publisher createSavepoint(String name) { + return createStatement("CREATE SAVEPOINT " + name).execute().then(); + } + + @Override + public RecordedStatement createStatement(String sql) { + + RecordedStatement statement = doCreateStatement(sql); + + createdStatements.add(statement); + + return statement; + } + + private RecordedStatement doCreateStatement(String sql) { + for (Map.Entry, Supplier>> entry : stubbings.entrySet()) { + + if (entry.getKey().test(sql) || entry.getKey().test(sql.toLowerCase()) + || entry.getKey().test(sql.toUpperCase())) { + return new RecordedStatement(sql, entry.getValue().get()); + } + + } + return new RecordedStatement(sql, Collections.emptyList()); + } + + @Override + public boolean isAutoCommit() { + throw new UnsupportedOperationException("isAutoCommit not yet supported"); + } + + @Override + public ConnectionMetadata getMetadata() { + throw new UnsupportedOperationException("getMetadata not yet supported"); + } + + @Override + public IsolationLevel getTransactionIsolationLevel() { + throw new UnsupportedOperationException("getTransactionIsolationLevel not yet supported"); + } + + @Override + public Publisher releaseSavepoint(String name) { + return createStatement("RELEASE SAVEPOINT " + name).execute().then(); + } + + @Override + public Publisher rollbackTransaction() { + return createStatement("ROLLBACK").execute().then(); + } + + @Override + public Publisher rollbackTransactionToSavepoint(String name) { + return createStatement("ROLLBACK TO " + name).execute().then(); + } + + @Override + public Publisher setAutoCommit(boolean autoCommit) { + return createStatement("SET AUTOCOMMIT " + autoCommit).execute().then(); + } + + @Override + public Publisher setTransactionIsolationLevel(IsolationLevel isolationLevel) { + return createStatement("SET TRANSACTION ISOLATION LEVEL " + isolationLevel.asSql()).execute().then(); + } + + @Override + public Publisher validate(ValidationDepth depth) { + return Mono.just(true); + } + } + + public class RecordedStatement implements Statement { + + private final String sql; + + private final List results; + + private final Map bindings = new LinkedHashMap<>(); + + public RecordedStatement(String sql, Result result) { + this(sql, Collections.singletonList(result)); + } + + public RecordedStatement(String sql, List results) { + this.sql = sql; + this.results = results; + } + + public Map getBindings() { + return bindings; + } + + public String getSql() { + return sql; + } + + @Override + public Statement add() { + return this; + } + + @Override + public Statement bind(int index, Object o) { + this.bindings.put(index, SettableValue.from(o)); + return this; + } + + @Override + public Statement bind(String identifier, Object o) { + this.bindings.put(identifier, SettableValue.from(o)); + return this; + } + + @Override + public Statement bindNull(int index, Class type) { + this.bindings.put(index, SettableValue.empty(type)); + return this; + } + + @Override + public Statement bindNull(String identifier, Class type) { + this.bindings.put(identifier, SettableValue.empty(type)); + return this; + } + + @Override + public Flux execute() { + return Flux.fromIterable(results).doOnSubscribe(subscription -> executedStatements.add(this)); + } + } + +}