#64 - Add Criteria API.

We now support Criteria creation and mapping to express where conditions with a fluent API.

databaseClient.select().from("legoset")
  .where(Criteria.of("name").like("John%").and("id").lessThanOrEquals(42055));

databaseClient.delete()
  .from(LegoSet.class)
  .where(Criteria.of("id").is(42055))
  .then()

databaseClient.delete()
  .from(LegoSet.class)
  .where(Criteria.of("id").is(42055))
  .fetch()
  .rowsUpdated()

Original pull request: #106.

src/main/java/org/springframework/data/r2dbc/domain/BindableOperation.java

@@ -1,4 +1,19 @@
-package org.springframework.data.r2dbc.function;
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.domain;
 
 import io.r2dbc.spi.Statement;
 

src/main/java/org/springframework/data/r2dbc/domain/Bindings.java

@@ -0,0 +1,264 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.domain;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.util.Streamable;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Value object representing value and {@code NULL} bindings for a {@link Statement} using {@link BindMarkers}.
+ *
+ * @author Mark Paluch
+ */
+public class Bindings implements Streamable<Bindings.Binding> {
+
+	private final Map<BindMarker, Binding> bindings;
+
+	/**
+	 * Create empty {@link Bindings}.
+	 */
+	public Bindings() {
+		this.bindings = Collections.emptyMap();
+	}
+
+	/**
+	 * Create {@link Bindings} from a {@link Map}.
+	 *
+	 * @param bindings must not be {@literal null}.
+	 */
+	public Bindings(Collection<Binding> bindings) {
+
+		Assert.notNull(bindings, "Bindings must not be null");
+
+		Map<BindMarker, Binding> mapping = new LinkedHashMap<>(bindings.size());
+		bindings.forEach(it -> mapping.put(it.getBindMarker(), it));
+		this.bindings = mapping;
+	}
+
+	Bindings(Map<BindMarker, Binding> bindings) {
+		this.bindings = bindings;
+	}
+
+	protected Map<BindMarker, Binding> getBindings() {
+		return bindings;
+	}
+
+	/**
+	 * Merge this bindings with an other {@link Bindings} object and create a new merged {@link Bindings} object.
+	 *
+	 * @param left the left object to merge with.
+	 * @param right the right object to merge with.
+	 * @return a new, merged {@link Bindings} object.
+	 */
+	public static Bindings merge(Bindings left, Bindings right) {
+
+		Assert.notNull(left, "Left side Bindings must not be null");
+		Assert.notNull(right, "Right side Bindings must not be null");
+
+		List<Binding> result = new ArrayList<>(left.getBindings().size() + right.getBindings().size());
+
+		result.addAll(left.getBindings().values());
+		result.addAll(right.getBindings().values());
+
+		return new Bindings(result);
+	}
+
+	/**
+	 * Apply the bindings to a {@link Statement}.
+	 *
+	 * @param statement the statement to apply to.
+	 */
+	public void apply(Statement statement) {
+
+		Assert.notNull(statement, "Statement must not be null");
+		this.bindings.forEach((marker, binding) -> binding.apply(statement));
+	}
+
+	/**
+	 * Performs the given action for each binding of this {@link Bindings} until all bindings have been processed or the
+	 * action throws an exception. Actions are performed in the order of iteration (if an iteration order is specified).
+	 * Exceptions thrown by the action are relayed to the
+	 *
+	 * @param action The action to be performed for each {@link Binding}.
+	 */
+	public void forEach(Consumer<? super Binding> action) {
+		this.bindings.forEach((marker, binding) -> action.accept(binding));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#iterator()
+	 */
+	@Override
+	public Iterator<Binding> iterator() {
+		return this.bindings.values().iterator();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#spliterator()
+	 */
+	@Override
+	public Spliterator<Binding> spliterator() {
+		return this.bindings.values().spliterator();
+	}
+
+	/**
+	 * Base class for value objects representing a value or a {@code NULL} binding.
+	 */
+	public abstract static class Binding {
+
+		private final BindMarker marker;
+
+		protected Binding(BindMarker marker) {
+			this.marker = marker;
+		}
+
+		/**
+		 * @return the associated {@link BindMarker}.
+		 */
+		public BindMarker getBindMarker() {
+			return marker;
+		}
+
+		/**
+		 * Return {@literal true} if there is a value present, otherwise {@literal false} for a {@code NULL} binding.
+		 *
+		 * @return {@literal true} if there is a value present, otherwise {@literal false} for a {@code NULL} binding.
+		 */
+		public abstract boolean hasValue();
+
+		/**
+		 * Return {@literal true} if this is is a {@code NULL} binding.
+		 *
+		 * @return {@literal true} if this is is a {@code NULL} binding.
+		 */
+		public boolean isNull() {
+			return !hasValue();
+		}
+
+		/**
+		 * Returns the value of this binding. Can be {@literal null} if this is a {@code NULL} binding.
+		 *
+		 * @return value of this binding. Can be {@literal null} if this is a {@code NULL} binding.
+		 */
+		@Nullable
+		public abstract Object getValue();
+
+		/**
+		 * Applies the binding to a {@link Statement}.
+		 *
+		 * @param statement the statement to apply to.
+		 */
+		public abstract void apply(Statement statement);
+	}
+
+	/**
+	 * Value binding.
+	 */
+	public static class ValueBinding extends Binding {
+
+		private final Object value;
+
+		public ValueBinding(BindMarker marker, Object value) {
+			super(marker);
+			this.value = value;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#hasValue()
+		 */
+		public boolean hasValue() {
+			return true;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#getValue()
+		 */
+		public Object getValue() {
+			return value;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#apply(io.r2dbc.spi.Statement)
+		 */
+		@Override
+		public void apply(Statement statement) {
+			getBindMarker().bind(statement, getValue());
+		}
+	}
+
+	/**
+	 * {@code NULL} binding.
+	 */
+	public static class NullBinding extends Binding {
+
+		private final Class<?> valueType;
+
+		public NullBinding(BindMarker marker, Class<?> valueType) {
+			super(marker);
+			this.valueType = valueType;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#hasValue()
+		 */
+		public boolean hasValue() {
+			return false;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#getValue()
+		 */
+		@Nullable
+		public Object getValue() {
+			return null;
+		}
+
+		public Class<?> getValueType() {
+			return valueType;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Bindings.Binding#apply(io.r2dbc.spi.Statement)
+		 */
+		@Override
+		public void apply(Statement statement) {
+			getBindMarker().bindNull(statement, getValueType());
+		}
+	}
+}

src/main/java/org/springframework/data/r2dbc/domain/MutableBindings.java

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.domain;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.LinkedHashMap;
+
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.util.Assert;
+
+/**
+ * Mutable extension to {@link Bindings} for Value and {@code NULL} bindings for a {@link Statement} using
+ * {@link BindMarkers}.
+ *
+ * @author Mark Paluch
+ */
+public class MutableBindings extends Bindings {
+
+	private final BindMarkers markers;
+
+	/**
+	 * Create new {@link MutableBindings}.
+	 *
+	 * @param markers must not be {@literal null}.
+	 */
+	public MutableBindings(BindMarkers markers) {
+
+		super(new LinkedHashMap<>());
+
+		Assert.notNull(markers, "BindMarkers must not be null");
+
+		this.markers = markers;
+	}
+
+	/**
+	 * Obtain the next {@link BindMarker}. Increments {@link BindMarkers} state.
+	 *
+	 * @return the next {@link BindMarker}.
+	 */
+	public BindMarker nextMarker() {
+		return markers.next();
+	}
+
+	/**
+	 * Obtain the next {@link BindMarker} with a name {@code hint}. Increments {@link BindMarkers} state.
+	 *
+	 * @param hint name hint.
+	 * @return the next {@link BindMarker}.
+	 */
+	public BindMarker nextMarker(String hint) {
+		return markers.next(hint);
+	}
+
+	/**
+	 * Bind a value to {@link BindMarker}.
+	 *
+	 * @param marker must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @return {@code this} {@link MutableBindings}.
+	 */
+	public MutableBindings bind(BindMarker marker, Object value) {
+
+		Assert.notNull(marker, "BindMarker must not be null");
+		Assert.notNull(value, "Value must not be null");
+
+		getBindings().put(marker, new ValueBinding(marker, value));
+
+		return this;
+	}
+
+	/**
+	 * Bind a value and return the related {@link BindMarker}. Increments {@link BindMarkers} state.
+	 *
+	 * @param value must not be {@literal null}.
+	 * @return {@code this} {@link MutableBindings}.
+	 */
+	public BindMarker bind(Object value) {
+
+		Assert.notNull(value, "Value must not be null");
+
+		BindMarker marker = nextMarker();
+		getBindings().put(marker, new ValueBinding(marker, value));
+
+		return marker;
+	}
+
+	/**
+	 * Bind a {@code NULL} value to {@link BindMarker}.
+	 *
+	 * @param marker must not be {@literal null}.
+	 * @param valueType must not be {@literal null}.
+	 * @return {@code this} {@link MutableBindings}.
+	 */
+	public MutableBindings bindNull(BindMarker marker, Class<?> valueType) {
+
+		Assert.notNull(marker, "BindMarker must not be null");
+		Assert.notNull(valueType, "Value type must not be null");
+
+		getBindings().put(marker, new NullBinding(marker, valueType));
+
+		return this;
+	}
+
+	/**
+	 * Bind a {@code NULL} value and return the related {@link BindMarker}. Increments {@link BindMarkers} state.
+	 *
+	 * @param valueType must not be {@literal null}.
+	 * @return {@code this} {@link MutableBindings}.
+	 */
+	public BindMarker bindNull(Class<?> valueType) {
+
+		Assert.notNull(valueType, "Value type must not be null");
+
+		BindMarker marker = nextMarker();
+		getBindings().put(marker, new NullBinding(marker, valueType));
+
+		return marker;
+	}
+}

src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java

@@ -30,6 +30,7 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.domain.PreparedOperation;
+import org.springframework.data.r2dbc.function.query.Criteria;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 
 /**
@@ -58,6 +59,11 @@ public interface DatabaseClient {
 	 */
 	InsertIntoSpec insert();
 
+	/**
+	 * Prepare an SQL DELETE call.
+	 */
+	DeleteFromSpec delete();
+
 	/**
 	 * Return a builder to mutate properties of this database client.
 	 */
@@ -262,7 +268,7 @@ public interface DatabaseClient {
 	}
 
 	/**
-	 * Contract for specifying {@code SELECT} options leading to the exchange.
+	 * Contract for specifying {@code INSERT} options leading to the exchange.
 	 */
 	interface InsertIntoSpec {
 
@@ -284,6 +290,29 @@ public interface DatabaseClient {
 		<T> TypedInsertSpec<T> into(Class<T> table);
 	}
 
+	/**
+	 * Contract for specifying {@code DELETE} options leading to the exchange.
+	 */
+	interface DeleteFromSpec {
+
+		/**
+		 * Specify the source {@literal table} to delete from.
+		 *
+		 * @param table must not be {@literal null} or empty.
+		 * @return a {@link GenericSelectSpec} for further configuration of the delete. Guaranteed to be not
+		 *         {@literal null}.
+		 */
+		DeleteSpec from(String table);
+
+		/**
+		 * Specify the source table to delete from to using the {@link Class entity class}.
+		 *
+		 * @param table must not be {@literal null}.
+		 * @return a {@link DeleteSpec} for further configuration of the delete. Guaranteed to be not {@literal null}.
+		 */
+		DeleteSpec from(Class<?> table);
+	}
+
 	/**
 	 * Contract for specifying {@code SELECT} options leading to the exchange.
 	 */
@@ -354,6 +383,13 @@ public interface DatabaseClient {
 		 */
 		S project(String... selectedFields);
 
+		/**
+		 * Configure a filter {@link Criteria}.
+		 *
+		 * @param criteria must not be {@literal null}.
+		 */
+		S where(Criteria criteria);
+
 		/**
 		 * Configure {@link Sort}.
 		 *
@@ -456,6 +492,31 @@ public interface DatabaseClient {
 		Mono<Void> then();
 	}
 
+	/**
+	 * Contract for specifying {@code DELETE} options leading to the exchange.
+	 */
+	interface DeleteSpec {
+
+		/**
+		 * Configure a filter {@link Criteria}.
+		 *
+		 * @param criteria must not be {@literal null}.
+		 */
+		DeleteSpec where(Criteria criteria);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		UpdatedRowsFetchSpec fetch();
+
+		/**
+		 * Perform the SQL call and return a {@link Mono} that completes without result on statement completion.
+		 *
+		 * @return a {@link Mono} ignoring its payload (actively dropping).
+		 */
+		Mono<Void> then();
+	}
+
 	/**
 	 * Contract for specifying parameter bindings.
 	 */

src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java

@@ -34,7 +34,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -53,13 +52,19 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.UncategorizedR2dbcException;
 import org.springframework.data.r2dbc.domain.BindTarget;
+import org.springframework.data.r2dbc.domain.BindableOperation;
 import org.springframework.data.r2dbc.domain.OutboundRow;
 import org.springframework.data.r2dbc.domain.PreparedOperation;
 import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
+import org.springframework.data.r2dbc.function.operation.BindableOperation;
+import org.springframework.data.r2dbc.function.query.BoundCondition;
+import org.springframework.data.r2dbc.function.query.Criteria;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
+import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.Insert;
+import org.springframework.data.relational.core.sql.Select;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -114,6 +119,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		return new DefaultInsertIntoSpec();
 	}
 
+	@Override
+	public DeleteFromSpec delete() {
+		return new DefaultDeleteFromSpec();
+	}
+
 	/**
 	 * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a
 	 * {@link Mono}. The connection is released after the {@link Mono} terminates (or the subscription is cancelled).
@@ -624,6 +634,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		final String table;
 		final List<String> projectedFields;
+		final @Nullable Criteria criteria;
 		final Sort sort;
 		final Pageable page;
 
@@ -633,6 +644,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			this.table = table;
 			this.projectedFields = Collections.emptyList();
+			this.criteria = null;
 			this.sort = Sort.unsorted();
 			this.page = Pageable.unpaged();
 		}
@@ -644,51 +656,50 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			projectedFields.addAll(this.projectedFields);
 			projectedFields.addAll(Arrays.asList(selectedFields));
 
-			return createInstance(table, projectedFields, sort, page);
+			return createInstance(table, projectedFields, criteria, sort, page);
+		}
+
+		public DefaultSelectSpecSupport where(Criteria whereCriteria) {
+
+			Assert.notNull(whereCriteria, "Criteria must not be null!");
+
+			return createInstance(table, projectedFields, whereCriteria, sort, page);
 		}
 
 		public DefaultSelectSpecSupport orderBy(Sort sort) {
 
 			Assert.notNull(sort, "Sort must not be null!");
 
-			return createInstance(table, projectedFields, sort, page);
+			return createInstance(table, projectedFields, criteria, sort, page);
 		}
 
 		public DefaultSelectSpecSupport page(Pageable page) {
 
 			Assert.notNull(page, "Pageable must not be null!");
 
-			return createInstance(table, projectedFields, sort, page);
+			return createInstance(table, projectedFields, criteria, sort, page);
 		}
 
-		<R> FetchSpec<R> execute(String sql, BiFunction<Row, RowMetadata, R> mappingFunction) {
-
-			Function<Connection, Statement> selectFunction = it -> {
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Executing SQL statement [" + sql + "]");
-				}
-
-				return it.createStatement(sql);
-			};
+		<R> FetchSpec<R> execute(PreparedOperation<?> preparedOperation, BiFunction<Row, RowMetadata, R> mappingFunction) {
 
+			Function<Connection, Statement> selectFunction = wrapPreparedOperation(preparedOperation);
 			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(selectFunction.apply(it).execute());
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
-					sql, //
+					preparedOperation.toQuery(), //
 					resultFunction, //
 					it -> Mono.error(new UnsupportedOperationException("Not available for SELECT")), //
 					mappingFunction);
 		}
 
-		protected abstract DefaultSelectSpecSupport createInstance(String table, List<String> projectedFields, Sort sort,
-				Pageable page);
+		protected abstract DefaultSelectSpecSupport createInstance(String table, List<String> projectedFields,
+				Criteria criteria, Sort sort, Pageable page);
 	}
 
 	private class DefaultGenericSelectSpec extends DefaultSelectSpecSupport implements GenericSelectSpec {
 
-		DefaultGenericSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page) {
-			super(table, projectedFields, sort, page);
+		DefaultGenericSelectSpec(String table, List<String> projectedFields, Criteria criteria, Sort sort, Pageable page) {
+			super(table, projectedFields, criteria, sort, page);
 		}
 
 		DefaultGenericSelectSpec(String table) {
@@ -700,7 +711,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.notNull(resultType, "Result type must not be null!");
 
-			return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, resultType,
+			return new DefaultTypedSelectSpec<>(table, projectedFields, criteria, sort, page, resultType,
 					dataAccessStrategy.getRowMapper(resultType));
 		}
 
@@ -717,6 +728,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return (DefaultGenericSelectSpec) super.project(selectedFields);
 		}
 
+		@Override
+		public DefaultGenericSelectSpec where(Criteria criteria) {
+			return (DefaultGenericSelectSpec) super.where(criteria);
+		}
+
 		@Override
 		public DefaultGenericSelectSpec orderBy(Sort sort) {
 			return (DefaultGenericSelectSpec) super.orderBy(sort);
@@ -734,15 +750,26 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
 
-			String select = dataAccessStrategy.select(table, new LinkedHashSet<>(this.projectedFields), sort, page);
+			PreparedOperation<Select> operation = dataAccessStrategy.getStatements().select(table, this.projectedFields,
+					(t, configurer) -> {
 
-			return execute(select, mappingFunction);
+						configurer.withPageRequest(page).withSort(sort);
+
+						if (criteria != null) {
+
+							BoundCondition boundCondition = dataAccessStrategy.getMappedCriteria(criteria, t);
+							configurer.withWhere(boundCondition.getCondition()).withBindings(boundCondition.getBindings());
+						}
+
+					});
+
+			return execute(operation, mappingFunction);
 		}
 
 		@Override
-		protected DefaultGenericSelectSpec createInstance(String table, List<String> projectedFields, Sort sort,
-				Pageable page) {
-			return new DefaultGenericSelectSpec(table, projectedFields, sort, page);
+		protected DefaultGenericSelectSpec createInstance(String table, List<String> projectedFields, Criteria criteria,
+				Sort sort, Pageable page) {
+			return new DefaultGenericSelectSpec(table, projectedFields, criteria, sort, page);
 		}
 	}
 
@@ -763,14 +790,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead);
 		}
 
-		DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page,
+		DefaultTypedSelectSpec(String table, List<String> projectedFields, Criteria criteria, Sort sort, Pageable page,
 				BiFunction<Row, RowMetadata, T> mappingFunction) {
-			this(table, projectedFields, sort, page, null, mappingFunction);
+			this(table, projectedFields, criteria, sort, page, null, mappingFunction);
 		}
 
-		DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page, Class<T> typeToRead,
-				BiFunction<Row, RowMetadata, T> mappingFunction) {
-			super(table, projectedFields, sort, page);
+		DefaultTypedSelectSpec(String table, List<String> projectedFields, Criteria criteria, Sort sort, Pageable page,
+				Class<T> typeToRead, BiFunction<Row, RowMetadata, T> mappingFunction) {
+			super(table, projectedFields, criteria, sort, page);
 			this.typeToRead = typeToRead;
 			this.mappingFunction = mappingFunction;
 		}
@@ -796,6 +823,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return (DefaultTypedSelectSpec<T>) super.project(selectedFields);
 		}
 
+		@Override
+		public DefaultTypedSelectSpec<T> where(Criteria criteria) {
+			return (DefaultTypedSelectSpec<T>) super.where(criteria);
+		}
+
 		@Override
 		public DefaultTypedSelectSpec<T> orderBy(Sort sort) {
 			return (DefaultTypedSelectSpec<T>) super.orderBy(sort);
@@ -821,15 +853,31 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				columns = this.projectedFields;
 			}
 
-			String select = dataAccessStrategy.select(table, new LinkedHashSet<>(columns), sort, page);
+			PreparedOperation<Select> operation = dataAccessStrategy.getStatements().select(table, columns,
+					(table, configurer) -> {
 
-			return execute(select, mappingFunction);
+						Sort sortToUse;
+						if (this.sort.isSorted()) {
+							sortToUse = dataAccessStrategy.getMappedSort(this.sort, this.typeToRead);
+						} else {
+							sortToUse = this.sort;
+						}
+
+						configurer.withPageRequest(page).withSort(sortToUse);
+
+						if (criteria != null) {
+							BoundCondition boundCondition = dataAccessStrategy.getMappedCriteria(criteria, table, this.typeToRead);
+							configurer.withWhere(boundCondition.getCondition()).withBindings(boundCondition.getBindings());
+						}
+					});
+
+			return execute(operation, mappingFunction);
 		}
 
 		@Override
-		protected DefaultTypedSelectSpec<T> createInstance(String table, List<String> projectedFields, Sort sort,
-				Pageable page) {
-			return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead, mappingFunction);
+		protected DefaultTypedSelectSpec<T> createInstance(String table, List<String> projectedFields, Criteria criteria,
+				Sort sort, Pageable page) {
+			return new DefaultTypedSelectSpec<>(table, projectedFields, criteria, sort, page, typeToRead, mappingFunction);
 		}
 	}
 
@@ -923,7 +971,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			};
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
-					sql, //
+					operation.toQuery(), //
 					resultFunction, //
 					it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), //
 					mappingFunction);
@@ -1042,7 +1090,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			};
 
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
-					sql, //
+					operation.toQuery(), //
 					resultFunction, //
 					it -> resultFunction //
 							.apply(it) //
@@ -1052,6 +1100,103 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 	}
 
+	/**
+	 * Default {@link DatabaseClient.DeleteFromSpec} implementation.
+	 */
+	class DefaultDeleteFromSpec implements DeleteFromSpec {
+
+		@Override
+		public DeleteSpec from(String table) {
+			return new DefaultDeleteSpec(null, table, null);
+		}
+
+		@Override
+		public DeleteSpec from(Class<?> table) {
+			return new DefaultDeleteSpec(table, null, null);
+		}
+	}
+
+	/**
+	 * Default implementation of {@link DatabaseClient.TypedInsertSpec}.
+	 */
+	@RequiredArgsConstructor
+	class DefaultDeleteSpec implements DeleteSpec {
+
+		private final @Nullable Class<?> typeToDelete;
+		private final @Nullable String table;
+		private final Criteria where;
+
+		@Override
+		public DeleteSpec where(Criteria criteria) {
+			return new DefaultDeleteSpec(this.typeToDelete, this.table, criteria);
+		}
+
+		@Override
+		public UpdatedRowsFetchSpec fetch() {
+
+			String table;
+
+			if (StringUtils.isEmpty(this.table)) {
+				table = dataAccessStrategy.getTableName(this.typeToDelete);
+			} else {
+				table = this.table;
+			}
+
+			return exchange(table);
+		}
+
+		@Override
+		public Mono<Void> then() {
+			return fetch().rowsUpdated().then();
+		}
+
+		private UpdatedRowsFetchSpec exchange(String table) {
+
+			PreparedOperation<Delete> operation = dataAccessStrategy.getStatements().delete(table, (t, configurer) -> {
+
+				if (this.where != null) {
+
+					BoundCondition condition;
+					if (this.table != null) {
+						condition = dataAccessStrategy.getMappedCriteria(this.where, t);
+					} else {
+						condition = dataAccessStrategy.getMappedCriteria(this.where, t, this.typeToDelete);
+					}
+
+					configurer.withWhere(condition.getCondition()).withBindings(condition.getBindings());
+				}
+			});
+
+			Function<Connection, Statement> deleteFunction = wrapPreparedOperation(operation);
+			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(deleteFunction.apply(it).execute());
+
+			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
+					operation.toQuery(), //
+					resultFunction, //
+					it -> resultFunction //
+							.apply(it) //
+							.flatMap(Result::getRowsUpdated) //
+							.collect(Collectors.summingInt(Integer::intValue)), //
+					(row, rowMetadata) -> rowMetadata);
+		}
+	}
+
+	private Function<Connection, Statement> wrapPreparedOperation(PreparedOperation<?> operation) {
+
+		return it -> {
+
+			String sql = operation.toQuery();
+			if (logger.isDebugEnabled()) {
+				logger.debug("Executing SQL statement [" + sql + "]");
+			}
+
+			Statement statement = it.createStatement(sql);
+			operation.bindTo(new StatementWrapper(statement));
+
+			return statement;
+		};
+	}
+
 	private static <T> Flux<T> doInConnectionMany(Connection connection, Function<Connection, Flux<T>> action) {
 
 		try {

src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java

@@ -19,21 +19,18 @@ import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.OptionalLong;
-import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.convert.CustomConversions.StoreConversions;
-import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.ArrayColumns;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.domain.OutboundRow;
@@ -42,15 +39,13 @@ import org.springframework.data.r2dbc.function.convert.EntityRowMapper;
 import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;
-import org.springframework.data.r2dbc.support.StatementRenderUtil;
+import org.springframework.data.r2dbc.function.query.BoundCondition;
+import org.springframework.data.r2dbc.function.query.Criteria;
+import org.springframework.data.r2dbc.function.query.CriteriaMapper;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.Expression;
-import org.springframework.data.relational.core.sql.OrderByField;
 import org.springframework.data.relational.core.sql.Select;
-import org.springframework.data.relational.core.sql.SelectBuilder.SelectFromAndOrderBy;
-import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.relational.core.sql.render.NamingStrategies;
 import org.springframework.data.relational.core.sql.render.RenderContext;
@@ -69,6 +64,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 	private final Dialect dialect;
 	private final R2dbcConverter converter;
+	private final CriteriaMapper criteriaMapper;
 	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 	private final StatementFactory statements;
 
@@ -94,14 +90,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return new MappingR2dbcConverter(context, customConversions);
 	}
 
-	public R2dbcConverter getConverter() {
-		return converter;
-	}
-
-	public MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
-		return mappingContext;
-	}
-
 	/**
 	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect} and {@link R2dbcConverter}.
 	 *
@@ -115,6 +103,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		Assert.notNull(converter, "RelationalConverter must not be null");
 
 		this.converter = converter;
+		this.criteriaMapper = new CriteriaMapper(converter);
 		this.mappingContext = (MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty>) this.converter
 				.getMappingContext();
 		this.dialect = dialect;
@@ -215,7 +204,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getMappedSort(java.lang.Class, org.springframework.data.domain.Sort)
 	 */
 	@Override
-	public Sort getMappedSort(Class<?> typeToRead, Sort sort) {
+	public Sort getMappedSort(Sort sort, Class<?> typeToRead) {
 
 		RelationalPersistentEntity<?> entity = getPersistentEntity(typeToRead);
 		if (entity == null) {
@@ -238,6 +227,30 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return Sort.by(mappedOrder);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getMappedCriteria(org.springframework.data.r2dbc.function.query.Criteria, org.springframework.data.relational.core.sql.Table)
+	 */
+	@Override
+	public BoundCondition getMappedCriteria(Criteria criteria, Table table) {
+		return getMappedCriteria(criteria, table, null);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getMappedCriteria(org.springframework.data.r2dbc.function.query.Criteria, org.springframework.data.relational.core.sql.Table, java.lang.Class)
+	 */
+	@Override
+	public BoundCondition getMappedCriteria(Criteria criteria, Table table, @Nullable Class<?> typeToRead) {
+
+		BindMarkers bindMarkers = this.dialect.getBindMarkersFactory().create();
+
+		RelationalPersistentEntity<?> entity = typeToRead != null ? mappingContext.getRequiredPersistentEntity(typeToRead)
+				: null;
+
+		return criteriaMapper.getMappedObject(bindMarkers, criteria, table, entity);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getRowMapper(java.lang.Class)
@@ -274,6 +287,18 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return dialect.getBindMarkersFactory();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getConverter()
+	 */
+	public R2dbcConverter getConverter() {
+		return converter;
+	}
+
+	public MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
+		return mappingContext;
+	}
+
 	private RelationalPersistentEntity<?> getRequiredPersistentEntity(Class<?> typeToRead) {
 		return mappingContext.getRequiredPersistentEntity(typeToRead);
 	}
@@ -282,56 +307,4 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	private RelationalPersistentEntity<?> getPersistentEntity(Class<?> typeToRead) {
 		return mappingContext.getPersistentEntity(typeToRead);
 	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#select(java.lang.String, java.util.Set, org.springframework.data.domain.Sort, org.springframework.data.domain.Pageable)
-	 */
-	@Override
-	public String select(String tableName, Set<String> columns, Sort sort, Pageable page) {
-
-		Table table = Table.create(tableName);
-
-		Collection<? extends Expression> selectList;
-
-		if (columns.isEmpty()) {
-			selectList = Collections.singletonList(table.asterisk());
-		} else {
-			selectList = table.columns(columns);
-		}
-
-		SelectFromAndOrderBy selectBuilder = StatementBuilder //
-				.select(selectList) //
-				.from(tableName) //
-				.orderBy(createOrderByFields(table, sort));
-
-		OptionalLong limit = OptionalLong.empty();
-		OptionalLong offset = OptionalLong.empty();
-
-		if (page.isPaged()) {
-			limit = OptionalLong.of(page.getPageSize());
-			offset = OptionalLong.of(page.getOffset());
-		}
-
-		// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
-		return StatementRenderUtil.render(selectBuilder.build(), limit, offset, this.dialect);
-	}
-
-	private Collection<? extends OrderByField> createOrderByFields(Table table, Sort sortToUse) {
-
-		List<OrderByField> fields = new ArrayList<>();
-
-		for (Order order : sortToUse) {
-
-			OrderByField orderByField = OrderByField.from(table.column(order.getProperty()));
-
-			if (order.getDirection() != null) {
-				fields.add(order.isAscending() ? orderByField.asc() : orderByField.desc());
-			} else {
-				fields.add(orderByField);
-			}
-		}
-
-		return fields;
-	}
 }

src/main/java/org/springframework/data/r2dbc/function/DefaultStatementFactory.java

@@ -24,18 +24,23 @@ import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.domain.Bindings;
+import org.springframework.data.r2dbc.domain.PreparedOperation;
 import org.springframework.data.r2dbc.domain.BindTarget;
 import org.springframework.data.r2dbc.domain.PreparedOperation;
 import org.springframework.data.r2dbc.domain.SettableValue;
+import org.springframework.data.r2dbc.support.StatementRenderUtil;
 import org.springframework.data.relational.core.sql.AssignValue;
 import org.springframework.data.relational.core.sql.Assignment;
 import org.springframework.data.relational.core.sql.Column;
@@ -44,6 +49,7 @@ import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.DeleteBuilder;
 import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.Insert;
+import org.springframework.data.relational.core.sql.OrderByField;
 import org.springframework.data.relational.core.sql.SQL;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.SelectBuilder;
@@ -89,6 +95,7 @@ class DefaultStatementFactory implements StatementFactory {
 		binderConsumer.accept(binderBuilder);
 
 		return withDialect((dialect, renderContext) -> {
+
 			Table table = Table.create(tableName);
 			List<Column> columns = table.columns(columnNames);
 			SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
@@ -111,6 +118,63 @@ class DefaultStatementFactory implements StatementFactory {
 		});
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.StatementFactory#select(java.lang.String, java.util.Collection, java.util.function.BiConsumer)
+	 */
+	@Override
+	public PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
+			BiConsumer<Table, SelectConfigurer> configurerConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notEmpty(columnNames, "Columns must not be empty");
+		Assert.notNull(configurerConsumer, "Configurer Consumer must not be null");
+
+		return withDialect((dialect, renderContext) -> {
+
+			DefaultSelectConfigurer configurer = new DefaultSelectConfigurer(dialect.getBindMarkersFactory().create());
+			Table table = Table.create(tableName);
+			configurerConsumer.accept(table, configurer);
+
+			List<Column> columns = table.columns(columnNames);
+			SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
+
+			if (configurer.condition != null) {
+				selectBuilder.where(configurer.condition);
+			}
+
+			if (configurer.sort != null) {
+				selectBuilder.orderBy(createOrderByFields(table, configurer.sort));
+			}
+
+			Select select = selectBuilder.build();
+			return new DefaultPreparedOperation<Select>(select, renderContext, configurer.bindings) {
+				@Override
+				public String toQuery() {
+					return StatementRenderUtil.render(select, configurer.limit, configurer.offset, dialect);
+				}
+			};
+		});
+	}
+
+	private Collection<? extends OrderByField> createOrderByFields(Table table, Sort sortToUse) {
+
+		List<OrderByField> fields = new ArrayList<>();
+
+		for (Sort.Order order : sortToUse) {
+
+			OrderByField orderByField = OrderByField.from(table.column(order.getProperty()));
+
+			if (order.getDirection() != null) {
+				fields.add(order.isAscending() ? orderByField.asc() : orderByField.desc());
+			} else {
+				fields.add(orderByField);
+			}
+		}
+
+		return fields;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.StatementFactory#insert(java.lang.String, java.util.Collection, java.util.function.Consumer)
@@ -155,7 +219,7 @@ class DefaultStatementFactory implements StatementFactory {
 			Insert insert = StatementBuilder.insert().into(table).columns(table.columns(binderBuilder.bindings.keySet()))
 					.values(expressions).build();
 
-			return new DefaultPreparedOperation<Insert>(insert, renderContext, binding);
+			return new DefaultPreparedOperation<Insert>(insert, renderContext, binding.toBindings());
 		});
 	}
 
@@ -204,7 +268,7 @@ class DefaultStatementFactory implements StatementFactory {
 				update = updateBuilder.build();
 			}
 
-			return new DefaultPreparedOperation<>(update, renderContext, binding);
+			return new DefaultPreparedOperation<>(update, renderContext, binding.toBindings());
 		});
 	}
 
@@ -242,7 +306,35 @@ class DefaultStatementFactory implements StatementFactory {
 				delete = deleteBuilder.build();
 			}
 
-			return new DefaultPreparedOperation<>(delete, renderContext, binding);
+			return new DefaultPreparedOperation<>(delete, renderContext, binding.toBindings());
+		});
+	}
+
+	@Override
+	public PreparedOperation<Delete> delete(String tableName, BiConsumer<Table, BindConfigurer> configurerConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notNull(configurerConsumer, "Configurer Consumer must not be null");
+
+		return withDialect((dialect, renderContext) -> {
+
+			Table table = Table.create(tableName);
+			DeleteBuilder.DeleteWhere deleteBuilder = StatementBuilder.delete().from(table);
+
+			BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
+			DefaultBindConfigurer configurer = new DefaultBindConfigurer(bindMarkers);
+
+			configurerConsumer.accept(table, configurer);
+
+			Delete delete;
+
+			if (configurer.condition != null) {
+				delete = deleteBuilder.where(configurer.condition).build();
+			} else {
+				delete = deleteBuilder.build();
+			}
+
+			return new DefaultPreparedOperation<>(delete, renderContext, configurer.bindings);
 		});
 	}
 
@@ -325,7 +417,6 @@ class DefaultStatementFactory implements StatementFactory {
 			});
 
 			return new Binding(values, nulls, conditionRef.get());
-
 		}
 
 		private static Condition toCondition(BindMarkers bindMarkers, Column column, SettableValue value,
@@ -410,6 +501,16 @@ class DefaultStatementFactory implements StatementFactory {
 			values.forEach((marker, value) -> marker.bind(to, value));
 			nulls.forEach((marker, value) -> marker.bindNull(to, value.getType()));
 		}
+
+		Bindings toBindings() {
+
+			List<Bindings.Binding> bindings = new ArrayList<>(values.size() + nulls.size());
+
+			values.forEach((marker, value) -> bindings.add(new Bindings.ValueBinding(marker, value)));
+			nulls.forEach((marker, value) -> bindings.add(new Bindings.NullBinding(marker, value.getType())));
+
+			return new Bindings(bindings);
+		}
 	}
 
 	/**
@@ -422,7 +523,7 @@ class DefaultStatementFactory implements StatementFactory {
 
 		private final T source;
 		private final RenderContext renderContext;
-		private final Binding binding;
+		private final Bindings bindings;
 
 		/*
 		 * (non-Javadoc)
@@ -463,7 +564,129 @@ class DefaultStatementFactory implements StatementFactory {
 
 		@Override
 		public void bindTo(BindTarget target) {
-			binding.apply(target);
+			bindings.apply(target);
+		}
+	}
+
+	/**
+	 * Default {@link SelectConfigurer} implementation.
+	 */
+	static class DefaultSelectConfigurer extends DefaultBindConfigurer implements SelectConfigurer {
+
+		OptionalLong limit = OptionalLong.empty();
+		OptionalLong offset = OptionalLong.empty();
+
+		Sort sort = Sort.unsorted();
+
+		DefaultSelectConfigurer(BindMarkers bindMarkers) {
+			super(bindMarkers);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withBindings(org.springframework.data.r2dbc.function.Bindings)
+		 */
+		@Override
+		public SelectConfigurer withBindings(Bindings bindings) {
+
+			super.withBindings(bindings);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withWhere(org.springframework.data.relational.core.sql.Condition)
+		 */
+		@Override
+		public SelectConfigurer withWhere(Condition condition) {
+
+			super.withWhere(condition);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withLimit(long)
+		 */
+		@Override
+		public SelectConfigurer withLimit(long limit) {
+
+			this.limit = OptionalLong.of(limit);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withOffset(long)
+		 */
+		@Override
+		public SelectConfigurer withOffset(long offset) {
+
+			this.offset = OptionalLong.of(offset);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withSort(org.springframework.data.domain.Sort)
+		 */
+		@Override
+		public SelectConfigurer withSort(Sort sort) {
+
+			Assert.notNull(sort, "Sort must not be null");
+
+			this.sort = sort;
+			return this;
+		}
+	}
+
+	/**
+	 * Default {@link SelectConfigurer} implementation.
+	 */
+	static class DefaultBindConfigurer implements BindConfigurer {
+
+		private final BindMarkers bindMarkers;
+
+		@Nullable Condition condition;
+		Bindings bindings = new Bindings();
+
+		DefaultBindConfigurer(BindMarkers bindMarkers) {
+			this.bindMarkers = bindMarkers;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#bindMarkers()
+		 */
+		@Override
+		public BindMarkers bindMarkers() {
+			return this.bindMarkers;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withBindings(org.springframework.data.r2dbc.function.Bindings)
+		 */
+		@Override
+		public BindConfigurer withBindings(Bindings bindings) {
+
+			Assert.notNull(bindings, "Bindings must not be null");
+
+			this.bindings = Bindings.merge(this.bindings, bindings);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.SelectConfigurer#withWhere(org.springframework.data.relational.core.sql.Condition)
+		 */
+		@Override
+		public BindConfigurer withWhere(Condition condition) {
+
+			Assert.notNull(condition, "Condition must not be null");
+
+			this.condition = condition;
+			return this;
 		}
 	}
 }

src/main/java/org/springframework/data/r2dbc/function/NamedParameterExpander.java

@@ -20,8 +20,10 @@ import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.domain.BindTarget;
+import org.springframework.data.r2dbc.domain.BindableOperation;
 
 /**
  * SQL translation support allowing the use of named parameters rather than native placeholders.

src/main/java/org/springframework/data/r2dbc/function/NamedParameterUtils.java

@@ -31,6 +31,7 @@ import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.domain.BindTarget;
+import org.springframework.data.r2dbc.domain.BindableOperation;
 import org.springframework.util.Assert;
 
 /**

src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java

@@ -19,15 +19,19 @@ import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
 import java.util.List;
-import java.util.Set;
 import java.util.function.BiFunction;
 
-import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.domain.BindableOperation;
+import org.springframework.data.r2dbc.domain.Bindings;
 import org.springframework.data.r2dbc.domain.OutboundRow;
 import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
+import org.springframework.data.r2dbc.function.query.BoundCondition;
+import org.springframework.data.r2dbc.function.query.Criteria;
+import org.springframework.data.relational.core.sql.Table;
 
 /**
  * Draft of a data access strategy that generalizes convenience operations using mapped entities. Typically used
@@ -56,11 +60,30 @@ public interface ReactiveDataAccessStrategy {
 	/**
 	 * Map the {@link Sort} object to apply field name mapping using {@link Class the type to read}.
 	 *
-	 * @param typeToRead
-	 * @param sort
+	 * @param sort must not be {@literal null}.
+	 * @param typeToRead must not be {@literal null}.
 	 * @return
 	 */
-	Sort getMappedSort(Class<?> typeToRead, Sort sort);
+	Sort getMappedSort(Sort sort, Class<?> typeToRead);
+
+	/**
+	 * Map the {@link Criteria} object to apply value mapping and return a {@link BoundCondition} with {@link Bindings}.
+	 *
+	 * @param criteria must not be {@literal null}.
+	 * @param table must not be {@literal null}.
+	 * @return
+	 */
+	BoundCondition getMappedCriteria(Criteria criteria, Table table);
+
+	/**
+	 * Map the {@link Criteria} object to apply value and field name mapping and return a {@link BoundCondition} with
+	 * {@link Bindings}.
+	 *
+	 * @param criteria must not be {@literal null}.
+	 * @param table must not be {@literal null}.
+	 * @return
+	 */
+	BoundCondition getMappedCriteria(Criteria criteria, Table table, Class<?> typeToRead);
 
 	// TODO: Broaden T to Mono<T>/Flux<T> for reactive relational data access?
 	<T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead);
@@ -71,6 +94,11 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	String getTableName(Class<?> type);
 
+	/**
+	 * Returns the {@link Dialect}-specific {@link StatementFactory}.
+	 *
+	 * @return the {@link Dialect}-specific {@link StatementFactory}.
+	 */
 	StatementFactory getStatements();
 
 	/**
@@ -87,20 +115,4 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	R2dbcConverter getConverter();
 
-	// -------------------------------------------------------------------------
-	// Methods creating SQL operations.
-	// Subject to be moved into a SQL creation DSL.
-	// -------------------------------------------------------------------------
-
-	/**
-	 * Create a {@code SELECT … ORDER BY … LIMIT …} operation for the given {@code table} using {@code columns} to
-	 * project.
-	 *
-	 * @param table the table to insert data to.
-	 * @param columns columns to return.
-	 * @param sort
-	 * @param page
-	 * @return
-	 */
-	String select(String table, Set<String> columns, Sort sort, Pageable page);
 }

src/main/java/org/springframework/data/r2dbc/function/StatementFactory.java

@@ -16,15 +16,24 @@
 package org.springframework.data.r2dbc.function;
 
 import java.util.Collection;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.domain.PreparedOperation;
+import org.springframework.data.r2dbc.domain.Bindings;
+import org.springframework.data.r2dbc.domain.PreparedOperation;
 import org.springframework.data.r2dbc.domain.SettableValue;
+import org.springframework.data.relational.core.sql.Condition;
 import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.Insert;
 import org.springframework.data.relational.core.sql.Select;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.relational.core.sql.Update;
+import org.springframework.util.Assert;
 
 /**
  * Interface declaring statement methods that are commonly used for {@code SELECT/INSERT/UPDATE/DELETE} operations.
@@ -47,6 +56,17 @@ public interface StatementFactory {
 	PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
 			Consumer<StatementBinderBuilder> binderConsumer);
 
+	/**
+	 * Creates a {@link Select} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param columnNames the columns to project, must not be {@literal null} or empty.
+	 * @param configurerConsumer customizer for {@link SelectConfigurer}.
+	 * @return the {@link PreparedOperation} to select the given columns.
+	 */
+	PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
+			BiConsumer<Table, SelectConfigurer> configurerConsumer);
+
 	/**
 	 * Creates a {@link Insert} statement.
 	 *
@@ -78,6 +98,15 @@ public interface StatementFactory {
 	 */
 	PreparedOperation<Delete> delete(String tableName, Consumer<StatementBinderBuilder> binderConsumer);
 
+	/**
+	 * Creates a {@link Delete} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param configurerConsumer customizer for {@link SelectConfigurer}.
+	 * @return the {@link PreparedOperation} to delete rows from {@code tableName}.
+	 */
+	PreparedOperation<Delete> delete(String tableName, BiConsumer<Table, BindConfigurer> configurerConsumer);
+
 	/**
 	 * Binder to specify parameter bindings by name. Bindings match to equals comparisons.
 	 */
@@ -100,4 +129,116 @@ public interface StatementFactory {
 		 */
 		void bind(String identifier, SettableValue settable);
 	}
+
+	/**
+	 * Binder to specify parameter bindings by name. Bindings match to equals comparisons.
+	 */
+	interface SelectConfigurer extends BindConfigurer {
+
+		/**
+		 * Returns the {@link BindMarkers} that are currently in use. Bind markers are stateful and represent the current
+		 * state.
+		 *
+		 * @return the {@link BindMarkers} that are currently in use.
+		 * @see #withBindings(Bindings)
+		 */
+		BindMarkers bindMarkers();
+
+		/**
+		 * Apply {@link Bindings} and merge these with already existing bindings.
+		 *
+		 * @param bindings must not be {@literal null}.
+		 * @return {@code this} {@link SelectConfigurer}.
+		 * @see #bindMarkers()
+		 */
+		SelectConfigurer withBindings(Bindings bindings);
+
+		/**
+		 * Apply a {@code WHERE} {@link Condition}. Replaces a previously configured {@link Condition}.
+		 *
+		 * @param condition must not be {@literal null}.
+		 * @return {@code this} {@link SelectConfigurer}.
+		 */
+		SelectConfigurer withWhere(Condition condition);
+
+		/**
+		 * Apply limit/offset and {@link Sort} from {@link Pageable}.
+		 *
+		 * @param pageable must not be {@literal null}.
+		 * @return {@code this} {@link SelectConfigurer}.
+		 */
+		default SelectConfigurer withPageRequest(Pageable pageable) {
+
+			Assert.notNull(pageable, "Pageable must not be null");
+
+			if (pageable.isPaged()) {
+
+				SelectConfigurer configurer = withLimit(pageable.getPageSize()).withOffset(pageable.getOffset());
+
+				if (pageable.getSort().isSorted()) {
+					return configurer.withSort(pageable.getSort());
+				}
+
+				return configurer;
+			}
+
+			return this;
+		}
+
+		/**
+		 * Apply a row limit.
+		 *
+		 * @param limit
+		 * @return {@code this} {@link SelectConfigurer}.
+		 */
+		SelectConfigurer withLimit(long limit);
+
+		/**
+		 * Apply a row offset.
+		 *
+		 * @param offset
+		 * @return {@code this} {@link SelectConfigurer}.
+		 */
+		SelectConfigurer withOffset(long offset);
+
+		/**
+		 * Apply an {@code ORDER BY} {@link Sort}. Replaces a previously configured {@link Sort}.
+		 *
+		 * @param sort must not be {@literal null}.
+		 * @return {@code this} {@link SelectConfigurer}.
+		 */
+		SelectConfigurer withSort(Sort sort);
+	}
+
+	/**
+	 * Binder to specify parameter bindings by name. Bindings match to equals comparisons.
+	 */
+	interface BindConfigurer {
+
+		/**
+		 * Returns the {@link BindMarkers} that are currently in use. Bind markers are stateful and represent the current
+		 * state.
+		 *
+		 * @return the {@link BindMarkers} that are currently in use.
+		 * @see #withBindings(Bindings)
+		 */
+		BindMarkers bindMarkers();
+
+		/**
+		 * Apply {@link Bindings} and merge these with already existing bindings.
+		 *
+		 * @param bindings must not be {@literal null}.
+		 * @return {@code this} {@link BindConfigurer}.
+		 * @see #bindMarkers()
+		 */
+		BindConfigurer withBindings(Bindings bindings);
+
+		/**
+		 * Apply a {@code WHERE} {@link Condition}. Replaces a previously configured {@link Condition}.
+		 *
+		 * @param condition must not be {@literal null}.
+		 * @return {@code this} {@link BindConfigurer}.
+		 */
+		BindConfigurer withWhere(Condition condition);
+	}
 }

src/main/java/org/springframework/data/r2dbc/function/query/BoundCondition.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function.query;
+
+import org.springframework.data.r2dbc.domain.Bindings;
+import org.springframework.data.relational.core.sql.Condition;
+import org.springframework.util.Assert;
+
+/**
+ * Value object representing a {@link Condition} with its {@link Bindings}.
+ *
+ * @author Mark Paluch
+ */
+public class BoundCondition {
+
+	private final Bindings bindings;
+
+	private final Condition condition;
+
+	public BoundCondition(Bindings bindings, Condition condition) {
+
+		Assert.notNull(bindings, "Bindings must not be null");
+		Assert.notNull(condition, "Condition must not be null");
+
+		this.bindings = bindings;
+		this.condition = condition;
+	}
+
+	public Bindings getBindings() {
+		return bindings;
+	}
+
+	public Condition getCondition() {
+		return condition;
+	}
+}

src/main/java/org/springframework/data/r2dbc/function/query/Criteria.java

@@ -0,0 +1,440 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function.query;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Central class for creating queries. It follows a fluent API style so that you can easily chain together multiple
+ * criteria. Static import of the {@code Criteria.property(…)} method will improve readability as in
+ * {@code where(property(…).is(…)}.
+ *
+ * @author Mark Paluch
+ */
+public class Criteria {
+
+	private final @Nullable Criteria previous;
+	private final Combinator combinator;
+
+	private final String property;
+	private final Comparator comparator;
+	private final @Nullable Object value;
+
+	private Criteria(String property, Comparator comparator, @Nullable Object value) {
+		this(null, Combinator.INITIAL, property, comparator, value);
+	}
+
+	private Criteria(@Nullable Criteria previous, Combinator combinator, String property, Comparator comparator,
+			@Nullable Object value) {
+		this.previous = previous;
+		this.combinator = combinator;
+		this.property = property;
+		this.comparator = comparator;
+		this.value = value;
+	}
+
+	/**
+	 * Static factory method to create a Criteria using the provided {@code property} name.
+	 *
+	 * @param property
+	 * @return a new {@link CriteriaStep} object to complete the first {@link Criteria}.
+	 */
+	public static CriteriaStep of(String property) {
+
+		Assert.notNull(property, "Property name must not be null!");
+
+		return new DefaultCriteriaStep(property);
+	}
+
+	/**
+	 * Create a new {@link Criteria} and combine it with {@code AND} using the provided {@code property} name.
+	 *
+	 * @param property
+	 * @return a new {@link CriteriaStep} object to complete the next {@link Criteria}.
+	 */
+	public CriteriaStep and(String property) {
+
+		Assert.notNull(property, "Property name must not be null!");
+
+		return new DefaultCriteriaStep(property) {
+			@Override
+			protected Criteria createCriteria(Comparator comparator, Object value) {
+				return new Criteria(Criteria.this, Combinator.AND, property, comparator, value);
+			}
+		};
+	}
+
+	/**
+	 * Create a new {@link Criteria} and combine it with {@code OR} using the provided {@code property} name.
+	 *
+	 * @param property
+	 * @return a new {@link CriteriaStep} object to complete the next {@link Criteria}.
+	 */
+	public CriteriaStep or(String property) {
+
+		Assert.notNull(property, "Property name must not be null!");
+
+		return new DefaultCriteriaStep(property) {
+			@Override
+			protected Criteria createCriteria(Comparator comparator, Object value) {
+				return new Criteria(Criteria.this, Combinator.OR, property, comparator, value);
+			}
+		};
+	}
+
+	/**
+	 * @return the previous {@link Criteria} object. Can be {@literal null} if there is no previous {@link Criteria}.
+	 * @see #hasPrevious()
+	 */
+	@Nullable
+	Criteria getPrevious() {
+		return previous;
+	}
+
+	/**
+	 * @return {@literal true} if this {@link Criteria} has a previous one.
+	 */
+	boolean hasPrevious() {
+		return previous != null;
+	}
+
+	/**
+	 * @return {@link Combinator} to combine this criteria with a previous one.
+	 */
+	Combinator getCombinator() {
+		return combinator;
+	}
+
+	/**
+	 * @return the property name.
+	 */
+	String getProperty() {
+		return property;
+	}
+
+	/**
+	 * @return {@link Comparator}.
+	 */
+	Comparator getComparator() {
+		return comparator;
+	}
+
+	/**
+	 * @return the comparison value. Can be {@literal null}.
+	 */
+	@Nullable
+	Object getValue() {
+		return value;
+	}
+
+	enum Comparator {
+		EQ, NEQ, LT, LTE, GT, GTE, IS_NULL, IS_NOT_NULL, LIKE, NOT_IN, IN,
+	}
+
+	enum Combinator {
+		INITIAL, AND, OR;
+	}
+
+	/**
+	 * Interface declaring terminal builder methods to build a {@link Criteria}.
+	 */
+	public interface CriteriaStep {
+
+		/**
+		 * Creates a {@link Criteria} using equality.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria is(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using equality (is not).
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria not(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using {@code IN}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria in(Object... values);
+
+		/**
+		 * Creates a {@link Criteria} using {@code IN}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria in(Collection<? extends Object> values);
+
+		/**
+		 * Creates a {@link Criteria} using {@code NOT IN}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria notIn(Object... values);
+
+		/**
+		 * Creates a {@link Criteria} using {@code NOT IN}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria notIn(Collection<? extends Object> values);
+
+		/**
+		 * Creates a {@link Criteria} using less-than ({@literal <}).
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria lessThan(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using less-than or equal to ({@literal <=}).
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria lessThanOrEquals(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using greater-than({@literal >}).
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria greaterThan(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using greater-than or equal to ({@literal >=}).
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria greaterThanOrEquals(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using {@code LIKE}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria like(Object value);
+
+		/**
+		 * Creates a {@link Criteria} using {@code IS NULL}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria isNull();
+
+		/**
+		 * Creates a {@link Criteria} using {@code IS NOT NULL}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		Criteria isNotNull();
+	}
+
+	/**
+	 * Default {@link CriteriaStep} implementation.
+	 */
+	@RequiredArgsConstructor
+	static class DefaultCriteriaStep implements CriteriaStep {
+
+		private final String property;
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#is(java.lang.Object)
+		 */
+		@Override
+		public Criteria is(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.EQ, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#not(java.lang.Object)
+		 */
+		@Override
+		public Criteria not(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.NEQ, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#in(java.lang.Object[])
+		 */
+		@Override
+		public Criteria in(Object... values) {
+
+			Assert.notNull(values, "Values must not be null!");
+
+			if (values.length > 1 && values[1] instanceof Collection) {
+				throw new InvalidDataAccessApiUsageException(
+						"You can only pass in one argument of type " + values[1].getClass().getName());
+			}
+
+			return createCriteria(Comparator.IN, Arrays.asList(values));
+		}
+
+		/**
+		 * @param values
+		 * @return
+		 */
+		@Override
+		public Criteria in(Collection<?> values) {
+
+			Assert.notNull(values, "Values must not be null!");
+
+			return createCriteria(Comparator.IN, values);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#notIn(java.lang.Object[])
+		 */
+		@Override
+		public Criteria notIn(Object... values) {
+
+			Assert.notNull(values, "Values must not be null!");
+
+			if (values.length > 1 && values[1] instanceof Collection) {
+				throw new InvalidDataAccessApiUsageException(
+						"You can only pass in one argument of type " + values[1].getClass().getName());
+			}
+
+			return createCriteria(Comparator.NOT_IN, Arrays.asList(values));
+		}
+
+		/**
+		 * @param values
+		 * @return
+		 */
+		@Override
+		public Criteria notIn(Collection<?> values) {
+
+			Assert.notNull(values, "Values must not be null!");
+
+			return createCriteria(Comparator.NOT_IN, values);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#lessThan(java.lang.Object)
+		 */
+		@Override
+		public Criteria lessThan(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.LT, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#lessThanOrEquals(java.lang.Object)
+		 */
+		@Override
+		public Criteria lessThanOrEquals(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.LTE, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#greaterThan(java.lang.Object)
+		 */
+		@Override
+		public Criteria greaterThan(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.GT, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#greaterThanOrEquals(java.lang.Object)
+		 */
+		@Override
+		public Criteria greaterThanOrEquals(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.GTE, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#like(java.lang.Object)
+		 */
+		@Override
+		public Criteria like(Object value) {
+
+			Assert.notNull(value, "Value must not be null");
+
+			return createCriteria(Comparator.LIKE, value);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#isNull()
+		 */
+		@Override
+		public Criteria isNull() {
+			return createCriteria(Comparator.IS_NULL, null);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.query.Criteria.CriteriaStep#isNotNull()
+		 */
+		@Override
+		public Criteria isNotNull() {
+			return createCriteria(Comparator.IS_NOT_NULL, null);
+		}
+
+		protected Criteria createCriteria(Comparator comparator, Object value) {
+			return new Criteria(property, comparator, value);
+		}
+	}
+}

src/main/java/org/springframework/data/r2dbc/function/query/CriteriaMapper.java

@@ -0,0 +1,413 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function.query;
+
+import static org.springframework.data.r2dbc.function.query.Criteria.*;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.mapping.PropertyPath;
+import org.springframework.data.mapping.PropertyReferenceException;
+import org.springframework.data.mapping.context.InvalidPersistentPropertyPath;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.domain.Bindings;
+import org.springframework.data.r2dbc.domain.MutableBindings;
+import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+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.SQL;
+import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Maps a {@link Criteria} to {@link Condition} considering mapping metadata.
+ *
+ * @author Mark Paluch
+ */
+public class CriteriaMapper {
+
+	private final R2dbcConverter converter;
+	private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+
+	/**
+	 * Creates a new {@link CriteriaMapper} with the given {@link R2dbcConverter}.
+	 *
+	 * @param converter must not be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	public CriteriaMapper(R2dbcConverter converter) {
+
+		Assert.notNull(converter, "R2dbcConverter must not be null!");
+
+		this.converter = converter;
+		this.mappingContext = (MappingContext) converter.getMappingContext();
+	}
+
+	/**
+	 * Map a {@link Criteria} object into {@link Condition} and consider value/{@code NULL} {@link Bindings}.
+	 *
+	 * @param markers bind markers object, must not be {@literal null}.
+	 * @param criteria criteria to map, must not be {@literal null}.
+	 * @param table must not be {@literal null}.
+	 * @param entity related {@link RelationalPersistentEntity}.
+	 * @return the mapped bindings.
+	 */
+	public BoundCondition getMappedObject(BindMarkers markers, Criteria criteria, Table table,
+			@Nullable RelationalPersistentEntity<?> entity) {
+
+		Assert.notNull(markers, "BindMarkers must not be null!");
+		Assert.notNull(criteria, "Criteria must not be null!");
+
+		Criteria current = criteria;
+		MutableBindings bindings = new MutableBindings(markers);
+
+		// reverse unroll criteria chain
+		Map<Criteria, Criteria> forwardChain = new HashMap<>();
+
+		while (current.hasPrevious()) {
+			forwardChain.put(current.getPrevious(), current);
+			current = current.getPrevious();
+		}
+
+		// perform the actual mapping
+		Condition mapped = getCondition(current, bindings, table, entity);
+		while (forwardChain.containsKey(current)) {
+
+			Criteria nextCriteria = forwardChain.get(current);
+
+			if (nextCriteria.getCombinator() == Combinator.AND) {
+				mapped = mapped.and(getCondition(nextCriteria, bindings, table, entity));
+			}
+
+			if (nextCriteria.getCombinator() == Combinator.OR) {
+				mapped = mapped.or(getCondition(nextCriteria, bindings, table, entity));
+			}
+
+			current = nextCriteria;
+		}
+
+		return new BoundCondition(bindings, mapped);
+	}
+
+	private Condition getCondition(Criteria criteria, MutableBindings bindings, Table table,
+			@Nullable RelationalPersistentEntity<?> entity) {
+
+		Field propertyField = createPropertyField(entity, criteria.getProperty(), this.mappingContext);
+		Column column = table.column(propertyField.getMappedColumnName());
+		Object mappedValue = convertValue(criteria.getValue(), propertyField.getTypeHint());
+
+		TypeInformation<?> actualType = propertyField.getTypeHint().getRequiredActualType();
+		return createCondition(column, mappedValue, actualType.getType(), bindings, criteria.getComparator());
+	}
+
+	@Nullable
+
+	private Object convertValue(@Nullable Object value, TypeInformation<?> typeInformation) {
+
+		if (value == null) {
+			return null;
+		}
+
+		if (typeInformation.isCollectionLike()) {
+			converter.writeValue(value, typeInformation);
+		} else if (value instanceof Iterable) {
+
+			List<Object> mapped = new ArrayList<>();
+
+			for (Object o : (Iterable<?>) value) {
+
+				mapped.add(converter.writeValue(o, typeInformation));
+			}
+			return mapped;
+		}
+
+		return converter.writeValue(value, typeInformation);
+	}
+
+	private Condition createCondition(Column column, @Nullable Object mappedValue, Class<?> valueType,
+			MutableBindings bindings, Comparator comparator) {
+
+		switch (comparator) {
+			case IS_NULL:
+				return column.isNull();
+			case IS_NOT_NULL:
+				return column.isNotNull();
+		}
+
+		if (comparator == Comparator.NOT_IN || comparator == Comparator.IN) {
+
+			Condition condition;
+			if (mappedValue instanceof Iterable) {
+
+				List<Expression> expressions = new ArrayList<>(
+						mappedValue instanceof Collection ? ((Collection) mappedValue).size() : 10);
+
+				for (Object o : (Iterable<?>) mappedValue) {
+
+					BindMarker bindMarker = bindings.nextMarker(column.getName());
+					expressions.add(bind(o, valueType, bindings, bindMarker));
+				}
+
+				condition = column.in(expressions.toArray(new Expression[0]));
+
+			} else {
+				BindMarker bindMarker = bindings.nextMarker(column.getName());
+				Expression expression = bind(mappedValue, valueType, bindings, bindMarker);
+
+				condition = column.in(expression);
+			}
+
+			if (comparator == Comparator.NOT_IN) {
+				condition = condition.not();
+			}
+
+			return condition;
+		}
+
+		BindMarker bindMarker = bindings.nextMarker(column.getName());
+		Expression expression = bind(mappedValue, valueType, bindings, bindMarker);
+
+		switch (comparator) {
+			case EQ:
+				return column.isEqualTo(expression);
+			case NEQ:
+				return column.isNotEqualTo(expression);
+			case LT:
+				return column.isLess(expression);
+			case LTE:
+				return column.isLessOrEqualTo(expression);
+			case GT:
+				return column.isGreater(expression);
+			case GTE:
+				return column.isGreaterOrEqualTo(expression);
+			case LIKE:
+				return column.like(expression);
+		}
+
+		throw new UnsupportedOperationException("Comparator " + comparator + " not supported");
+	}
+
+	protected Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, String key,
+			MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
+		return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext);
+	}
+
+	private Expression bind(@Nullable Object mappedValue, Class<?> valueType, MutableBindings bindings,
+			BindMarker bindMarker) {
+
+		if (mappedValue != null) {
+			bindings.bind(bindMarker, mappedValue);
+		} else {
+			bindings.bindNull(bindMarker, valueType);
+		}
+
+		return SQL.bindMarker(bindMarker.getPlaceholder());
+	}
+
+	/**
+	 * Value object to represent a field and its meta-information.
+	 */
+	protected static class Field {
+
+		protected final String name;
+
+		/**
+		 * Creates a new {@link Field} without meta-information but the given name.
+		 *
+		 * @param name must not be {@literal null} or empty.
+		 */
+		public Field(String name) {
+
+			Assert.hasText(name, "Name must not be null!");
+			this.name = name;
+		}
+
+		/**
+		 * Returns the underlying {@link RelationalPersistentProperty} backing the field. For path traversals this will be
+		 * the property that represents the value to handle. This means it'll be the leaf property for plain paths or the
+		 * association property in case we refer to an association somewhere in the path.
+		 *
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public RelationalPersistentProperty getProperty() {
+			return null;
+		}
+
+		/**
+		 * Returns the {@link RelationalPersistentEntity} that field is owned by.
+		 *
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public RelationalPersistentEntity<?> getPropertyEntity() {
+			return null;
+		}
+
+		/**
+		 * Returns the key to be used in the mapped document eventually.
+		 *
+		 * @return
+		 */
+		public String getMappedColumnName() {
+			return name;
+		}
+
+		public TypeInformation<?> getTypeHint() {
+			return ClassTypeInformation.OBJECT;
+		}
+	}
+
+	/**
+	 * Extension of {@link Field} to be backed with mapping metadata.
+	 */
+	protected static class MetadataBackedField extends Field {
+
+		private static final String INVALID_ASSOCIATION_REFERENCE = "Invalid path reference %s! Associations can only be pointed to directly or via their id property!";
+
+		private final RelationalPersistentEntity<?> entity;
+		private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+		private final RelationalPersistentProperty property;
+		private final @Nullable PersistentPropertyPath<RelationalPersistentProperty> path;
+
+		/**
+		 * Creates a new {@link MetadataBackedField} with the given name, {@link RelationalPersistentEntity} and
+		 * {@link MappingContext}.
+		 *
+		 * @param name must not be {@literal null} or empty.
+		 * @param entity must not be {@literal null}.
+		 * @param context must not be {@literal null}.
+		 */
+		protected MetadataBackedField(String name, RelationalPersistentEntity<?> entity,
+				MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context) {
+			this(name, entity, context, null);
+		}
+
+		/**
+		 * Creates a new {@link MetadataBackedField} with the given name, {@link RelationalPersistentEntity} and
+		 * {@link MappingContext} with the given {@link RelationalPersistentProperty}.
+		 *
+		 * @param name must not be {@literal null} or empty.
+		 * @param entity must not be {@literal null}.
+		 * @param context must not be {@literal null}.
+		 * @param property may be {@literal null}.
+		 */
+		protected MetadataBackedField(String name, RelationalPersistentEntity<?> entity,
+				MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context,
+				@Nullable RelationalPersistentProperty property) {
+
+			super(name);
+
+			Assert.notNull(entity, "MongoPersistentEntity must not be null!");
+
+			this.entity = entity;
+			this.mappingContext = context;
+
+			this.path = getPath(name);
+			this.property = path == null ? property : path.getLeafProperty();
+		}
+
+		@Override
+		public RelationalPersistentProperty getProperty() {
+			return property;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.convert.QueryMapper.Field#getEntity()
+		 */
+		@Override
+		public RelationalPersistentEntity<?> getPropertyEntity() {
+			RelationalPersistentProperty property = getProperty();
+			return property == null ? null : mappingContext.getPersistentEntity(property);
+		}
+
+		@Override
+		public String getMappedColumnName() {
+			return path == null ? name : path.toDotPath(RelationalPersistentProperty::getColumnName);
+		}
+
+		@Nullable
+		protected PersistentPropertyPath<RelationalPersistentProperty> getPath() {
+			return path;
+		}
+
+		/**
+		 * Returns the {@link PersistentPropertyPath} for the given {@code pathExpression}.
+		 *
+		 * @param pathExpression
+		 * @return
+		 */
+		@Nullable
+		private PersistentPropertyPath<RelationalPersistentProperty> getPath(String pathExpression) {
+
+			try {
+
+				PropertyPath path = PropertyPath.from(pathExpression, entity.getTypeInformation());
+
+				if (isPathToJavaLangClassProperty(path)) {
+					return null;
+				}
+
+				return mappingContext.getPersistentPropertyPath(path);
+			} catch (PropertyReferenceException | InvalidPersistentPropertyPath e) {
+				return null;
+			}
+		}
+
+		private boolean isPathToJavaLangClassProperty(PropertyPath path) {
+
+			if (path.getType().equals(Class.class) && path.getLeafProperty().getOwningType().getType().equals(Class.class)) {
+				return true;
+			}
+			return false;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.convert.QueryMapper.Field#getTypeHint()
+		 */
+		@Override
+		public TypeInformation<?> getTypeHint() {
+
+			RelationalPersistentProperty property = getProperty();
+
+			if (property == null) {
+				return super.getTypeHint();
+			}
+
+			if (property.getActualType().isInterface()
+					|| java.lang.reflect.Modifier.isAbstract(property.getActualType().getModifiers())) {
+				return ClassTypeInformation.OBJECT;
+			}
+
+			return property.getTypeInformation();
+		}
+	}
+}

src/main/java/org/springframework/data/r2dbc/function/query/package-info.java

@@ -0,0 +1,6 @@
+/**
+ * Query and update support.
+ */
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.data.r2dbc.function.query;

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

@@ -31,7 +31,6 @@ import org.springframework.data.r2dbc.domain.PreparedOperation;
 import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
-import org.springframework.data.r2dbc.function.StatementFactory;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
 import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.Functions;
@@ -123,8 +122,6 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
 		String idColumnName = getIdColumnName();
 
-		StatementFactory statements;
-
 		PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(), columns,
 				binder -> {
 					binder.filterBy(idColumnName, SettableValue.from(id));

src/test/java/org/springframework/data/r2dbc/domain/BindingsUnitTests.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.domain;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+
+/**
+ * Unit tests for {@link Bindings}.
+ *
+ * @author Mark Paluch
+ */
+public class BindingsUnitTests {
+
+	BindMarkersFactory markersFactory = BindMarkersFactory.indexed("$", 1);
+	Statement statementMock = mock(Statement.class);
+
+	@Test // gh-64
+	public void shouldCreateBindings() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		bindings.bind(bindings.nextMarker(), "foo");
+		bindings.bindNull(bindings.nextMarker(), String.class);
+
+		assertThat(bindings.stream()).hasSize(2);
+	}
+
+	@Test // gh-64
+	public void shouldApplyValueBinding() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		bindings.bind(bindings.nextMarker(), "foo");
+		bindings.apply(statementMock);
+
+		verify(statementMock).bind(0, "foo");
+	}
+
+	@Test // gh-64
+	public void shouldApplySimpleValueBinding() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		BindMarker marker = bindings.bind("foo");
+		bindings.apply(statementMock);
+
+		assertThat(marker.getPlaceholder()).isEqualTo("$1");
+		verify(statementMock).bind(0, "foo");
+	}
+
+	@Test // gh-64
+	public void shouldApplyNullBinding() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		bindings.bindNull(bindings.nextMarker(), String.class);
+
+		bindings.apply(statementMock);
+
+		verify(statementMock).bindNull(0, String.class);
+	}
+
+	@Test // gh-64
+	public void shouldApplySimpleNullBinding() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		BindMarker marker = bindings.bindNull(String.class);
+		bindings.apply(statementMock);
+
+		assertThat(marker.getPlaceholder()).isEqualTo("$1");
+		verify(statementMock).bindNull(0, String.class);
+	}
+
+	@Test // gh-64
+	public void shouldConsumeBindings() {
+
+		MutableBindings bindings = new MutableBindings(markersFactory.create());
+
+		bindings.bind(bindings.nextMarker(), "foo");
+		bindings.bindNull(bindings.nextMarker(), String.class);
+
+		AtomicInteger counter = new AtomicInteger();
+
+		bindings.forEach(binding -> {
+
+			if (binding.hasValue()) {
+				counter.incrementAndGet();
+				assertThat(binding.getValue()).isEqualTo("foo");
+				assertThat(binding.getBindMarker().getPlaceholder()).isEqualTo("$1");
+			}
+
+			if (binding.isNull()) {
+				counter.incrementAndGet();
+
+				assertThat(((Bindings.NullBinding) binding).getValueType()).isEqualTo(String.class);
+				assertThat(binding.getBindMarker().getPlaceholder()).isEqualTo("$2");
+			}
+		});
+
+		assertThat(counter).hasValue(2);
+	}
+
+	@Test // gh-64
+	public void shouldMergeBindings() {
+
+		BindMarkers markers = markersFactory.create();
+
+		BindMarker shared = markers.next();
+		BindMarker leftMarker = markers.next();
+		List<Bindings.Binding> left = new ArrayList<>();
+		left.add(new Bindings.NullBinding(shared, String.class));
+		left.add(new Bindings.ValueBinding(leftMarker, "left"));
+
+		BindMarker rightMarker = markers.next();
+		List<Bindings.Binding> right = new ArrayList<>();
+		left.add(new Bindings.ValueBinding(shared, "override"));
+		left.add(new Bindings.ValueBinding(rightMarker, "right"));
+
+		Bindings merged = Bindings.merge(new Bindings(left), new Bindings(right));
+
+		assertThat(merged).hasSize(3);
+
+		merged.apply(statementMock);
+		verify(statementMock).bind(0, "override");
+		verify(statementMock).bind(1, "left");
+		verify(statementMock).bind(2, "right");
+	}
+
+}

src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java

@@ -32,6 +32,7 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.r2dbc.function.query.Criteria;
 import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
 import org.springframework.data.relational.core.mapping.Table;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -204,6 +205,43 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 		assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
 	}
 
+	@Test // gh-64
+	public void deleteUntyped() {
+
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42064, 'FORSCHUNGSSCHIFF', 13)");
+
+		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
+
+		databaseClient.delete() //
+				.from("legoset") //
+				.where(Criteria.of("id").is(42055)) //
+				.fetch() //
+				.rowsUpdated() //
+				.as(StepVerifier::create) //
+				.expectNext(1).verifyComplete();
+
+		assertThat(jdbc.queryForList("SELECT id AS count FROM legoset")).hasSize(1);
+	}
+
+	@Test // gh-64
+	public void deleteTyped() {
+
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42064, 'FORSCHUNGSSCHIFF', 13)");
+
+		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
+
+		databaseClient.delete() //
+				.from(LegoSet.class) //
+				.where(Criteria.of("id").is(42055)) //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		assertThat(jdbc.queryForList("SELECT id AS count FROM legoset")).hasSize(1);
+	}
+
 	@Test // gh-2
 	public void selectAsMap() {
 
@@ -241,6 +279,44 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 				.verifyComplete();
 	}
 
+	@Test // gh-8
+	public void selectWithCriteria() {
+
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
+
+		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
+
+		databaseClient.select().from("legoset") //
+				.project("id", "name", "manual") //
+				.orderBy(Sort.by("id")) //
+				.where(Criteria.of("id").greaterThanOrEquals(42055).and("id").lessThanOrEquals(42055))
+				.map((r, md) -> r.get("id", Integer.class)) //
+				.all() //
+				.as(StepVerifier::create) //
+				.expectNext(42055) //
+				.verifyComplete();
+	}
+
+	@Test // gh-64
+	public void selectWithCriteriaIn() {
+
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42064, 'FORSCHUNGSSCHIFF', 13)");
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42068, 'FLUGHAFEN-LÖSCHFAHRZEUG', 13)");
+
+		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
+
+		databaseClient.select().from(LegoSet.class) //
+				.orderBy(Sort.by("id")) //
+				.where(Criteria.of("id").in(42055, 42064)) //
+				.map((r, md) -> r.get("id", Integer.class)) //
+				.all() //
+				.as(StepVerifier::create) //
+				.expectNext(42055) //
+				.expectNext(42064) //
+				.verifyComplete();
+	}
+
 	@Test // gh-2
 	public void selectOrderByIdDesc() {
 

src/test/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientUnitTests.java

@@ -30,6 +30,7 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
+
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 
 /**

src/test/java/org/springframework/data/r2dbc/function/NamedParameterUtilsUnitTests.java

@@ -27,6 +27,7 @@ import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
 import org.springframework.data.r2dbc.dialect.SqlServerDialect;
 import org.springframework.data.r2dbc.domain.BindTarget;
+import org.springframework.data.r2dbc.domain.BindableOperation;
 
 /**
  * Unit tests for {@link NamedParameterUtils}.

src/test/java/org/springframework/data/r2dbc/function/query/CriteriaMapperUnitTests.java

@@ -0,0 +1,218 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function.query;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.r2dbc.spi.Statement;
+
+import org.junit.Test;
+
+import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
+import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
+import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.sql.Table;
+
+/**
+ * Unit tests for {@link CriteriaMapper}.
+ *
+ * @author Mark Paluch
+ */
+public class CriteriaMapperUnitTests {
+
+	R2dbcConverter converter = new MappingR2dbcConverter(new RelationalMappingContext());
+	CriteriaMapper mapper = new CriteriaMapper(converter);
+	Statement statementMock = mock(Statement.class);
+
+	@Test // gh-64
+	public void shouldMapSimpleCriteria() {
+
+		Criteria criteria = Criteria.of("name").is("foo");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name = ?[$1]");
+
+		bindings.getBindings().apply(statementMock);
+		verify(statementMock).bind(0, "foo");
+	}
+
+	@Test // gh-64
+	public void shouldConsiderColumnName() {
+
+		Criteria criteria = Criteria.of("alternative").is("foo");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.another_name = ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapAndCriteria() {
+
+		Criteria criteria = Criteria.of("name").is("foo").and("bar").is("baz");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name = ?[$1] AND person.bar = ?[$2]");
+
+		bindings.getBindings().apply(statementMock);
+		verify(statementMock).bind(0, "foo");
+		verify(statementMock).bind(1, "baz");
+	}
+
+	@Test // gh-64
+	public void shouldMapOrCriteria() {
+
+		Criteria criteria = Criteria.of("name").is("foo").or("bar").is("baz");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name = ?[$1] OR person.bar = ?[$2]");
+	}
+
+	@Test // gh-64
+	public void shouldMapAndOrCriteria() {
+
+		Criteria criteria = Criteria.of("name").is("foo") //
+				.and("name").isNotNull() //
+				.or("bar").is("baz") //
+				.and("anotherOne").is("alternative");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo(
+				"person.name = ?[$1] AND person.name IS NOT NULL OR person.bar = ?[$2] AND person.anotherOne = ?[$3]");
+	}
+
+	@Test // gh-64
+	public void shouldMapNeq() {
+
+		Criteria criteria = Criteria.of("name").not("foo");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name != ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsNull() {
+
+		Criteria criteria = Criteria.of("name").isNull();
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name IS NULL");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsNotNull() {
+
+		Criteria criteria = Criteria.of("name").isNotNull();
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name IS NOT NULL");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsIn() {
+
+		Criteria criteria = Criteria.of("name").in("a", "b", "c");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name IN (?[$1], ?[$2], ?[$3])");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsNotIn() {
+
+		Criteria criteria = Criteria.of("name").notIn("a", "b", "c");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("NOT person.name IN (?[$1], ?[$2], ?[$3])");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsGt() {
+
+		Criteria criteria = Criteria.of("name").greaterThan("a");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name > ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsGte() {
+
+		Criteria criteria = Criteria.of("name").greaterThanOrEquals("a");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name >= ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsLt() {
+
+		Criteria criteria = Criteria.of("name").lessThan("a");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name < ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsLte() {
+
+		Criteria criteria = Criteria.of("name").lessThanOrEquals("a");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name <= ?[$1]");
+	}
+
+	@Test // gh-64
+	public void shouldMapIsLike() {
+
+		Criteria criteria = Criteria.of("name").like("a");
+
+		BoundCondition bindings = map(criteria);
+
+		assertThat(bindings.getCondition().toString()).isEqualTo("person.name LIKE ?[$1]");
+	}
+
+	@SuppressWarnings("unchecked")
+	private BoundCondition map(Criteria criteria) {
+
+		BindMarkersFactory markers = BindMarkersFactory.indexed("$", 1);
+
+		return mapper.getMappedObject(markers.create(), criteria, Table.create("person"),
+				converter.getMappingContext().getRequiredPersistentEntity(Person.class));
+	}
+
+	static class Person {
+
+		String name;
+		@Column("another_name") String alternative;
+	}
+}

src/test/java/org/springframework/data/r2dbc/function/query/CriteriaUnitTests.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function.query;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.r2dbc.function.query.Criteria.*;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+
+import org.springframework.data.r2dbc.function.query.Criteria.*;
+
+/**
+ * Unit tests for {@link Criteria}.
+ *
+ * @author Mark Paluch
+ */
+public class CriteriaUnitTests {
+
+	@Test // gh-64
+	public void andChainedCriteria() {
+
+		Criteria criteria = of("foo").is("bar").and("baz").isNotNull();
+
+		assertThat(criteria.getProperty()).isEqualTo("baz");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NOT_NULL);
+		assertThat(criteria.getValue()).isNull();
+		assertThat(criteria.getPrevious()).isNotNull();
+		assertThat(criteria.getCombinator()).isEqualTo(Combinator.AND);
+
+		criteria = criteria.getPrevious();
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.EQ);
+		assertThat(criteria.getValue()).isEqualTo("bar");
+	}
+
+	@Test // gh-64
+	public void orChainedCriteria() {
+
+		Criteria criteria = of("foo").is("bar").or("baz").isNotNull();
+
+		assertThat(criteria.getProperty()).isEqualTo("baz");
+		assertThat(criteria.getCombinator()).isEqualTo(Combinator.OR);
+
+		criteria = criteria.getPrevious();
+
+		assertThat(criteria.getPrevious()).isNull();
+		assertThat(criteria.getValue()).isEqualTo("bar");
+	}
+
+	@Test // gh-64
+	public void shouldBuildEqualsCriteria() {
+
+		Criteria criteria = of("foo").is("bar");
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.EQ);
+		assertThat(criteria.getValue()).isEqualTo("bar");
+	}
+
+	@Test // gh-64
+	public void shouldBuildNotEqualsCriteria() {
+
+		Criteria criteria = of("foo").not("bar");
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.NEQ);
+		assertThat(criteria.getValue()).isEqualTo("bar");
+	}
+
+	@Test // gh-64
+	public void shouldBuildInCriteria() {
+
+		Criteria criteria = of("foo").in("bar", "baz");
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.IN);
+		assertThat(criteria.getValue()).isEqualTo(Arrays.asList("bar", "baz"));
+	}
+
+	@Test // gh-64
+	public void shouldBuildNotInCriteria() {
+
+		Criteria criteria = of("foo").notIn("bar", "baz");
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.NOT_IN);
+		assertThat(criteria.getValue()).isEqualTo(Arrays.asList("bar", "baz"));
+	}
+
+	@Test // gh-64
+	public void shouldBuildGtCriteria() {
+
+		Criteria criteria = of("foo").greaterThan(1);
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.GT);
+		assertThat(criteria.getValue()).isEqualTo(1);
+	}
+
+	@Test // gh-64
+	public void shouldBuildGteCriteria() {
+
+		Criteria criteria = of("foo").greaterThanOrEquals(1);
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.GTE);
+		assertThat(criteria.getValue()).isEqualTo(1);
+	}
+
+	@Test // gh-64
+	public void shouldBuildLtCriteria() {
+
+		Criteria criteria = of("foo").lessThan(1);
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.LT);
+		assertThat(criteria.getValue()).isEqualTo(1);
+	}
+
+	@Test // gh-64
+	public void shouldBuildLteCriteria() {
+
+		Criteria criteria = of("foo").lessThanOrEquals(1);
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.LTE);
+		assertThat(criteria.getValue()).isEqualTo(1);
+	}
+
+	@Test // gh-64
+	public void shouldBuildLikeCriteria() {
+
+		Criteria criteria = of("foo").like("hello%");
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.LIKE);
+		assertThat(criteria.getValue()).isEqualTo("hello%");
+	}
+
+	@Test // gh-64
+	public void shouldBuildIsNullCriteria() {
+
+		Criteria criteria = of("foo").isNull();
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NULL);
+	}
+
+	@Test // gh-64
+	public void shouldBuildIsNotNullCriteria() {
+
+		Criteria criteria = of("foo").isNotNull();
+
+		assertThat(criteria.getProperty()).isEqualTo("foo");
+		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NOT_NULL);
+	}
+}