diff --git a/src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.java b/src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.java
deleted file mode 100644
index 71f437c..0000000
--- a/src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package org.springframework.data.r2dbc.function;
-
-import io.r2dbc.spi.Statement;
-
-/**
- * Extension to {@link BindableOperation} for operations that allow parameter substitution for a single {@code id}
- * column that accepts either a single value or multiple values, depending on the underlying operation.
- *
- * @author Mark Paluch
- * @see Statement#bind
- * @see Statement#bindNull
- */
-public interface BindIdOperation extends BindableOperation {
-
-	/**
-	 * Bind the given {@code value} to the {@link Statement} using the underlying binding strategy.
-	 *
-	 * @param statement the statement to bind the value to.
-	 * @param value the actual value. Must not be {@literal null}.
-	 * @see Statement#bind
-	 */
-	void bindId(Statement statement, Object value);
-
-	/**
-	 * Bind the given {@code values} to the {@link Statement} using the underlying binding strategy.
-	 *
-	 * @param statement the statement to bind the value to.
-	 * @param values the actual values.
-	 * @see Statement#bind
-	 */
-	void bindIds(Statement statement, Iterable<? extends Object> values);
-}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
index 7062fca..df95db7 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
@@ -26,6 +26,7 @@ import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
@@ -137,6 +138,9 @@ public interface DatabaseClient {
 	 * Contract for specifying a SQL call along with options leading to the exchange. The SQL string can contain either
 	 * native parameter bind markers (e.g. {@literal $1, $2} for Postgres, {@literal @P0, @P1} for SQL Server) or named
 	 * parameters (e.g. {@literal :foo, :bar}) when {@link NamedParameterExpander} is enabled.
+	 * <p>
+	 * Accepts {@link PreparedOperation} as SQL and binding {@link Supplier}.
+	 * </p>
 	 *
 	 * @see NamedParameterExpander
 	 * @see DatabaseClient.Builder#namedParameters(NamedParameterExpander)
@@ -156,6 +160,7 @@ public interface DatabaseClient {
 		 *
 		 * @param sqlSupplier must not be {@literal null}.
 		 * @return a new {@link GenericExecuteSpec}.
+		 * @see PreparedOperation
 		 */
 		GenericExecuteSpec sql(Supplier<String> sqlSupplier);
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
index 0ccfd57..e837816 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
@@ -37,7 +37,6 @@ import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -49,6 +48,7 @@ import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.UncategorizedR2dbcException;
@@ -57,6 +57,7 @@ import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
+import org.springframework.data.relational.core.sql.Insert;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -336,9 +337,17 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					logger.debug("Executing SQL statement [" + sql + "]");
 				}
 
+				if (sqlSupplier instanceof PreparedOperation<?>) {
+					return ((PreparedOperation<?>) sqlSupplier).bind(it.createStatement(sql));
+				}
+
 				BindableOperation operation = namedParameters.expand(sql, dataAccessStrategy.getBindMarkersFactory(),
 						new MapBindParameterSource(byName));
 
+				if (logger.isTraceEnabled()) {
+					logger.trace("Expanded SQL [" + operation.toQuery() + "]");
+				}
+
 				Statement statement = it.createStatement(operation.toQuery());
 
 				byName.forEach((name, o) -> {
@@ -366,6 +375,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		public ExecuteSpecSupport bind(int index, Object value) {
 
+			assertNotPreparedOperation();
 			Assert.notNull(value, () -> String.format("Value at index %d must not be null. Use bindNull(…) instead.", index));
 
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
@@ -376,6 +386,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		public ExecuteSpecSupport bindNull(int index, Class<?> type) {
 
+			assertNotPreparedOperation();
+
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
 			byIndex.put(index, SettableValue.empty(type));
 
@@ -384,6 +396,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		public ExecuteSpecSupport bind(String name, Object value) {
 
+			assertNotPreparedOperation();
+
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 			Assert.notNull(value,
 					() -> String.format("Value for parameter %s must not be null. Use bindNull(…) instead.", name));
@@ -396,6 +410,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		public ExecuteSpecSupport bindNull(String name, Class<?> type) {
 
+			assertNotPreparedOperation();
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
@@ -404,6 +419,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
 
+		private void assertNotPreparedOperation() {
+			if (sqlSupplier instanceof PreparedOperation<?>) {
+				throw new InvalidDataAccessApiUsageException("Cannot add bindings to a PreparedOperation");
+			}
+		}
+
 		protected ExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
 				Supplier<String> sqlSupplier) {
 			return new ExecuteSpecSupport(byIndex, byName, sqlSupplier);
@@ -881,20 +902,19 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				throw new IllegalStateException("Insert fields is empty!");
 			}
 
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet());
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						byName.forEach(it::bind);
+					});
 
-			String sql = bindableInsert.toQuery();
+			String sql = operation.toQuery();
 			Function<Connection, Statement> insertFunction = it -> {
 
 				if (logger.isDebugEnabled()) {
 					logger.debug("Executing SQL statement [" + sql + "]");
 				}
 
-				Statement statement = it.createStatement(sql).returnGeneratedValues();
-
-				byName.forEach((k, v) -> bindableInsert.bind(statement, k, v));
-
-				return statement;
+				return operation.bind(it.createStatement(sql));
 			};
 
 			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
@@ -998,18 +1018,17 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			OutboundRow outboundRow = dataAccessStrategy.getOutboundRow(toInsert);
 
-			Set<String> columns = new LinkedHashSet<>();
+			PreparedOperation<Insert> operation = dataAccessStrategy.getStatements().insert(table, Collections.emptyList(),
+					it -> {
+						outboundRow.forEach((k, v) -> {
 
-			outboundRow.forEach((k, v) -> {
+							if (v.hasValue()) {
+								it.bind(k, v);
+							}
+						});
+					});
 
-				if (v.hasValue()) {
-					columns.add(k);
-				}
-			});
-
-			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, columns);
-
-			String sql = bindableInsert.toQuery();
+			String sql = operation.toQuery();
 
 			Function<Connection, Statement> insertFunction = it -> {
 
@@ -1017,15 +1036,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					logger.debug("Executing SQL statement [" + sql + "]");
 				}
 
-				Statement statement = it.createStatement(sql).returnGeneratedValues();
-
-				outboundRow.forEach((k, v) -> {
-					if (v.hasValue()) {
-						bindableInsert.bind(statement, k, v);
-					}
-				});
-
-				return statement;
+				return operation.bind(it.createStatement(sql));
 			};
 
 			Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
index d36ea69..17f3ddb 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
@@ -17,14 +17,11 @@ package org.springframework.data.r2dbc.function;
 
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
-import io.r2dbc.spi.Statement;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
 import java.util.function.BiFunction;
@@ -37,8 +34,6 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.ArrayColumns;
-import org.springframework.data.r2dbc.dialect.BindMarker;
-import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.domain.OutboundRow;
@@ -53,13 +48,17 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentEnti
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.OrderByField;
+import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.SelectBuilder.SelectFromAndOrderBy;
 import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.core.sql.render.NamingStrategies;
+import org.springframework.data.relational.core.sql.render.RenderContext;
+import org.springframework.data.relational.core.sql.render.RenderNamingStrategy;
+import org.springframework.data.relational.core.sql.render.SelectRenderContext;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * Default {@link ReactiveDataAccessStrategy} implementation.
@@ -71,6 +70,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	private final Dialect dialect;
 	private final R2dbcConverter converter;
 	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
+	private final StatementFactory statements;
 
 	/**
 	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect}.
@@ -118,6 +118,30 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		this.mappingContext = (MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty>) this.converter
 				.getMappingContext();
 		this.dialect = dialect;
+
+		RenderContext renderContext = new RenderContext() {
+			@Override
+			public RenderNamingStrategy getNamingStrategy() {
+				return NamingStrategies.asIs();
+			}
+
+			@Override
+			public SelectRenderContext getSelect() {
+				return new SelectRenderContext() {
+					@Override
+					public Function<Select, ? extends CharSequence> afterSelectList() {
+						return it -> "";
+					}
+
+					@Override
+					public Function<Select, ? extends CharSequence> afterOrderBy(boolean hasOrderBy) {
+						return it -> "";
+					}
+				};
+			}
+		};
+
+		this.statements = new DefaultStatementFactory(this.dialect, renderContext);
 	}
 
 	/*
@@ -218,7 +242,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getRowMapper(java.lang.Class)
 	 */
-	@SuppressWarnings("unchecked")
 	@Override
 	public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
 		return new EntityRowMapper<>(typeToRead, converter);
@@ -233,6 +256,15 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return getRequiredPersistentEntity(type).getTableName();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getStatements()
+	 */
+	@Override
+	public StatementFactory getStatements() {
+		return this.statements;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getBindMarkersFactory()
@@ -251,15 +283,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return mappingContext.getPersistentEntity(typeToRead);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#insertAndReturnGeneratedKeys(java.lang.String, java.util.Set)
-	 */
-	@Override
-	public BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns) {
-		return new DefaultBindableInsert(dialect.getBindMarkersFactory().create(), table, columns);
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#select(java.lang.String, java.util.Set, org.springframework.data.domain.Sort, org.springframework.data.domain.Pageable)
@@ -290,6 +313,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 			offset = OptionalLong.of(page.getOffset());
 		}
 
+		// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
 		return StatementRenderUtil.render(selectBuilder.build(), limit, offset, this.dialect);
 	}
 
