#2 - Introduce R2DBC exception translation.

We now provide exception translation for R2DBC exceptions based on Spring JDBC's SQLErrorCodes.

src/main/java/org/springframework/data/r2dbc/BadSqlGrammarException.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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;
+
+import io.r2dbc.spi.R2dbcException;
+
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+
+/**
+ * Exception thrown when SQL specified is invalid. Such exceptions always have a {@link io.r2dbc.spi.R2dbcException}
+ * root cause.
+ * <p>
+ * It would be possible to have subclasses for no such table, no such column etc. A custom
+ * {@link org.springframework.data.r2dbc.support.R2dbcExceptionTranslator} could create such more specific exceptions,
+ * without affecting code using this class.
+ *
+ * @author Mark Paluch
+ */
+public class BadSqlGrammarException extends InvalidDataAccessResourceUsageException {
+
+	private final String sql;
+
+	/**
+	 * Creates a new {@link BadSqlGrammarException}.
+	 *
+	 * @param task name of current task.
+	 * @param sql the offending SQL statement.
+	 * @param ex the root cause.
+	 */
+	public BadSqlGrammarException(String task, String sql, R2dbcException ex) {
+
+		super(task + "; bad SQL grammar [" + sql + "]", ex);
+
+		this.sql = sql;
+	}
+
+	/**
+	 * Return the wrapped {@link R2dbcException}.
+	 */
+	public R2dbcException getR2dbcException() {
+		return (R2dbcException) getCause();
+	}
+
+	/**
+	 * Return the SQL that caused the problem.
+	 */
+	public String getSql() {
+		return this.sql;
+	}
+}

src/main/java/org/springframework/data/r2dbc/InvalidResultAccessException.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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;
+
+import io.r2dbc.spi.R2dbcException;
+
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.jdbc.BadSqlGrammarException;
+import org.springframework.lang.Nullable;
+
+/**
+ * Exception thrown when a {@link io.r2dbc.spi.Result} has been accessed in an invalid fashion. Such exceptions always
+ * have a {@link io.r2dbc.spi.R2dbcException} root cause.
+ * <p>
+ * This typically happens when an invalid {@link Result} column index or name has been specified.
+ *
+ * @author Mark Paluch
+ * @see BadSqlGrammarException
+ */
+@SuppressWarnings("serial")
+public class InvalidResultAccessException extends InvalidDataAccessResourceUsageException {
+
+	private final @Nullable String sql;
+
+	/**
+	 * Creates a new {@link InvalidResultAccessException}.
+	 *
+	 * @param task name of current task.
+	 * @param sql the offending SQL statement.
+	 * @param ex the root cause.
+	 */
+	public InvalidResultAccessException(String task, String sql, R2dbcException ex) {
+
+		super(task + "; invalid Result access for SQL [" + sql + "]", ex);
+
+		this.sql = sql;
+	}
+
+	/**
+	 * Creates a new {@link InvalidResultAccessException}.
+	 *
+	 * @param ex the root cause.
+	 */
+	public InvalidResultAccessException(R2dbcException ex) {
+
+		super(ex.getMessage(), ex);
+
+		this.sql = null;
+	}
+
+	/**
+	 * Return the wrapped {@link R2dbcException}.
+	 */
+	public R2dbcException getR2dbcException() {
+		return (R2dbcException) getCause();
+	}
+
+	/**
+	 * Return the SQL that caused the problem.
+	 *
+	 * @return the offending SQL, if known.
+	 */
+	@Nullable
+	public String getSql() {
+		return this.sql;
+	}
+}

src/main/java/org/springframework/data/r2dbc/UncategorizedR2dbcException.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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;
+
+import io.r2dbc.spi.R2dbcException;
+
+import org.springframework.dao.UncategorizedDataAccessException;
+import org.springframework.lang.Nullable;
+
+/**
+ * Exception thrown when we can't classify a {@link R2dbcException} into one of our generic data access exceptions.
+ *
+ * @author Mark Paluch
+ */
+public class UncategorizedR2dbcException extends UncategorizedDataAccessException {
+
+	/**
+	 * SQL that led to the problem
+	 */
+	private final @Nullable String sql;
+
+	/**
+	 * Creates a new {@link UncategorizedR2dbcException}.
+	 *
+	 * @param task name of current task
+	 * @param sql the offending SQL statement
+	 * @param ex the root cause
+	 */
+	public UncategorizedR2dbcException(String task, @Nullable String sql, R2dbcException ex) {
+
+		super(String.format("%s; uncategorized R2dbcException%s; %s", task, sql != null ? " for SQL [" + sql + "]" : "",
+				ex.getMessage()), ex);
+		this.sql = sql;
+	}
+
+	/**
+	 * Returns the original {@link R2dbcException}.
+	 *
+	 * @return the original {@link R2dbcException}.
+	 */
+	public R2dbcException getR2dbcException() {
+		return (R2dbcException) getCause();
+	}
+
+	/**
+	 * Return the SQL that led to the problem (if known).
+	 */
+	@Nullable
+	public String getSql() {
+		return this.sql;
+	}
+}

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

