From 01eccbbbd7b5c478108226c0c11c27b40435f514 Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Fri, 24 Jan 2020 14:35:58 +0100
Subject: [PATCH] #220 - Refactor StatementMapper.

Use limit/offset instead of Page and accept Expression objects to declare a select list. Use SqlIdentifier in Update, Query, Criteria and fluent API.

Original pull request: #287.
---
 .../r2dbc/core/DefaultDatabaseClient.java     |  12 +-
 .../r2dbc/core/DefaultStatementMapper.java    |  46 +++---
 .../r2dbc/core/FluentR2dbcOperations.java     |   2 +-
 .../data/r2dbc/core/R2dbcEntityTemplate.java  |  55 +++---
 .../r2dbc/core/ReactiveDeleteOperation.java   |  19 ++-
 .../core/ReactiveDeleteOperationSupport.java  |  24 +--
 .../r2dbc/core/ReactiveInsertOperation.java   |  19 ++-
 .../core/ReactiveInsertOperationSupport.java  |  13 +-
 .../r2dbc/core/ReactiveSelectOperation.java   |  19 ++-
 .../core/ReactiveSelectOperationSupport.java  |  29 ++--
 .../r2dbc/core/ReactiveUpdateOperation.java   |  19 ++-
 .../core/ReactiveUpdateOperationSupport.java  |  22 +--
 .../data/r2dbc/core/StatementMapper.java      | 156 ++++++++++++++----
 .../data/r2dbc/query/Criteria.java            |  25 +--
 .../data/r2dbc/query/Query.java               |  46 ++++--
 .../data/r2dbc/query/QueryMapper.java         |  72 ++++----
 .../data/r2dbc/query/Update.java              |  28 +++-
 .../data/r2dbc/query/UpdateMapper.java        |  11 +-
 .../support/SimpleR2dbcRepository.java        |   6 -
 .../core/R2dbcEntityTemplateUnitTests.java    |   2 +-
 .../ReactiveSelectOperationUnitTests.java     |   2 +-
 .../data/r2dbc/query/CriteriaUnitTests.java   |  34 ++--
 .../r2dbc/query/UpdateMapperUnitTests.java    |  17 +-
 23 files changed, 434 insertions(+), 244 deletions(-)

diff --git a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java
index 912887b0..f02a8446 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java
@@ -818,8 +818,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			StatementMapper mapper = dataAccessStrategy.getStatementMapper();
 
-			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table).withProjection(this.projectedFields)
-					.withSort(this.sort).withPage(this.page);
+			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table)
+					.withProjection(this.projectedFields.toArray(new SqlIdentifier[0])).withSort(this.sort).withPage(this.page);
 
 			if (this.criteria != null) {
 				selectSpec = selectSpec.withCriteria(this.criteria);
@@ -931,8 +931,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 				columns = this.projectedFields;
 			}
 
