From 32f1770529c5fd4916644e8f46afdcce953773b5 Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Wed, 22 May 2019 10:07:18 +0200
Subject: [PATCH] #55 - Reuse Dialect support provided by Spring Data
 Relational.

We now reuse the existing Dialect infrastructure provided by Spring Data Relational to enhance it for R2DBC specifics such as bind markers.

Original pull request: #125.
---
 .../config/AbstractR2dbcConfiguration.java    | 20 ++---
 .../r2dbc/convert/MappingR2dbcConverter.java  | 10 +--
 .../data/r2dbc/convert/R2dbcConverter.java    |  2 +-
 .../core/DefaultDatabaseClientBuilder.java    |  4 +-
 .../DefaultReactiveDataAccessStrategy.java    | 63 +++++-----------
 .../r2dbc/core/DefaultStatementMapper.java    | 38 ++--------
 .../core/ReactiveDataAccessStrategy.java      |  5 +-
 .../data/r2dbc/core/StatementMapper.java      |  4 +-
 .../data/r2dbc/dialect/ArrayColumns.java      | 53 --------------
 .../data/r2dbc/dialect/Database.java          | 20 ++---
 .../data/r2dbc/dialect/LimitClause.java       | 42 -----------
 .../data/r2dbc/dialect/MySqlDialect.java      | 51 +------------
 .../data/r2dbc/dialect/PostgresDialect.java   | 73 ++++---------------
 .../{Dialect.java => R2dbcDialect.java}       | 22 +-----
 .../data/r2dbc/dialect/SqlServerDialect.java  | 42 +----------
 .../r2dbc/support/StatementRenderUtil.java    | 66 -----------------
 ...ReactiveDataAccessStrategyTestSupport.java |  4 +-
 .../dialect/PostgresDialectUnitTests.java     |  1 +
 .../dialect/SqlServerDialectUnitTests.java    |  1 +
 19 files changed, 81 insertions(+), 440 deletions(-)
 delete mode 100644 src/main/java/org/springframework/data/r2dbc/dialect/ArrayColumns.java
 delete mode 100644 src/main/java/org/springframework/data/r2dbc/dialect/LimitClause.java
 rename src/main/java/org/springframework/data/r2dbc/dialect/{Dialect.java => R2dbcDialect.java} (67%)
 delete mode 100644 src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java

diff --git a/src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java b/src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java
index f5c048c..2f50917 100644
--- a/src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java
+++ b/src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java
@@ -34,7 +34,7 @@ import org.springframework.data.r2dbc.core.DatabaseClient;
 import org.springframework.data.r2dbc.core.DefaultReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.dialect.Database;
-import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.support.R2dbcExceptionSubclassTranslator;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.data.r2dbc.support.SqlStateR2dbcExceptionTranslator;
@@ -78,15 +78,15 @@ public abstract class AbstractR2dbcConfiguration implements ApplicationContextAw
 	public abstract ConnectionFactory connectionFactory();
 
 	/**
-	 * Return a {@link Dialect} for the given {@link ConnectionFactory}. This method attempts to resolve a {@link Dialect}
-	 * from {@link io.r2dbc.spi.ConnectionFactoryMetadata}. Override this method to specify a dialect instead of
-	 * attempting to resolve one.
+	 * Return a {@link R2dbcDialect} for the given {@link ConnectionFactory}. This method attempts to resolve a
+	 * {@link R2dbcDialect} from {@link io.r2dbc.spi.ConnectionFactoryMetadata}. Override this method to specify a dialect
+	 * instead of attempting to resolve one.
 	 *
 	 * @param connectionFactory the configured {@link ConnectionFactory}.
-	 * @return the resolved {@link Dialect}.
-	 * @throws UnsupportedOperationException if the {@link Dialect} cannot be determined.
+	 * @return the resolved {@link R2dbcDialect}.
+	 * @throws UnsupportedOperationException if the {@link R2dbcDialect} cannot be determined.
 	 */