@@ -28,7 +28,7 @@ import java.util.function.Supplier;
 import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
-import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 
 /**
  * A non-blocking, reactive client for performing database calls requests with Reactive Streams back pressure. Provides
@@ -91,12 +91,12 @@ public interface DatabaseClient {
 		Builder connectionFactory(ConnectionFactory factory);
 
 		/**
-		 * Configures a {@link SQLExceptionTranslator}.
+		 * Configures a {@link R2dbcExceptionTranslator}.
 		 *
 		 * @param exceptionTranslator must not be {@literal null}.
 		 * @return {@code this} {@link Builder}.
 		 */
-		Builder exceptionTranslator(SQLExceptionTranslator exceptionTranslator);
+		Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator);
 
 		/**
 		 * Configures a {@link ReactiveDataAccessStrategy}.

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

@@ -17,6 +17,7 @@ package org.springframework.data.r2dbc.function;
 
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
@@ -29,7 +30,6 @@ import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -48,16 +48,16 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
 import org.springframework.dao.DataAccessException;
-import org.springframework.dao.UncategorizedDataAccessException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.NullHandling;
 import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.r2dbc.UncategorizedR2dbcException;
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
+import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.data.util.Pair;
 import org.springframework.jdbc.core.SqlProvider;
-import org.springframework.jdbc.support.SQLExceptionTranslator;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -74,13 +74,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 	private final ConnectionFactory connector;
 
-	private final SQLExceptionTranslator exceptionTranslator;
+	private final R2dbcExceptionTranslator exceptionTranslator;
 
 	private final ReactiveDataAccessStrategy dataAccessStrategy;
 
 	private final DefaultDatabaseClientBuilder builder;
 
-	DefaultDatabaseClient(ConnectionFactory connector, SQLExceptionTranslator exceptionTranslator,
+	DefaultDatabaseClient(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,
 			ReactiveDataAccessStrategy dataAccessStrategy, DefaultDatabaseClientBuilder builder) {
 
 		this.connector = connector;
@@ -133,7 +133,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			return doInConnection(connectionToUse, action);
 		}, this::closeConnection, this::closeConnection, this::closeConnection) //
-				.onErrorMap(SQLException.class, ex -> translateException("execute", getSql(action), ex));
+				.onErrorMap(R2dbcException.class, ex -> translateException("execute", getSql(action), ex));
 	}
 
 	/**
@@ -160,7 +160,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			return doInConnectionMany(connectionToUse, action);
 		}, this::closeConnection, this::closeConnection, this::closeConnection) //
-				.onErrorMap(SQLException.class, ex -> translateException("executeMany", getSql(action), ex));
+				.onErrorMap(R2dbcException.class, ex -> translateException("executeMany", getSql(action), ex));
 	}
 
 	/**
@@ -185,7 +185,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	/**
 	 * Obtain the {@link ConnectionFactory} for actual use.
 	 *
-	 * @return the ConnectionFactory (never {@code null})
+	 * @return the ConnectionFactory (never {@literal null})
 	 * @throws IllegalStateException in case of no DataSource set
 	 */
 	protected ConnectionFactory obtainConnectionFactory() {
@@ -204,17 +204,17 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	}
 
 	/**
-	 * Translate the given {@link SQLException} into a generic {@link DataAccessException}.
+	 * Translate the given {@link R2dbcException} into a generic {@link DataAccessException}.
 	 *
-	 * @param task readable text describing the task being attempted
-	 * @param sql SQL query or update that caused the problem (may be {@code null})
-	 * @param ex the offending {@code SQLException}
-	 * @return a DataAccessException wrapping the {@code SQLException} (never {@code null})
+	 * @param task readable text describing the task being attempted.
+	 * @param sql SQL query or update that caused the problem (may be {@literal null}).
+	 * @param ex the offending {@link R2dbcException}.
+	 * @return a DataAccessException wrapping the {@link R2dbcException} (never {@literal null}).
 	 */
