Avoid creation of multiple connections during schema migration

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/BatchingConnection.java

@@ -11,7 +11,6 @@
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 
-
 import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
 
 import io.vertx.sqlclient.spi.DatabaseMetadata;
@@ -192,11 +191,6 @@ public CompletionStage<ResultSet> selectJdbc(String sql, Object[] paramValues) {
 				: delegate.selectJdbc( sql, paramValues );
 	}
 
-	@Override
-	public CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues) {
-		return delegate.selectJdbcOutsideTransaction( sql, paramValues );
-	}
-
 	public <T> CompletionStage<T> selectIdentifier(String sql, Object[] paramValues, Class<T> idClass) {
 		// Do not want to execute the batch here
 		// because we want to be able to select

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/ReactiveConnection.java

@@ -60,23 +60,6 @@ interface Expectation {
 
 	CompletionStage<ResultSet> selectJdbc(String sql, Object[] paramValues);
 
-	/**
-	 * This method is intended to be used only for queries returning
-	 * a ResultSet that must be executed outside of any "current"
-	 * transaction (i.e with autocommit=true).
-	 * <p/>
-	 * For example, it would be appropriate to use this method when
-	 * performing queries on information_schema or system tables in
-	 * order to obtain metadata information about catalogs, schemas,
-	 * tables, etc.
-	 *
-	 * @param sql - the query to execute outside of a transaction
-	 * @param paramValues - a non-null array of parameter values
-	 *
-	 * @return the CompletionStage<ResultSet> from executing the query.
-	 */
-	CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues);
-
 	<T> CompletionStage<T> insertAndSelectIdentifier(String sql, Object[] paramValues, Class<T> idClass, String idColumnName);
 	CompletionStage<ResultSet> insertAndSelectIdentifierAsResultSet(String sql, Object[] paramValues, Class<?> idClass, String idColumnName);
 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/ReactiveConnectionPool.java

@@ -10,8 +10,10 @@
 import org.hibernate.reactive.provider.ReactiveServiceRegistryBuilder;
 import org.hibernate.service.Service;
 
+import java.sql.ResultSet;
 import java.util.concurrent.CompletionStage;
 
+
 /**
  * A Hibernate {@link Service} that provides access to pooled
  * {@link ReactiveConnection reactive connections}.
@@ -63,6 +65,8 @@ public interface ReactiveConnectionPool extends Service {
 	 */
 	CompletionStage<ReactiveConnection> getConnection(String tenantId, SqlExceptionHelper sqlExceptionHelper);
 
+	CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues);
+
 	/**
 	 * The shutdown of the pool is actually asynchronous but the
 	 * core service registry won't return the {@link CompletionStage}.

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/ExternalSqlClientPool.java

@@ -5,15 +5,26 @@
  */
 package org.hibernate.reactive.pool.impl;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Objects;
 import java.util.concurrent.CompletionStage;
 
+import org.hibernate.engine.jdbc.internal.FormatStyle;
 import org.hibernate.engine.jdbc.spi.SqlExceptionHelper;
 import org.hibernate.engine.jdbc.spi.SqlStatementLogger;
+import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
 import org.hibernate.reactive.mutiny.Mutiny;
 import org.hibernate.reactive.stage.Stage;
 import org.hibernate.reactive.util.impl.CompletionStages;
 
+import io.vertx.sqlclient.DatabaseException;
 import io.vertx.sqlclient.Pool;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
+import io.vertx.sqlclient.Tuple;
+
+import static org.hibernate.reactive.util.impl.CompletionStages.rethrow;
 
 /**
  * A pool of reactive connections backed by a Vert.x {@link Pool}.
@@ -82,4 +93,41 @@ public SqlExceptionHelper getSqlExceptionHelper() {
 	public CompletionStage<Void> getCloseFuture() {
 		return CompletionStages.voidFuture();
 	}
+
+
+	@Override
+	public CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues) {
+		return preparedQueryOutsideTransaction( sql, Tuple.wrap( paramValues ) )
+				.thenApply( ResultSetAdaptor::new );
+	}
+
+	public CompletionStage<RowSet<Row>> preparedQueryOutsideTransaction(String sql, Tuple parameters) {
+		feedback( sql );
+		return getPool().preparedQuery( sql ).execute( parameters ).toCompletionStage()
+				.handle( (rows, throwable) -> convertException( rows, sql, throwable ) );
+	}
+
+	/**
+	 * Similar to {@link org.hibernate.exception.internal.SQLExceptionTypeDelegate#convert(SQLException, String, String)}
+	 */
+	private <T> T convertException(T rows, String sql, Throwable sqlException) {
+		if ( sqlException == null ) {
+			return rows;
+		}
+		if ( sqlException instanceof DatabaseException ) {
+			DatabaseException de = (DatabaseException) sqlException;
+			sqlException = sqlExceptionHelper
+					.convert( new SQLException( de.getMessage(), de.getSqlState(), de.getErrorCode() ), "error executing SQL statement", sql );
+		}
+		return rethrow( sqlException );
+	}
+
+	private void feedback(String sql) {
+		Objects.requireNonNull( sql, "SQL query cannot be null" );
+		// DDL already gets formatted by the client, so don't reformat it
+		FormatStyle formatStyle = sqlStatementLogger.isFormat() && !sql.contains( System.lineSeparator() )
+				? FormatStyle.BASIC
+				: FormatStyle.NONE;
+		sqlStatementLogger.logStatement( sql, formatStyle.getFormatter() );
+	}
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientConnection.java

