From 6ebcfaf67eac5e282c68f642a7434f750ecae8d1 Mon Sep 17 00:00:00 2001 From: terrymanu Date: Fri, 31 May 2024 00:15:26 +0800 Subject: [PATCH 1/2] Move executeUpdate to DriverExecutor --- .../driver/executor/DriverExecutor.java | 101 +++++++++++++++++- .../ShardingSpherePreparedStatement.java | 57 ++-------- .../statement/ShardingSphereStatement.java | 54 ++-------- 3 files changed, 110 insertions(+), 102 deletions(-) diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java index 10878e41d365a..b6f2f39a794af 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java @@ -19,6 +19,7 @@ import lombok.Getter; import org.apache.shardingsphere.driver.executor.callback.ExecuteQueryCallback; +import org.apache.shardingsphere.driver.executor.callback.ExecuteUpdateCallback; import org.apache.shardingsphere.driver.executor.callback.impl.PreparedStatementExecuteQueryCallback; import org.apache.shardingsphere.driver.executor.callback.impl.StatementExecuteQueryCallback; import org.apache.shardingsphere.driver.jdbc.core.connection.ShardingSphereConnection; @@ -29,20 +30,26 @@ import org.apache.shardingsphere.infra.binder.context.statement.dml.SelectStatementContext; import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey; import org.apache.shardingsphere.infra.connection.kernel.KernelProcessor; +import org.apache.shardingsphere.infra.database.core.type.DatabaseType; import org.apache.shardingsphere.infra.database.core.type.DatabaseTypeRegistry; +import org.apache.shardingsphere.infra.exception.dialect.SQLExceptionTransformEngine; import org.apache.shardingsphere.infra.executor.audit.SQLAuditEngine; import org.apache.shardingsphere.infra.executor.kernel.ExecutorEngine; import org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroup; import org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroupContext; import org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroupReportContext; import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext; +import org.apache.shardingsphere.infra.executor.sql.execute.engine.ConnectionMode; import org.apache.shardingsphere.infra.executor.sql.execute.engine.SQLExecutorExceptionHandler; import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutionUnit; import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutor; +import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutorCallback; import org.apache.shardingsphere.infra.executor.sql.execute.engine.raw.RawExecutor; import org.apache.shardingsphere.infra.executor.sql.execute.engine.raw.RawSQLExecutionUnit; import org.apache.shardingsphere.infra.executor.sql.execute.engine.raw.callback.RawSQLExecutorCallback; +import org.apache.shardingsphere.infra.executor.sql.execute.result.ExecuteResult; import org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryResult; +import org.apache.shardingsphere.infra.executor.sql.execute.result.update.UpdateResult; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.JDBCDriverType; import org.apache.shardingsphere.infra.executor.sql.prepare.raw.RawExecutionPrepareEngine; @@ -55,11 +62,13 @@ import org.apache.shardingsphere.infra.rule.attribute.raw.RawExecutionRuleAttribute; import org.apache.shardingsphere.infra.session.query.QueryContext; import org.apache.shardingsphere.mode.metadata.MetaDataContexts; +import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement; import org.apache.shardingsphere.sqlfederation.engine.SQLFederationEngine; import org.apache.shardingsphere.sqlfederation.executor.context.SQLFederationContext; import org.apache.shardingsphere.traffic.executor.TrafficExecutor; import org.apache.shardingsphere.traffic.executor.TrafficExecutorCallback; import org.apache.shardingsphere.traffic.rule.TrafficRule; +import org.apache.shardingsphere.transaction.implicit.ImplicitTransactionCallback; import java.sql.Connection; import java.sql.PreparedStatement; @@ -249,17 +258,99 @@ private List getResultSets() throws SQLException { * @param queryContext query context * @param prepareEngine prepare engine * @param trafficCallback traffic callback + * @param updateCallback update callback + * @param isNeedImplicitCommitTransaction is need implicit commit transaction + * @param statementReplayCallback statement replay callback * @return updated row count * @throws SQLException SQL exception */ - public Optional executeAdvanceUpdate(final ShardingSphereMetaData metaData, final ShardingSphereDatabase database, final QueryContext queryContext, - final DriverExecutionPrepareEngine prepareEngine, - final TrafficExecutorCallback trafficCallback) throws SQLException { + @SuppressWarnings("rawtypes") + public int executeAdvanceUpdate(final ShardingSphereMetaData metaData, final ShardingSphereDatabase database, final QueryContext queryContext, + final DriverExecutionPrepareEngine prepareEngine, final TrafficExecutorCallback trafficCallback, + final ExecuteUpdateCallback updateCallback, final boolean isNeedImplicitCommitTransaction, + final StatementReplayCallback statementReplayCallback) throws SQLException { Optional trafficInstanceId = connection.getTrafficInstanceId(metaData.getGlobalRuleMetaData().getSingleRule(TrafficRule.class), queryContext); if (trafficInstanceId.isPresent()) { - return Optional.of(trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, trafficCallback)); + return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, trafficCallback); } - return Optional.empty(); + ExecutionContext executionContext = createExecutionContext(metaData, database, queryContext); + return database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty() + ? executeUpdate(database, updateCallback, queryContext.getSqlStatementContext(), executionContext, prepareEngine, isNeedImplicitCommitTransaction, statementReplayCallback) + : accumulate(rawExecutor.execute(createRawExecutionGroupContext(metaData, database, executionContext), queryContext, new RawSQLExecutorCallback())); + } + + @SuppressWarnings("rawtypes") + private int executeUpdate(final ShardingSphereDatabase database, final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final ExecutionContext executionContext, + final DriverExecutionPrepareEngine prepareEngine, final boolean isNeedImplicitCommitTransaction, + final StatementReplayCallback statementReplayCallback) throws SQLException { + return isNeedImplicitCommitTransaction + ? executeWithImplicitCommitTransaction(() -> useDriverToExecuteUpdate( + database, updateCallback, sqlStatementContext, executionContext, prepareEngine, statementReplayCallback), connection, database.getProtocolType()) + : useDriverToExecuteUpdate(database, updateCallback, sqlStatementContext, executionContext, prepareEngine, statementReplayCallback); + } + + private T executeWithImplicitCommitTransaction(final ImplicitTransactionCallback callback, final Connection connection, final DatabaseType databaseType) throws SQLException { + T result; + try { + connection.setAutoCommit(false); + result = callback.execute(); + connection.commit(); + // CHECKSTYLE:OFF + } catch (final Exception ex) { + // CHECKSTYLE:ON + connection.rollback(); + throw SQLExceptionTransformEngine.toSQLException(ex, databaseType); + } finally { + connection.setAutoCommit(true); + } + return result; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private int useDriverToExecuteUpdate(final ShardingSphereDatabase database, final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, + final ExecutionContext executionContext, final DriverExecutionPrepareEngine prepareEngine, + final StatementReplayCallback statementReplayCallback) throws SQLException { + ExecutionGroupContext executionGroupContext = createExecutionGroupContext(database, executionContext, prepareEngine); + for (ExecutionGroup each : executionGroupContext.getInputGroups()) { + statements.addAll(getStatements(each)); + if (JDBCDriverType.PREPARED_STATEMENT.equals(prepareEngine.getType())) { + parameterSets.addAll(getParameterSets(each)); + } + } + statementReplayCallback.replay(statements, parameterSets); + JDBCExecutorCallback callback = createExecuteUpdateCallback(database, updateCallback, sqlStatementContext, prepareEngine.getType()); + return regularExecutor.executeUpdate(executionGroupContext, executionContext.getQueryContext(), executionContext.getRouteContext().getRouteUnits(), callback); + } + + private ExecutionGroupContext createExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext, + final DriverExecutionPrepareEngine prepareEngine) throws SQLException { + return prepareEngine.prepare(executionContext.getRouteContext(), executionContext.getExecutionUnits(), + new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", ""))); + } + + private JDBCExecutorCallback createExecuteUpdateCallback(final ShardingSphereDatabase database, + final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final String jdbcDriverType) { + boolean isExceptionThrown = SQLExecutorExceptionHandler.isExceptionThrown(); + return new JDBCExecutorCallback(database.getProtocolType(), database.getResourceMetaData(), sqlStatementContext.getSqlStatement(), isExceptionThrown) { + + @Override + protected Integer executeSQL(final String sql, final Statement statement, final ConnectionMode connectionMode, final DatabaseType storageType) throws SQLException { + return JDBCDriverType.STATEMENT.equals(jdbcDriverType) ? updateCallback.executeUpdate(sql, statement) : ((PreparedStatement) statement).executeUpdate(); + } + + @Override + protected Optional getSaneResult(final SQLStatement sqlStatement, final SQLException ex) { + return Optional.empty(); + } + }; + } + + private int accumulate(final Collection results) { + int result = 0; + for (ExecuteResult each : results) { + result += ((UpdateResult) each).getUpdateCount(); + } + return result; } /** diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java index f029ee03fd19c..67f5ab7e98deb 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java @@ -59,7 +59,6 @@ import org.apache.shardingsphere.infra.executor.sql.execute.result.ExecuteResult; import org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryResult; import org.apache.shardingsphere.infra.executor.sql.execute.result.query.impl.driver.jdbc.type.stream.JDBCStreamQueryResult; -import org.apache.shardingsphere.infra.executor.sql.execute.result.update.UpdateResult; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.JDBCDriverType; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption; @@ -277,18 +276,16 @@ public int executeUpdate() throws SQLException { QueryContext queryContext = createQueryContext(); handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement()); ShardingSphereDatabase database = metaDataContexts.getMetaData().getDatabase(databaseName); - Optional updatedCount = executor.executeAdvanceUpdate(metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), - (statement, sql) -> ((PreparedStatement) statement).executeUpdate()); - if (updatedCount.isPresent()) { - return updatedCount.get(); - } ExecutionContext executionContext = createExecutionContext(queryContext); - if (hasRawExecutionRule()) { - Collection results = - executor.getRawExecutor().execute(createRawExecutionGroupContext(executionContext), executionContext.getQueryContext(), new RawSQLExecutorCallback()); - return accumulate(results); + boolean isNeedImplicitCommitTransaction = isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1); + final int result = executor.executeAdvanceUpdate(metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), + (statement, sql) -> ((PreparedStatement) statement).executeUpdate(), null, isNeedImplicitCommitTransaction, (StatementReplayCallback) this::replay); + for (Statement each : executor.getStatements()) { + statements.add((PreparedStatement) each); } - return executeUpdateWithExecutionContext(executionContext); + parameterSets.addAll(executor.getParameterSets()); + findGeneratedKey().ifPresent(optional -> generatedValues.addAll(optional.getGeneratedValues())); + return result; // CHECKSTYLE:OFF } catch (final RuntimeException ex) { // CHECKSTYLE:ON @@ -299,38 +296,6 @@ public int executeUpdate() throws SQLException { } } - private int useDriverToExecuteUpdate(final ExecutionContext executionContext) throws SQLException { - ExecutionGroupContext executionGroupContext = createExecutionGroupContext(executionContext); - cacheStatements(executionGroupContext.getInputGroups()); - return executor.getRegularExecutor().executeUpdate(executionGroupContext, - executionContext.getQueryContext(), executionContext.getRouteContext().getRouteUnits(), createExecuteUpdateCallback()); - } - - private int accumulate(final Collection results) { - int result = 0; - for (ExecuteResult each : results) { - result += ((UpdateResult) each).getUpdateCount(); - } - return result; - } - - private JDBCExecutorCallback createExecuteUpdateCallback() { - boolean isExceptionThrown = SQLExecutorExceptionHandler.isExceptionThrown(); - return new JDBCExecutorCallback(metaDataContexts.getMetaData().getDatabase(databaseName).getProtocolType(), - metaDataContexts.getMetaData().getDatabase(databaseName).getResourceMetaData(), sqlStatement, isExceptionThrown) { - - @Override - protected Integer executeSQL(final String sql, final Statement statement, final ConnectionMode connectionMode, final DatabaseType storageType) throws SQLException { - return ((PreparedStatement) statement).executeUpdate(); - } - - @Override - protected Optional getSaneResult(final SQLStatement sqlStatement, final SQLException ex) { - return Optional.empty(); - } - }; - } - @Override public boolean execute() throws SQLException { try { @@ -370,12 +335,6 @@ private boolean executeWithExecutionContext(final ExecutionContext executionCont : useDriverToExecute(executionContext); } - private int executeUpdateWithExecutionContext(final ExecutionContext executionContext) throws SQLException { - return isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1) - ? executeWithImplicitCommitTransaction(() -> useDriverToExecuteUpdate(executionContext), connection, metaDataContexts.getMetaData().getDatabase(databaseName).getProtocolType()) - : useDriverToExecuteUpdate(executionContext); - } - private boolean useDriverToExecute(final ExecutionContext executionContext) throws SQLException { ExecutionGroupContext executionGroupContext = createExecutionGroupContext(executionContext); cacheStatements(executionGroupContext.getInputGroups()); diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java index 1065c06855788..7ab7a2792bb2b 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java @@ -54,7 +54,6 @@ import org.apache.shardingsphere.infra.executor.sql.execute.result.ExecuteResult; import org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryResult; import org.apache.shardingsphere.infra.executor.sql.execute.result.query.impl.driver.jdbc.type.stream.JDBCStreamQueryResult; -import org.apache.shardingsphere.infra.executor.sql.execute.result.update.UpdateResult; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.JDBCDriverType; import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption; @@ -232,13 +231,6 @@ public int executeUpdate(final String sql, final String[] columnNames) throws SQ } } - private int executeUpdate(final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final ExecutionContext executionContext) throws SQLException { - return isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1) - ? executeWithImplicitCommitTransaction(() -> useDriverToExecuteUpdate(updateCallback, sqlStatementContext, executionContext), connection, - metaDataContexts.getMetaData().getDatabase(databaseName).getProtocolType()) - : useDriverToExecuteUpdate(updateCallback, sqlStatementContext, executionContext); - } - private int executeUpdate0(final String sql, final ExecuteUpdateCallback updateCallback, final TrafficExecutorCallback trafficCallback) throws SQLException { QueryContext queryContext = createQueryContext(sql); handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement()); @@ -246,47 +238,13 @@ private int executeUpdate0(final String sql, final ExecuteUpdateCallback updateC connection.getDatabaseConnectionManager().getConnectionContext().setCurrentDatabase(databaseName); ShardingSphereDatabase database = metaDataContexts.getMetaData().getDatabase(databaseName); sqlStatementContext = queryContext.getSqlStatementContext(); - Optional updatedCount = executor.executeAdvanceUpdate(metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), trafficCallback); - if (updatedCount.isPresent()) { - return updatedCount.get(); - } ExecutionContext executionContext = createExecutionContext(queryContext); - if (!metaDataContexts.getMetaData().getDatabase(databaseName).getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty()) { - Collection results = executor.getRawExecutor().execute(createRawExecutionGroupContext(executionContext), queryContext, new RawSQLExecutorCallback()); - return accumulate(results); - } - return executeUpdate(updateCallback, queryContext.getSqlStatementContext(), executionContext); - } - - private int useDriverToExecuteUpdate(final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final ExecutionContext executionContext) throws SQLException { - ExecutionGroupContext executionGroupContext = createExecutionGroupContext(executionContext); - cacheStatements(executionGroupContext.getInputGroups()); - JDBCExecutorCallback callback = createExecuteUpdateCallback(updateCallback, sqlStatementContext); - return executor.getRegularExecutor().executeUpdate(executionGroupContext, executionContext.getQueryContext(), executionContext.getRouteContext().getRouteUnits(), callback); - } - - private JDBCExecutorCallback createExecuteUpdateCallback(final ExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext) { - boolean isExceptionThrown = SQLExecutorExceptionHandler.isExceptionThrown(); - return new JDBCExecutorCallback(metaDataContexts.getMetaData().getDatabase(databaseName).getProtocolType(), - metaDataContexts.getMetaData().getDatabase(databaseName).getResourceMetaData(), sqlStatementContext.getSqlStatement(), isExceptionThrown) { - - @Override - protected Integer executeSQL(final String sql, final Statement statement, final ConnectionMode connectionMode, final DatabaseType storageType) throws SQLException { - return updateCallback.executeUpdate(sql, statement); - } - - @Override - protected Optional getSaneResult(final SQLStatement sqlStatement, final SQLException ex) { - return Optional.empty(); - } - }; - } - - private int accumulate(final Collection results) { - int result = 0; - for (ExecuteResult each : results) { - result += ((UpdateResult) each).getUpdateCount(); - } + boolean isNeedImplicitCommitTransaction = isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1); + int result = executor.executeAdvanceUpdate( + metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), trafficCallback, updateCallback, isNeedImplicitCommitTransaction, + (StatementReplayCallback) (statements, parameterSets) -> replay(statements)); + statements.addAll(executor.getStatements()); + replay(statements); return result; } From 367b0797bf1251bea3b83137e428af7487dc91f8 Mon Sep 17 00:00:00 2001 From: terrymanu Date: Fri, 31 May 2024 00:55:36 +0800 Subject: [PATCH 2/2] Move executeUpdate to DriverExecutor --- .../apache/shardingsphere/driver/executor/DriverExecutor.java | 4 ++-- .../jdbc/core/statement/ShardingSpherePreparedStatement.java | 4 ++-- .../driver/jdbc/core/statement/ShardingSphereStatement.java | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java index b6f2f39a794af..0b464cd1da704 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/DriverExecutor.java @@ -261,6 +261,7 @@ private List getResultSets() throws SQLException { * @param updateCallback update callback * @param isNeedImplicitCommitTransaction is need implicit commit transaction * @param statementReplayCallback statement replay callback + * @param executionContext execution context * @return updated row count * @throws SQLException SQL exception */ @@ -268,12 +269,11 @@ private List getResultSets() throws SQLException { public int executeAdvanceUpdate(final ShardingSphereMetaData metaData, final ShardingSphereDatabase database, final QueryContext queryContext, final DriverExecutionPrepareEngine prepareEngine, final TrafficExecutorCallback trafficCallback, final ExecuteUpdateCallback updateCallback, final boolean isNeedImplicitCommitTransaction, - final StatementReplayCallback statementReplayCallback) throws SQLException { + final StatementReplayCallback statementReplayCallback, final ExecutionContext executionContext) throws SQLException { Optional trafficInstanceId = connection.getTrafficInstanceId(metaData.getGlobalRuleMetaData().getSingleRule(TrafficRule.class), queryContext); if (trafficInstanceId.isPresent()) { return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, trafficCallback); } - ExecutionContext executionContext = createExecutionContext(metaData, database, queryContext); return database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty() ? executeUpdate(database, updateCallback, queryContext.getSqlStatementContext(), executionContext, prepareEngine, isNeedImplicitCommitTransaction, statementReplayCallback) : accumulate(rawExecutor.execute(createRawExecutionGroupContext(metaData, database, executionContext), queryContext, new RawSQLExecutorCallback())); diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java index 67f5ab7e98deb..f185ba74bc9c1 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java @@ -279,12 +279,12 @@ public int executeUpdate() throws SQLException { ExecutionContext executionContext = createExecutionContext(queryContext); boolean isNeedImplicitCommitTransaction = isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1); final int result = executor.executeAdvanceUpdate(metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), - (statement, sql) -> ((PreparedStatement) statement).executeUpdate(), null, isNeedImplicitCommitTransaction, (StatementReplayCallback) this::replay); + (statement, sql) -> ((PreparedStatement) statement).executeUpdate(), null, isNeedImplicitCommitTransaction, (StatementReplayCallback) this::replay, + executionContext); for (Statement each : executor.getStatements()) { statements.add((PreparedStatement) each); } parameterSets.addAll(executor.getParameterSets()); - findGeneratedKey().ifPresent(optional -> generatedValues.addAll(optional.getGeneratedValues())); return result; // CHECKSTYLE:OFF } catch (final RuntimeException ex) { diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java index 7ab7a2792bb2b..3486275a6efad 100644 --- a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java +++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java @@ -242,7 +242,7 @@ private int executeUpdate0(final String sql, final ExecuteUpdateCallback updateC boolean isNeedImplicitCommitTransaction = isNeedImplicitCommitTransaction(connection, sqlStatementContext.getSqlStatement(), executionContext.getExecutionUnits().size() > 1); int result = executor.executeAdvanceUpdate( metaDataContexts.getMetaData(), database, queryContext, createDriverExecutionPrepareEngine(database), trafficCallback, updateCallback, isNeedImplicitCommitTransaction, - (StatementReplayCallback) (statements, parameterSets) -> replay(statements)); + (StatementReplayCallback) (statements, parameterSets) -> replay(statements), executionContext); statements.addAll(executor.getStatements()); replay(statements); return result;