-	public Dialect getDialect(ConnectionFactory connectionFactory) {
+	public R2dbcDialect getDialect(ConnectionFactory connectionFactory) {
 
 		return Database.findDatabase(connectionFactory)
 				.orElseThrow(() -> new UnsupportedOperationException(
@@ -172,13 +172,13 @@ public abstract class AbstractR2dbcConfiguration implements ApplicationContextAw
 	}
 
 	/**
-	 * Returns the {@link Dialect}-specific {@link StoreConversions}.
+	 * Returns the {@link R2dbcDialect}-specific {@link StoreConversions}.
 	 *
-	 * @return the {@link Dialect}-specific {@link StoreConversions}.
+	 * @return the {@link R2dbcDialect}-specific {@link StoreConversions}.
 	 */
 	protected StoreConversions getStoreConversions() {
 
-		Dialect dialect = getDialect(lookupConnectionFactory());
+		R2dbcDialect dialect = getDialect(lookupConnectionFactory());
 		return StoreConversions.of(dialect.getSimpleTypeHolder(), R2dbcCustomConversions.STORE_CONVERTERS);
 	}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java b/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
index 72e63f0..d337eba 100644
--- a/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
+++ b/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
@@ -38,11 +38,11 @@ import org.springframework.data.mapping.PreferredConstructor.Parameter;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.ParameterValueProvider;
-import org.springframework.data.r2dbc.dialect.ArrayColumns;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.util.ClassTypeInformation;
@@ -84,7 +84,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	// Entity reading
 	// ----------------------------------
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.convert.EntityReader#read(java.lang.Class, S)
 	 */
@@ -93,7 +93,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 		return read(type, row, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.convert.R2dbcConverter#read(java.lang.Class, io.r2dbc.spi.Row, io.r2dbc.spi.RowMetadata)
 	 */
@@ -233,7 +233,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	// Entity writing
 	// ----------------------------------
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.convert.EntityWriter#write(java.lang.Object, java.lang.Object)
 	 */
@@ -337,7 +337,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 		return Enum.class.isAssignableFrom(value.getClass()) ? ((Enum<?>) value).name() : value;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.convert.R2dbcConverter#getArrayValue(org.springframework.data.r2dbc.dialect.ArrayColumns, org.springframework.data.relational.core.mapping.RelationalPersistentProperty, java.lang.Object)
 	 */
diff --git a/src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java b/src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java
index ed9d345..81015a8 100644
--- a/src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java
+++ b/src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java
@@ -24,9 +24,9 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.data.convert.EntityReader;
 import org.springframework.data.convert.EntityWriter;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.r2dbc.dialect.ArrayColumns;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 
diff --git a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientBuilder.java b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientBuilder.java
index 8edd56d..8853f5b 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientBuilder.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClientBuilder.java
@@ -22,7 +22,7 @@ import java.util.function.Consumer;
 
 import org.springframework.data.r2dbc.core.DatabaseClient.Builder;
 import org.springframework.data.r2dbc.dialect.Database;
-import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.support.R2dbcExceptionSubclassTranslator;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.lang.Nullable;
@@ -121,7 +121,7 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 		if (accessStrategy == null) {
 
-			Dialect dialect = Database.findDatabase(this.connectionFactory)
+			R2dbcDialect dialect = Database.findDatabase(this.connectionFactory)
 					.orElseThrow(() -> new UnsupportedOperationException(
 							"Cannot determine a Dialect. Configure the dialect by providing DefaultReactiveDataAccessStrategy(Dialect)"))
 					.defaultDialect();
diff --git a/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java
index 3385f88..e7ee999 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java
@@ -23,9 +23,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.BiFunction;
-import java.util.function.Function;
 
-import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.convert.CustomConversions.StoreConversions;
 import org.springframework.data.mapping.context.MappingContext;
@@ -33,26 +31,19 @@ import org.springframework.data.r2dbc.convert.EntityRowMapper;
 import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
-import org.springframework.data.r2dbc.dialect.ArrayColumns;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
-import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 import org.springframework.data.r2dbc.query.UpdateMapper;
-import org.springframework.data.relational.core.mapping.NamingStrategy;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
+import org.springframework.data.relational.core.dialect.RenderContextFactory;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.mapping.Table;
-import org.springframework.data.relational.core.sql.Select;
-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.
@@ -61,36 +52,36 @@ import org.springframework.util.StringUtils;
  */
 public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
 
-	private final Dialect dialect;
+	private final R2dbcDialect dialect;
 	private final R2dbcConverter converter;
 	private final UpdateMapper updateMapper;
 	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 	private final StatementMapper statementMapper;
 
 	/**
-	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect} and optional
+	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link R2dbcDialect} and optional
 	 * {@link org.springframework.core.convert.converter.Converter}s.
 	 *
-	 * @param dialect the {@link Dialect} to use.
+	 * @param dialect the {@link R2dbcDialect} to use.
 	 */
-	public DefaultReactiveDataAccessStrategy(Dialect dialect) {
+	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect) {
 		this(dialect, Collections.emptyList());
 	}
 
 	/**
-	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect} and optional
+	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link R2dbcDialect} and optional
 	 * {@link org.springframework.core.convert.converter.Converter}s.
 	 *
-	 * @param dialect the {@link Dialect} to use.
+	 * @param dialect the {@link R2dbcDialect} to use.
 	 * @param converters custom converters to register, must not be {@literal null}.
 	 * @see R2dbcCustomConversions
 	 * @see org.springframework.core.convert.converter.Converter
 	 */
-	public DefaultReactiveDataAccessStrategy(Dialect dialect, Collection<?> converters) {
+	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect, Collection<?> converters) {
 		this(dialect, createConverter(dialect, converters));
 	}
 
-	private static R2dbcConverter createConverter(Dialect dialect, Collection<?> converters) {
+	private static R2dbcConverter createConverter(R2dbcDialect dialect, Collection<?> converters) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(converters, "Converters must not be null");
@@ -105,13 +96,13 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	}
 
 	/**
-	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect} and {@link R2dbcConverter}.
+	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link R2dbcDialect} and {@link R2dbcConverter}.
 	 *
-	 * @param dialect the {@link Dialect} to use.
+	 * @param dialect the {@link R2dbcDialect} to use.
 	 * @param converter must not be {@literal null}.
 	 */
 	@SuppressWarnings("unchecked")
-	public DefaultReactiveDataAccessStrategy(Dialect dialect, R2dbcConverter converter) {
+	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect, R2dbcConverter converter) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(converter, "RelationalConverter must not be null");
@@ -122,29 +113,9 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 				.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.statementMapper = new DefaultStatementMapper(dialect, renderContext, this.updateMapper, this.mappingContext);
+		RenderContextFactory factory = new RenderContextFactory(dialect);
+		this.statementMapper = new DefaultStatementMapper(dialect, factory.createRenderContext(), this.updateMapper,
+				this.mappingContext);
 	}
 
 	/*
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 ae2d806..9bff339 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java
@@ -20,7 +20,6 @@ import lombok.RequiredArgsConstructor;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.OptionalLong;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -28,28 +27,14 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindTarget;
 import org.springframework.data.r2dbc.dialect.Bindings;
-import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.query.BoundAssignments;
 import org.springframework.data.r2dbc.query.BoundCondition;
 import org.springframework.data.r2dbc.query.UpdateMapper;
-import org.springframework.data.r2dbc.support.StatementRenderUtil;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-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.Delete;
-import org.springframework.data.relational.core.sql.DeleteBuilder;
-import org.springframework.data.relational.core.sql.Insert;
-import org.springframework.data.relational.core.sql.InsertBuilder;
+import org.springframework.data.relational.core.sql.*;
 import org.springframework.data.relational.core.sql.InsertBuilder.InsertValuesWithBuild;
-import org.springframework.data.relational.core.sql.OrderByField;
-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;
@@ -63,7 +48,7 @@ import org.springframework.util.Assert;
 @RequiredArgsConstructor
 class DefaultStatementMapper implements StatementMapper {
 
-	private final Dialect dialect;
+	private final R2dbcDialect dialect;
 	private final RenderContext renderContext;
 	private final UpdateMapper updateMapper;
 	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
@@ -116,26 +101,15 @@ class DefaultStatementMapper implements StatementMapper {
 			selectBuilder.orderBy(createOrderByFields(table, mappedSort));
 		}
 
-		OptionalLong limit;
-		OptionalLong offset;
-
 		if (selectSpec.getPage().isPaged()) {
 
 			Pageable page = selectSpec.getPage();
-			limit = OptionalLong.of(page.getPageSize());
-			offset = OptionalLong.of(page.getOffset());
-		} else {
-			limit = OptionalLong.empty();
-			offset = OptionalLong.empty();
+
+			selectBuilder.limitOffset(page.getPageSize(), page.getOffset());
 		}
 
 		Select select = selectBuilder.build();
-		return new DefaultPreparedOperation<Select>(select, this.renderContext, bindings) {
-			@Override
-			public String toQuery() {
-				return StatementRenderUtil.render(select, limit, offset, DefaultStatementMapper.this.dialect);
-			}
-		};
+		return new DefaultPreparedOperation<>(select, this.renderContext, bindings);
 	}
 
 	private Collection<? extends OrderByField> createOrderByFields(Table table, Sort sortToUse) {
diff --git a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java
index 0070cd6..cd98bb3 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java
@@ -23,7 +23,6 @@ import java.util.function.BiFunction;
 
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
-import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 
@@ -67,9 +66,9 @@ public interface ReactiveDataAccessStrategy {
 	String getTableName(Class<?> type);
 
 	/**
-	 * Returns the {@link Dialect}-specific {@link StatementMapper}.
+	 * Returns the {@link org.springframework.data.r2dbc.dialect.R2dbcDialect}-specific {@link StatementMapper}.
 	 *
-	 * @return the {@link Dialect}-specific {@link StatementMapper}.
+	 * @return the {@link org.springframework.data.r2dbc.dialect.R2dbcDialect}-specific {@link StatementMapper}.
 	 */
 	StatementMapper getStatementMapper();
 
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 6641eb5..94d5ba8 100644
--- a/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
@@ -25,7 +25,6 @@ import java.util.Map;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
-import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 import org.springframework.data.r2dbc.query.Criteria;
 import org.springframework.data.r2dbc.query.Update;
@@ -33,7 +32,8 @@ import org.springframework.lang.Nullable;
 
 /**
  * Mapper for statement specifications to {@link PreparedOperation}. Statement mapping applies a
- * {@link Dialect}-specific transformation considering {@link BindMarkers} and vendor-specific SQL differences.
+ * {@link org.springframework.data.r2dbc.dialect.R2dbcDialect}-specific transformation considering {@link BindMarkers}
+ * and vendor-specific SQL differences.
  *
  * @author Mark Paluch
  */
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/ArrayColumns.java b/src/main/java/org/springframework/data/r2dbc/dialect/ArrayColumns.java
deleted file mode 100644
index 5259af2..0000000
--- a/src/main/java/org/springframework/data/r2dbc/dialect/ArrayColumns.java
+++ /dev/null
@@ -1,53 +0,0 @@
-package org.springframework.data.r2dbc.dialect;
-
-/**
- * Interface declaring methods that express how a dialect supports array-typed columns.
- *
- * @author Mark Paluch
- */
-public interface ArrayColumns {
-
-	/**
-	 * Returns {@literal true} if the dialect supports array-typed columns.
-	 *
-	 * @return {@literal true} if the dialect supports array-typed columns.
-	 */
-	boolean isSupported();
-
-	/**
-	 * Translate the {@link Class user type} of an array into the dialect-specific type. This method considers only the
-	 * component type.
-	 *
-	 * @param userType component type of the array.
-	 * @return the dialect-supported array type.
-	 * @throws UnsupportedOperationException if array typed columns are not supported.
-	 * @throws IllegalArgumentException if the {@code userType} is not a supported array type.
-	 */
-	Class<?> getArrayType(Class<?> userType);
-
-	/**
-	 * Default {@link ArrayColumns} implementation for dialects that do not support array-typed columns.
-	 */
-	enum Unsupported implements ArrayColumns {
-
-		INSTANCE;
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#isSupported()
-		 */
-		@Override
-		public boolean isSupported() {
-			return false;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#getArrayType(java.lang.Class)
-		 */
-		@Override
-		public Class<?> getArrayType(Class<?> userType) {
-			throw new UnsupportedOperationException("Array types not supported");
-		}
-	}
-}
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/Database.java b/src/main/java/org/springframework/data/r2dbc/dialect/Database.java
index 0eb27bf..64f7f95 100644
--- a/src/main/java/org/springframework/data/r2dbc/dialect/Database.java
+++ b/src/main/java/org/springframework/data/r2dbc/dialect/Database.java
@@ -10,9 +10,9 @@ import java.util.Optional;
 import org.springframework.util.Assert;
 
 /**
- * Enumeration of known Databases for offline {@link Dialect} resolution. R2DBC {@link io.r2dbc.spi.ConnectionFactory}
- * provides {@link io.r2dbc.spi.ConnectionFactoryMetadata metadata} that allows resolving an appropriate {@link Dialect}
- * if none was configured explicitly.
+ * Enumeration of known Databases for offline {@link R2dbcDialect} resolution. R2DBC
+ * {@link io.r2dbc.spi.ConnectionFactory} provides {@link io.r2dbc.spi.ConnectionFactoryMetadata metadata} that allows
+ * resolving an appropriate {@link R2dbcDialect} if none was configured explicitly.
  *
  * @author Mark Paluch
  * @author Jens Schauder
@@ -26,7 +26,7 @@ public enum Database {
 		}
 
 		@Override
-		public Dialect defaultDialect() {
+		public R2dbcDialect defaultDialect() {
 			return PostgresDialect.INSTANCE;
 		}
 	},
@@ -38,7 +38,7 @@ public enum Database {
 		}
 
 		@Override
-		public Dialect defaultDialect() {
+		public R2dbcDialect defaultDialect() {
 			return SqlServerDialect.INSTANCE;
 		}
 	},
@@ -50,7 +50,7 @@ public enum Database {
 		}
 
 		@Override
-		public Dialect defaultDialect() {
+		public R2dbcDialect defaultDialect() {
 			return H2Dialect.INSTANCE;
 		}
 	},
@@ -62,7 +62,7 @@ public enum Database {
 		}
 
 		@Override
-		public Dialect defaultDialect() {
+		public R2dbcDialect defaultDialect() {
 			return MySqlDialect.INSTANCE;
 		}
 	};
@@ -96,10 +96,10 @@ public enum Database {
 	public abstract String driverName();
 
 	/**
-	 * Returns the latest {@link Dialect} for the underlying database.
+	 * Returns the latest {@link R2dbcDialect} for the underlying database.
 	 *
-	 * @return the latest {@link Dialect} for the underlying database.
+	 * @return the latest {@link R2dbcDialect} for the underlying database.
 	 */
-	public abstract Dialect defaultDialect();
+	public abstract R2dbcDialect defaultDialect();
 
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/LimitClause.java b/src/main/java/org/springframework/data/r2dbc/dialect/LimitClause.java
deleted file mode 100644
index 7d1eb48..0000000
--- a/src/main/java/org/springframework/data/r2dbc/dialect/LimitClause.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package org.springframework.data.r2dbc.dialect;
-
-/**
- * A clause representing Dialect-specific {@code LIMIT}.
- *
- * @author Mark Paluch
- */
-public interface LimitClause {
-
-	/**
-	 * Returns the {@code LIMIT} clause
-	 *
-	 * @param limit the actual limit to use.
-	 * @return rendered limit clause.
-	 */
-	String getClause(long limit);
-
-	/**
-	 * Returns the {@code LIMIT} clause
-	 *
-	 * @param limit the actual limit to use.
-	 * @param offset the offset to start from.
-	 * @return rendered limit clause.
-	 */
-	String getClause(long limit, long offset);
-
-	/**
-	 * Returns the {@link Position} where to apply the {@link #getClause(long) clause}.
-	 */
-	Position getClausePosition();
-
-	/**
-	 * Enumeration of where to render the clause within the SQL statement.
-	 */
-	enum Position {
-
-		/**
-		 * Append the clause at the end of the statement.
-		 */
-		END
-	}
-}
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/MySqlDialect.java b/src/main/java/org/springframework/data/r2dbc/dialect/MySqlDialect.java
index d3cd260..78f2c01 100644
--- a/src/main/java/org/springframework/data/r2dbc/dialect/MySqlDialect.java
+++ b/src/main/java/org/springframework/data/r2dbc/dialect/MySqlDialect.java
@@ -29,7 +29,8 @@ import java.util.UUID;
  *
  * @author Mark Paluch
  */
-public class MySqlDialect implements Dialect {
+public class MySqlDialect extends org.springframework.data.relational.core.dialect.MySqlDialect
+		implements R2dbcDialect {
 
 	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(
 			Arrays.asList(UUID.class, URL.class, URI.class, InetAddress.class));
@@ -41,36 +42,6 @@ public class MySqlDialect implements Dialect {
 
 	private static final BindMarkersFactory ANONYMOUS = BindMarkersFactory.anonymous("?");
 
-	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
-		 */
-		@Override
-		public String getClause(long limit, long offset) {
-			return String.format("LIMIT %d,%d", limit, offset);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
-		 */
-		@Override
-		public String getClause(long limit) {
-			return "LIMIT " + limit;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
-		 */
-		@Override
-		public Position getClausePosition() {
-			return Position.END;
-		}
-	};
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
@@ -88,22 +59,4 @@ public class MySqlDialect implements Dialect {
 	public Collection<? extends Class<?>> getSimpleTypes() {
 		return SIMPLE_TYPES;
 	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()
-	 */
-	@Override
-	public LimitClause limit() {
-		return LIMIT_CLAUSE;
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.dialect.Dialect#getArraySupport()
-	 */
-	@Override
-	public ArrayColumns getArraySupport() {
-		return ArrayColumns.Unsupported.INSTANCE;
-	}
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java b/src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java
index 9ccd41b..162aae7 100644
--- a/src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java
+++ b/src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java
@@ -12,7 +12,8 @@ import java.util.Set;
 import java.util.UUID;
 
 import org.springframework.data.mapping.model.SimpleTypeHolder;
-import org.springframework.util.Assert;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
+import org.springframework.data.util.Lazy;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -20,7 +21,8 @@ import org.springframework.util.ClassUtils;
  *
  * @author Mark Paluch
  */
-public class PostgresDialect implements Dialect {
+public class PostgresDialect extends org.springframework.data.relational.core.dialect.PostgresDialect
+		implements R2dbcDialect {
 
 	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(
 			Arrays.asList(UUID.class, URL.class, URI.class, InetAddress.class));
@@ -32,37 +34,9 @@ public class PostgresDialect implements Dialect {
 
 	private static final BindMarkersFactory INDEXED = BindMarkersFactory.indexed("$", 1);
 
-	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
-		 */
-		@Override
-		public String getClause(long limit, long offset) {
-			return String.format("LIMIT %d OFFSET %d", limit, offset);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
-		 */
-		@Override
-		public String getClause(long limit) {
-			return "LIMIT " + limit;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
-		 */
-		@Override
-		public Position getClausePosition() {
-			return Position.END;
-		}
-	};
-
-	private final PostgresArrayColumns ARRAY_COLUMNS = new PostgresArrayColumns(getSimpleTypeHolder());
+	private final Lazy<ArrayColumns> arrayColumns = Lazy.of(() -> new R2dbcArrayColumns(
+			org.springframework.data.relational.core.dialect.PostgresDialect.INSTANCE.getArraySupport(),
+			getSimpleTypeHolder()));
 
 	/*
 	 * (non-Javadoc)
@@ -82,52 +56,35 @@ public class PostgresDialect implements Dialect {
 		return SIMPLE_TYPES;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()
-	 */
-	@Override
-	public LimitClause limit() {
-		return LIMIT_CLAUSE;
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#getArraySupport()
 	 */
 	@Override
 	public ArrayColumns getArraySupport() {
-		return ARRAY_COLUMNS;
+		return this.arrayColumns.get();
 	}
 
 	@RequiredArgsConstructor
-	static class PostgresArrayColumns implements ArrayColumns {
+	private static class R2dbcArrayColumns implements ArrayColumns {
 
-		private final SimpleTypeHolder simpleTypes;
+		private final ArrayColumns delegate;
+		private final SimpleTypeHolder simpleTypeHolder;
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#isSupported()
-		 */
 		@Override
 		public boolean isSupported() {
-			return true;
+			return this.delegate.isSupported();
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#getArrayType(java.lang.Class)
-		 */
 		@Override
 		public Class<?> getArrayType(Class<?> userType) {
 
-			Assert.notNull(userType, "Array component type must not be null");
-
-			if (!simpleTypes.isSimpleType(userType)) {
+			if (!simpleTypeHolder.isSimpleType(userType)) {
 				throw new IllegalArgumentException("Unsupported array type: " + ClassUtils.getQualifiedName(userType));
 			}
 
-			return ClassUtils.resolvePrimitiveIfNecessary(userType);
+			return this.delegate.getArrayType(userType);
 		}
 	}
+
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/Dialect.java b/src/main/java/org/springframework/data/r2dbc/dialect/R2dbcDialect.java
similarity index 67%
rename from src/main/java/org/springframework/data/r2dbc/dialect/Dialect.java
rename to src/main/java/org/springframework/data/r2dbc/dialect/R2dbcDialect.java
index 7edc6e8..510fb7f 100644
--- a/src/main/java/org/springframework/data/r2dbc/dialect/Dialect.java
+++ b/src/main/java/org/springframework/data/r2dbc/dialect/R2dbcDialect.java
@@ -6,16 +6,16 @@ import java.util.HashSet;
 import java.util.Set;
 
 import org.springframework.data.mapping.model.SimpleTypeHolder;
-import org.springframework.data.r2dbc.dialect.ArrayColumns.Unsupported;
 import org.springframework.data.r2dbc.mapping.R2dbcSimpleTypeHolder;
+import org.springframework.data.relational.core.dialect.Dialect;
 
 /**
- * Represents a dialect that is implemented by a particular database.
+ * R2DBC-specific extension to {@link Dialect}. Represents a dialect that is implemented by a particular database.
  *
  * @author Mark Paluch
  * @author Jens Schauder
  */
-public interface Dialect {
+public interface R2dbcDialect extends Dialect {
 
 	/**
 	 * Returns the {@link BindMarkersFactory} used by this dialect.
@@ -48,20 +48,4 @@ public interface Dialect {
 
 		return new SimpleTypeHolder(simpleTypes, true);
 	}
-
-	/**
-	 * Return the {@link LimitClause} used by this dialect.
-	 *
-	 * @return the {@link LimitClause} used by this dialect.
-	 */
-	LimitClause limit();
-
-	/**
-	 * Returns the array support object that describes how array-typed columns are supported by this dialect.
-	 *
-	 * @return the array support object that describes how array-typed columns are supported by this dialect.
-	 */
-	default ArrayColumns getArraySupport() {
-		return Unsupported.INSTANCE;
-	}
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java b/src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java
index d1b0a9a..33ea66c 100644
--- a/src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java
+++ b/src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java
@@ -11,7 +11,8 @@ import java.util.UUID;
  *
  * @author Mark Paluch
  */
-public class SqlServerDialect implements Dialect {
+public class SqlServerDialect extends org.springframework.data.relational.core.dialect.SqlServerDialect
+		implements R2dbcDialect {
 
 	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(Collections.singletonList(UUID.class));
 
@@ -23,36 +24,6 @@ public class SqlServerDialect implements Dialect {
 	private static final BindMarkersFactory NAMED = BindMarkersFactory.named("@", "P", 32,
 			SqlServerDialect::filterBindMarker);
 
-	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
-		 */
-		@Override
-		public String getClause(long limit) {
-			return "OFFSET 0 ROWS FETCH NEXT " + limit + " ROWS ONLY";
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
-		 */
-		@Override
-		public String getClause(long limit, long offset) {
-			return String.format("OFFSET %d ROWS FETCH NEXT %d ROWS ONLY", offset, limit);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
-		 */
-		@Override
-		public Position getClausePosition() {
-			return Position.END;
-		}
-	};
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
@@ -71,15 +42,6 @@ public class SqlServerDialect implements Dialect {
 		return SIMPLE_TYPES;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()
-	 */
-	@Override
-	public LimitClause limit() {
-		return LIMIT_CLAUSE;
-	}
-
 	private static String filterBindMarker(CharSequence input) {
 
 		StringBuilder builder = new StringBuilder();
diff --git a/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java b/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
deleted file mode 100644
index b0fb55e..0000000
--- a/src/main/java/org/springframework/data/r2dbc/support/StatementRenderUtil.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.support;
-
-import java.util.OptionalLong;
-
-import org.springframework.data.r2dbc.dialect.Dialect;
-import org.springframework.data.r2dbc.dialect.LimitClause;
-import org.springframework.data.relational.core.sql.Select;
-import org.springframework.data.relational.core.sql.render.SqlRenderer;
-
-/**
- * Utility class to assist with SQL rendering. Mainly for internal use within the framework.
- *
- * @author Mark Paluch
- */
-public abstract class StatementRenderUtil {
-
-	/**
-	 * Render {@link Select} to SQL considering {@link Dialect} specifics.
-	 *
-	 * @param select must not be {@literal null}.
-	 * @param limit must not be {@literal null}.
-	 * @param offset must not be {@literal null}.
-	 * @param dialect must not be {@literal null}.
-	 * @return the rendered SQL statement.
-	 */
-	public static String render(Select select, OptionalLong limit, OptionalLong offset, Dialect dialect) {
-
-		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();
-
-			String clause;
-			if (offset.isPresent()) {
-				clause = limitClause.getClause(limit.getAsLong(), offset.getAsLong());
-			} else {
-				clause = limitClause.getClause(limit.getAsLong());
-			}
-
-			return sql + " " + clause;
-		}
-
-		return sql;
-	}
-
-	private StatementRenderUtil() {}
-
-}
diff --git a/src/test/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategyTestSupport.java b/src/test/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategyTestSupport.java
index 8be51db..d9db558 100644
--- a/src/test/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategyTestSupport.java
+++ b/src/test/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategyTestSupport.java
@@ -36,11 +36,11 @@ import java.util.function.Function;
 
 import org.junit.Test;
 
-import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 import org.springframework.data.r2dbc.mapping.SettableValue;
 
 /**
- * Abstract base class for {@link Dialect}-aware {@link DefaultReactiveDataAccessStrategy} tests.
+ * Abstract base class for {@link R2dbcDialect}-aware {@link DefaultReactiveDataAccessStrategy} tests.
  *
  * @author Mark Paluch
  */
diff --git a/src/test/java/org/springframework/data/r2dbc/dialect/PostgresDialectUnitTests.java b/src/test/java/org/springframework/data/r2dbc/dialect/PostgresDialectUnitTests.java
index e1fe860..a079bfc 100644
--- a/src/test/java/org/springframework/data/r2dbc/dialect/PostgresDialectUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/dialect/PostgresDialectUnitTests.java
@@ -7,6 +7,7 @@ import java.util.List;
 
 import org.junit.Test;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
 
 /**
  * Unit tests for {@link PostgresDialect}.
diff --git a/src/test/java/org/springframework/data/r2dbc/dialect/SqlServerDialectUnitTests.java b/src/test/java/org/springframework/data/r2dbc/dialect/SqlServerDialectUnitTests.java
index cb39be6..502083e 100644
--- a/src/test/java/org/springframework/data/r2dbc/dialect/SqlServerDialectUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/dialect/SqlServerDialectUnitTests.java
@@ -6,6 +6,7 @@ import java.util.UUID;
 
 import org.junit.Test;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
 
 /**
  * Unit tests for {@link SqlServerDialect}.