@@ -128,12 +128,6 @@ public CompletionStage<ResultSet> selectJdbc(String sql, Object[] paramValues) {
 				.thenApply( ResultSetAdaptor::new );
 	}
 
-	@Override
-	public CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues) {
-		return preparedQueryOutsideTransaction( sql, Tuple.wrap( paramValues ) )
-				.thenApply( ResultSetAdaptor::new );
-	}
-
 	@Override
 	public CompletionStage<Void> execute(String sql) {
 		return preparedQuery( sql )
@@ -278,12 +272,6 @@ public CompletionStage<RowSet<Row>> preparedQueryOutsideTransaction(String sql)
 				.handle( (rows, throwable) -> convertException( rows, sql, throwable ) );
 	}
 
-	public CompletionStage<RowSet<Row>> preparedQueryOutsideTransaction(String sql, Tuple parameters) {
-		feedback( sql );
-		return pool.preparedQuery( sql ).execute( parameters ).toCompletionStage()
-				.handle( (rows, throwable) -> convertException( rows, sql, throwable ) );
-	}
-
 	private void feedback(String sql) {
 		Objects.requireNonNull( sql, "SQL query cannot be null" );
 		// DDL already gets formatted by the client, so don't reformat it

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientPool.java

@@ -5,18 +5,29 @@
  */
 package org.hibernate.reactive.pool.impl;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 
+import org.hibernate.engine.jdbc.internal.FormatStyle;
 import org.hibernate.engine.jdbc.spi.SqlExceptionHelper;
 import org.hibernate.engine.jdbc.spi.SqlStatementLogger;
+import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
 import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.pool.ReactiveConnectionPool;
 
 import io.vertx.core.Future;
+import io.vertx.sqlclient.DatabaseException;
 import io.vertx.sqlclient.Pool;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
 import io.vertx.sqlclient.SqlConnection;
+import io.vertx.sqlclient.Tuple;
+
+import static org.hibernate.reactive.util.impl.CompletionStages.rethrow;
 
 /**
  * A pool of reactive connections backed by a supplier of
@@ -99,6 +110,56 @@
 		);
 	}
 
+	/**
+	 * This method is intended to be used only for queries returning
+	 * a ResultSet that must be executed outside any "current"
+	 * transaction (i.e. with autocommit=true).
+	 * <p/>
+	 * For example, it would be appropriate to use this method when
+	 * performing queries on information_schema or system tables in
+	 * order to obtain metadata information about catalogs, schemas,
+	 * tables, etc.
+	 *
+	 * @param sql - the query to execute outside a transaction
+	 * @param paramValues - a non-null array of parameter values
+	 *
+	 * @return the CompletionStage<ResultSet> from executing the query.
+	 */
+	public CompletionStage<ResultSet> selectJdbcOutsideTransaction(String sql, Object[] paramValues) {
+		return preparedQueryOutsideTransaction( sql, Tuple.wrap( paramValues ) )
+				.thenApply( ResultSetAdaptor::new );
+	}
+
+	private CompletionStage<RowSet<Row>> preparedQueryOutsideTransaction(String sql, Tuple parameters) {
+		feedback( sql );
+		return getPool().preparedQuery( sql ).execute( parameters ).toCompletionStage()
+				.handle( (rows, throwable) -> convertException( rows, sql, throwable ) );
+	}
+
+	/**
+	 * Similar to {@link org.hibernate.exception.internal.SQLExceptionTypeDelegate#convert(SQLException, String, String)}
+	 */
+	private <T> T convertException(T rows, String sql, Throwable sqlException) {
+		if ( sqlException == null ) {
+			return rows;
+		}
+		if ( sqlException instanceof DatabaseException ) {
+			DatabaseException de = (DatabaseException) sqlException;
+			sqlException = getSqlExceptionHelper()
+					.convert( new SQLException( de.getMessage(), de.getSqlState(), de.getErrorCode() ), "error executing SQL statement", sql );
+		}
+		return rethrow( sqlException );
+	}
+
+	private void feedback(String sql) {
+		Objects.requireNonNull( sql, "SQL query cannot be null" );
+		// DDL already gets formatted by the client, so don't reformat it
+		FormatStyle formatStyle = getSqlStatementLogger().isFormat() && !sql.contains( System.lineSeparator() )
+				? FormatStyle.BASIC
+				: FormatStyle.NONE;
+		getSqlStatementLogger().logStatement( sql, formatStyle.getFormatter() );
+	}
+
 	/**
 	 * @param onCancellation invoke when converted {@link java.util.concurrent.CompletionStage} cancellation.
 	 */