@@ -310,304 +334,4 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 		return fields;
 	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#updateById(java.lang.String, java.util.Set, java.lang.String)
-	 */
-	@Override
-	public BindIdOperation updateById(String table, Set<String> columns, String idColumn) {
-		return new DefaultBindableUpdate(dialect.getBindMarkersFactory().create(), table, columns, idColumn);
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteById(java.lang.String, java.lang.String)
-	 */
-	@Override
-	public BindIdOperation deleteById(String table, String idColumn) {
-
-		return new DefaultBindIdOperation(dialect.getBindMarkersFactory().create(),
-				marker -> String.format("DELETE FROM %s WHERE %s = %s", table, idColumn, marker.getPlaceholder()));
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteByIdIn(java.lang.String, java.lang.String)
-	 */
-	@Override
-	public BindIdOperation deleteByIdIn(String table, String idColumn) {
-
-		String query = String.format("DELETE FROM %s", table);
-		return new DefaultBindIdIn(dialect.getBindMarkersFactory().create(), query, idColumn);
-	}
-
-	/**
-	 * Default {@link BindableOperation} implementation for a {@code INSERT} operation.
-	 */
-	static class DefaultBindableInsert implements BindableOperation {
-
-		private final Map<String, BindMarker> markers = new LinkedHashMap<>();
-		private final String query;
-
-		DefaultBindableInsert(BindMarkers bindMarkers, String table, Collection<String> columns) {
-
-			StringBuilder builder = new StringBuilder();
-			List<String> placeholders = new ArrayList<>(columns.size());
-
-			for (String column : columns) {
-				BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
-				placeholders.add(marker.getPlaceholder());
-			}
-
-			String columnsString = StringUtils.collectionToDelimitedString(columns, ", ");
-			String placeholdersString = StringUtils.collectionToDelimitedString(placeholders, ", ");
-
-			builder.append("INSERT INTO ").append(table).append(" (").append(columnsString).append(")").append(" VALUES(")
-					.append(placeholdersString).append(")");
-
-			this.query = builder.toString();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
-		 */
-		@Override
-		public void bind(Statement statement, String identifier, Object value) {
-			markers.get(identifier).bind(statement, value);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
-		 */
-		@Override
-		public void bindNull(Statement statement, String identifier, Class<?> valueType) {
-			markers.get(identifier).bindNull(statement, valueType);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
-		 */
-		@Override
-		public String toQuery() {
-			return this.query;
-		}
-	}
-
-	/**
-	 * Default {@link BindIdOperation} implementation for a {@code UPDATE} operation using a single key.
-	 */
-	static class DefaultBindableUpdate implements BindIdOperation {
-
-		private final Map<String, BindMarker> markers = new LinkedHashMap<>();
-		private final BindMarker idMarker;
-		private final String query;
-
-		DefaultBindableUpdate(BindMarkers bindMarkers, String tableName, Set<String> columns, String idColumnName) {
-
-			this.idMarker = bindMarkers.next();
-
-			StringBuilder setClause = new StringBuilder();
-
-			for (String column : columns) {
-
-				BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
-
-				if (setClause.length() != 0) {
-					setClause.append(", ");
-				}
-
-				setClause.append(column).append(" = ").append(marker.getPlaceholder());
-			}
-
-			this.query = String.format("UPDATE %s SET %s WHERE %s = %s", tableName, setClause, idColumnName,
-					idMarker.getPlaceholder());
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
-		 */
-		@Override
-		public void bind(Statement statement, String identifier, Object value) {
-			markers.get(identifier).bind(statement, value);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
-		 */
-		@Override
-		public void bindNull(Statement statement, String identifier, Class<?> valueType) {
-			markers.get(identifier).bindNull(statement, valueType);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
-		 */
-		@Override
-		public void bindId(Statement statement, Object value) {
-			idMarker.bind(statement, value);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
-		 */
-		@Override
-		public void bindIds(Statement statement, Iterable<?> values) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
-		 */
-		@Override
-		public String toQuery() {
-			return this.query;
-		}
-	}
-
-	/**
-	 * Default {@link BindIdOperation} implementation for a {@code SELECT} or {@code DELETE} operation using a single key
-	 * in the {@code WHERE} predicate.
-	 */
-	static class DefaultBindIdOperation implements BindIdOperation {
-
-		private final BindMarker idMarker;
-		private final String query;
-
-		DefaultBindIdOperation(BindMarkers bindMarkers, Function<BindMarker, String> queryFunction) {
-
-			this.idMarker = bindMarkers.next();
-			this.query = queryFunction.apply(this.idMarker);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
-		 */
-		@Override
-		public void bind(Statement statement, String identifier, Object value) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
-		 */
-		@Override
-		public void bindNull(Statement statement, String identifier, Class<?> valueType) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
-		 */
-		@Override
-		public void bindId(Statement statement, Object value) {
-			idMarker.bind(statement, value);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
-		 */
-		@Override
-		public void bindIds(Statement statement, Iterable<?> values) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
-		 */
-		@Override
-		public String toQuery() {
-			return this.query;
-		}
-	}
-
-	/**
-	 * Default {@link BindIdOperation} implementation for a {@code SELECT … WHERE id IN (…)} or
-	 * {@code DELETE … WHERE id IN (…)}.
-	 */
-	static class DefaultBindIdIn implements BindIdOperation {
-
-		private final List<String> markers = new ArrayList<>();
-		private final BindMarkers bindMarkers;
-		private final String baseQuery;
-		private final String idColumnName;
-
-		DefaultBindIdIn(BindMarkers bindMarkers, String baseQuery, String idColumnName) {
-
-			this.bindMarkers = bindMarkers;
-			this.baseQuery = baseQuery;
-			this.idColumnName = idColumnName;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
-		 */
-		@Override
-		public void bind(Statement statement, String identifier, Object value) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
-		 */
-		@Override
-		public void bindNull(Statement statement, String identifier, Class<?> valueType) {
-			throw new UnsupportedOperationException();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
-		 */
-		@Override
-		public void bindId(Statement statement, Object value) {
-
-			BindMarker bindMarker = bindMarkers.next();
-			markers.add(bindMarker.getPlaceholder());
-			bindMarker.bind(statement, value);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
-		 */
-		@Override
-		public void bindIds(Statement statement, Iterable<?> values) {
-
-			for (Object value : values) {
-				bindId(statement, value);
-			}
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
-		 */
-		@Override
-		public String toQuery() {
-
-			if (this.markers.isEmpty()) {
-				throw new UnsupportedOperationException();
-			}
-
-			String in = StringUtils.collectionToDelimitedString(this.markers, ", ");
-
-			return String.format("%s WHERE %s IN (%s)", this.baseQuery, this.idColumnName, in);
-		}
-	}
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultStatementFactory.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultStatementFactory.java
new file mode 100644
index 0000000..6028fd1
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultStatementFactory.java
@@ -0,0 +1,474 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import io.r2dbc.spi.Statement;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.r2dbc.dialect.BindMarker;
+import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.domain.SettableValue;
+import org.springframework.data.relational.core.sql.AssignValue;
+import org.springframework.data.relational.core.sql.Assignment;
+import org.springframework.data.relational.core.sql.Column;
+import org.springframework.data.relational.core.sql.Condition;
+import org.springframework.data.relational.core.sql.Delete;
+import org.springframework.data.relational.core.sql.DeleteBuilder;
+import org.springframework.data.relational.core.sql.Expression;
+import org.springframework.data.relational.core.sql.Insert;
+import org.springframework.data.relational.core.sql.SQL;
+import org.springframework.data.relational.core.sql.Select;
+import org.springframework.data.relational.core.sql.SelectBuilder;
+import org.springframework.data.relational.core.sql.StatementBuilder;
+import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.core.sql.Update;
+import org.springframework.data.relational.core.sql.UpdateBuilder;
+import org.springframework.data.relational.core.sql.render.RenderContext;
+import org.springframework.data.relational.core.sql.render.SqlRenderer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Default {@link StatementFactory} implementation.
+ *
+ * @author Mark Paluch
+ */
+@RequiredArgsConstructor
+class DefaultStatementFactory implements StatementFactory {
+
+	private final Dialect dialect;
+	private final RenderContext renderContext;
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.StatementFactory#select(java.lang.String, java.util.Collection, java.util.function.Consumer)
+	 */
+	@Override
+	public PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
+			Consumer<StatementBinderBuilder> binderConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notEmpty(columnNames, "Columns must not be empty");
+		Assert.notNull(binderConsumer, "Binder Consumer must not be null");
+
+		DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
+			@Override
+			public void bind(String identifier, SettableValue settable) {
+				throw new InvalidDataAccessApiUsageException("Binding for SELECT not supported. Use filterBy(…)");
+			}
+		};
+
+		binderConsumer.accept(binderBuilder);
+
+		return withDialect((dialect, renderContext) -> {
+			Table table = Table.create(tableName);
+			List<Column> columns = table.columns(columnNames);
+			SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
+
+			BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
+			Binding binding = binderBuilder.build(table, bindMarkers);
+			Select select;
+
+			if (binding.hasCondition()) {
+				select = selectBuilder.where(binding.getCondition()).build();
+			} else {
+				select = selectBuilder.build();
+			}
+
+			return new DefaultPreparedOperation<>(select, renderContext, binding);
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.StatementFactory#insert(java.lang.String, java.util.Collection, java.util.function.Consumer)
+	 */
+	@Override
+	public PreparedOperation<Insert> insert(String tableName, Collection<String> generatedKeysNames,
+			Consumer<StatementBinderBuilder> binderConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notNull(generatedKeysNames, "Generated key names must not be null");
+		Assert.notNull(binderConsumer, "Binder Consumer must not be null");
+
+		DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
+			@Override
+			public void filterBy(String identifier, SettableValue settable) {
+				throw new InvalidDataAccessApiUsageException("Filter-Binding for INSERT not supported. Use bind(…)");
+			}
+		};
+
+		binderConsumer.accept(binderBuilder);
+
+		return withDialect((dialect, renderContext) -> {
+
+			BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
+			Table table = Table.create(tableName);
+
+			Map<BindMarker, SettableValue> expressionBindings = new LinkedHashMap<>();
+			List<Expression> expressions = new ArrayList<>();
+			binderBuilder.forEachBinding((column, settableValue) -> {
+
+				BindMarker bindMarker = bindMarkers.next(column);
+
+				expressions.add(SQL.bindMarker(bindMarker.getPlaceholder()));
+				expressionBindings.put(bindMarker, settableValue);
+			});
+
+			if (expressions.isEmpty()) {
+				throw new IllegalStateException("INSERT contains no value expressions");
+			}
+
+			Binding binding = binderBuilder.build(table, bindMarkers).withBindings(expressionBindings);
+			Insert insert = StatementBuilder.insert().into(table).columns(table.columns(binderBuilder.bindings.keySet()))
+					.values(expressions).build();
+
+			return new DefaultPreparedOperation<Insert>(insert, renderContext, binding) {
+				@Override
+				public Statement bind(Statement to) {
+					return super.bind(to).returnGeneratedValues(generatedKeysNames.toArray(new String[0]));
+				}
+			};
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.StatementFactory#update(java.lang.String, java.util.function.Consumer)
+	 */
+	@Override
+	public PreparedOperation<Update> update(String tableName, Consumer<StatementBinderBuilder> binderConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notNull(binderConsumer, "Binder Consumer must not be null");
+
+		DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder();
+
+		binderConsumer.accept(binderBuilder);
+
+		return withDialect((dialect, renderContext) -> {
+
+			BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
+			Table table = Table.create(tableName);
+
+			Map<BindMarker, SettableValue> assignmentBindings = new LinkedHashMap<>();
+			List<Assignment> assignments = new ArrayList<>();
+			binderBuilder.forEachBinding((column, settableValue) -> {
+
+				BindMarker bindMarker = bindMarkers.next(column);
+				AssignValue assignment = table.column(column).set(SQL.bindMarker(bindMarker.getPlaceholder()));
+
+				assignments.add(assignment);
+				assignmentBindings.put(bindMarker, settableValue);
+			});
+
+			if (assignments.isEmpty()) {
+				throw new IllegalStateException("UPDATE contains no assignments");
+			}
+
+			UpdateBuilder.UpdateWhere updateBuilder = StatementBuilder.update(table).set(assignments);
+
+			Binding binding = binderBuilder.build(table, bindMarkers).withBindings(assignmentBindings);
+			Update update;
+
+			if (binding.hasCondition()) {
+				update = updateBuilder.where(binding.getCondition()).build();
+			} else {
+				update = updateBuilder.build();
+			}
+
+			return new DefaultPreparedOperation<>(update, renderContext, binding);
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.StatementFactory#delete(java.lang.String, java.util.function.Consumer)
+	 */
+	@Override
+	public PreparedOperation<Delete> delete(String tableName, Consumer<StatementBinderBuilder> binderConsumer) {
+
+		Assert.hasText(tableName, "Table must not be empty");
+		Assert.notNull(binderConsumer, "Binder Consumer must not be null");
+
+		DefaultBinderBuilder binderBuilder = new DefaultBinderBuilder() {
+			@Override
+			public void bind(String identifier, SettableValue settable) {
+				throw new InvalidDataAccessApiUsageException("Binding for DELETE not supported. Use filterBy(…)");
+			}
+		};
+
+		binderConsumer.accept(binderBuilder);
+
+		return withDialect((dialect, renderContext) -> {
+
+			Table table = Table.create(tableName);
+			DeleteBuilder.DeleteWhere deleteBuilder = StatementBuilder.delete().from(table);
+
+			BindMarkers bindMarkers = dialect.getBindMarkersFactory().create();
+			Binding binding = binderBuilder.build(table, bindMarkers);
+			Delete delete;
+
+			if (binding.hasCondition()) {
+				delete = deleteBuilder.where(binding.getCondition()).build();
+			} else {
+				delete = deleteBuilder.build();
+			}
+
+			return new DefaultPreparedOperation<>(delete, renderContext, binding);
+		});
+	}
+
+	private <T> T withDialect(BiFunction<Dialect, RenderContext, T> action) {
+
+		Assert.notNull(action, "Action must not be null");
+
+		return action.apply(this.dialect, this.renderContext);
+	}
+
+	/**
+	 * Default {@link StatementBinderBuilder} implementation.
+	 */
+	static class DefaultBinderBuilder implements StatementBinderBuilder {
+
+		final Map<String, SettableValue> filters = new LinkedHashMap<>();
+		final Map<String, SettableValue> bindings = new LinkedHashMap<>();
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.StatementBinderBuilder#filterBy(java.lang.String, org.springframework.data.r2dbc.domain.SettableValue)
+		 */
+		@Override
+		public void filterBy(String identifier, SettableValue settable) {
+
+			Assert.hasText(identifier, "FilterBy identifier must not be empty");
+			Assert.notNull(settable, "SettableValue for Filter must not be null");
+			this.filters.put(identifier, settable);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.StatementFactory.StatementBinderBuilder#bind(java.lang.String, org.springframework.data.r2dbc.domain.SettableValue)
+		 */
+		@Override
+		public void bind(String identifier, SettableValue settable) {
+
+			Assert.hasText(identifier, "Bind value identifier must not be empty");
+			Assert.notNull(settable, "SettableValue must not be null");
+
+			this.bindings.put(identifier, settable);
+		}
+
+		/**
+		 * Call {@link BiConsumer} for each filter binding.
+		 *
+		 * @param consumer the consumer to notify.
+		 */
+		void forEachFilter(BiConsumer<String, SettableValue> consumer) {
+			filters.forEach(consumer);
+		}
+
+		/**
+		 * Call {@link BiConsumer} for each value binding.
+		 *
+		 * @param consumer the consumer to notify.
+		 */
+		void forEachBinding(BiConsumer<String, SettableValue> consumer) {
+			bindings.forEach(consumer);
+		}
+
+		Binding build(Table table, BindMarkers bindMarkers) {
+
+			Map<BindMarker, Object> values = new LinkedHashMap<>();
+			Map<BindMarker, SettableValue> nulls = new LinkedHashMap<>();
+
+			AtomicReference<Condition> conditionRef = new AtomicReference<>();
+
+			forEachFilter((k, v) -> {
+
+				Condition condition = toCondition(bindMarkers, table.column(k), v, values, nulls);
+				Condition current = conditionRef.get();
+				if (current == null) {
+					current = condition;
+				} else {
+					current = current.and(condition);
+				}
+
+				conditionRef.set(current);
+			});
+
+			return new Binding(values, nulls, conditionRef.get());
+
+		}
+
+		private static Condition toCondition(BindMarkers bindMarkers, Column column, SettableValue value,
+				Map<BindMarker, Object> values, Map<BindMarker, SettableValue> nulls) {
+
+			if (value.hasValue()) {
+
+				Object bindValue = value.getValue();
+
+				if (bindValue instanceof Iterable) {
+
+					Iterable<?> iterable = (Iterable<?>) bindValue;
+					List<Expression> expressions = new ArrayList<>();
+
+					for (Object o : iterable) {
+						BindMarker marker = bindMarkers.next(column.getName());
+						if (o == null) {
+							nulls.put(marker, value);
+						} else {
+							values.put(marker, o);
+						}
+						expressions.add(SQL.bindMarker(marker.getPlaceholder()));
+					}
+
+					return column.in(expressions.toArray(new Expression[0]));
+				}
+
+				BindMarker marker = bindMarkers.next(column.getName());
+				values.put(marker, value.getValue());
+				return column.isEqualTo(SQL.bindMarker(marker.getPlaceholder()));
+			}
+
+			return column.isNull();
+		}
+	}
+
+	/**
+	 * Value object holding value and {@code NULL} bindings.
+	 *
+	 * @see SettableValue
+	 */
+	@RequiredArgsConstructor
+	@Getter
+	static class Binding {
+
+		private final Map<BindMarker, Object> values;
+		private final Map<BindMarker, SettableValue> nulls;
+
+		private final @Nullable Condition condition;
+
+		boolean hasCondition() {
+			return condition != null;
+		}
+
+		/**
+		 * Append bindings.
+		 *
+		 * @param assignmentBindings
+		 * @return
+		 */
+		Binding withBindings(Map<BindMarker, SettableValue> assignmentBindings) {
+
+			assignmentBindings.forEach(((bindMarker, settableValue) -> {
+
+				if (settableValue.isEmpty()) {
+					nulls.put(bindMarker, settableValue);
+				} else {
+					values.put(bindMarker, settableValue.getValue());
+				}
+			}));
+
+			return this;
+		}
+
+		/**
+		 * Apply bindings to a {@link Statement}.
+		 *
+		 * @param to
+		 */
+		void apply(Statement to) {
+
+			values.forEach((marker, value) -> marker.bind(to, value));
+			nulls.forEach((marker, value) -> marker.bindNull(to, value.getType()));
+		}
+	}
+
+	/**
+	 * Default implementation of {@link PreparedOperation}.
+	 *
+	 * @param <T>
+	 */
+	@RequiredArgsConstructor
+	static class DefaultPreparedOperation<T> implements PreparedOperation<T> {
+
+		private final T source;
+		private final RenderContext renderContext;
+		private final Binding binding;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.PreparedOperation#getSource()
+		 */
+		@Override
+		public T getSource() {
+			return this.source;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
+		 */
+		@Override
+		public String toQuery() {
+
+			SqlRenderer sqlRenderer = SqlRenderer.create(renderContext);
+
+			if (this.source instanceof Select) {
+				return sqlRenderer.render((Select) this.source);
+			}
+
+			if (this.source instanceof Insert) {
+				return sqlRenderer.render((Insert) this.source);
+			}
+
+			if (this.source instanceof Update) {
+				return sqlRenderer.render((Update) this.source);
+			}
+
+			if (this.source instanceof Delete) {
+				return sqlRenderer.render((Delete) this.source);
+			}
+
+			throw new IllegalStateException("Cannot render " + this.getSource());
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.function.PreparedOperation#bind(io.r2dbc.spi.Statement)
+		 */
+		@Override
+		public Statement bind(Statement to) {
+
+			binding.apply(to);
+			return to;
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/PreparedOperation.java b/src/main/java/org/springframework/data/r2dbc/function/PreparedOperation.java
new file mode 100644
index 0000000..865ab26
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/function/PreparedOperation.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.function.Supplier;
+
+/**
+ * Extension to {@link QueryOperation} for a prepared SQL query {@link Supplier} with bound parameters. Contains
+ * parameter bindings that can be {@link #bind(Statement)} bound to a {@link Statement}.
+ * <p>
+ * Can be executed with {@link DatabaseClient}.
+ * </p>
+ *
+ * @param <T> underlying operation source.
+ * @author Mark Paluch
+ * @see DatabaseClient
+ * @see DatabaseClient.SqlSpec#sql(Supplier)
+ */
+public interface PreparedOperation<T> extends QueryOperation {
+
+	/**
+	 * @return the query source, such as a statement/criteria object.
+	 */
+	T getSource();
+
+	/**
+	 * Bind query parameters to a {@link Statement}.
+	 *
+	 * @param to the target statement to bind parameters to.
+	 * @return the bound statement.
+	 */
+	Statement bind(Statement to);
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
index 2e82f36..0395448 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
@@ -17,7 +17,6 @@ package org.springframework.data.r2dbc.function;
 
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
-import io.r2dbc.spi.Statement;
 
 import java.util.List;
 import java.util.Set;
@@ -72,6 +71,8 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	String getTableName(Class<?> type);
 
+	StatementFactory getStatements();
+
 	/**
 	 * Returns the configured {@link BindMarkersFactory} to create native parameter placeholder markers.
 	 *
@@ -91,15 +92,6 @@ public interface ReactiveDataAccessStrategy {
 	// Subject to be moved into a SQL creation DSL.
 	// -------------------------------------------------------------------------
 
-	/**
-	 * Create an {@code INSERT} operation for the given {@code table} to insert {@code columns}.
-	 *
-	 * @param table the table to insert data to.
-	 * @param columns column names that will be bound.
-	 * @return the {@link BindableOperation} representing the {@code INSERT} statement.
-	 */
-	BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns);
-
 	/**
 	 * Create a {@code SELECT … ORDER BY … LIMIT …} operation for the given {@code table} using {@code columns} to
 	 * project.
@@ -111,35 +103,4 @@ public interface ReactiveDataAccessStrategy {
 	 * @return
 	 */
 	String select(String table, Set<String> columns, Sort sort, Pageable page);
-
-	/**
-	 * Create a {@code UPDATE … SET … WHERE id = ?} operation for the given {@code table} updating {@code columns} and
-	 * {@code idColumn}.
-	 *
-	 * @param table the table to insert data to.
-	 * @param columns columns to update.
-	 * @param idColumn name of the primary key.
-	 * @return
-	 */
-	BindIdOperation updateById(String table, Set<String> columns, String idColumn);
-
-	/**
-	 * Create a {@code DELETE … WHERE id = ?} operation for the given {@code table} and {@code idColumn}.
-	 *
-	 * @param table the table to insert data to.
-	 * @param idColumn name of the primary key.
-	 * @return
-	 */
-	BindIdOperation deleteById(String table, String idColumn);
-
-	/**
-	 * Create a {@code DELETE … WHERE id IN (?)} operation for the given {@code table} and {@code idColumn}. The actual
-	 * {@link BindableOperation#toQuery() query} string depends on {@link BindIdOperation#bindIds(Statement, Iterable)
-	 * bound parameters}.
-	 *
-	 * @param table the table to insert data to.
-	 * @param idColumn name of the primary key.
-	 * @return
-	 */
-	BindIdOperation deleteByIdIn(String table, String idColumn);
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/function/StatementFactory.java b/src/main/java/org/springframework/data/r2dbc/function/StatementFactory.java
new file mode 100644
index 0000000..21380ed
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/function/StatementFactory.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import java.util.Collection;
+import java.util.function.Consumer;
+
+import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.domain.SettableValue;
+import org.springframework.data.relational.core.sql.Delete;
+import org.springframework.data.relational.core.sql.Insert;
+import org.springframework.data.relational.core.sql.Select;
+import org.springframework.data.relational.core.sql.Update;
+
+/**
+ * Interface declaring statement methods that are commonly used for {@code SELECT/INSERT/UPDATE/DELETE} operations.
+ * These methods consider {@link Dialect} specifics and accept bind parameters with values.
+ *
+ * @author Mark Paluch
+ * @see PreparedOperation
+ */
+public interface StatementFactory {
+
+	/**
+	 * Creates a {@link Select} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param columnNames the columns to project, must not be {@literal null} or empty.
+	 * @param binderConsumer customizer for bindings. Supports only
+	 *          {@link StatementBinderBuilder#filterBy(String, SettableValue)} bindings.
+	 * @return the {@link PreparedOperation} to select the given columns.
+	 */
+	PreparedOperation<Select> select(String tableName, Collection<String> columnNames,
+			Consumer<StatementBinderBuilder> binderConsumer);
+
+	/**
+	 * Creates a {@link Insert} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param generatedKeysNames must not be {@literal null}.
+	 * @param binderConsumer customizer for bindings. Supports only
+	 *          {@link StatementBinderBuilder#bind(String, SettableValue)} bindings.
+	 * @return the {@link PreparedOperation} to update values in {@code tableName} assigning bound values.
+	 */
+	PreparedOperation<Insert> insert(String tableName, Collection<String> generatedKeysNames,
+			Consumer<StatementBinderBuilder> binderConsumer);
+
+	/**
+	 * Creates a {@link Update} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param binderConsumer customizer for bindings.
+	 * @return the {@link PreparedOperation} to update values in {@code tableName} assigning bound values.
+	 */
+	PreparedOperation<Update> update(String tableName, Consumer<StatementBinderBuilder> binderConsumer);
+
+	/**
+	 * Creates a {@link Delete} statement.
+	 *
+	 * @param tableName must not be {@literal null} or empty.
+	 * @param binderConsumer customizer for bindings. Supports only
+	 *          {@link StatementBinderBuilder#filterBy(String, SettableValue)} bindings.
+	 * @return the {@link PreparedOperation} to delete rows from {@code tableName}.
+	 */
+	PreparedOperation<Delete> delete(String tableName, Consumer<StatementBinderBuilder> binderConsumer);
+
+	/**
+	 * Binder to specify parameter bindings by name. Bindings match to equals comparisons.
+	 */
+	interface StatementBinderBuilder {
+
+		/**
+		 * Bind the given Id {@code value} to this builder using the underlying binding strategy to express a filter
+		 * condition. {@link Collection} type values translate to {@code IN} matching.
+		 *
+		 * @param identifier named identifier that is considered by the underlying binding strategy.
+		 * @param settable must not be {@literal null}. Use {@link SettableValue#empty(Class)} for {@code NULL} values.
+		 */
+		void filterBy(String identifier, SettableValue settable);
+
+		/**
+		 * Bind the given {@code value} to this builder using the underlying binding strategy.
+		 *
+		 * @param identifier named identifier that is considered by the underlying binding strategy.
+		 * @param settable must not be {@literal null}. Use {@link SettableValue#empty(Class)} for {@code NULL} values.
+		 */
+		void bind(String identifier, SettableValue settable);
+	}
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
index 1963616..cda23d9 100644
--- a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
@@ -20,29 +20,25 @@ import lombok.RequiredArgsConstructor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedHashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
 import org.reactivestreams.Publisher;
 
-import org.springframework.data.r2dbc.dialect.BindMarker;
-import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.domain.SettableValue;
-import org.springframework.data.r2dbc.function.BindIdOperation;
 import org.springframework.data.r2dbc.function.DatabaseClient;
-import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
+import org.springframework.data.r2dbc.function.PreparedOperation;
 import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
+import org.springframework.data.r2dbc.function.StatementFactory;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
-import org.springframework.data.relational.core.sql.Conditions;
-import org.springframework.data.relational.core.sql.Expression;
+import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.Functions;
-import org.springframework.data.relational.core.sql.SQL;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.core.sql.Update;
 import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.relational.repository.query.RelationalEntityInformation;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
@@ -83,15 +79,13 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		Map<String, SettableValue> columns = accessStrategy.getOutboundRow(objectToSave);
 		columns.remove(getIdColumnName()); // do not update the Id column.
 		String idColumnName = getIdColumnName();
-		BindIdOperation update = accessStrategy.updateById(entity.getTableName(), columns.keySet(), idColumnName);
 
-		GenericExecuteSpec exec = databaseClient.execute().sql(update);
+		PreparedOperation<Update> operation = accessStrategy.getStatements().update(entity.getTableName(), binder -> {
+			columns.forEach(binder::bind);
+			binder.filterBy(idColumnName, SettableValue.from(id));
+		});
 
-		BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(exec);
-		columns.forEach((k, v) -> update.bind(wrapper, k, v));
-		update.bindId(wrapper, id);
-
-		return wrapper.getBoundOperation().as(entity.getJavaType()) //
+		return databaseClient.execute().sql(operation).as(entity.getJavaType()) //
 				.then() //
 				.thenReturn(objectToSave);
 	}
@@ -129,18 +123,14 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
 		String idColumnName = getIdColumnName();
 
-		BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
-		BindMarker bindMarker = bindMarkers.next("id");
+		StatementFactory statements;
 
-		Table table = Table.create(entity.getTableName());
-		Select select = StatementBuilder //
-				.select(table.columns(columns)) //
-				.from(table) //
-				.where(Conditions.isEqual(table.column(idColumnName), SQL.bindMarker(bindMarker.getPlaceholder()))) //
-				.build();
+		PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(), columns,
+				binder -> {
+					binder.filterBy(idColumnName, SettableValue.from(id));
+				});
 
-		return databaseClient.execute().sql(SqlRenderer.toString(select)) //
-				.bind(0, id) //
+		return databaseClient.execute().sql(operation) //
 				.as(entity.getJavaType()) //
 				.fetch() //
 				.one();
@@ -164,18 +154,12 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 
 		String idColumnName = getIdColumnName();
 
-		BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
-		BindMarker bindMarker = bindMarkers.next("id");
+		PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(),
+				Collections.singleton(idColumnName), binder -> {
+					binder.filterBy(idColumnName, SettableValue.from(id));
+				});
 
-		Table table = Table.create(entity.getTableName());
-		Select select = StatementBuilder //
-				.select(table.column(idColumnName)) //
-				.from(table) //
-				.where(Conditions.isEqual(table.column(idColumnName), SQL.bindMarker(bindMarker.getPlaceholder()))) //
-				.build();
-
-		return databaseClient.execute().sql(SqlRenderer.toString(select)) //
-				.bind(0, id) //
+		return databaseClient.execute().sql(operation) //
 				.map((r, md) -> r) //
 				.first() //
 				.hasElement();
@@ -225,25 +209,12 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 			Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
 			String idColumnName = getIdColumnName();
 
-			BindMarkers bindMarkers = accessStrategy.getBindMarkersFactory().create();
+			PreparedOperation<Select> operation = accessStrategy.getStatements().select(entity.getTableName(), columns,
+					binder -> {
+						binder.filterBy(idColumnName, SettableValue.from(ids));
+					});
 
-			List<Expression> markers = new ArrayList<>();
-
-			for (int i = 0; i < ids.size(); i++) {
-				markers.add(SQL.bindMarker(bindMarkers.next("id").getPlaceholder()));
-			}
-
-			Table table = Table.create(entity.getTableName());
-			Select select = StatementBuilder.select(table.columns(columns)).from(table)
-					.where(Conditions.in(table.column(idColumnName), markers)).build();
-
-			GenericExecuteSpec executeSpec = databaseClient.execute().sql(SqlRenderer.toString(select));
-
-			for (int i = 0; i < ids.size(); i++) {
-				executeSpec = executeSpec.bind(i, ids.get(i));
-			}
-
-			return executeSpec.as(entity.getJavaType()).fetch().all();
+			return databaseClient.execute().sql(operation).as(entity.getJavaType()).fetch().all();
 		});
 	}
 
@@ -274,12 +245,11 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 
 		Assert.notNull(id, "Id must not be null!");
 
-		BindIdOperation delete = accessStrategy.deleteById(entity.getTableName(), getIdColumnName());
-		BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(databaseClient.execute().sql(delete));
+		PreparedOperation<Delete> delete = accessStrategy.getStatements().delete(entity.getTableName(), binder -> {
+			binder.filterBy(getIdColumnName(), SettableValue.from(id));
+		});
 
-		delete.bindId(wrapper, id);
-
-		return wrapper.getBoundOperation() //
+		return databaseClient.execute().sql(delete) //
 				.fetch() //
 				.rowsUpdated() //
 				.then();
@@ -299,13 +269,11 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 				return Flux.empty();
 			}
 
-			String idColumnName = getIdColumnName();
-			BindIdOperation delete = accessStrategy.deleteByIdIn(entity.getTableName(), idColumnName);
+			PreparedOperation<Delete> delete = accessStrategy.getStatements().delete(entity.getTableName(), binder -> {
+				binder.filterBy(getIdColumnName(), SettableValue.from(ids));
+			});
 
-			BindSpecAdapter<GenericExecuteSpec> wrapper = BindSpecAdapter.create(databaseClient.execute().sql(delete));
-			delete.bindIds(wrapper, ids);
-
-			return wrapper.getBoundOperation().as(entity.getJavaType()).fetch().rowsUpdated();
+			return databaseClient.execute().sql(delete).then();
 		}).then();
 	}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java b/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
index ae8c7dd..b0fb55e 100644
--- a/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
+++ b/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
@@ -43,6 +43,7 @@ public abstract class StatementRenderUtil {
 		String sql = SqlRenderer.toString(select);
 
 		// TODO: Replace with proper {@link Dialect} rendering for limit/offset.
+		// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
 		if (limit.isPresent()) {
 
 			LimitClause limitClause = dialect.limit();
diff --git a/src/test/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategyUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategyUnitTests.java
deleted file mode 100644
index 2d6c221..0000000
--- a/src/test/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategyUnitTests.java
+++ /dev/null
@@ -1,103 +0,0 @@
-package org.springframework.data.r2dbc.function;
-
-import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
-import io.r2dbc.spi.Statement;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-
-import org.junit.Test;
-
-import org.springframework.data.r2dbc.dialect.PostgresDialect;
-import org.springframework.data.r2dbc.domain.SettableValue;
-
-/**
- * Unit tests for {@link DefaultReactiveDataAccessStrategy}.
- *
- * @author Mark Paluch
- */
-public class DefaultReactiveDataAccessStrategyUnitTests {
-
-	DefaultReactiveDataAccessStrategy strategy = new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE);
-
-	@Test // gh-20
-	public void shouldRenderInsertAndReturnGeneratedKeysQuery() {
-
-		BindableOperation operation = strategy.insertAndReturnGeneratedKeys("table",
-				new HashSet<>(Arrays.asList("firstname", "lastname")));
-
-		assertThat(operation.toQuery()).isEqualTo("INSERT INTO table (firstname, lastname) VALUES($1, $2)");
-	}
-
-	@Test // gh-20
-	public void shouldRenderUpdateByIdQuery() {
-
-		BindableOperation operation = strategy.updateById("table", new HashSet<>(Arrays.asList("firstname", "lastname")),
-				"id");
-
-		assertThat(operation.toQuery()).isEqualTo("UPDATE table SET firstname = $2, lastname = $3 WHERE id = $1");
-	}
-
-	@Test // gh-20
-	public void shouldRenderDeleteByIdQuery() {
-
-		BindableOperation operation = strategy.deleteById("table", "id");
-
-		assertThat(operation.toQuery()).isEqualTo("DELETE FROM table WHERE id = $1");
-	}
-
-	@Test // gh-20
-	public void shouldRenderDeleteByIdInQuery() {
-
-		Statement statement = mock(Statement.class);
-		BindIdOperation operation = strategy.deleteByIdIn("table", "id");
-
-		operation.bindId(statement, Collections.singleton("foo"));
-		assertThat(operation.toQuery()).isEqualTo("DELETE FROM table WHERE id IN ($1)");
-
-		operation.bindId(statement, "bar");
-		assertThat(operation.toQuery()).isEqualTo("DELETE FROM table WHERE id IN ($1, $2)");
-	}
-
-	@Test // gh-22
-	public void shouldUpdateArray() {
-
-		Map<String, SettableValue> columnsToUpdate = strategy
-				.getOutboundRow(new WithCollectionTypes(new String[] { "one", "two" }, null));
-
-		Object stringArray = columnsToUpdate.get("string_array").getValue();
-
-		assertThat(stringArray).isInstanceOf(String[].class);
-		assertThat((String[]) stringArray).hasSize(2).contains("one", "two");
-	}
-
-	@Test // gh-22
-	public void shouldConvertListToArray() {
-
-		Map<String, SettableValue> columnsToUpdate = strategy
-				.getOutboundRow(new WithCollectionTypes(null, Arrays.asList("one", "two")));
-
-		Object stringArray = columnsToUpdate.get("string_collection").getValue();
-
-		assertThat(stringArray).isInstanceOf(String[].class);
-		assertThat((String[]) stringArray).hasSize(2).contains("one", "two");
-	}
-
-	static class WithCollectionTypes {
-
-		String[] stringArray;
-
-		List<String> stringCollection;
-
-		WithCollectionTypes(String[] stringArray, List<String> stringCollection) {
-
-			this.stringArray = stringArray;
-			this.stringCollection = stringCollection;
-		}
-	}
-}
diff --git a/src/test/java/org/springframework/data/r2dbc/function/StatementFactoryUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/StatementFactoryUnitTests.java
new file mode 100644
index 0000000..b658745
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/function/StatementFactoryUnitTests.java
@@ -0,0 +1,247 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.r2dbc.spi.Statement;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.junit.Test;
+
+import org.springframework.data.r2dbc.dialect.PostgresDialect;
+import org.springframework.data.r2dbc.domain.SettableValue;
+import org.springframework.data.relational.core.dialect.RenderContextFactory;
+import org.springframework.data.relational.core.sql.Delete;
+import org.springframework.data.relational.core.sql.Insert;
+import org.springframework.data.relational.core.sql.Select;
+import org.springframework.data.relational.core.sql.Update;
+
+/**
+ * Unit tests for {@link StatementFactory}.
+ *
+ * @author Mark Paluch
+ */
+public class StatementFactoryUnitTests {
+
+	// See https://github.com/spring-projects/spring-data-r2dbc/issues/55
+	DefaultStatementFactory statements = new DefaultStatementFactory(PostgresDialect.INSTANCE,
+			new RenderContextFactory(org.springframework.data.relational.core.dialect.PostgresDialect.INSTANCE)
+					.createRenderContext());
+
+	Statement statementMock = mock(Statement.class);
+
+	@Test
+	public void shouldToQuerySimpleSelectWithoutBindings() {
+
+		PreparedOperation<Select> select = statements.select("foo", Arrays.asList("bar", "baz"), it -> {});
+
+		assertThat(select.getSource()).isInstanceOf(Select.class);
+		assertThat(select.toQuery()).isEqualTo("SELECT foo.bar, foo.baz FROM foo");
+
+		select.bind(statementMock);
+		verifyZeroInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleSelectWithSimpleFilter() {
+
+		PreparedOperation<Select> select = statements.select("foo", Arrays.asList("bar", "baz"), it -> {
+			it.filterBy("doe", SettableValue.from("John"));
+		});
+
+		assertThat(select.getSource()).isInstanceOf(Select.class);
+		assertThat(select.toQuery()).isEqualTo("SELECT foo.bar, foo.baz FROM foo WHERE foo.doe = $1");
+
+		select.bind(statementMock);
+		verify(statementMock).bind(0, "John");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleSelectWithMultipleFilters() {
+
+		PreparedOperation<Select> select = statements.select("foo", Arrays.asList("bar", "baz"), it -> {
+			it.filterBy("doe", SettableValue.from("John"));
+			it.filterBy("baz", SettableValue.from("Jake"));
+		});
+
+		assertThat(select.getSource()).isInstanceOf(Select.class);
+		assertThat(select.toQuery()).isEqualTo("SELECT foo.bar, foo.baz FROM foo WHERE foo.doe = $1 AND foo.baz = $2");
+
+		select.bind(statementMock);
+		verify(statementMock).bind(0, "John");
+		verify(statementMock).bind(1, "Jake");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleSelectWithNullFilter() {
+
+		PreparedOperation<Select> select = statements.select("foo", Arrays.asList("bar", "baz"), it -> {
+			it.filterBy("doe", SettableValue.empty(String.class));
+		});
+
+		assertThat(select.getSource()).isInstanceOf(Select.class);
+		assertThat(select.toQuery()).isEqualTo("SELECT foo.bar, foo.baz FROM foo WHERE foo.doe IS NULL");
+
+		select.bind(statementMock);
+		verifyZeroInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleSelectWithIterableFilter() {
+
+		PreparedOperation<Select> select = statements.select("foo", Arrays.asList("bar", "baz"), it -> {
+			it.filterBy("doe", SettableValue.from(Arrays.asList("John", "Jake")));
+		});
+
+		assertThat(select.getSource()).isInstanceOf(Select.class);
+		assertThat(select.toQuery()).isEqualTo("SELECT foo.bar, foo.baz FROM foo WHERE foo.doe IN ($1, $2)");
+
+		select.bind(statementMock);
+		verify(statementMock).bind(0, "John");
+		verify(statementMock).bind(1, "Jake");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldFailInsertToQueryingWithoutValueBindings() {
+
+		assertThatThrownBy(() -> statements.insert("foo", Collections.emptyList(), it -> {}))
+				.isInstanceOf(IllegalStateException.class);
+	}
+
+	@Test
+	public void shouldToQuerySimpleInsert() {
+
+		PreparedOperation<Insert> insert = statements.insert("foo", Collections.emptyList(), it -> {
+			it.bind("bar", SettableValue.from("Foo"));
+		});
+
+		assertThat(insert.getSource()).isInstanceOf(Insert.class);
+		assertThat(insert.toQuery()).isEqualTo("INSERT INTO foo (bar) VALUES ($1)");
+
+		insert.bind(statementMock);
+		verify(statementMock).bind(0, "Foo");
+		verify(statementMock).returnGeneratedValues(any(String[].class));
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldFailUpdateToQueryingWithoutValueBindings() {
+
+		assertThatThrownBy(() -> statements.update("foo", it -> it.filterBy("foo", SettableValue.empty(Object.class))))
+				.isInstanceOf(IllegalStateException.class);
+	}
+
+	@Test
+	public void shouldToQuerySimpleUpdate() {
+
+		PreparedOperation<Update> update = statements.update("foo", it -> {
+			it.bind("bar", SettableValue.from("Foo"));
+		});
+
+		assertThat(update.getSource()).isInstanceOf(Update.class);
+		assertThat(update.toQuery()).isEqualTo("UPDATE foo SET bar = $1");
+
+		update.bind(statementMock);
+		verify(statementMock).bind(0, "Foo");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQueryNullUpdate() {
+
+		PreparedOperation<Update> update = statements.update("foo", it -> {
+			it.bind("bar", SettableValue.empty(String.class));
+		});
+
+		assertThat(update.getSource()).isInstanceOf(Update.class);
+		assertThat(update.toQuery()).isEqualTo("UPDATE foo SET bar = $1");
+
+		update.bind(statementMock);
+		verify(statementMock).bindNull(0, String.class);
+
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQueryUpdateWithFilter() {
+
+		PreparedOperation<Update> update = statements.update("foo", it -> {
+			it.bind("bar", SettableValue.from("Foo"));
+			it.filterBy("baz", SettableValue.from("Baz"));
+		});
+
+		assertThat(update.getSource()).isInstanceOf(Update.class);
+		assertThat(update.toQuery()).isEqualTo("UPDATE foo SET bar = $1 WHERE foo.baz = $2");
+
+		update.bind(statementMock);
+		verify(statementMock).bind(0, "Foo");
+		verify(statementMock).bind(1, "Baz");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleDeleteWithSimpleFilter() {
+
+		PreparedOperation<Delete> delete = statements.delete("foo", it -> {
+			it.filterBy("doe", SettableValue.from("John"));
+		});
+
+		assertThat(delete.getSource()).isInstanceOf(Delete.class);
+		assertThat(delete.toQuery()).isEqualTo("DELETE FROM foo WHERE foo.doe = $1");
+
+		delete.bind(statementMock);
+		verify(statementMock).bind(0, "John");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleDeleteWithMultipleFilters() {
+
+		PreparedOperation<Delete> delete = statements.delete("foo", it -> {
+			it.filterBy("doe", SettableValue.from("John"));
+			it.filterBy("baz", SettableValue.from("Jake"));
+		});
+
+		assertThat(delete.getSource()).isInstanceOf(Delete.class);
+		assertThat(delete.toQuery()).isEqualTo("DELETE FROM foo WHERE foo.doe = $1 AND foo.baz = $2");
+
+		delete.bind(statementMock);
+		verify(statementMock).bind(0, "John");
+		verify(statementMock).bind(1, "Jake");
+		verifyNoMoreInteractions(statementMock);
+	}
+
+	@Test
+	public void shouldToQuerySimpleDeleteWithNullFilter() {
+
+		PreparedOperation<Delete> delete = statements.delete("foo", it -> {
+			it.filterBy("doe", SettableValue.empty(String.class));
+		});
+
+		assertThat(delete.getSource()).isInstanceOf(Delete.class);
+		assertThat(delete.toQuery()).isEqualTo("DELETE FROM foo WHERE foo.doe IS NULL");
+
+		delete.bind(statementMock);
+		verifyZeroInteractions(statementMock);
+	}
+}