diff --git a/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java b/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java
new file mode 100644
index 00000000..30f0c6da
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2018-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;
+
+/**
+ * Stripped down interface providing access to a fluent API that specifies a basic set of reactive R2DBC operations.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ * @see R2dbcEntityOperations
+ */
+public interface FluentR2dbcOperations
+ extends ReactiveSelectOperation, ReactiveInsertOperation, ReactiveUpdateOperation, ReactiveDeleteOperation {}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java
new file mode 100644
index 00000000..ac258567
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2018-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.Mono;
+
+import org.springframework.data.r2dbc.query.Query;
+
+/**
+ * The {@link ReactiveDeleteOperation} interface allows creation and execution of {@code DELETE} operations in a fluent
+ * API style.
+ *
+ * The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching}. By default,
+ * the table to operate on is derived from the initial {@literal domainType} and can be defined there via
+ * {@link org.springframework.data.relational.core.mapping.Table} annotation. Using {@code inTable} allows to override
+ * the table name for the execution.
+ *
+ *
+ *
+ * delete(Jedi.class)
+ * .from("star_wars")
+ * .matching(query(where("firstname").is("luke")))
+ * .all();
+ *
+ *
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+public interface ReactiveDeleteOperation {
+
+ /**
+ * Begin creating a {@code DELETE} operation for the given {@link Class domainType}.
+ *
+ * @param domainType {@link Class type} of domain object to delete; must not be {@literal null}.
+ * @return new instance of {@link ReactiveDelete}.
+ * @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
+ * @see ReactiveDelete
+ */
+ ReactiveDelete delete(Class> domainType);
+
+ /**
+ * Table override (optional).
+ */
+ interface DeleteWithTable {
+
+ /**
+ * Explicitly set the {@link String name} of the table on which to perform the delete.
+ *
+ * Skip this step to use the default table derived from the {@link Class domain type}.
+ *
+ * @param table {@link String name} of the table; must not be {@literal null} or empty.
+ * @return new instance of {@link DeleteWithQuery}.
+ * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
+ * @see DeleteWithQuery
+ */
+ DeleteWithQuery from(String table);
+ }
+
+ /**
+ * Required {@link Query filter}.
+ */
+ interface DeleteWithQuery {
+
+ /**
+ * Define the {@link Query} used to filter elements in the delete.
+ *
+ * @param query {@link Query} used as the filter in the delete; must not be {@literal null}.
+ * @return new instance of {@link TerminatingDelete}.
+ * @throws IllegalArgumentException if {@link Query} is {@literal null}.
+ * @see TerminatingDelete
+ * @see Query
+ */
+ TerminatingDelete matching(Query query);
+ }
+
+ /**
+ * Trigger {@code DELETE} operation by calling one of the terminating methods.
+ */
+ interface TerminatingDelete {
+
+ /**
+ * Remove all matching rows.
+ *
+ * @return the number of affected rows; never {@literal null}.
+ * @see Mono
+ */
+ Mono all();
+ }
+
+ /**
+ * The {@link ReactiveDelete} interface provides methods for constructing {@code DELETE} operations in a fluent way.
+ */
+ interface ReactiveDelete extends DeleteWithTable, DeleteWithQuery {}
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java
new file mode 100644
index 00000000..479538e7
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2018-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.Mono;
+
+import org.springframework.data.r2dbc.query.Query;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link ReactiveDeleteOperation}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
+
+ private final R2dbcEntityTemplate template;
+
+ ReactiveDeleteOperationSupport(R2dbcEntityTemplate template) {
+ this.template = template;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation#delete(java.lang.Class)
+ */
+ @Override
+ public ReactiveDelete delete(Class> domainType) {
+
+ Assert.notNull(domainType, "DomainType must not be null");
+
+ return new ReactiveDeleteSupport(this.template, domainType, Query.empty(), null);
+ }
+
+ static class ReactiveDeleteSupport implements ReactiveDelete, TerminatingDelete {
+
+ private final R2dbcEntityTemplate template;
+
+ private final Class> domainType;
+
+ private final Query query;
+
+ private final @Nullable String tableName;
+
+ ReactiveDeleteSupport(R2dbcEntityTemplate template, Class> domainType, Query query, @Nullable String tableName) {
+ this.template = template;
+ this.domainType = domainType;
+ this.query = query;
+ this.tableName = tableName;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.DeleteWithTable#from(java.lang.String)
+ */
+ @Override
+ public DeleteWithQuery from(String tableName) {
+
+ Assert.hasText(tableName, "Table name must not be null or empty");
+
+ return new ReactiveDeleteSupport(this.template, this.domainType, this.query, tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.DeleteWithQuery#matching(org.springframework.data.r2dbc.query.Query)
+ */
+ @Override
+ public TerminatingDelete matching(Query query) {
+
+ Assert.notNull(query, "Query must not be null");
+
+ return new ReactiveDeleteSupport(this.template, this.domainType, query, this.tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.TerminatingDelete#all()
+ */
+ public Mono all() {
+ return this.template.doDelete(this.query, this.domainType, getTableName());
+ }
+
+ private String getTableName() {
+ return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java
new file mode 100644
index 00000000..bd514ad7
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2018-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.Mono;
+
+/**
+ * The {@link ReactiveInsertOperation} interface allows creation and execution of {@code INSERT} operations in a fluent
+ * API style.
+ *
+ * By default,the table to operate on is derived from the initial {@link Class domainType} and can be defined there via
+ * {@link org.springframework.data.relational.core.mapping.Table} annotation. Using {@code inTable} allows to override
+ * the table name for the execution.
+ *
+ *
+ *
+ * insert(Jedi.class)
+ * .into("star_wars")
+ * .using(luke);
+ *
+ *
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+public interface ReactiveInsertOperation {
+
+ /**
+ * Begin creating an {@code INSERT} operation for given {@link Class domainType}.
+ *
+ * @param {@link Class type} of the application domain object.
+ * @param domainType {@link Class type} of the domain object to insert; must not be {@literal null}.
+ * @return new instance of {@link ReactiveInsert}.
+ * @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
+ * @see ReactiveInsert
+ */
+ ReactiveInsert insert(Class domainType);
+
+ /**
+ * Table override (optional).
+ */
+ interface InsertWithTable extends TerminatingInsert {
+
+ /**
+ * Explicitly set the {@link String name} of the table.
+ *
+ * Skip this step to use the default table derived from the {@link Class domain type}.
+ *
+ * @param table {@link String name} of the table; must not be {@literal null} or empty.
+ * @return new instance of {@link TerminatingInsert}.
+ * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
+ */
+ TerminatingInsert into(String table);
+ }
+
+ /**
+ * Trigger {@code INSERT} execution by calling one of the terminating methods.
+ */
+ interface TerminatingInsert {
+
+ /**
+ * Insert exactly one {@link Object}.
+ *
+ * @param object {@link Object} to insert; must not be {@literal null}.
+ * @return the write result for this operation.
+ * @throws IllegalArgumentException if {@link Object} is {@literal null}.
+ * @see Mono
+ */
+ Mono using(T object);
+ }
+
+ /**
+ * The {@link ReactiveInsert} interface provides methods for constructing {@code INSERT} operations in a fluent way.
+ */
+ interface ReactiveInsert extends InsertWithTable {}
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java
new file mode 100644
index 00000000..11fc23e9
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2018-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.Mono;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link ReactiveInsertOperation}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
+
+ private final R2dbcEntityTemplate template;
+
+ ReactiveInsertOperationSupport(R2dbcEntityTemplate template) {
+ this.template = template;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation#insert(java.lang.Class)
+ */
+ @Override
+ public ReactiveInsert insert(Class domainType) {
+
+ Assert.notNull(domainType, "DomainType must not be null");
+
+ return new ReactiveInsertSupport<>(this.template, domainType, null);
+ }
+
+ static class ReactiveInsertSupport implements ReactiveInsert {
+
+ private final R2dbcEntityTemplate template;
+
+ private final Class domainType;
+
+ private final @Nullable String tableName;
+
+ ReactiveInsertSupport(R2dbcEntityTemplate template, Class domainType, String tableName) {
+ this.template = template;
+ this.domainType = domainType;
+ this.tableName = tableName;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation.InsertWithTable#into(java.lang.String)
+ */
+ @Override
+ public TerminatingInsert into(String tableName) {
+
+ Assert.notNull(tableName, "Table name must not be null");
+
+ return new ReactiveInsertSupport<>(this.template, this.domainType, tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation.TerminatingInsert#one(java.lang.Object)
+ */
+ @Override
+ public Mono using(T object) {
+
+ Assert.notNull(object, "Object to insert must not be null");
+
+ return this.template.doInsert(object, getTableName());
+ }
+
+ private String getTableName() {
+ return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java
new file mode 100644
index 00000000..15dae6f5
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2018-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.data.r2dbc.query.Query;
+
+/**
+ * The {@link ReactiveSelectOperation} interface allows creation and execution of {@code SELECT} operations in a fluent
+ * API style.
+ *
+ * The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching}. By default,
+ * the originating {@literal domainType} is also used for mapping back the result from the {@link io.r2dbc.spi.Row}.
+ * However, it is possible to define an different {@literal returnType} via {@code as} to mapping the result.
+ *
+ * By default, the table to operate on is derived from the initial {@literal domainType} and can be defined there via
+ * the {@link org.springframework.data.relational.core.mapping.Table} annotation. Using {@code inTable} allows to
+ * override the table name for the execution.
+ *
+ *
+ *
+ * select(Human.class)
+ * .from("star_wars")
+ * .as(Jedi.class)
+ * .matching(query(where("firstname").is("luke")))
+ * .all();
+ *
+ *
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+public interface ReactiveSelectOperation {
+
+ /**
+ * Begin creating a {@code SELECT} operation for the given {@link Class domainType}.
+ *
+ * @param {@link Class type} of the application domain object.
+ * @param domainType {@link Class type} of the domain object to query; must not be {@literal null}.
+ * @return new instance of {@link ReactiveSelect}.
+ * @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
+ * @see ReactiveSelect
+ */
+ ReactiveSelect select(Class domainType);
+
+ /**
+ * Table override (optional).
+ */
+ interface SelectWithTable extends SelectWithQuery {
+
+ /**
+ * Explicitly set the {@link String name} of the table on which to perform the query.
+ *
+ * Skip this step to use the default table derived from the {@link Class domain type}.
+ *
+ * @param table {@link String name} of the table; must not be {@literal null} or empty.
+ * @return new instance of {@link SelectWithProjection}.
+ * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
+ * @see SelectWithProjection
+ */
+ SelectWithProjection from(String table);
+ }
+
+ /**
+ * Result type override (optional).
+ */
+ interface SelectWithProjection extends SelectWithQuery {
+
+ /**
+ * Define the {@link Class result target type} that the fields should be mapped to.
+ *
+ * Skip this step if you are only interested in the original {@link Class domain type}.
+ *
+ * @param {@link Class type} of the result.
+ * @param resultType desired {@link Class type} of the result; must not be {@literal null}.
+ * @return new instance of {@link SelectWithQuery}.
+ * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
+ * @see SelectWithQuery
+ */
+ SelectWithQuery as(Class resultType);
+ }
+
+ /**
+ * Define a {@link Query} used as the filter for the {@code SELECT}.
+ */
+ interface SelectWithQuery extends TerminatingSelect {
+
+ /**
+ * Set the {@link Query} used as a filter in the {@code SELECT} statement.
+ *
+ * @param query {@link Query} used as a filter; must not be {@literal null}.
+ * @return new instance of {@link TerminatingSelect}.
+ * @throws IllegalArgumentException if {@link Query} is {@literal null}.
+ * @see Query
+ * @see TerminatingSelect
+ */
+ TerminatingSelect matching(Query query);
+ }
+
+ /**
+ * Trigger {@code SELECT} execution by calling one of the terminating methods.
+ */
+ interface TerminatingSelect {
+
+ /**
+ * Get the number of matching elements.
+ *
+ * @return a {@link Mono} emitting the total number of matching elements; never {@literal null}.
+ * @see Mono
+ */
+ Mono count();
+
+ /**
+ * Check for the presence of matching elements.
+ *
+ * @return a {@link Mono} emitting {@literal true} if at least one matching element exists; never {@literal null}.
+ * @see Mono
+ */
+ Mono exists();
+
+ /**
+ * Get the first result or no result.
+ *
+ * @return the first result or {@link Mono#empty()} if no match found; never {@literal null}.
+ * @see Mono
+ */
+ Mono first();
+
+ /**
+ * Get exactly zero or one result.
+ *
+ * @return exactly one result or {@link Mono#empty()} if no match found; never {@literal null}.
+ * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+ * @see Mono
+ */
+ Mono one();
+
+ /**
+ * Get all matching elements.
+ *
+ * @return all matching elements; never {@literal null}.
+ * @see Flux
+ */
+ Flux all();
+ }
+
+ /**
+ * The {@link ReactiveSelect} interface provides methods for constructing {@code SELECT} operations in a fluent way.
+ */
+ interface ReactiveSelect extends SelectWithTable, SelectWithProjection {}
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
new file mode 100644
index 00000000..d1e9be24
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
@@ -0,0 +1,157 @@
+/*
+ * Copyright 2018-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.data.r2dbc.query.Query;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link ReactiveSelectOperation}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
+
+ private final R2dbcEntityTemplate template;
+
+ ReactiveSelectOperationSupport(R2dbcEntityTemplate template) {
+ this.template = template;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation#select(java.lang.Class)
+ */
+ @Override
+ public ReactiveSelect select(Class domainType) {
+
+ Assert.notNull(domainType, "DomainType must not be null");
+
+ return new ReactiveSelectSupport<>(this.template, domainType, domainType, Query.empty(), null);
+ }
+
+ static class ReactiveSelectSupport implements ReactiveSelect {
+
+ private final R2dbcEntityTemplate template;
+
+ private final Class> domainType;
+
+ private final Class returnType;
+
+ private final Query query;
+
+ private final @Nullable String tableName;
+
+ ReactiveSelectSupport(R2dbcEntityTemplate template, Class> domainType, Class returnType, Query query,
+ @Nullable String tableName) {
+ this.template = template;
+ this.domainType = domainType;
+ this.returnType = returnType;
+ this.query = query;
+ this.tableName = tableName;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithTable#from(java.lang.String)
+ */
+ @Override
+ public SelectWithProjection from(String tableName) {
+
+ Assert.notNull(tableName, "Table name must not be null");
+
+ return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.query, tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithProjection#as(java.lang.Class)
+ */
+ @Override
+ public SelectWithQuery as(Class returnType) {
+
+ Assert.notNull(returnType, "ReturnType must not be null");
+
+ return new ReactiveSelectSupport<>(this.template, this.domainType, returnType, this.query, this.tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithQuery#matching(org.springframework.data.r2dbc.query.Query)
+ */
+ @Override
+ public TerminatingSelect matching(Query query) {
+
+ Assert.notNull(query, "Query must not be null");
+
+ return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, query, this.tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#count()
+ */
+ @Override
+ public Mono count() {
+ return this.template.doCount(this.query, this.domainType, getTableName());
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#exists()
+ */
+ @Override
+ public Mono exists() {
+ return this.template.doExists(this.query, this.domainType, getTableName());
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#first()
+ */
+ @Override
+ public Mono first() {
+ return this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType).first();
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#one()
+ */
+ @Override
+ public Mono one() {
+ return this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType).one();
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#all()
+ */
+ @Override
+ public Flux all() {
+ return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType).all();
+ }
+
+ private String getTableName() {
+ return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
+ }
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java
new file mode 100644
index 00000000..1e9baa95
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java
@@ -0,0 +1,113 @@
+/*
+ * Copyright 2018-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.Mono;
+
+import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.r2dbc.query.Update;
+
+/**
+ * The {@link ReactiveUpdateOperation} interface allows creation and execution of {@code UPDATE} operations in a fluent
+ * API style.
+ *
+ * The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching}, as well as
+ * the {@link Update} via {@code apply}.
+ *
+ * By default, the table to operate on is derived from the initial {@literal domainType} and can be defined there via
+ * the {@link org.springframework.data.relational.core.mapping.Table} annotation. Using {@code inTable} allows a
+ * developer to override the table name for the execution.
+ *
+ *
+ *
+ * update(Jedi.class)
+ * .table("star_wars")
+ * .matching(query(where("firstname").is("luke")))
+ * .apply(update("lastname", "skywalker"))
+ * .all();
+ *
+ *
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+public interface ReactiveUpdateOperation {
+
+ /**
+ * Begin creating an {@code UPDATE} operation for the given {@link Class domainType}.
+ *
+ * @param domainType {@link Class type} of domain object to update; must not be {@literal null}.
+ * @return new instance of {@link ReactiveUpdate}.
+ * @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
+ * @see ReactiveUpdate
+ */
+ ReactiveUpdate update(Class> domainType);
+
+ /**
+ * Table override (optional).
+ */
+ interface UpdateWithTable {
+
+ /**
+ * Explicitly set the {@link String name} of the table on which to perform the update.
+ *
+ * Skip this step to use the default table derived from the {@link Class domain type}.
+ *
+ * @param table {@link String name} of the table; must not be {@literal null} or empty.
+ * @return new instance of {@link UpdateWithQuery}.
+ * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
+ * @see UpdateWithQuery
+ */
+ UpdateWithQuery inTable(String table);
+ }
+
+ /**
+ * Define a {@link Query} used as the filter for the {@link Update}.
+ */
+ interface UpdateWithQuery {
+
+ /**
+ * Filter rows to update by the given {@link Query}.
+ *
+ * @param query {@link Query} used as a filter in the update; must not be {@literal null}.
+ * @return new instance of {@link TerminatingUpdate}.
+ * @throws IllegalArgumentException if {@link Query} is {@literal null}.
+ * @see Query
+ * @see TerminatingUpdate
+ */
+ TerminatingUpdate matching(Query query);
+ }
+
+ /**
+ * Trigger {@code UPDATE} execution by calling one of the terminating methods.
+ */
+ interface TerminatingUpdate {
+
+ /**
+ * Update all matching rows in the table.
+ *
+ * @return the number of affected rows by the update; never {@literal null}.
+ * @see Mono
+ */
+ Mono apply(Update update);
+ }
+
+ /**
+ * The {@link ReactiveUpdate} interface provides methods for constructing {@code UPDATE} operations in a fluent way.
+ */
+ interface ReactiveUpdate extends UpdateWithTable, UpdateWithQuery {}
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java
new file mode 100644
index 00000000..90a7908c
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2018-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.Mono;
+
+import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.r2dbc.query.Update;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link ReactiveUpdateOperation}.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ */
+class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
+
+ private final R2dbcEntityTemplate template;
+
+ ReactiveUpdateOperationSupport(R2dbcEntityTemplate template) {
+ this.template = template;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation#update(java.lang.Class)
+ */
+ @Override
+ public ReactiveUpdate update(Class> domainType) {
+
+ Assert.notNull(domainType, "DomainType must not be null");
+
+ return new ReactiveUpdateSupport(this.template, domainType, Query.empty(), null);
+ }
+
+ static class ReactiveUpdateSupport implements ReactiveUpdate, TerminatingUpdate {
+
+ private final R2dbcEntityTemplate template;
+
+ private final Class> domainType;
+
+ private final Query query;
+
+ private final @Nullable String tableName;
+
+ ReactiveUpdateSupport(R2dbcEntityTemplate template, Class> domainType, Query query, @Nullable String tableName) {
+ this.template = template;
+ this.domainType = domainType;
+ this.query = query;
+ this.tableName = tableName;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.UpdateWithTable#inTable(java.lang.String)
+ */
+ @Override
+ public UpdateWithQuery inTable(String tableName) {
+
+ Assert.notNull(tableName, "Table name must not be null");
+
+ return new ReactiveUpdateSupport(this.template, this.domainType, this.query, tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.UpdateWithQuery#matching(org.springframework.data.r2dbc.query.Query)
+ */
+ @Override
+ public TerminatingUpdate matching(Query query) {
+
+ Assert.notNull(query, "Query must not be null");
+
+ return new ReactiveUpdateSupport(this.template, this.domainType, query, this.tableName);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.TerminatingUpdate#apply(org.springframework.data.r2dbc.query.Update)
+ */
+ @Override
+ public Mono apply(Update update) {
+
+ Assert.notNull(update, "Update must not be null");
+
+ return this.template.doUpdate(this.query, update, this.domainType, getTableName());
+ }
+
+ private String getTableName() {
+ return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
+ }
+ }
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationUnitTests.java
new file mode 100644
index 00000000..0abb72b7
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationUnitTests.java
@@ -0,0 +1,109 @@
+/*
+ * 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 static org.springframework.data.r2dbc.query.Criteria.*;
+import static org.springframework.data.r2dbc.query.Query.*;
+
+import io.r2dbc.spi.test.MockResult;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.mapping.SettableValue;
+import org.springframework.data.r2dbc.testing.StatementRecorder;
+import org.springframework.data.relational.core.mapping.Column;
+
+/**
+ * Unit test for {@link ReactiveDeleteOperation}.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveDeleteOperationUnitTests {
+
+ 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 shouldDelete() {
+
+ MockResult result = MockResult.builder().rowsUpdated(1).build();
+
+ recorder.addStubbing(s -> s.startsWith("DELETE"), result);
+
+ entityTemplate.delete(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .all() //
+ .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 shouldDeleteInTable() {
+
+ MockResult result = MockResult.builder().rowsUpdated(1).build();
+
+ recorder.addStubbing(s -> s.startsWith("DELETE"), result);
+
+ entityTemplate.delete(Person.class) //
+ .from("other_table") //
+ .matching(query(where("name").is("Walter"))) //
+ .all() //
+ .as(StepVerifier::create) //
+ .expectNext(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("DELETE"));
+
+ assertThat(statement.getSql()).isEqualTo("DELETE FROM other_table WHERE other_table.THE_NAME = $1");
+ }
+
+ 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/core/ReactiveInsertOperationUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationUnitTests.java
new file mode 100644
index 00000000..f5b98f8d
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationUnitTests.java
@@ -0,0 +1,126 @@
+/*
+ * 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 org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.mapping.SettableValue;
+import org.springframework.data.r2dbc.testing.StatementRecorder;
+import org.springframework.data.relational.core.mapping.Column;
+
+/**
+ * Unit test for {@link ReactiveInsertOperation}.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveInsertOperationUnitTests {
+
+ 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 shouldInsert() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, 42).build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("INSERT"), result);
+
+ Person person = new Person();
+ person.setName("Walter");
+
+ entityTemplate.insert(Person.class) //
+ .using(person) //
+ .as(StepVerifier::create) //
+ .consumeNextWith(actual -> {
+
+ assertThat(actual.id).isEqualTo("42");
+ }) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("INSERT"));
+
+ assertThat(statement.getSql()).isEqualTo("INSERT INTO person (THE_NAME) VALUES ($1)");
+ assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
+ }
+
+ @Test // gh-220
+ public void shouldUpdateInTable() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, 42).build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("INSERT"), result);
+
+ Person person = new Person();
+ person.setName("Walter");
+
+ entityTemplate.insert(Person.class) //
+ .into("the_table") //
+ .using(person) //
+ .as(StepVerifier::create) //
+ .consumeNextWith(actual -> {
+
+ assertThat(actual.id).isEqualTo("42");
+ }) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("INSERT"));
+
+ assertThat(statement.getSql()).isEqualTo("INSERT INTO the_table (THE_NAME) VALUES ($1)");
+ }
+
+ 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/core/ReactiveSelectOperationUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java
new file mode 100644
index 00000000..ec2593bd
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java
@@ -0,0 +1,232 @@
+/*
+ * 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 static org.springframework.data.r2dbc.query.Criteria.*;
+import static org.springframework.data.r2dbc.query.Query.*;
+
+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 org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.testing.StatementRecorder;
+import org.springframework.data.relational.core.mapping.Column;
+
+/**
+ * Unit test for {@link ReactiveSelectOperation}.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveSelectOperationUnitTests {
+
+ 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 shouldSelectAll() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .matching(query(where("name").is("Walter")).limit(10).offset(20)) //
+ .all() //
+ .as(StepVerifier::create) //
+ .expectNextCount(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+ assertThat(statement.getSql())
+ .isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 LIMIT 10 OFFSET 20");
+ }
+
+ @Test // gh-220
+ public void shouldSelectAs() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .as(PersonProjection.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .all() //
+ .as(StepVerifier::create) //
+ .expectNextCount(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+ assertThat(statement.getSql()).isEqualTo("SELECT person.THE_NAME FROM person WHERE person.THE_NAME = $1");
+ }
+
+ @Test // gh-220
+ public void shouldSelectFromTable() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .from("the_table") //
+ .matching(query(where("name").is("Walter"))) //
+ .all() //
+ .as(StepVerifier::create) //
+ .expectNextCount(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+ assertThat(statement.getSql()).isEqualTo("SELECT the_table.* FROM the_table WHERE the_table.THE_NAME = $1");
+ }
+
+ @Test // gh-220
+ public void shouldSelectFirst() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .first() //
+ .as(StepVerifier::create) //
+ .expectNextCount(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+ assertThat(statement.getSql()).isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 LIMIT 1");
+ }
+
+ @Test // gh-220
+ public void shouldSelectOne() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .one() //
+ .as(StepVerifier::create) //
+ .expectNextCount(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+ assertThat(statement.getSql()).isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 LIMIT 2");
+ }
+
+ @Test // gh-220
+ public void shouldSelectExists() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").build())
+ .build();
+ MockResult result = MockResult.builder().rowMetadata(metadata)
+ .row(MockRow.builder().identified("id", Object.class, "Walter").build()).build();
+
+ recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+ entityTemplate.select(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .exists() //
+ .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");
+ }
+
+ @Test // gh-220
+ public void shouldSelectCount() {
+
+ MockRowMetadata metadata = MockRowMetadata.builder().columnMetadata(MockColumnMetadata.builder().name("id").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.select(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .count() //
+ .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");
+ }
+
+ static class Person {
+
+ @Id String id;
+
+ @Column("THE_NAME") String name;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+ }
+
+ interface PersonProjection {
+
+ String getName();
+ }
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationUnitTests.java
new file mode 100644
index 00000000..7be1107e
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationUnitTests.java
@@ -0,0 +1,111 @@
+/*
+ * 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 static org.springframework.data.r2dbc.query.Criteria.*;
+import static org.springframework.data.r2dbc.query.Query.*;
+
+import io.r2dbc.spi.test.MockResult;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.mapping.SettableValue;
+import org.springframework.data.r2dbc.query.Update;
+import org.springframework.data.r2dbc.testing.StatementRecorder;
+import org.springframework.data.relational.core.mapping.Column;
+
+/**
+ * Unit test for {@link ReactiveUpdateOperation}.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveUpdateOperationUnitTests {
+
+ 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 shouldUpdate() {
+
+ MockResult result = MockResult.builder().rowsUpdated(1).build();
+
+ recorder.addStubbing(s -> s.startsWith("UPDATE"), result);
+
+ entityTemplate.update(Person.class) //
+ .matching(query(where("name").is("Walter"))) //
+ .apply(Update.update("name", "Heisenberg")) //
+ .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 shouldUpdateInTable() {
+
+ MockResult result = MockResult.builder().rowsUpdated(1).build();
+
+ recorder.addStubbing(s -> s.startsWith("UPDATE"), result);
+
+ entityTemplate.update(Person.class) //
+ .inTable("the_table") //
+ .matching(query(where("name").is("Walter"))) //
+ .apply(Update.update("name", "Heisenberg")) //
+ .as(StepVerifier::create) //
+ .expectNext(1) //
+ .verifyComplete();
+
+ StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("UPDATE"));
+
+ assertThat(statement.getSql()).isEqualTo("UPDATE the_table SET THE_NAME = $1 WHERE the_table.THE_NAME = $2");
+ }
+
+ static class Person {
+
+ @Id String id;
+
+ @Column("THE_NAME") String name;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+ }
+
+}