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}.