-			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table).withProjection(columns)
-					.withPage(this.page).withSort(this.sort);
+			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table)
+					.withProjection(columns.toArray(new SqlIdentifier[0])).withPage(this.page).withSort(this.sort);
 
 			if (this.criteria != null) {
 				selectSpec = selectSpec.withCriteria(this.criteria);
@@ -1038,7 +1038,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			StatementMapper.InsertSpec insert = mapper.createInsert(this.table);
 
 			for (SqlIdentifier column : this.byName.keySet()) {
-				insert = insert.withColumn(dataAccessStrategy.toSql(column), this.byName.get(column));
+				insert = insert.withColumn(column, this.byName.get(column));
 			}
 
 			PreparedOperation<?> operation = mapper.getMappedObject(insert);
@@ -1161,7 +1161,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			for (SqlIdentifier column : outboundRow.keySet()) {
 				SettableValue settableValue = outboundRow.get(column);
 				if (settableValue.hasValue()) {
-					insert = insert.withColumn(dataAccessStrategy.toSql(column), settableValue);
+					insert = insert.withColumn(column, settableValue);
 				}
 			}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java b/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java
index a6756ecb..441b5f2f 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java
@@ -16,11 +16,8 @@
 package org.springframework.data.r2dbc.core;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindTarget;
@@ -84,9 +81,9 @@ class DefaultStatementMapper implements StatementMapper {
 	private PreparedOperation<Select> getMappedObject(SelectSpec selectSpec,
 			@Nullable RelationalPersistentEntity<?> entity) {
 
-		Table table = Table.create(toSql(selectSpec.getTable()));
-		List<Column> columns = table.columns(toSql(selectSpec.getProjectedFields()));
-		SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
+		Table table = selectSpec.getTable();
+		SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(getSelectList(selectSpec, entity))
+				.from(table);
 
 		BindMarkers bindMarkers = this.dialect.getBindMarkersFactory().create();
 		Bindings bindings = Bindings.empty();
@@ -102,37 +99,36 @@ class DefaultStatementMapper implements StatementMapper {
 
 		if (selectSpec.getSort().isSorted()) {
 
-			Sort mappedSort = this.updateMapper.getMappedObject(selectSpec.getSort(), entity);
-			selectBuilder.orderBy(createOrderByFields(table, mappedSort));
+			List<OrderByField> sort = this.updateMapper.getMappedSort(table, selectSpec.getSort(), entity);
+			selectBuilder.orderBy(sort);
 		}
 
-		if (selectSpec.getPage().isPaged()) {
+		if (selectSpec.getLimit() > 0) {
+			selectBuilder.limit(selectSpec.getLimit());
+		}
 
-			Pageable page = selectSpec.getPage();
-
-			selectBuilder.limitOffset(page.getPageSize(), page.getOffset());
+		if (selectSpec.getOffset() > 0) {
+			selectBuilder.offset(selectSpec.getOffset());
 		}
 
 		Select select = selectBuilder.build();
 		return new DefaultPreparedOperation<>(select, this.renderContext, bindings);
 	}
 
-	private Collection<? extends OrderByField> createOrderByFields(Table table, Sort sortToUse) {
+	protected List<Expression> getSelectList(SelectSpec selectSpec, @Nullable RelationalPersistentEntity<?> entity) {
 
-		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);
-			}
+		if (entity == null) {
+			return selectSpec.getSelectList();
 		}
 
-		return fields;
+		List<Expression> selectList = selectSpec.getSelectList();
+		List<Expression> mapped = new ArrayList<>(selectList.size());
+
+		for (Expression expression : selectList) {
+			mapped.add(updateMapper.getMappedObject(expression, entity));
+		}
+
+		return mapped;
 	}
 
 	/*
diff --git a/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java b/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java
index 30f0c6da..8ab66f6c 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java
index 115bc161..2c1e2dfe 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java
@@ -37,13 +37,15 @@ import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionInformation;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
-import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
 import org.springframework.data.r2dbc.query.Criteria;
 import org.springframework.data.r2dbc.query.Query;
 import org.springframework.data.r2dbc.query.Update;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.Functions;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.util.ProxyUtils;
 import org.springframework.util.Assert;
 
@@ -74,13 +76,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 	 * @param databaseClient
 	 */
 	public R2dbcEntityTemplate(DatabaseClient databaseClient) {
-
-		Assert.notNull(databaseClient, "DatabaseClient must not be null");
-
-		this.databaseClient = databaseClient;
-		this.dataAccessStrategy = getDataAccessStrategy(databaseClient);
-		this.mappingContext = getMappingContext(this.dataAccessStrategy);
-		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this(databaseClient, getDataAccessStrategy(databaseClient));
 	}
 
 	/**
@@ -174,7 +170,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doCount(query, entityClass, getTableName(entityClass));
 	}
 
-	Mono<Long> doCount(Query query, Class<?> entityClass, String tableName) {
+	Mono<Long> doCount(Query query, Class<?> entityClass, SqlIdentifier tableName) {
 
 		RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
@@ -211,16 +207,18 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doExists(query, entityClass, getTableName(entityClass));
 	}
 
-	Mono<Boolean> doExists(Query query, Class<?> entityClass, String tableName) {
+	Mono<Boolean> doExists(Query query, Class<?> entityClass, SqlIdentifier tableName) {
 
 		RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
 
-		String columnName = entity.hasIdProperty() ? entity.getRequiredIdProperty().getColumnName() : "*";
+		SqlIdentifier columnName = entity.hasIdProperty() ? entity.getRequiredIdProperty().getColumnName()
+				: SqlIdentifier.unquoted("*");
 
 		StatementMapper.SelectSpec selectSpec = statementMapper //
 				.createSelect(tableName) //
-				.withProjection(columnName);
+				.withProjection(columnName) //
+				.limit(1);
 
 		Optional<Criteria> criteria = query.getCriteria();
 		if (criteria.isPresent()) {
@@ -248,14 +246,13 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doSelect(query, entityClass, getTableName(entityClass), entityClass).all();
 	}
 
-	<T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityClass, String tableName, Class<T> returnType) {
+	<T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityClass, SqlIdentifier tableName, Class<T> returnType) {
 
-		RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
 
 		StatementMapper.SelectSpec selectSpec = statementMapper //
 				.createSelect(tableName) //
-				.withProjection(getSelectProjection(query, returnType));
+				.doWithTable((table, spec) -> spec.withProjection(getSelectProjection(table, query, returnType)));
 
 		if (query.getLimit() > 0) {
 			selectSpec = selectSpec.limit(query.getLimit());
@@ -310,7 +307,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doUpdate(query, update, entityClass, getTableName(entityClass));
 	}
 
-	Mono<Integer> doUpdate(Query query, Update update, Class<?> entityClass, String tableName) {
+	Mono<Integer> doUpdate(Query query, Update update, Class<?> entityClass, SqlIdentifier tableName) {
 
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
 
@@ -339,7 +336,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doDelete(query, entityClass, getTableName(entityClass));
 	}
 
-	Mono<Integer> doDelete(Query query, Class<?> entityClass, String tableName) {
+	Mono<Integer> doDelete(Query query, Class<?> entityClass, SqlIdentifier tableName) {
 
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
 
@@ -371,7 +368,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return doInsert(entity, getRequiredEntity(entity).getTableName());
 	}
 
-	<T> Mono<T> doInsert(T entity, String tableName) {
+	<T> Mono<T> doInsert(T entity, SqlIdentifier tableName) {
 
 		RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
 
@@ -434,7 +431,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return Query.query(Criteria.where(persistentEntity.getRequiredIdProperty().getName()).is(id));
 	}
 
-	String getTableName(Class<?> entityClass) {
+	SqlIdentifier getTableName(Class<?> entityClass) {
 		return getRequiredEntity(entityClass).getTableName();
 	}
 
@@ -447,7 +444,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return (RelationalPersistentEntity) getRequiredEntity(entityType);
 	}
 
-	private <T> List<String> getSelectProjection(Query query, Class<T> returnType) {
+	private <T> List<Expression> getSelectProjection(Table table, Query query, Class<T> returnType) {
 
 		if (query.getColumns().isEmpty()) {
 
@@ -456,19 +453,21 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 				ProjectionInformation projectionInformation = projectionFactory.getProjectionInformation(returnType);
 
 				if (projectionInformation.isClosed()) {
-					return projectionInformation.getInputProperties().stream().map(FeatureDescriptor::getName)
+					return projectionInformation.getInputProperties().stream().map(FeatureDescriptor::getName).map(table::column)
 							.collect(Collectors.toList());
 				}
 			}
 
-			return Collections.singletonList("*");
+			return Collections.singletonList(table.asterisk());
 		}
 
-		return query.getColumns();
+		return query.getColumns().stream().map(table::column).collect(Collectors.toList());
 	}
 
 	private static ReactiveDataAccessStrategy getDataAccessStrategy(DatabaseClient databaseClient) {
 
+		Assert.notNull(databaseClient, "DatabaseClient must not be null");
+
 		if (databaseClient instanceof DefaultDatabaseClient) {
 
 			DefaultDatabaseClient client = (DefaultDatabaseClient) databaseClient;
@@ -478,14 +477,4 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		throw new IllegalStateException("Cannot obtain ReactiveDataAccessStrategy");
 	}
 
-	private static MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext(
-			ReactiveDataAccessStrategy strategy) {
-
-		if (strategy instanceof DefaultReactiveDataAccessStrategy) {
-			DefaultReactiveDataAccessStrategy strategy1 = (DefaultReactiveDataAccessStrategy) strategy;
-			return strategy1.getMappingContext();
-		}
-		return new R2dbcMappingContext();
-	}
-
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java
index ac258567..dbc046dd 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperation.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.data.r2dbc.core;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 
 /**
  * The {@link ReactiveDeleteOperation} interface allows creation and execution of {@code DELETE} operations in a fluent
@@ -67,7 +68,21 @@ public interface ReactiveDeleteOperation {
 		 * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
 		 * @see DeleteWithQuery
 		 */
-		DeleteWithQuery from(String table);
+		default DeleteWithQuery from(String table) {
+			return from(SqlIdentifier.unquoted(table));
+		}
+
+		/**
+		 * Explicitly set the {@link SqlIdentifier name} of the table on which to perform the delete.
+		 * <p>
+		 * Skip this step to use the default table derived from the {@link Class domain type}.
+		 *
+		 * @param table {@link SqlIdentifier name} of the table; must not be {@literal null}.
+		 * @return new instance of {@link DeleteWithQuery}.
+		 * @throws IllegalArgumentException if {@link SqlIdentifier table} is {@literal null}.
+		 * @see DeleteWithQuery
+		 */
+		DeleteWithQuery from(SqlIdentifier table);
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java
index 479538e7..bc3d25af 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationSupport.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.data.r2dbc.core;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -35,7 +36,7 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
 		this.template = template;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation#delete(java.lang.Class)
 	 */
@@ -55,28 +56,29 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
 
 		private final Query query;
 
-		private final @Nullable String tableName;
+		private final @Nullable SqlIdentifier tableName;
 
-		ReactiveDeleteSupport(R2dbcEntityTemplate template, Class<?> domainType, Query query, @Nullable String tableName) {
+		ReactiveDeleteSupport(R2dbcEntityTemplate template, Class<?> domainType, Query query,
+				@Nullable SqlIdentifier tableName) {
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
 			this.tableName = tableName;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.DeleteWithTable#from(java.lang.String)
+		 * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.DeleteWithTable#from(SqlIdentifier)
 		 */
 		@Override
-		public DeleteWithQuery from(String tableName) {
+		public DeleteWithQuery from(SqlIdentifier tableName) {
 
-			Assert.hasText(tableName, "Table name must not be null or empty");
+			Assert.notNull(tableName, "Table name must not be null");
 
 			return new ReactiveDeleteSupport(this.template, this.domainType, this.query, tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.DeleteWithQuery#matching(org.springframework.data.r2dbc.query.Query)
 		 */
@@ -88,7 +90,7 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
 			return new ReactiveDeleteSupport(this.template, this.domainType, query, this.tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation.TerminatingDelete#all()
 		 */
@@ -96,7 +98,7 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
 			return this.template.doDelete(this.query, this.domainType, getTableName());
 		}
 
-		private String getTableName() {
+		private SqlIdentifier getTableName() {
 			return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java
index bd514ad7..89b44f85 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperation.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,8 @@ package org.springframework.data.r2dbc.core;
 
 import reactor.core.publisher.Mono;
 
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+
 /**
  * The {@link ReactiveInsertOperation} interface allows creation and execution of {@code INSERT} operations in a fluent
  * API style.
@@ -63,7 +65,20 @@ public interface ReactiveInsertOperation {
 		 * @return new instance of {@link TerminatingInsert}.
 		 * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
 		 */
-		TerminatingInsert<T> into(String table);
+		default TerminatingInsert<T> into(String table) {
+			return into(SqlIdentifier.unquoted(table));
+		}
+
+		/**
+		 * Explicitly set the {@link SqlIdentifier name} of the table.
+		 * <p>
+		 * Skip this step to use the default table derived from the {@link Class domain type}.
+		 *
+		 * @param table {@link SqlIdentifier name} of the table; must not be {@literal null}.
+		 * @return new instance of {@link TerminatingInsert}.
+		 * @throws IllegalArgumentException if {@link SqlIdentifier table} is {@literal null}.
+		 */
+		TerminatingInsert<T> into(SqlIdentifier table);
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java
index 11fc23e9..593040dd 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationSupport.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@ package org.springframework.data.r2dbc.core;
 
 import reactor.core.publisher.Mono;
 
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -52,9 +53,9 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
 
 		private final Class<T> domainType;
 
-		private final @Nullable String tableName;
+		private final @Nullable SqlIdentifier tableName;
 
-		ReactiveInsertSupport(R2dbcEntityTemplate template, Class<T> domainType, String tableName) {
+		ReactiveInsertSupport(R2dbcEntityTemplate template, Class<T> domainType, @Nullable SqlIdentifier tableName) {
 			this.template = template;
 			this.domainType = domainType;
 			this.tableName = tableName;
@@ -62,10 +63,10 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
 
 		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation.InsertWithTable#into(java.lang.String)
+		 * @see org.springframework.data.r2dbc.core.ReactiveInsertOperation.InsertWithTable#into(SqlIdentifier)
 		 */
 		@Override
-		public TerminatingInsert<T> into(String tableName) {
+		public TerminatingInsert<T> into(SqlIdentifier tableName) {
 
 			Assert.notNull(tableName, "Table name must not be null");
 
@@ -84,7 +85,7 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
 			return this.template.doInsert(object, getTableName());
 		}
 
-		private String getTableName() {
+		private SqlIdentifier getTableName() {
 			return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java
index 15dae6f5..c67de55c 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperation.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 
 /**
  * The {@link ReactiveSelectOperation} interface allows creation and execution of {@code SELECT} operations in a fluent
@@ -73,7 +74,21 @@ public interface ReactiveSelectOperation {
 		 * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
 		 * @see SelectWithProjection
 		 */
-		SelectWithProjection<T> from(String table);
+		default SelectWithProjection<T> from(String table) {
+			return from(SqlIdentifier.unquoted(table));
+		}
+
+		/**
+		 * Explicitly set the {@link SqlIdentifier name} of the table on which to perform the query.
+		 * <p>
+		 * Skip this step to use the default table derived from the {@link Class domain type}.
+		 *
+		 * @param table {@link SqlIdentifier name} of the table; must not be {@literal null}.
+		 * @return new instance of {@link SelectWithProjection}.
+		 * @throws IllegalArgumentException if {@link SqlIdentifier table} is {@literal null}.
+		 * @see SelectWithProjection
+		 */
+		SelectWithProjection<T> from(SqlIdentifier table);
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
index d1e9be24..f17713c4 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -36,7 +37,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 		this.template = template;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation#select(java.lang.Class)
 	 */
@@ -58,10 +59,10 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 		private final Query query;
 
-		private final @Nullable String tableName;
+		private final @Nullable SqlIdentifier tableName;
 
 		ReactiveSelectSupport(R2dbcEntityTemplate template, Class<?> domainType, Class<T> returnType, Query query,
-				@Nullable String tableName) {
+				@Nullable SqlIdentifier tableName) {
 			this.template = template;
 			this.domainType = domainType;
 			this.returnType = returnType;
@@ -69,19 +70,19 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			this.tableName = tableName;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithTable#from(java.lang.String)
 		 */
 		@Override
-		public SelectWithProjection<T> from(String tableName) {
+		public SelectWithProjection<T> from(SqlIdentifier tableName) {
 
 			Assert.notNull(tableName, "Table name must not be null");
 
 			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.query, tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithProjection#as(java.lang.Class)
 		 */
@@ -93,7 +94,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return new ReactiveSelectSupport<>(this.template, this.domainType, returnType, this.query, this.tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.SelectWithQuery#matching(org.springframework.data.r2dbc.query.Query)
 		 */
@@ -105,7 +106,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, query, this.tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#count()
 		 */
@@ -114,7 +115,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return this.template.doCount(this.query, this.domainType, getTableName());
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#exists()
 		 */
@@ -123,7 +124,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return this.template.doExists(this.query, this.domainType, getTableName());
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#first()
 		 */
@@ -132,7 +133,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType).first();
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#one()
 		 */
@@ -141,7 +142,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType).one();
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveSelectOperation.TerminatingSelect#all()
 		 */
@@ -150,7 +151,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 			return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType).all();
 		}
 
-		private String getTableName() {
+		private SqlIdentifier getTableName() {
 			return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java
index 1e9baa95..c779f628 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperation.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
 import org.springframework.data.r2dbc.query.Update;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 
 /**
  * The {@link ReactiveUpdateOperation} interface allows creation and execution of {@code UPDATE} operations in a fluent
@@ -71,7 +72,21 @@ public interface ReactiveUpdateOperation {
 		 * @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
 		 * @see UpdateWithQuery
 		 */
-		UpdateWithQuery inTable(String table);
+		default UpdateWithQuery inTable(String table) {
+			return inTable(SqlIdentifier.unquoted(table));
+		}
+
+		/**
+		 * Explicitly set the {@link SqlIdentifier name} of the table on which to perform the update.
+		 * <p>
+		 * Skip this step to use the default table derived from the {@link Class domain type}.
+		 *
+		 * @param table {@link SqlIdentifier name} of the table; must not be {@literal null}.
+		 * @return new instance of {@link UpdateWithQuery}.
+		 * @throws IllegalArgumentException if {@link SqlIdentifier table} is {@literal null}.
+		 * @see UpdateWithQuery
+		 */
+		UpdateWithQuery inTable(SqlIdentifier table);
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java
index 90a7908c..668ca9ee 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationSupport.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.data.r2dbc.query.Query;
 import org.springframework.data.r2dbc.query.Update;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -36,7 +37,7 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
 		this.template = template;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation#update(java.lang.Class)
 	 */
@@ -56,28 +57,29 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
 
 		private final Query query;
 
-		private final @Nullable String tableName;
+		private final @Nullable SqlIdentifier tableName;
 
-		ReactiveUpdateSupport(R2dbcEntityTemplate template, Class<?> domainType, Query query, @Nullable String tableName) {
+		ReactiveUpdateSupport(R2dbcEntityTemplate template, Class<?> domainType, Query query,
+				@Nullable SqlIdentifier tableName) {
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
 			this.tableName = tableName;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.UpdateWithTable#inTable(java.lang.String)
+		 * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.UpdateWithTable#inTable(SqlIdentifier)
 		 */
 		@Override
-		public UpdateWithQuery inTable(String tableName) {
+		public UpdateWithQuery inTable(SqlIdentifier tableName) {
 
 			Assert.notNull(tableName, "Table name must not be null");
 
 			return new ReactiveUpdateSupport(this.template, this.domainType, this.query, tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.UpdateWithQuery#matching(org.springframework.data.r2dbc.query.Query)
 		 */
@@ -89,7 +91,7 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
 			return new ReactiveUpdateSupport(this.template, this.domainType, query, this.tableName);
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation.TerminatingUpdate#apply(org.springframework.data.r2dbc.query.Update)
 		 */
@@ -101,7 +103,7 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
 			return this.template.doUpdate(this.query, update, this.domainType, getTableName());
 		}
 
-		private String getTableName() {
+		private SqlIdentifier getTableName() {
 			return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java b/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
index b348a304..9932f878 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
@@ -22,6 +22,7 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
 import org.springframework.data.domain.Pageable;
@@ -30,7 +31,9 @@ import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 import org.springframework.data.r2dbc.query.Criteria;
 import org.springframework.data.r2dbc.query.Update;
+import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.lang.Nullable;
 
 /**
@@ -179,19 +182,23 @@ public interface StatementMapper {
 	 */
 	class SelectSpec {
 
-		private final SqlIdentifier table;
-		private final List<SqlIdentifier> projectedFields;
+		private final Table table;
+		private final List<String> projectedFields;
+		private final List<Expression> selectList;
 		private final @Nullable Criteria criteria;
 		private final Sort sort;
-		private final Pageable page;
+		private final long offset;
+		private final int limit;
 
-		protected SelectSpec(SqlIdentifier table, List<SqlIdentifier> projectedFields, @Nullable Criteria criteria,
-				Sort sort, Pageable page) {
+		protected SelectSpec(Table table, List<String> projectedFields, List<Expression> selectList,
+				@Nullable Criteria criteria, Sort sort, int limit, long offset) {
 			this.table = table;
 			this.projectedFields = projectedFields;
+			this.selectList = selectList;
 			this.criteria = criteria;
 			this.sort = sort;
-			this.page = page;
+			this.offset = offset;
+			this.limit = limit;
 		}
 
 		/**
@@ -212,17 +219,12 @@ public interface StatementMapper {
 		 * @since 1.1
 		 */
 		public static SelectSpec create(SqlIdentifier table) {
-			return new SelectSpec(table, Collections.emptyList(), null, Sort.unsorted(), Pageable.unpaged());
+			return new SelectSpec(Table.create(table), Collections.emptyList(), Collections.emptyList(), null,
+					Sort.unsorted(), -1, -1);
 		}
 
-		/**
-		 * Associate {@code projectedFields} with the select and create a new {@link SelectSpec}.
-		 *
-		 * @param projectedFields
-		 * @return the {@link SelectSpec}.
-		 */
-		public SelectSpec withProjection(String... projectedFields) {
-			return withProjection(Arrays.stream(projectedFields).map(SqlIdentifier::unquoted).collect(Collectors.toList()));
+		public SelectSpec doWithTable(BiFunction<Table, SelectSpec, SelectSpec> function) {
+			return function.apply(getTable(), this);
 		}
 
 		/**
@@ -232,12 +234,50 @@ public interface StatementMapper {
 		 * @return the {@link SelectSpec}.
 		 * @since 1.1
 		 */
-		public SelectSpec withProjection(Collection<SqlIdentifier> projectedFields) {
+		public SelectSpec withProjection(String... projectedFields) {
+			return withProjection(Arrays.stream(projectedFields).map(table::column).collect(Collectors.toList()));
+		}
 
-			List<SqlIdentifier> fields = new ArrayList<>(this.projectedFields);
-			fields.addAll(projectedFields);
+		/**
+		 * Associate {@code projectedFields} with the select and create a new {@link SelectSpec}.
+		 *
+		 * @param projectedFields
+		 * @return the {@link SelectSpec}.
+		 * @since 1.1
+		 */
+		public SelectSpec withProjection(SqlIdentifier... projectedFields) {
+			return withProjection(Arrays.stream(projectedFields).map(table::column).collect(Collectors.toList()));
+		}
 
-			return new SelectSpec(this.table, fields, this.criteria, this.sort, this.page);
+		/**
+		 * Associate {@code expressions} with the select list and create a new {@link SelectSpec}.
+		 *
+		 * @param expressions
+		 * @return the {@link SelectSpec}.
+		 * @since 1.1
+		 */
+		public SelectSpec withProjection(Expression... expressions) {
+
+			List<Expression> selectList = new ArrayList<>(this.selectList);
+			selectList.addAll(Arrays.asList(expressions));
+
+			return new SelectSpec(this.table, projectedFields, selectList, this.criteria, this.sort, this.limit, this.offset);
+		}
+
+		/**
+		 * Associate {@code projectedFields} with the select and create a new {@link SelectSpec}.
+		 *
+		 * @param projectedFields
+		 * @return the {@link SelectSpec}.
+		 * @since 1.1
+		 */
+		public SelectSpec withProjection(Collection<Expression> projectedFields) {
+
+			List<Expression> selectList = new ArrayList<>(this.selectList);
+			selectList.addAll(projectedFields);
+
+			return new SelectSpec(this.table, this.projectedFields, selectList, this.criteria, this.sort, this.limit,
+					this.offset);
 		}
 
 		/**
@@ -247,7 +287,8 @@ public interface StatementMapper {
 		 * @return the {@link SelectSpec}.
 		 */
 		public SelectSpec withCriteria(Criteria criteria) {
-			return new SelectSpec(this.table, this.projectedFields, criteria, this.sort, this.page);
+			return new SelectSpec(this.table, this.projectedFields, this.selectList, criteria, this.sort, this.limit,
+					this.offset);
 		}
 
 		/**
@@ -259,10 +300,12 @@ public interface StatementMapper {
 		public SelectSpec withSort(Sort sort) {
 
 			if (sort.isSorted()) {
-				return new SelectSpec(this.table, this.projectedFields, this.criteria, sort, this.page);
+				return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria, sort, this.limit,
+						this.offset);
 			}
 
-			return new SelectSpec(this.table, this.projectedFields, this.criteria, this.sort, this.page);
+			return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria, this.sort, this.limit,
+					this.offset);
 		}
 
 		/**
@@ -277,21 +320,53 @@ public interface StatementMapper {
 
 				Sort sort = page.getSort();
 
-				return new SelectSpec(this.table, this.projectedFields, this.criteria, sort.isSorted() ? sort : this.sort,
-						page);
+				return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria,
+						sort.isSorted() ? sort : this.sort, page.getPageSize(), page.getOffset());
 			}
 
-			return new SelectSpec(this.table, this.projectedFields, this.criteria, this.sort, page);
+			return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria, this.sort, this.limit,
+					this.offset);
 		}
 
-		public SqlIdentifier getTable() {
+		/**
+		 * Associate a result offset with the select and create a new {@link SelectSpec}.
+		 *
+		 * @param page
+		 * @return the {@link SelectSpec}.
+		 */
+		public SelectSpec offset(long offset) {
+			return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria, this.sort, this.limit,
+					offset);
+		}
+
+		/**
+		 * Associate a result limit with the select and create a new {@link SelectSpec}.
+		 *
+		 * @param page
+		 * @return the {@link SelectSpec}.
+		 */
+		public SelectSpec limit(int limit) {
+			return new SelectSpec(this.table, this.projectedFields, this.selectList, this.criteria, this.sort, limit,
+					this.offset);
+		}
+
+		public Table getTable() {
 			return this.table;
 		}
 
-		public List<SqlIdentifier> getProjectedFields() {
+		/**
+		 * @return
+		 * @deprecated since 1.1, use {@link #getSelectList()} instead.
+		 */
+		@Deprecated
+		public List<String> getProjectedFields() {
 			return Collections.unmodifiableList(this.projectedFields);
 		}
 
+		public List<Expression> getSelectList() {
+			return Collections.unmodifiableList(selectList);
+		}
+
 		@Nullable
 		public Criteria getCriteria() {
 			return this.criteria;
@@ -301,8 +376,12 @@ public interface StatementMapper {
 			return this.sort;
 		}
 
-		public Pageable getPage() {
-			return this.page;
+		public long getOffset() {
+			return this.offset;
+		}
+
+		public int getLimit() {
+			return this.limit;
 		}
 	}
 
@@ -312,9 +391,9 @@ public interface StatementMapper {
 	class InsertSpec {
 
 		private final SqlIdentifier table;
-		private final Map<String, SettableValue> assignments;
+		private final Map<SqlIdentifier, SettableValue> assignments;
 
-		protected InsertSpec(SqlIdentifier table, Map<String, SettableValue> assignments) {
+		protected InsertSpec(SqlIdentifier table, Map<SqlIdentifier, SettableValue> assignments) {
 			this.table = table;
 			this.assignments = assignments;
 		}
@@ -348,8 +427,19 @@ public interface StatementMapper {
 		 * @return the {@link InsertSpec}.
 		 */
 		public InsertSpec withColumn(String column, SettableValue value) {
+			return withColumn(SqlIdentifier.unquoted(column), value);
+		}
 
-			Map<String, SettableValue> values = new LinkedHashMap<>(this.assignments);
+		/**
+		 * Associate a column with a {@link SettableValue} and create a new {@link InsertSpec}.
+		 *
+		 * @param column
+		 * @param value
+		 * @return the {@link InsertSpec}.
+		 */
+		public InsertSpec withColumn(SqlIdentifier column, SettableValue value) {
+
+			Map<SqlIdentifier, SettableValue> values = new LinkedHashMap<>(this.assignments);
 			values.put(column, value);
 
 			return new InsertSpec(this.table, values);
@@ -359,7 +449,7 @@ public interface StatementMapper {
 			return this.table;
 		}
 
-		public Map<String, SettableValue> getAssignments() {
+		public Map<SqlIdentifier, SettableValue> getAssignments() {
 			return Collections.unmodifiableMap(this.assignments);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/query/Criteria.java b/src/main/java/org/springframework/data/r2dbc/query/Criteria.java
index 73712ee3..b737479e 100644
--- a/src/main/java/org/springframework/data/r2dbc/query/Criteria.java
+++ b/src/main/java/org/springframework/data/r2dbc/query/Criteria.java
@@ -19,6 +19,7 @@ import java.util.Arrays;
 import java.util.Collection;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -35,15 +36,15 @@ public class Criteria {
 	private final @Nullable Criteria previous;
 	private final Combinator combinator;
 
-	private final String column;
+	private final SqlIdentifier column;
 	private final Comparator comparator;
 	private final @Nullable Object value;
 
-	private Criteria(String column, Comparator comparator, @Nullable Object value) {
+	private Criteria(SqlIdentifier column, Comparator comparator, @Nullable Object value) {
 		this(null, Combinator.INITIAL, column, comparator, value);
 	}
 
-	private Criteria(@Nullable Criteria previous, Combinator combinator, String column, Comparator comparator,
+	private Criteria(@Nullable Criteria previous, Combinator combinator, SqlIdentifier column, Comparator comparator,
 			@Nullable Object value) {
 
 		this.previous = previous;
@@ -63,7 +64,7 @@ public class Criteria {
 
 		Assert.hasText(column, "Column name must not be null or empty!");
 
-		return new DefaultCriteriaStep(column);
+		return new DefaultCriteriaStep(SqlIdentifier.unquoted(column));
 	}
 
 	/**
@@ -76,10 +77,10 @@ public class Criteria {
 
 		Assert.hasText(column, "Column name must not be null or empty!");
 
-		return new DefaultCriteriaStep(column) {
+		return new DefaultCriteriaStep(SqlIdentifier.unquoted(column)) {
 			@Override
 			protected Criteria createCriteria(Comparator comparator, Object value) {
-				return new Criteria(Criteria.this, Combinator.AND, column, comparator, value);
+				return new Criteria(Criteria.this, Combinator.AND, SqlIdentifier.unquoted(column), comparator, value);
 			}
 		};
 	}
@@ -94,10 +95,10 @@ public class Criteria {
 
 		Assert.hasText(column, "Column name must not be null or empty!");
 
-		return new DefaultCriteriaStep(column) {
+		return new DefaultCriteriaStep(SqlIdentifier.unquoted(column)) {
 			@Override
 			protected Criteria createCriteria(Comparator comparator, Object value) {
-				return new Criteria(Criteria.this, Combinator.OR, column, comparator, value);
+				return new Criteria(Criteria.this, Combinator.OR, SqlIdentifier.unquoted(column), comparator, value);
 			}
 		};
 	}
@@ -126,9 +127,9 @@ public class Criteria {
 	}
 
 	/**
-	 * @return the property name.
+	 * @return the column/property name.
 	 */
-	String getColumn() {
+	SqlIdentifier getColumn() {
 		return column;
 	}
 
@@ -268,9 +269,9 @@ public class Criteria {
 	 */
 	static class DefaultCriteriaStep implements CriteriaStep {
 
-		private final String property;
+		private final SqlIdentifier property;
 
-		DefaultCriteriaStep(String property) {
+		DefaultCriteriaStep(SqlIdentifier property) {
 			this.property = property;
 		}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/query/Query.java b/src/main/java/org/springframework/data/r2dbc/query/Query.java
index 5b06edb0..d2558180 100644
--- a/src/main/java/org/springframework/data/r2dbc/query/Query.java
+++ b/src/main/java/org/springframework/data/r2dbc/query/Query.java
@@ -21,9 +21,11 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -41,8 +43,7 @@ public class Query {
 
 	private final @Nullable Criteria criteria;
 
-	// TODO: select list should be List<Expression>
-	private final List<String> columns;
+	private final List<SqlIdentifier> columns;
 	private final Sort sort;
 	private final int limit;
 	private final long offset;
@@ -63,14 +64,10 @@ public class Query {
 	 * @param criteria must not be {@literal null}.
 	 */
 	private Query(@Nullable Criteria criteria) {
-		this.criteria = criteria;
-		this.sort = Sort.unsorted();
-		this.columns = Collections.emptyList();
-		this.limit = -1;
-		this.offset = -1;
+		this(criteria, Collections.emptyList(), Sort.unsorted(), -1, -1);
 	}
 
-	private Query(Criteria criteria, List<String> columns, Sort sort, int limit, long offset) {
+	private Query(@Nullable Criteria criteria, List<SqlIdentifier> columns, Sort sort, int limit, long offset) {
 		this.criteria = criteria;
 		this.columns = columns;
 		this.sort = sort;
@@ -97,7 +94,7 @@ public class Query {
 
 		Assert.notNull(columns, "Columns must not be null");
 
-		return columns(Arrays.asList(columns));
+		return withColumns(Arrays.stream(columns).map(SqlIdentifier::unquoted).collect(Collectors.toList()));
 	}
 
 	/**
@@ -110,7 +107,34 @@ public class Query {
 
 		Assert.notNull(columns, "Columns must not be null");
 
-		List<String> newColumns = new ArrayList<>(this.columns);
+		return withColumns(columns.stream().map(SqlIdentifier::unquoted).collect(Collectors.toList()));
+	}
+
+	/**
+	 * Add columns to the query.
+	 *
+	 * @param columns
+	 * @return a new {@link Query} object containing the former settings with {@code columns} applied.
+	 * @since 1.1
+	 */
+	public Query columns(SqlIdentifier... columns) {
+
+		Assert.notNull(columns, "Columns must not be null");
+
+		return withColumns(Arrays.asList(columns));
+	}
+
+	/**
+	 * Add columns to the query.
+	 *
+	 * @param columns
+	 * @return a new {@link Query} object containing the former settings with {@code columns} applied.
+	 */
+	private Query withColumns(Collection<SqlIdentifier> columns) {
+
+		Assert.notNull(columns, "Columns must not be null");
+
+		List<SqlIdentifier> newColumns = new ArrayList<>(this.columns);
 		newColumns.addAll(columns);
 		return new Query(this.criteria, newColumns, this.sort, this.limit, offset);
 	}
@@ -186,7 +210,7 @@ public class Query {
 	 *
 	 * @return
 	 */
-	public List<String> getColumns() {
+	public List<SqlIdentifier> getColumns() {
 		return columns;
 	}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java b/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java
index d22e093e..4b70489f 100644
--- a/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java
@@ -22,7 +22,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.UnaryOperator;
 
-import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PropertyPath;
@@ -40,15 +39,7 @@ import org.springframework.data.r2dbc.query.Criteria.Combinator;
 import org.springframework.data.r2dbc.query.Criteria.Comparator;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.Aliased;
-import org.springframework.data.relational.core.sql.Column;
-import org.springframework.data.relational.core.sql.Condition;
-import org.springframework.data.relational.core.sql.Expression;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
-import org.springframework.data.relational.core.sql.SQL;
-import org.springframework.data.relational.core.sql.SqlIdentifier;
-import org.springframework.data.relational.core.sql.SimpleFunction;
-import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.core.sql.*;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
@@ -114,7 +105,7 @@ public class QueryMapper {
 
 		for (Sort.Order order : sort) {
 
-			Field field = createPropertyField(entity, order.getProperty(), this.mappingContext);
+			Field field = createPropertyField(entity, SqlIdentifier.unquoted(order.getProperty()), this.mappingContext);
 			mappedOrder.add(
 					Sort.Order.by(toSql(field.getMappedColumnName())).with(order.getNullHandling()).with(order.getDirection()));
 		}
@@ -122,6 +113,29 @@ public class QueryMapper {
 		return Sort.by(mappedOrder);
 	}
 
+	/**
+	 * Map the {@link Sort} object to apply field name mapping using {@link Class the type to read}.
+	 *
+	 * @param sort must not be {@literal null}.
+	 * @param entity related {@link RelationalPersistentEntity}, can be {@literal null}.
+	 * @return
+	 * @since 1.1
+	 */
+	public List<OrderByField> getMappedSort(Table table, Sort sort, @Nullable RelationalPersistentEntity<?> entity) {
+
+		List<OrderByField> mappedOrder = new ArrayList<>();
+
+		for (Sort.Order order : sort) {
+
+			Field field = createPropertyField(entity, SqlIdentifier.unquoted(order.getProperty()), this.mappingContext);
+			OrderByField orderBy = OrderByField.from(table.column(field.getMappedColumnName()))
+					.withNullHandling(order.getNullHandling());
+			mappedOrder.add(order.isAscending() ? orderBy.asc() : orderBy.desc());
+		}
+
+		return mappedOrder;
+	}
+
 	/**
 	 * Map the {@link Expression} object to apply field name mapping using {@link Class the type to read}.
 	 *
@@ -132,7 +146,7 @@ public class QueryMapper {
 	 */
 	public Expression getMappedObject(Expression expression, @Nullable RelationalPersistentEntity<?> entity) {
 
-		if (entity == null) {
+		if (entity == null || expression instanceof AsteriskFromTable) {
 			return expression;
 		}
 
@@ -140,19 +154,17 @@ public class QueryMapper {
 
 			Column column = (Column) expression;
 			Field field = createPropertyField(entity, column.getName());
+			Table table = column.getTable();
 
-			return column instanceof Aliased
-					? Column.aliased(field.getMappedColumnName(), column.getTable(), ((Aliased) column).getAlias())
-					: Column.create(field.getMappedColumnName(), column.getTable());
+			return column instanceof Aliased ? table.column(field.getMappedColumnName()).as(((Aliased) column).getAlias())
+					: table.column(field.getMappedColumnName());
 		}
 
 		if (expression instanceof SimpleFunction) {
 
-			// Revisit after https://jira.spring.io/browse/DATAJDBC-478
 			SimpleFunction function = (SimpleFunction) expression;
-			DirectFieldAccessor accessor = new DirectFieldAccessor(function);
 
-			List<Expression> arguments = (List<Expression>) accessor.getPropertyValue("expressions");
+			List<Expression> arguments = function.getExpressions();
 			List<Expression> mappedArguments = new ArrayList<>(arguments.size());
 
 			for (Expression argument : arguments) {
@@ -218,7 +230,7 @@ public class QueryMapper {
 			@Nullable RelationalPersistentEntity<?> entity) {
 
 		Field propertyField = createPropertyField(entity, criteria.getColumn(), this.mappingContext);
-		Column column = table.column(toSql(propertyField.getMappedColumnName()));
+		Column column = table.column(propertyField.getMappedColumnName());
 		TypeInformation<?> actualType = propertyField.getTypeHint().getRequiredActualType();
 
 		Object mappedValue;
@@ -291,7 +303,7 @@ public class QueryMapper {
 
 				for (Object o : (Iterable<?>) mappedValue) {
 
-					BindMarker bindMarker = bindings.nextMarker(column.getName());
+					BindMarker bindMarker = bindings.nextMarker(column.getName().getReference());
 					expressions.add(bind(o, valueType, bindings, bindMarker));
 				}
 
@@ -299,7 +311,7 @@ public class QueryMapper {
 
 			} else {
 
-				BindMarker bindMarker = bindings.nextMarker(column.getName());
+				BindMarker bindMarker = bindings.nextMarker(column.getName().getReference());
 				Expression expression = bind(mappedValue, valueType, bindings, bindMarker);
 
 				condition = column.in(expression);
@@ -312,7 +324,7 @@ public class QueryMapper {
 			return condition;
 		}
 
-		BindMarker bindMarker = bindings.nextMarker(column.getName());
+		BindMarker bindMarker = bindings.nextMarker(column.getName().getReference());
 		Expression expression = bind(mappedValue, valueType, bindings, bindMarker);
 
 		switch (comparator) {
@@ -335,11 +347,11 @@ public class QueryMapper {
 		}
 	}
 
-	Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, String key) {
+	Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, SqlIdentifier key) {
 		return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext);
 	}
 
-	Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, String key,
+	Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, SqlIdentifier key,
 			MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
 		return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext);
 	}
@@ -374,14 +386,14 @@ public class QueryMapper {
 	 */
 	protected static class Field {
 
-		protected final String name;
+		protected final SqlIdentifier 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) {
+		public Field(SqlIdentifier name) {
 
 			Assert.notNull(name, "Name must not be null!");
 			this.name = name;
@@ -393,7 +405,7 @@ public class QueryMapper {
 		 * @return
 		 */
 		public SqlIdentifier getMappedColumnName() {
-			return new PassThruIdentifier(this.name);
+			return this.name;
 		}
 
 		public TypeInformation<?> getTypeHint() {
@@ -419,7 +431,7 @@ public class QueryMapper {
 		 * @param entity must not be {@literal null}.
 		 * @param context must not be {@literal null}.
 		 */
-		protected MetadataBackedField(String name, RelationalPersistentEntity<?> entity,
+		protected MetadataBackedField(SqlIdentifier name, RelationalPersistentEntity<?> entity,
 				MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context) {
 			this(name, entity, context, null);
 		}
@@ -433,7 +445,7 @@ public class QueryMapper {
 		 * @param context must not be {@literal null}.
 		 * @param property may be {@literal null}.
 		 */
-		protected MetadataBackedField(String name, RelationalPersistentEntity<?> entity,
+		protected MetadataBackedField(SqlIdentifier name, RelationalPersistentEntity<?> entity,
 				MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context,
 				@Nullable RelationalPersistentProperty property) {
 
@@ -444,7 +456,7 @@ public class QueryMapper {
 			this.entity = entity;
 			this.mappingContext = context;
 
-			this.path = getPath(name);
+			this.path = getPath(name.getReference());
 			this.property = this.path == null ? property : this.path.getLeafProperty();
 		}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/query/Update.java b/src/main/java/org/springframework/data/r2dbc/query/Update.java
index f265aa2f..068ff523 100644
--- a/src/main/java/org/springframework/data/r2dbc/query/Update.java
+++ b/src/main/java/org/springframework/data/r2dbc/query/Update.java
@@ -19,6 +19,7 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -32,9 +33,9 @@ public class Update {
 
 	private static final Update EMPTY = new Update(Collections.emptyMap());
 
-	private final Map<String, Object> columnsToUpdate;
+	private final Map<SqlIdentifier, Object> columnsToUpdate;
 
-	private Update(Map<String, Object> columnsToUpdate) {
+	private Update(Map<SqlIdentifier, Object> columnsToUpdate) {
 		this.columnsToUpdate = columnsToUpdate;
 	}
 
@@ -57,6 +58,21 @@ public class Update {
 	 * @return
 	 */
 	public Update set(String column, @Nullable Object value) {
+
+		Assert.hasText(column, "Column for update must not be null or blank");
+
+		return addMultiFieldOperation(SqlIdentifier.unquoted(column), value);
+	}
+
+	/**
+	 * Update a column by assigning a value.
+	 *
+	 * @param column must not be {@literal null}.
+	 * @param value can be {@literal null}.
+	 * @return
+	 * @since 1.1
+	 */
+	public Update set(SqlIdentifier column, @Nullable Object value) {
 		return addMultiFieldOperation(column, value);
 	}
 
@@ -65,15 +81,15 @@ public class Update {
 	 *
 	 * @return
 	 */
-	public Map<String, Object> getAssignments() {
+	public Map<SqlIdentifier, Object> getAssignments() {
 		return Collections.unmodifiableMap(this.columnsToUpdate);
 	}
 
-	private Update addMultiFieldOperation(String key, Object value) {
+	private Update addMultiFieldOperation(SqlIdentifier key, @Nullable Object value) {
 
-		Assert.hasText(key, "Column for update must not be null or blank");
+		Assert.notNull(key, "Column for update must not be null");
 
-		Map<String, Object> updates = new LinkedHashMap<>(this.columnsToUpdate);
+		Map<SqlIdentifier, Object> updates = new LinkedHashMap<>(this.columnsToUpdate);
 		updates.put(key, value);
 
 		return new Update(updates);
diff --git a/src/main/java/org/springframework/data/r2dbc/query/UpdateMapper.java b/src/main/java/org/springframework/data/r2dbc/query/UpdateMapper.java
index 21d69b92..260df541 100644
--- a/src/main/java/org/springframework/data/r2dbc/query/UpdateMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/query/UpdateMapper.java
@@ -32,6 +32,7 @@ import org.springframework.data.relational.core.sql.Assignment;
 import org.springframework.data.relational.core.sql.Assignments;
 import org.springframework.data.relational.core.sql.Column;
 import org.springframework.data.relational.core.sql.SQL;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
@@ -77,8 +78,8 @@ public class UpdateMapper extends QueryMapper {
 	 * @param entity related {@link RelationalPersistentEntity}, can be {@literal null}.
 	 * @return the mapped {@link BoundAssignments}.
 	 */
-	public BoundAssignments getMappedObject(BindMarkers markers, Map<String, ? extends Object> assignments, Table table,
-			@Nullable RelationalPersistentEntity<?> entity) {
+	public BoundAssignments getMappedObject(BindMarkers markers, Map<SqlIdentifier, ? extends Object> assignments,
+			Table table, @Nullable RelationalPersistentEntity<?> entity) {
 
 		Assert.notNull(markers, "BindMarkers must not be null!");
 		Assert.notNull(assignments, "Assignments must not be null!");
@@ -95,11 +96,11 @@ public class UpdateMapper extends QueryMapper {
 		return new BoundAssignments(bindings, result);
 	}
 
-	private Assignment getAssignment(String columnName, Object value, MutableBindings bindings, Table table,
+	private Assignment getAssignment(SqlIdentifier columnName, Object value, MutableBindings bindings, Table table,
 			@Nullable RelationalPersistentEntity<?> entity) {
 
 		Field propertyField = createPropertyField(entity, columnName, getMappingContext());
-		Column column = table.column(toSql(propertyField.getMappedColumnName()));
+		Column column = table.column(propertyField.getMappedColumnName());
 		TypeInformation<?> actualType = propertyField.getTypeHint().getRequiredActualType();
 
 		Object mappedValue;
@@ -128,7 +129,7 @@ public class UpdateMapper extends QueryMapper {
 
 	private Assignment createAssignment(Column column, Object value, Class<?> type, MutableBindings bindings) {
 
-		BindMarker bindMarker = bindings.nextMarker(column.getName());
+		BindMarker bindMarker = bindings.nextMarker(column.getName().getReference());
 		AssignValue assignValue = Assignments.value(column, SQL.bindMarker(bindMarker.getPlaceholder()));
 
 		if (value == null) {
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 191152e3..4ff1b44f 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
@@ -28,12 +28,6 @@ import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.query.Criteria;
 import org.springframework.data.r2dbc.query.Query;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.Functions;
-import org.springframework.data.relational.core.sql.Select;
-import org.springframework.data.relational.core.sql.SqlIdentifier;
-import org.springframework.data.relational.core.sql.StatementBuilder;
-import org.springframework.data.relational.core.sql.Table;
-import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.relational.repository.query.RelationalEntityInformation;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.data.util.Lazy;
diff --git a/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java
index dd5031be..bfdab1ea 100644
--- a/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java
@@ -96,7 +96,7 @@ public class R2dbcEntityTemplateUnitTests {
 
 		StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
 
-		assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1");
+		assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1 LIMIT 1");
 		assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
 	}
 
diff --git a/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java
index ec2593bd..1a4c8b29 100644
--- a/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java
@@ -185,7 +185,7 @@ public class ReactiveSelectOperationUnitTests {
 
 		StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
 
-		assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1");
+		assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1 LIMIT 1");
 	}
 
 	@Test // gh-220
diff --git a/src/test/java/org/springframework/data/r2dbc/query/CriteriaUnitTests.java b/src/test/java/org/springframework/data/r2dbc/query/CriteriaUnitTests.java
index d8e82ea3..887585ba 100644
--- a/src/test/java/org/springframework/data/r2dbc/query/CriteriaUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/query/CriteriaUnitTests.java
@@ -21,9 +21,9 @@ import static org.springframework.data.r2dbc.query.Criteria.*;
 import java.util.Arrays;
 
 import org.junit.Test;
-import org.springframework.data.r2dbc.query.Criteria;
-import org.springframework.data.r2dbc.query.Criteria.Combinator;
-import org.springframework.data.r2dbc.query.Criteria.Comparator;
+
+import org.springframework.data.r2dbc.query.Criteria.*;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 
 /**
  * Unit tests for {@link Criteria}.
@@ -37,7 +37,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").is("bar").and("baz").isNotNull();
 
-		assertThat(criteria.getColumn()).isEqualTo("baz");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("baz"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NOT_NULL);
 		assertThat(criteria.getValue()).isNull();
 		assertThat(criteria.getPrevious()).isNotNull();
@@ -45,7 +45,7 @@ public class CriteriaUnitTests {
 
 		criteria = criteria.getPrevious();
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.EQ);
 		assertThat(criteria.getValue()).isEqualTo("bar");
 	}
@@ -55,7 +55,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").is("bar").or("baz").isNotNull();
 
-		assertThat(criteria.getColumn()).isEqualTo("baz");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("baz"));
 		assertThat(criteria.getCombinator()).isEqualTo(Combinator.OR);
 
 		criteria = criteria.getPrevious();
@@ -69,7 +69,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").is("bar");
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.EQ);
 		assertThat(criteria.getValue()).isEqualTo("bar");
 	}
@@ -79,7 +79,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").not("bar");
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.NEQ);
 		assertThat(criteria.getValue()).isEqualTo("bar");
 	}
@@ -89,7 +89,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").in("bar", "baz");
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.IN);
 		assertThat(criteria.getValue()).isEqualTo(Arrays.asList("bar", "baz"));
 	}
@@ -99,7 +99,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").notIn("bar", "baz");
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.NOT_IN);
 		assertThat(criteria.getValue()).isEqualTo(Arrays.asList("bar", "baz"));
 	}
@@ -109,7 +109,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").greaterThan(1);
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.GT);
 		assertThat(criteria.getValue()).isEqualTo(1);
 	}
@@ -119,7 +119,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").greaterThanOrEquals(1);
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.GTE);
 		assertThat(criteria.getValue()).isEqualTo(1);
 	}
@@ -129,7 +129,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").lessThan(1);
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.LT);
 		assertThat(criteria.getValue()).isEqualTo(1);
 	}
@@ -139,7 +139,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").lessThanOrEquals(1);
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.LTE);
 		assertThat(criteria.getValue()).isEqualTo(1);
 	}
@@ -149,7 +149,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").like("hello%");
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.LIKE);
 		assertThat(criteria.getValue()).isEqualTo("hello%");
 	}
@@ -159,7 +159,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").isNull();
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NULL);
 	}
 
@@ -168,7 +168,7 @@ public class CriteriaUnitTests {
 
 		Criteria criteria = where("foo").isNotNull();
 
-		assertThat(criteria.getColumn()).isEqualTo("foo");
+		assertThat(criteria.getColumn()).isEqualTo(SqlIdentifier.unquoted("foo"));
 		assertThat(criteria.getComparator()).isEqualTo(Comparator.IS_NOT_NULL);
 	}
 }
diff --git a/src/test/java/org/springframework/data/r2dbc/query/UpdateMapperUnitTests.java b/src/test/java/org/springframework/data/r2dbc/query/UpdateMapperUnitTests.java
index 26679073..9a747475 100644
--- a/src/test/java/org/springframework/data/r2dbc/query/UpdateMapperUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/query/UpdateMapperUnitTests.java
@@ -34,6 +34,7 @@ import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.sql.AssignValue;
 import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.SQL;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.Table;
 
 /**
@@ -54,10 +55,10 @@ public class UpdateMapperUnitTests {
 
 		BoundAssignments mapped = map(update);
 
-		Map<String, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
+		Map<SqlIdentifier, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
 				.collect(Collectors.toMap(k -> k.getColumn().getName(), AssignValue::getValue));
 
-		assertThat(assignments).containsEntry("another_name", SQL.bindMarker("$1"));
+		assertThat(assignments).containsEntry(SqlIdentifier.unquoted("another_name"), SQL.bindMarker("$1"));
 	}
 
 	@Test // gh-64
@@ -67,10 +68,10 @@ public class UpdateMapperUnitTests {
 
 		BoundAssignments mapped = map(update);
 
-		Map<String, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
+		Map<SqlIdentifier, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
 				.collect(Collectors.toMap(k -> k.getColumn().getName(), AssignValue::getValue));
 
-		assertThat(assignments).containsEntry("another_name", SQL.bindMarker("$1"));
+		assertThat(assignments).containsEntry(SqlIdentifier.unquoted("another_name"), SQL.bindMarker("$1"));
 
 		mapped.getBindings().apply(bindTarget);
 		verify(bindTarget).bindNull(0, String.class);
@@ -87,7 +88,7 @@ public class UpdateMapperUnitTests {
 		assertThat(mapped.getAssignments().get(0).toString()).isEqualTo("person.another_name = NULL");
 
 		mapped.getBindings().apply(bindTarget);
-		verifyZeroInteractions(bindTarget);
+		verifyNoInteractions(bindTarget);
 	}
 
 	@Test // gh-195
@@ -97,12 +98,12 @@ public class UpdateMapperUnitTests {
 
 		BoundAssignments mapped = map(update);
 
-		Map<String, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
+		Map<SqlIdentifier, Expression> assignments = mapped.getAssignments().stream().map(it -> (AssignValue) it)
 				.collect(Collectors.toMap(k -> k.getColumn().getName(), AssignValue::getValue));
 
 		assertThat(update.getAssignments()).hasSize(3);
-		assertThat(assignments).hasSize(3).containsEntry("c1", SQL.bindMarker("$1")).containsEntry("c2",
-				SQL.bindMarker("$2"));
+		assertThat(assignments).hasSize(3).containsEntry(SqlIdentifier.unquoted("c1"), SQL.bindMarker("$1"))
+				.containsEntry(SqlIdentifier.unquoted("c2"), SQL.bindMarker("$2"));
 	}
 
 	private BoundAssignments map(Update update) {