Fix HQLUpdateQueryTest

This is done by following same pattern used in the other
*QueryPlan classes: make a reactive copy of the ORM
implementations

With this change HQLUpdateQueryTest now passes and no
new failures occur

Author: geoand

hibernate-reactive-core/src/main/java/org/hibernate/reactive/query/sqm/iternal/ReactiveSimpleInsertQueryPlan.java

@@ -6,12 +6,35 @@
 package org.hibernate.reactive.query.sqm.iternal;
 
 
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletionStage;
 
+import org.hibernate.action.internal.BulkOperationCleanupAction;
+import org.hibernate.engine.jdbc.spi.JdbcServices;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.MappingModelExpressible;
 import org.hibernate.query.spi.DomainQueryExecutionContext;
+import org.hibernate.query.spi.QueryEngine;
+import org.hibernate.query.spi.QueryParameterImplementor;
 import org.hibernate.query.sqm.internal.DomainParameterXref;
+import org.hibernate.query.sqm.internal.SqmJdbcExecutionContextAdapter;
+import org.hibernate.query.sqm.internal.SqmUtil;
+import org.hibernate.query.sqm.spi.SqmParameterMappingModelResolutionAccess;
+import org.hibernate.query.sqm.sql.SqmTranslation;
+import org.hibernate.query.sqm.sql.SqmTranslator;
+import org.hibernate.query.sqm.sql.SqmTranslatorFactory;
+import org.hibernate.query.sqm.tree.expression.SqmParameter;
 import org.hibernate.query.sqm.tree.insert.SqmInsertStatement;
 import org.hibernate.reactive.query.sql.spi.ReactiveNonSelectQueryPlan;
+import org.hibernate.reactive.sql.exec.internal.StandardReactiveJdbcMutationExecutor;
+import org.hibernate.sql.ast.SqlAstTranslator;
+import org.hibernate.sql.ast.spi.FromClauseAccess;
+import org.hibernate.sql.ast.tree.expression.JdbcParameter;
+import org.hibernate.sql.ast.tree.insert.InsertStatement;
+import org.hibernate.sql.exec.spi.JdbcOperationQueryInsert;
+import org.hibernate.sql.exec.spi.JdbcParameterBindings;
 
 /**
  * @see org.hibernate.query.sqm.internal.SimpleInsertQueryPlan
@@ -22,6 +45,12 @@ public class ReactiveSimpleInsertQueryPlan implements ReactiveNonSelectQueryPlan
 
 	private final DomainParameterXref domainParameterXref;
 
+	private Map<SqmParameter<?>, MappingModelExpressible<?>> paramTypeResolutions;
+
+	private JdbcOperationQueryInsert jdbcInsert;
+	private FromClauseAccess tableGroupAccess;
+	private Map<QueryParameterImplementor<?>, Map<SqmParameter<?>, List<List<JdbcParameter>>>> jdbcParamsXref;
+
 	public ReactiveSimpleInsertQueryPlan(
 			SqmInsertStatement<?> sqmInsert,
 			DomainParameterXref domainParameterXref) {
@@ -31,6 +60,83 @@ public ReactiveSimpleInsertQueryPlan(
 
 	@Override
 	public CompletionStage<Integer> executeReactiveUpdate(DomainQueryExecutionContext executionContext) {
-		throw new UnsupportedOperationException();
+		BulkOperationCleanupAction.schedule( executionContext.getSession(), sqmInsert );
+		final SharedSessionContractImplementor session = executionContext.getSession();
+		final SessionFactoryImplementor factory = session.getFactory();
+		final JdbcServices jdbcServices = factory.getJdbcServices();
+		SqlAstTranslator<JdbcOperationQueryInsert> insertTranslator = null;
+		if ( jdbcInsert == null ) {
+			insertTranslator = createInsertTranslator( executionContext );
+		}
+
+		final JdbcParameterBindings jdbcParameterBindings = SqmUtil.createJdbcParameterBindings(
+				executionContext.getQueryParameterBindings(),
+				domainParameterXref,
+				jdbcParamsXref,
+				factory.getRuntimeMetamodels().getMappingMetamodel(),
+				tableGroupAccess::findTableGroup,
+				new SqmParameterMappingModelResolutionAccess() {
+					@Override @SuppressWarnings("unchecked")
+					public <T> MappingModelExpressible<T> getResolvedMappingModelType(SqmParameter<T> parameter) {
+						return (MappingModelExpressible<T>) paramTypeResolutions.get(parameter);
+					}
+				},
+				session
+		);
+
+		if ( jdbcInsert != null && !jdbcInsert.isCompatibleWith(
+				jdbcParameterBindings,
+				executionContext.getQueryOptions()
+		) ) {
+			insertTranslator = createInsertTranslator( executionContext );
+		}
+
+		if ( insertTranslator != null ) {
+			jdbcInsert = insertTranslator.translate( jdbcParameterBindings, executionContext.getQueryOptions() );
+		}
+		else {
+			jdbcInsert.bindFilterJdbcParameters( jdbcParameterBindings );
+		}
+
+		return StandardReactiveJdbcMutationExecutor.INSTANCE
+				.executeReactive(
+						jdbcInsert,
+						jdbcParameterBindings,
+						session.getJdbcCoordinator().getStatementPreparer()::prepareStatement,
+						(i, ps) -> {},
+						SqmJdbcExecutionContextAdapter.omittingLockingAndPaging( executionContext )
+				);
+	}
+
+	//TODO: reuse from ORM
+	private SqlAstTranslator<JdbcOperationQueryInsert> createInsertTranslator(DomainQueryExecutionContext executionContext) {
+		final SessionFactoryImplementor factory = executionContext.getSession().getFactory();
+		final QueryEngine queryEngine = factory.getQueryEngine();
+
+		final SqmTranslatorFactory translatorFactory = queryEngine.getSqmTranslatorFactory();
+		final SqmTranslator<InsertStatement> translator = translatorFactory.createInsertTranslator(
+				sqmInsert,
+				executionContext.getQueryOptions(),
+				domainParameterXref,
+				executionContext.getQueryParameterBindings(),
+				executionContext.getSession().getLoadQueryInfluencers(),
+				factory
+		);
+
+		final SqmTranslation<InsertStatement> sqmInterpretation = translator.translate();
+
+		tableGroupAccess = sqmInterpretation.getFromClauseAccess();
+
+		this.jdbcParamsXref = SqmUtil.generateJdbcParamsXref(
+				domainParameterXref,
+				sqmInterpretation::getJdbcParamsBySqmParam
+		);
+
+		this.paramTypeResolutions = sqmInterpretation.getSqmParameterMappingModelTypeResolutions();
+
+		return factory.getJdbcServices()
+				.getJdbcEnvironment()
+				.getSqlAstTranslatorFactory()
+				.buildInsertTranslator( factory, sqmInterpretation.getSqlAst() );
 	}
 }