-	protected DataAccessException translateException(String task, @Nullable String sql, SQLException ex) {
+	protected DataAccessException translateException(String task, @Nullable String sql, R2dbcException ex) {
 
 		DataAccessException dae = exceptionTranslator.translate(task, sql, ex);
-		return (dae != null ? dae : new UncategorizedSQLException(task, sql, ex));
+		return (dae != null ? dae : new UncategorizedR2dbcException(task, sql, ex));
 	}
 
 	private static void doBind(Statement statement, Map<String, Optional<Object>> byName,
@@ -992,10 +992,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		try {
 			return action.apply(connection);
-		} catch (RuntimeException e) {
+		} catch (R2dbcException e) {
 
 			String sql = getSql(action);
-			return Flux.error(new DefaultDatabaseClient.UncategorizedSQLException("doInConnectionMany", sql, e) {});
+			return Flux.error(new UncategorizedR2dbcException("doInConnectionMany", sql, e) {});
 		}
 	}
 
@@ -1003,10 +1003,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		try {
 			return action.apply(connection);
-		} catch (RuntimeException e) {
+		} catch (R2dbcException e) {
 
 			String sql = getSql(action);
-			return Mono.error(new DefaultDatabaseClient.UncategorizedSQLException("doInConnection", sql, e) {});
+			return Mono.error(new UncategorizedR2dbcException("doInConnection", sql, e) {});
 		}
 	}
 
@@ -1014,7 +1014,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	 * Determine SQL from potential provider object.
 	 *
 	 * @param sqlProvider object that's potentially a SqlProvider
-	 * @return the SQL string, or {@code null}
+	 * @return the SQL string, or {@literal null}
 	 * @see SqlProvider
 	 */
 	@Nullable
@@ -1078,31 +1078,4 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			}
 		}
 	}
-
-	private static class UncategorizedSQLException extends UncategorizedDataAccessException implements SqlProvider {
-
-		/** SQL that led to the problem */
-		@Nullable private final String sql;
-
-		/**
-		 * Constructor for UncategorizedSQLException.
-		 *
-		 * @param task name of current task
-		 * @param sql the offending SQL statement
-		 * @param ex the root cause
-		 */
-		public UncategorizedSQLException(String task, @Nullable String sql, Exception ex) {
-			super(String.format("%s; uncategorized SQLException%s; %s", task, sql != null ? " for SQL [" + sql + "]" : "",
-					ex.getMessage()), ex);
-			this.sql = sql;
-		}
-
-		/**
-		 * Return the SQL that led to the problem (if known).
-		 */
-		@Nullable
-		public String getSql() {
-			return this.sql;
-		}
-	}
 }

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

@@ -21,8 +21,8 @@ import io.r2dbc.spi.ConnectionFactory;
 import java.util.function.Consumer;
 
 import org.springframework.data.r2dbc.function.DatabaseClient.Builder;
-import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
-import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
+import org.springframework.data.r2dbc.support.SqlErrorCodeR2dbcExceptionTranslator;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -34,7 +34,7 @@ import org.springframework.util.Assert;
 class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 
 	private @Nullable ConnectionFactory connector;
-	private SQLExceptionTranslator exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator();
+	private @Nullable R2dbcExceptionTranslator exceptionTranslator;
 	private ReactiveDataAccessStrategy accessStrategy = new DefaultReactiveDataAccessStrategy();
 
 	DefaultDatabaseClientBuilder() {}
@@ -57,9 +57,9 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 	}
 
 	@Override
-	public Builder exceptionTranslator(SQLExceptionTranslator exceptionTranslator) {
+	public Builder exceptionTranslator(R2dbcExceptionTranslator exceptionTranslator) {
 
-		Assert.notNull(exceptionTranslator, "SQLExceptionTranslator must not be null!");
+		Assert.notNull(exceptionTranslator, "R2dbcExceptionTranslator must not be null!");
 
 		this.exceptionTranslator = exceptionTranslator;
 		return this;
@@ -77,6 +77,12 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
 	@Override
 	public DatabaseClient build() {
 
+		R2dbcExceptionTranslator exceptionTranslator = this.exceptionTranslator;
+
+		if (exceptionTranslator == null) {
+			exceptionTranslator = new SqlErrorCodeR2dbcExceptionTranslator(connector);
+		}
+
 		return new DefaultDatabaseClient(this.connector, exceptionTranslator, accessStrategy,
 				new DefaultDatabaseClientBuilder(this));
 	}

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

@@ -25,6 +25,8 @@ import reactor.core.publisher.Mono;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
+import org.springframework.jdbc.core.SqlProvider;
+
 /**
  * Default {@link SqlResult} implementation.
  *
@@ -47,7 +49,27 @@ class DefaultSqlResult<T> implements SqlResult<T> {
 		this.updatedRowsFunction = updatedRowsFunction;
 
 		this.fetchSpec = new DefaultFetchSpec<>(connectionAccessor, sql,
-				it -> resultFunction.apply(it).flatMap(result -> result.map(mappingFunction)), updatedRowsFunction);
+				new SqlFunction<Connection, Flux<T>>() {
+					@Override
+					public Flux<T> apply(Connection connection) {
+						return resultFunction.apply(connection).flatMap(result -> result.map(mappingFunction));
+					}
+
+					@Override
+					public String getSql() {
+						return sql;
+					}
+				}, new SqlFunction<Connection, Mono<Integer>>() {
+					@Override
+					public Mono<Integer> apply(Connection connection) {
+						return updatedRowsFunction.apply(connection);
+					}
+
+					@Override
+					public String getSql() {
+						return sql;
+					}
+				});
 	}
 
 	/* (non-Javadoc)
@@ -89,4 +111,13 @@ class DefaultSqlResult<T> implements SqlResult<T> {
 	public Mono<Integer> rowsUpdated() {
 		return fetchSpec.rowsUpdated();
 	}
+
+	/**
+	 * Union type combining {@link Function} and {@link SqlProvider} to expose the SQL that is related to the underlying
+	 * action.
+	 *
+	 * @param <T> the type of the input to the function.
+	 * @param <R> the type of the result of the function.
+	 */
+	interface SqlFunction<T, R> extends Function<T, R>, SqlProvider {}
 }

src/main/java/org/springframework/data/r2dbc/support/AbstractFallbackR2dbcExceptionTranslator.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 io.r2dbc.spi.R2dbcException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.r2dbc.UncategorizedR2dbcException;
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for {@link R2dbcExceptionTranslator} implementations that allow for fallback to some other
+ * {@link R2dbcExceptionTranslator}.
+ *
+ * @author Mark Paluch
+ */
+public abstract class AbstractFallbackR2dbcExceptionTranslator implements R2dbcExceptionTranslator {
+
+	/** Logger available to subclasses */
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	@Nullable private R2dbcExceptionTranslator fallbackTranslator;
+
+	/**
+	 * Override the default SQL state fallback translator (typically a {@link R2dbcExceptionTranslator}).
+	 */
+	public void setFallbackTranslator(@Nullable R2dbcExceptionTranslator fallback) {
+		this.fallbackTranslator = fallback;
+	}
+
+	/**
+	 * Return the fallback exception translator, if any.
+	 */
+	@Nullable
+	public R2dbcExceptionTranslator getFallbackTranslator() {
+		return this.fallbackTranslator;
+	}
+
+	/**
+	 * Pre-checks the arguments, calls {@link #doTranslate}, and invokes the {@link #getFallbackTranslator() fallback
+	 * translator} if necessary.
+	 */
+	@Override
+	@NonNull
+	public DataAccessException translate(String task, @Nullable String sql, R2dbcException ex) {
+
+		Assert.notNull(ex, "Cannot translate a null R2dbcException");
+
+		DataAccessException dae = doTranslate(task, sql, ex);
+		if (dae != null) {
+			// Specific exception match found.
+			return dae;
+		}
+
+		// Looking for a fallback...
+		R2dbcExceptionTranslator fallback = getFallbackTranslator();
+		if (fallback != null) {
+			dae = fallback.translate(task, sql, ex);
+			if (dae != null) {
+				// Fallback exception match found.
+				return dae;
+			}
+		}
+
+		// We couldn't identify it more precisely.
+		return new UncategorizedR2dbcException(task, sql, ex);
+	}
+
+	/**
+	 * Template method for actually translating the given exception.
+	 * <p>
+	 * The passed-in arguments will have been pre-checked. Furthermore, this method is allowed to return {@literal null}
+	 * to indicate that no exception match has been found and that fallback translation should kick in.
+	 *
+	 * @param task readable text describing the task being attempted.
+	 * @param sql SQL query or update that caused the problem (if known).
+	 * @param ex the offending {@link R2dbcException}.
+	 * @return the DataAccessException, wrapping the {@link R2dbcException}; or {@literal null} if no exception match
+	 *         found.
+	 */
+	@Nullable
+	protected abstract DataAccessException doTranslate(String task, @Nullable String sql, R2dbcException ex);
+
+	/**
+	 * Build a message {@code String} for the given {@link java.sql.R2dbcException}.
+	 * <p>
+	 * To be called by translator subclasses when creating an instance of a generic
+	 * {@link org.springframework.dao.DataAccessException} class.
+	 *
+	 * @param task readable text describing the task being attempted.
+	 * @param sql the SQL statement that caused the problem.
+	 * @param ex the offending {@link R2dbcException}.
+	 * @return the message {@code String} to use.
+	 */
+	protected String buildMessage(String task, @Nullable String sql, R2dbcException ex) {
+		return task + "; " + (sql != null ? "SQL [" + sql : "]; " + "") + ex.getMessage();
+	}
+}

src/main/java/org/springframework/data/r2dbc/support/R2dbcExceptionTranslator.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 io.r2dbc.spi.R2dbcException;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.lang.Nullable;
+
+/**
+ * Strategy interface for translating between {@link io.r2dbc.spi.R2dbcException R2dbcExceptions} and Spring's data
+ * access strategy-agnostic {@link DataAccessException} hierarchy.
+ * <p>
+ * Implementations can be generic (for example, using {@link io.r2dbc.spi.R2dbcException#getSqlState() SQLState} codes
+ * for R2DBC) or wholly proprietary (for example, using Oracle error codes) for greater precision.
+ *
+ * @author Mark Paluch
+ * @see org.springframework.dao.DataAccessException
+ * @see SqlStateR2dbcExceptionTranslator
+ * @see SqlErrorCodeR2dbcExceptionTranslator
+ */
+@FunctionalInterface
+public interface R2dbcExceptionTranslator {
+
+	/**
+	 * Translate the given {@link R2dbcException} into a generic {@link DataAccessException}.
+	 * <p>
+	 * The returned DataAccessException is supposed to contain the original {@link R2dbcException} as root cause. However,
+	 * client code may not generally rely on this due to DataAccessExceptions possibly being caused by other resource APIs
+	 * as well. That said, a {@code getRootCause() instanceof R2dbcException} check (and subsequent cast) is considered
+	 * reliable when expecting R2DBC-based access to have happened.
+	 *
+	 * @param task readable text describing the task being attempted.
+	 * @param sql SQL query or update that caused the problem (if known).
+	 * @param ex the offending {@link R2dbcException}.
+	 * @return the DataAccessException wrapping the {@code R2dbcException}, or {@literal null} if no translation could be
+	 *         applied (in a custom translator; the default translators always throw an
+	 *         {@link org.springframework.data.r2dbc.UncategorizedR2dbcException} in such a case).
+	 * @see org.springframework.dao.DataAccessException#getRootCause()
+	 */
+	@Nullable
+	DataAccessException translate(String task, @Nullable String sql, R2dbcException ex);
+
+}

src/main/java/org/springframework/data/r2dbc/support/SqlErrorCodeR2dbcExceptionTranslator.java

@@ -0,0 +1,273 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.R2dbcException;
+
+import java.sql.SQLException;
+import java.util.Arrays;
+
+import org.springframework.dao.CannotAcquireLockException;
+import org.springframework.dao.CannotSerializeTransactionException;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DeadlockLoserDataAccessException;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.dao.PermissionDeniedDataAccessException;
+import org.springframework.dao.TransientDataAccessResourceException;
+import org.springframework.data.r2dbc.BadSqlGrammarException;
+import org.springframework.data.r2dbc.InvalidResultAccessException;
+import org.springframework.jdbc.support.SQLErrorCodes;
+import org.springframework.jdbc.support.SQLErrorCodesFactory;
+import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.lang.Nullable;
+
+/**
+ * Implementation of {@link R2dbcExceptionTranslator} that analyzes vendor-specific error codes. More precise than an
+ * implementation based on SQL state, but heavily vendor-specific.
+ * <p>
+ * This class applies the following matching rules:
+ * <ul>
+ * <li>Try custom translation implemented by any subclass. Note that this class is concrete and is typically used
+ * itself, in which case this rule doesn't apply.
+ * <li>Apply error code matching. Error codes are obtained from the SQLErrorCodesFactory by default. This factory loads
+ * a "sql-error-codes.xml" file from the class path, defining error code mappings for database names from database
+ * meta-data.
+ * <li>Fallback to a fallback translator. {@link SqlStateR2dbcExceptionTranslator} is the default fallback translator,
+ * analyzing the exception's SQL state only.
+ * </ul>
+ * <p>
+ * The configuration file named "sql-error-codes.xml" is by default read from the
+ * {@code org.springframework.jdbc.support} package. It can be overridden through a file of the same name in the root of
+ * the class path (e.g. in the "/WEB-INF/classes" directory), as long as the Spring JDBC package is loaded from the same
+ * ClassLoader.
+ *
+ * @author Mark Paluch
+ * @see SQLErrorCodesFactory
+ * @see SqlStateR2dbcExceptionTranslator
+ */
+public class SqlErrorCodeR2dbcExceptionTranslator extends AbstractFallbackR2dbcExceptionTranslator {
+
+	/** Error codes used by this translator */
+	@Nullable private SQLErrorCodes sqlErrorCodes;
+
+	/**
+	 * Creates a new {@link SqlErrorCodeR2dbcExceptionTranslator}. The {@link SQLErrorCodes} or
+	 * {@link io.r2dbc.spi.ConnectionFactory} property must be set.
+	 */
+	public SqlErrorCodeR2dbcExceptionTranslator() {}
+
+	/**
+	 * Create a SQL error code translator for the given DataSource. Invoking this constructor will cause a Connection to
+	 * be obtained from the DataSource to get the meta-data.
+	 *
+	 * @param connectionFactory {@link ConnectionFactory} to use to find meta-data and establish which error codes are
+	 *          usable.
+	 * @see SQLErrorCodesFactory
+	 */
+	public SqlErrorCodeR2dbcExceptionTranslator(ConnectionFactory connectionFactory) {
+		this();
+		setConnectionFactory(connectionFactory);
+	}
+
+	/**
+	 * Create a SQL error code translator for the given database product name. Invoking this constructor will avoid
+	 * obtaining a Connection from the DataSource to get the meta-data.
+	 *
+	 * @param dbName the database product name that identifies the error codes entry
+	 * @see SQLErrorCodesFactory
+	 * @see java.sql.DatabaseMetaData#getDatabaseProductName()
+	 */
+	public SqlErrorCodeR2dbcExceptionTranslator(String dbName) {
+		this();
+		setDatabaseProductName(dbName);
+	}
+
+	/**
+	 * Create a SQLErrorCode translator given these error codes. Does not require a database meta-data lookup to be
+	 * performed using a connection.
+	 *
+	 * @param sec error codes
+	 */
+	public SqlErrorCodeR2dbcExceptionTranslator(SQLErrorCodes sec) {
+		this();
+		this.sqlErrorCodes = sec;
+	}
+
+	/**
+	 * Set the DataSource for this translator.
+	 * <p>
+	 * Setting this property will cause a Connection to be obtained from the DataSource to get the meta-data.
+	 *
+	 * @param connectionFactory {@link ConnectionFactory} to use to find meta-data and establish which error codes are
+	 *          usable.
+	 * @see SQLErrorCodesFactory#getErrorCodes(String)
+	 * @see io.r2dbc.spi.ConnectionFactoryMetadata#getName()
+	 */
+	public void setConnectionFactory(ConnectionFactory connectionFactory) {
+		this.sqlErrorCodes = SQLErrorCodesFactory.getInstance().getErrorCodes(connectionFactory.getMetadata().getName());
+	}
+
+	/**
+	 * Set the database product name for this translator.
+	 * <p>
+	 * Setting this property will avoid obtaining a Connection from the DataSource to get the meta-data.
+	 *
+	 * @param dbName the database product name that identifies the error codes entry.
+	 * @see SQLErrorCodesFactory#getErrorCodes(String)
+	 * @see io.r2dbc.spi.ConnectionFactoryMetadata#getName()
+	 */
+	public void setDatabaseProductName(String dbName) {
+		this.sqlErrorCodes = SQLErrorCodesFactory.getInstance().getErrorCodes(dbName);
+	}
+
+	/**
+	 * Set custom error codes to be used for translation.
+	 *
+	 * @param sec custom error codes to use.
+	 */
+	public void setSqlErrorCodes(@Nullable SQLErrorCodes sec) {
+		this.sqlErrorCodes = sec;
+	}
+
+	/**
+	 * Return the error codes used by this translator. Usually determined via a DataSource.
+	 *
+	 * @see #setConnectionFactory
+	 */
+	@Nullable
+	public SQLErrorCodes getSqlErrorCodes() {
+		return this.sqlErrorCodes;
+	}
+
+	@Override
+	@Nullable
+	protected DataAccessException doTranslate(String task, @Nullable String sql, R2dbcException ex) {
+
+		R2dbcException translated = ex;
+
+		// First, try custom translation from overridden method.
+		DataAccessException dex = customTranslate(task, sql, translated);
+		if (dex != null) {
+			return dex;
+		}
+
+		// Next, try the custom SQLExceptionTranslator, if available.
+		if (this.sqlErrorCodes != null) {
+			SQLExceptionTranslator customTranslator = this.sqlErrorCodes.getCustomSqlExceptionTranslator();
+			if (customTranslator != null) {
+				DataAccessException customDex = customTranslator.translate(task, sql,
+						new SQLException(ex.getMessage(), ex.getSqlState(), ex));
+				if (customDex != null) {
+					return customDex;
+				}
+			}
+		}
+
+		// Check SQLErrorCodes with corresponding error code, if available.
+		if (this.sqlErrorCodes != null) {
+			String errorCode;
+			if (this.sqlErrorCodes.isUseSqlStateForTranslation()) {
+				errorCode = translated.getSqlState();
+			} else {
+				// Try to find R2dbcException with actual error code, looping through the causes.
+				R2dbcException current = translated;
+				while (current.getErrorCode() == 0 && current.getCause() instanceof R2dbcException) {
+					current = (R2dbcException) current.getCause();
+				}
+				errorCode = Integer.toString(current.getErrorCode());
+			}
+
+			if (errorCode != null) {
+				// Look for grouped error codes.
+				if (Arrays.binarySearch(this.sqlErrorCodes.getBadSqlGrammarCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new BadSqlGrammarException(task, (sql != null ? sql : ""), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getInvalidResultSetAccessCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new InvalidResultAccessException(task, (sql != null ? sql : ""), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getDuplicateKeyCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new DuplicateKeyException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getDataIntegrityViolationCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new DataIntegrityViolationException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getPermissionDeniedCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new PermissionDeniedDataAccessException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getDataAccessResourceFailureCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new DataAccessResourceFailureException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getTransientDataAccessResourceCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new TransientDataAccessResourceException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getCannotAcquireLockCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new CannotAcquireLockException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getDeadlockLoserCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new DeadlockLoserDataAccessException(buildMessage(task, sql, translated), translated);
+				} else if (Arrays.binarySearch(this.sqlErrorCodes.getCannotSerializeTransactionCodes(), errorCode) >= 0) {
+					logTranslation(task, sql, translated);
+					return new CannotSerializeTransactionException(buildMessage(task, sql, translated), translated);
+				}
+			}
+		}
+
+		// We couldn't identify it more precisely - let's hand it over to the SQLState fallback translator.
+		if (logger.isDebugEnabled()) {
+			String codes;
+			if (this.sqlErrorCodes != null && this.sqlErrorCodes.isUseSqlStateForTranslation()) {
+				codes = "SQL state '" + translated.getSqlState() + "', error code '" + translated.getErrorCode();
+			} else {
+				codes = "Error code '" + translated.getErrorCode() + "'";
+			}
+			logger.debug("Unable to translate R2dbcException with " + codes + ", will now try the fallback translator");
+		}
+
+		return null;
+	}
+
+	/**
+	 * Subclasses can override this method to attempt a custom mapping from {@link R2dbcException} to
+	 * {@link DataAccessException}.
+	 *
+	 * @param task readable text describing the task being attempted
+	 * @param sql SQL query or update that caused the problem. May be {@literal null}.
+	 * @param ex the offending {@link R2dbcException}.
+	 * @return null if no custom translation was possible, otherwise a {@link DataAccessException} resulting from custom
+	 *         translation. This exception should include the {@link R2dbcException} parameter as a nested root cause.
+	 *         This implementation always returns null, meaning that the translator always falls back to the default error
+	 *         codes.
+	 */
+	@Nullable
+	protected DataAccessException customTranslate(String task, @Nullable String sql, R2dbcException ex) {
+		return null;
+	}
+
+	private void logTranslation(String task, @Nullable String sql, R2dbcException exception) {
+
+		if (logger.isDebugEnabled()) {
+
+			String intro = "Translating";
+			logger.debug(intro + " R2dbcException with SQL state '" + exception.getSqlState() + "', error code '"
+					+ exception.getErrorCode() + "', message [" + exception.getMessage() + "]"
+					+ (sql != null ? "; SQL was [" + sql + "]" : "") + " for task [" + task + "]");
+		}
+	}
+}

src/main/java/org/springframework/data/r2dbc/support/SqlStateR2dbcExceptionTranslator.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 io.r2dbc.spi.R2dbcException;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.springframework.dao.ConcurrencyFailureException;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.TransientDataAccessResourceException;
+import org.springframework.data.r2dbc.BadSqlGrammarException;
+import org.springframework.lang.Nullable;
+
+/**
+ * {@link R2dbcExceptionTranslator} implementation that analyzes the SQL state in the {@link R2dbcException} based on
+ * the first two digits (the SQL state "class"). Detects standard SQL state values and well-known vendor-specific SQL
+ * states.
+ * <p>
+ * Not able to diagnose all problems, but is portable between databases and does not require special initialization (no
+ * database vendor detection, etc.). For more precise translation, consider
+ * {@link SqlErrorCodeR2dbcExceptionTranslator}.
+ *
+ * @author Mark Paluch
+ * @see io.r2dbc.spi.R2dbcException#getSqlState()
+ * @see SqlErrorCodeR2dbcExceptionTranslator
+ */
+public class SqlStateR2dbcExceptionTranslator extends AbstractFallbackR2dbcExceptionTranslator {
+
+	private static final Set<String> BAD_SQL_GRAMMAR_CODES = new HashSet<>(8);
+	private static final Set<String> DATA_INTEGRITY_VIOLATION_CODES = new HashSet<>(8);
+	private static final Set<String> DATA_ACCESS_RESOURCE_FAILURE_CODES = new HashSet<>(8);
+	private static final Set<String> TRANSIENT_DATA_ACCESS_RESOURCE_CODES = new HashSet<>(8);
+	private static final Set<String> CONCURRENCY_FAILURE_CODES = new HashSet<>(4);
+
+	static {
+		BAD_SQL_GRAMMAR_CODES.add("07"); // Dynamic SQL error
+		BAD_SQL_GRAMMAR_CODES.add("21"); // Cardinality violation
+		BAD_SQL_GRAMMAR_CODES.add("2A"); // Syntax error direct SQL
+		BAD_SQL_GRAMMAR_CODES.add("37"); // Syntax error dynamic SQL
+		BAD_SQL_GRAMMAR_CODES.add("42"); // General SQL syntax error
+		BAD_SQL_GRAMMAR_CODES.add("65"); // Oracle: unknown identifier
+
+		DATA_INTEGRITY_VIOLATION_CODES.add("01"); // Data truncation
+		DATA_INTEGRITY_VIOLATION_CODES.add("02"); // No data found
+		DATA_INTEGRITY_VIOLATION_CODES.add("22"); // Value out of range
+		DATA_INTEGRITY_VIOLATION_CODES.add("23"); // Integrity constraint violation
+		DATA_INTEGRITY_VIOLATION_CODES.add("27"); // Triggered data change violation
+		DATA_INTEGRITY_VIOLATION_CODES.add("44"); // With check violation
+
+		DATA_ACCESS_RESOURCE_FAILURE_CODES.add("08"); // Connection exception
+		DATA_ACCESS_RESOURCE_FAILURE_CODES.add("53"); // PostgreSQL: insufficient resources (e.g. disk full)
+		DATA_ACCESS_RESOURCE_FAILURE_CODES.add("54"); // PostgreSQL: program limit exceeded (e.g. statement too complex)
+		DATA_ACCESS_RESOURCE_FAILURE_CODES.add("57"); // DB2: out-of-memory exception / database not started
+		DATA_ACCESS_RESOURCE_FAILURE_CODES.add("58"); // DB2: unexpected system error
+
+		TRANSIENT_DATA_ACCESS_RESOURCE_CODES.add("JW"); // Sybase: internal I/O error
+		TRANSIENT_DATA_ACCESS_RESOURCE_CODES.add("JZ"); // Sybase: unexpected I/O error
+		TRANSIENT_DATA_ACCESS_RESOURCE_CODES.add("S1"); // DB2: communication failure
+
+		CONCURRENCY_FAILURE_CODES.add("40"); // Transaction rollback
+		CONCURRENCY_FAILURE_CODES.add("61"); // Oracle: deadlock
+	}
+
+	@Override
+	@Nullable
+	protected DataAccessException doTranslate(String task, @Nullable String sql, R2dbcException ex) {
+
+		// First, the getSQLState check...
+		String sqlState = getSqlState(ex);
+		if (sqlState != null && sqlState.length() >= 2) {
+
+			String classCode = sqlState.substring(0, 2);
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Extracted SQL state class '" + classCode + "' from value '" + sqlState + "'");
+			}
+
+			if (BAD_SQL_GRAMMAR_CODES.contains(classCode)) {
+				return new BadSqlGrammarException(task, (sql != null ? sql : ""), ex);
+			} else if (DATA_INTEGRITY_VIOLATION_CODES.contains(classCode)) {
+				return new DataIntegrityViolationException(buildMessage(task, sql, ex), ex);
+			} else if (DATA_ACCESS_RESOURCE_FAILURE_CODES.contains(classCode)) {
+				return new DataAccessResourceFailureException(buildMessage(task, sql, ex), ex);
+			} else if (TRANSIENT_DATA_ACCESS_RESOURCE_CODES.contains(classCode)) {
+				return new TransientDataAccessResourceException(buildMessage(task, sql, ex), ex);
+			} else if (CONCURRENCY_FAILURE_CODES.contains(classCode)) {
+				return new ConcurrencyFailureException(buildMessage(task, sql, ex), ex);
+			}
+		}
+
+		// Couldn't resolve anything proper - resort to UncategorizedR2dbcException.
+		return null;
+	}
+
+	/**
+	 * Gets the SQL state code from the supplied {@link R2dbcException exception}.
+	 * <p>
+	 * Some R2DBC drivers nest the actual exception from a batched update, so we might need to dig down into the nested
+	 * exception.
+	 *
+	 * @param ex the exception from which the {@link R2dbcException#getSqlState() SQL state} is to be extracted.
+	 * @return the SQL state code.
+	 */
+	@Nullable
+	private String getSqlState(R2dbcException ex) {
+
+		String sqlState = ex.getSqlState();
+
+		if (sqlState == null) {
+
+			for (Throwable throwable : ex.getSuppressed()) {
+
+				if (!(throwable instanceof R2dbcException)) {
+					continue;
+				}
+
+				R2dbcException r2dbcException = (R2dbcException) throwable;
+				if (r2dbcException.getSqlState() != null) {
+					sqlState = r2dbcException.getSqlState();
+					break;
+				}
+			}
+		}
+
+		return sqlState;
+	}
+}