You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2024/06/10 07:54:58 UTC
(shardingsphere) branch master updated: Add DriverPushDownExecuteUpdateExecutor (#31645)
This is an automated email from the ASF dual-hosted git repository.
zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 93ae1d8c742 Add DriverPushDownExecuteUpdateExecutor (#31645)
93ae1d8c742 is described below
commit 93ae1d8c74280a95d94b26b6ae9d304ae021bf98
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Mon Jun 10 15:54:52 2024 +0800
Add DriverPushDownExecuteUpdateExecutor (#31645)
* Add DriverPushDownExecuteUpdateExecutor
* Add DriverPushDownExecuteUpdateExecutor
* Add DriverPushDownExecuteUpdateExecutor
---
.../executor/engine/DriverExecuteExecutor.java | 26 +--
.../engine/DriverExecuteUpdateExecutor.java | 208 ++-------------------
.../DriverPushDownExecuteUpdateExecutor.java} | 207 +++++++++-----------
.../transaction/rule/TransactionRule.java | 27 +++
4 files changed, 126 insertions(+), 342 deletions(-)
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteExecutor.java
index 3d666aea4a9..3c8349075a2 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteExecutor.java
@@ -55,13 +55,10 @@ import org.apache.shardingsphere.infra.rule.attribute.raw.RawExecutionRuleAttrib
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.mode.metadata.refresher.MetaDataRefreshEngine;
import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.DMLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.SelectStatement;
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.rule.TrafficRule;
-import org.apache.shardingsphere.transaction.api.TransactionType;
import org.apache.shardingsphere.transaction.implicit.ImplicitTransactionCallback;
import org.apache.shardingsphere.transaction.rule.TransactionRule;
@@ -128,9 +125,10 @@ public final class DriverExecuteExecutor {
Collection<ExecuteResult> results = rawExecutor.execute(createRawExecutionGroupContext(database, executionContext), queryContext, new RawSQLExecutorCallback());
return results.iterator().next() instanceof QueryResult;
}
- boolean isNeedImplicitCommitTransaction = isNeedImplicitCommitTransaction(
- connection, queryContext.getSqlStatementContext().getSqlStatement(), executionContext.getExecutionUnits().size() > 1);
- return executeWithExecutionContext(database, executeCallback, executionContext, prepareEngine, isNeedImplicitCommitTransaction, addCallback, replayCallback);
+ boolean isImplicitCommitTransaction = metaData.getGlobalRuleMetaData().getSingleRule(TransactionRule.class).isImplicitCommitTransaction(
+ executionContext.getSqlStatementContext().getSqlStatement(),
+ connection.getDatabaseConnectionManager().getConnectionTransaction(), connection.getAutoCommit(), executionContext.getExecutionUnits().size() > 1);
+ return executeWithExecutionContext(database, executeCallback, executionContext, prepareEngine, isImplicitCommitTransaction, addCallback, replayCallback);
}
private ExecuteQueryCallback getExecuteQueryCallback(final ShardingSphereDatabase database, final QueryContext queryContext, final String jdbcDriverType) {
@@ -250,22 +248,6 @@ public final class DriverExecuteExecutor {
};
}
- private boolean isNeedImplicitCommitTransaction(final ShardingSphereConnection connection, final SQLStatement sqlStatement, final boolean multiExecutionUnits) {
- if (!connection.getAutoCommit()) {
- return false;
- }
- TransactionType transactionType = connection.getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(TransactionRule.class).getDefaultType();
- boolean isInTransaction = connection.getDatabaseConnectionManager().getConnectionTransaction().isInTransaction();
- if (!TransactionType.isDistributedTransaction(transactionType) || isInTransaction) {
- return false;
- }
- return isWriteDMLStatement(sqlStatement) && multiExecutionUnits;
- }
-
- private boolean isWriteDMLStatement(final SQLStatement sqlStatement) {
- return sqlStatement instanceof DMLStatement && !(sqlStatement instanceof SelectStatement);
- }
-
/**
* Get result set.
*
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java
index 0ab5cbf1bfe..b607dbb70e9 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java
@@ -21,57 +21,22 @@ import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.driver.executor.callback.add.StatementAddCallback;
import org.apache.shardingsphere.driver.executor.callback.execute.StatementExecuteUpdateCallback;
import org.apache.shardingsphere.driver.executor.callback.replay.StatementReplayCallback;
+import org.apache.shardingsphere.driver.executor.engine.pushdown.DriverPushDownExecuteUpdateExecutor;
import org.apache.shardingsphere.driver.jdbc.core.connection.ShardingSphereConnection;
-import org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
-import org.apache.shardingsphere.infra.binder.context.type.TableAvailable;
-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.exception.dialect.SQLExceptionTransformEngine;
-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.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;
-import org.apache.shardingsphere.infra.executor.sql.process.ProcessEngine;
import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
-import org.apache.shardingsphere.infra.metadata.user.Grantee;
-import org.apache.shardingsphere.infra.route.context.RouteUnit;
-import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
-import org.apache.shardingsphere.infra.rule.attribute.datanode.DataNodeRuleAttribute;
-import org.apache.shardingsphere.infra.rule.attribute.raw.RawExecutionRuleAttribute;
import org.apache.shardingsphere.infra.session.query.QueryContext;
-import org.apache.shardingsphere.mode.metadata.refresher.MetaDataRefreshEngine;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.DMLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.SelectStatement;
import org.apache.shardingsphere.traffic.executor.TrafficExecutor;
import org.apache.shardingsphere.traffic.rule.TrafficRule;
-import org.apache.shardingsphere.transaction.api.TransactionType;
-import org.apache.shardingsphere.transaction.implicit.ImplicitTransactionCallback;
-import org.apache.shardingsphere.transaction.rule.TransactionRule;
import java.sql.Connection;
-import java.sql.PreparedStatement;
import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
import java.util.Optional;
/**
@@ -84,12 +49,18 @@ public final class DriverExecuteUpdateExecutor {
private final ShardingSphereMetaData metaData;
- private final JDBCExecutor jdbcExecutor;
-
- private final RawExecutor rawExecutor;
+ private final DriverPushDownExecuteUpdateExecutor pushDownExecuteUpdateExecutor;
private final TrafficExecutor trafficExecutor;
+ public DriverExecuteUpdateExecutor(final ShardingSphereConnection connection, final ShardingSphereMetaData metaData, final JDBCExecutor jdbcExecutor, final RawExecutor rawExecutor,
+ final TrafficExecutor trafficExecutor) {
+ this.connection = connection;
+ this.metaData = metaData;
+ pushDownExecuteUpdateExecutor = new DriverPushDownExecuteUpdateExecutor(connection, metaData, jdbcExecutor, rawExecutor);
+ this.trafficExecutor = trafficExecutor;
+ }
+
/**
* Execute update.
*
@@ -111,163 +82,6 @@ public final class DriverExecuteUpdateExecutor {
}
ExecutionContext executionContext = new KernelProcessor().generateExecutionContext(
queryContext, database, metaData.getGlobalRuleMetaData(), metaData.getProps(), connection.getDatabaseConnectionManager().getConnectionContext());
- return hasRawExecutionRule(database)
- ? accumulate(rawExecutor.execute(createRawExecutionGroupContext(database, executionContext), queryContext, new RawSQLExecutorCallback()))
- : executeUpdate(database, updateCallback, queryContext.getSqlStatementContext(), executionContext, prepareEngine, isNeedImplicitCommitTransaction(
- connection, queryContext.getSqlStatementContext().getSqlStatement(), executionContext.getExecutionUnits().size() > 1), addCallback, replayCallback);
- }
-
- @SuppressWarnings("rawtypes")
- private int executeUpdate(final ShardingSphereDatabase database, final StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext,
- final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine, final boolean isNeedImplicitCommitTransaction,
- final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
- return isNeedImplicitCommitTransaction
- ? executeWithImplicitCommitTransaction(() -> useDriverToExecuteUpdate(
- database, updateCallback, sqlStatementContext, executionContext, prepareEngine, addCallback, replayCallback), connection, database.getProtocolType())
- : useDriverToExecuteUpdate(database, updateCallback, sqlStatementContext, executionContext, prepareEngine, addCallback, replayCallback);
- }
-
- private boolean hasRawExecutionRule(final ShardingSphereDatabase database) {
- return !database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty();
- }
-
- private Collection<Statement> getStatements(final ExecutionGroup<JDBCExecutionUnit> executionGroup) {
- Collection<Statement> result = new LinkedList<>();
- for (JDBCExecutionUnit each : executionGroup.getInputs()) {
- result.add(each.getStorageResource());
- }
- return result;
- }
-
- private Collection<List<Object>> getParameterSets(final ExecutionGroup<JDBCExecutionUnit> executionGroup) {
- Collection<List<Object>> result = new LinkedList<>();
- for (JDBCExecutionUnit each : executionGroup.getInputs()) {
- result.add(each.getExecutionUnit().getSqlUnit().getParameters());
- }
- return result;
- }
-
- private ExecutionGroupContext<RawSQLExecutionUnit> createRawExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext) throws SQLException {
- int maxConnectionsSizePerQuery = metaData.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
- return new RawExecutionPrepareEngine(maxConnectionsSizePerQuery, database.getRuleMetaData().getRules()).prepare(
- executionContext.getRouteContext(), executionContext.getExecutionUnits(), new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", "")));
- }
-
- private <T> T executeWithImplicitCommitTransaction(final ImplicitTransactionCallback<T> 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 StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext,
- final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
- final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
- ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext = createExecutionGroupContext(database, executionContext, prepareEngine);
- for (ExecutionGroup<JDBCExecutionUnit> each : executionGroupContext.getInputGroups()) {
- addCallback.add(getStatements(each), JDBCDriverType.PREPARED_STATEMENT.equals(prepareEngine.getType()) ? getParameterSets(each) : Collections.emptyList());
- }
- replayCallback.replay();
- JDBCExecutorCallback<Integer> callback = createExecuteUpdateCallback(database, updateCallback, sqlStatementContext, prepareEngine.getType());
- return useDriverToExecuteUpdate(database, executionGroupContext, executionContext.getQueryContext(), executionContext.getRouteContext().getRouteUnits(), callback);
- }
-
- private int useDriverToExecuteUpdate(final ShardingSphereDatabase database, final ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext,
- final QueryContext queryContext, final Collection<RouteUnit> routeUnits, final JDBCExecutorCallback<Integer> callback) throws SQLException {
- ProcessEngine processEngine = new ProcessEngine();
- try {
- processEngine.executeSQL(executionGroupContext, queryContext);
- List<Integer> results = doExecute(database, executionGroupContext, queryContext, routeUnits, callback);
- return isNeedAccumulate(database.getRuleMetaData().getRules(), queryContext.getSqlStatementContext()) ? accumulate(results) : results.get(0);
- } finally {
- processEngine.completeSQLExecution(executionGroupContext.getReportContext().getProcessId());
- }
- }
-
- private boolean isNeedAccumulate(final Collection<ShardingSphereRule> rules, final SQLStatementContext sqlStatementContext) {
- if (!(sqlStatementContext instanceof TableAvailable)) {
- return false;
- }
- for (ShardingSphereRule each : rules) {
- Optional<DataNodeRuleAttribute> ruleAttribute = each.getAttributes().findAttribute(DataNodeRuleAttribute.class);
- if (ruleAttribute.isPresent() && ruleAttribute.get().isNeedAccumulate(((TableAvailable) sqlStatementContext).getTablesContext().getTableNames())) {
- return true;
- }
- }
- return false;
- }
-
- private int accumulate(final List<Integer> updateResults) {
- int result = 0;
- for (Integer each : updateResults) {
- result += null == each ? 0 : each;
- }
- return result;
- }
-
- private int accumulate(final Collection<ExecuteResult> results) {
- int result = 0;
- for (ExecuteResult each : results) {
- result += ((UpdateResult) each).getUpdateCount();
- }
- return result;
- }
-
- private <T> List<T> doExecute(final ShardingSphereDatabase database, final ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext, final QueryContext queryContext,
- final Collection<RouteUnit> routeUnits, final JDBCExecutorCallback<T> callback) throws SQLException {
- List<T> results = jdbcExecutor.execute(executionGroupContext, callback);
- new MetaDataRefreshEngine(
- connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps()).refresh(queryContext.getSqlStatementContext(), routeUnits);
- return results;
- }
-
- private ExecutionGroupContext<JDBCExecutionUnit> createExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext,
- final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine) throws SQLException {
- return prepareEngine.prepare(executionContext.getRouteContext(), executionContext.getExecutionUnits(),
- new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", "")));
- }
-
- private JDBCExecutorCallback<Integer> createExecuteUpdateCallback(final ShardingSphereDatabase database,
- final StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final String jdbcDriverType) {
- boolean isExceptionThrown = SQLExecutorExceptionHandler.isExceptionThrown();
- return new JDBCExecutorCallback<Integer>(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<Integer> getSaneResult(final SQLStatement sqlStatement, final SQLException ex) {
- return Optional.empty();
- }
- };
- }
-
- private boolean isNeedImplicitCommitTransaction(final ShardingSphereConnection connection, final SQLStatement sqlStatement, final boolean multiExecutionUnits) {
- if (!connection.getAutoCommit()) {
- return false;
- }
- TransactionType transactionType = connection.getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(TransactionRule.class).getDefaultType();
- boolean isInTransaction = connection.getDatabaseConnectionManager().getConnectionTransaction().isInTransaction();
- if (!TransactionType.isDistributedTransaction(transactionType) || isInTransaction) {
- return false;
- }
- return isWriteDMLStatement(sqlStatement) && multiExecutionUnits;
- }
-
- private boolean isWriteDMLStatement(final SQLStatement sqlStatement) {
- return sqlStatement instanceof DMLStatement && !(sqlStatement instanceof SelectStatement);
+ return pushDownExecuteUpdateExecutor.executeUpdate(database, executionContext, prepareEngine, updateCallback, addCallback, replayCallback);
}
}
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/DriverPushDownExecuteUpdateExecutor.java
similarity index 53%
copy from jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java
copy to jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/DriverPushDownExecuteUpdateExecutor.java
index 0ab5cbf1bfe..954b3d91048 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/DriverExecuteUpdateExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/DriverPushDownExecuteUpdateExecutor.java
@@ -15,17 +15,16 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.driver.executor.engine;
+package org.apache.shardingsphere.driver.executor.engine.pushdown;
-import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.driver.executor.callback.add.StatementAddCallback;
import org.apache.shardingsphere.driver.executor.callback.execute.StatementExecuteUpdateCallback;
import org.apache.shardingsphere.driver.executor.callback.replay.StatementReplayCallback;
import org.apache.shardingsphere.driver.jdbc.core.connection.ShardingSphereConnection;
import org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
import org.apache.shardingsphere.infra.binder.context.type.TableAvailable;
+import org.apache.shardingsphere.infra.config.props.ConfigurationProperties;
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.exception.dialect.SQLExceptionTransformEngine;
import org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroup;
@@ -48,19 +47,13 @@ import org.apache.shardingsphere.infra.executor.sql.prepare.raw.RawExecutionPrep
import org.apache.shardingsphere.infra.executor.sql.process.ProcessEngine;
import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.infra.metadata.database.rule.RuleMetaData;
import org.apache.shardingsphere.infra.metadata.user.Grantee;
-import org.apache.shardingsphere.infra.route.context.RouteUnit;
import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
import org.apache.shardingsphere.infra.rule.attribute.datanode.DataNodeRuleAttribute;
import org.apache.shardingsphere.infra.rule.attribute.raw.RawExecutionRuleAttribute;
-import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.mode.metadata.refresher.MetaDataRefreshEngine;
import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.DMLStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.SelectStatement;
-import org.apache.shardingsphere.traffic.executor.TrafficExecutor;
-import org.apache.shardingsphere.traffic.rule.TrafficRule;
-import org.apache.shardingsphere.transaction.api.TransactionType;
import org.apache.shardingsphere.transaction.implicit.ImplicitTransactionCallback;
import org.apache.shardingsphere.transaction.rule.TransactionRule;
@@ -75,26 +68,36 @@ import java.util.List;
import java.util.Optional;
/**
- * Driver execute update executor.
+ * Driver push down execute update executor.
*/
-@RequiredArgsConstructor
-public final class DriverExecuteUpdateExecutor {
+public final class DriverPushDownExecuteUpdateExecutor {
private final ShardingSphereConnection connection;
- private final ShardingSphereMetaData metaData;
+ private final String processId;
+
+ private final RuleMetaData globalRuleMetaData;
+
+ private final ConfigurationProperties props;
private final JDBCExecutor jdbcExecutor;
private final RawExecutor rawExecutor;
- private final TrafficExecutor trafficExecutor;
+ public DriverPushDownExecuteUpdateExecutor(final ShardingSphereConnection connection, final ShardingSphereMetaData metaData, final JDBCExecutor jdbcExecutor, final RawExecutor rawExecutor) {
+ this.connection = connection;
+ processId = connection.getProcessId();
+ globalRuleMetaData = metaData.getGlobalRuleMetaData();
+ props = metaData.getProps();
+ this.jdbcExecutor = jdbcExecutor;
+ this.rawExecutor = rawExecutor;
+ }
/**
* Execute update.
*
* @param database database
- * @param queryContext query context
+ * @param executionContext execution context
* @param prepareEngine prepare engine
* @param updateCallback statement execute update callback
* @param replayCallback statement replay callback
@@ -103,32 +106,43 @@ public final class DriverExecuteUpdateExecutor {
* @throws SQLException SQL exception
*/
@SuppressWarnings("rawtypes")
- public int executeUpdate(final ShardingSphereDatabase database, final QueryContext queryContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
+ public int executeUpdate(final ShardingSphereDatabase database, final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
final StatementExecuteUpdateCallback updateCallback, final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
- Optional<String> trafficInstanceId = connection.getTrafficInstanceId(metaData.getGlobalRuleMetaData().getSingleRule(TrafficRule.class), queryContext);
- if (trafficInstanceId.isPresent()) {
- return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, updateCallback::executeUpdate);
- }
- ExecutionContext executionContext = new KernelProcessor().generateExecutionContext(
- queryContext, database, metaData.getGlobalRuleMetaData(), metaData.getProps(), connection.getDatabaseConnectionManager().getConnectionContext());
- return hasRawExecutionRule(database)
- ? accumulate(rawExecutor.execute(createRawExecutionGroupContext(database, executionContext), queryContext, new RawSQLExecutorCallback()))
- : executeUpdate(database, updateCallback, queryContext.getSqlStatementContext(), executionContext, prepareEngine, isNeedImplicitCommitTransaction(
- connection, queryContext.getSqlStatementContext().getSqlStatement(), executionContext.getExecutionUnits().size() > 1), addCallback, replayCallback);
+ return database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty()
+ ? executeJDBCUpdate(database, executionContext, prepareEngine, updateCallback, addCallback, replayCallback)
+ : executeRawUpdate(database, executionContext);
}
@SuppressWarnings("rawtypes")
- private int executeUpdate(final ShardingSphereDatabase database, final StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext,
- final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine, final boolean isNeedImplicitCommitTransaction,
- final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
- return isNeedImplicitCommitTransaction
- ? executeWithImplicitCommitTransaction(() -> useDriverToExecuteUpdate(
- database, updateCallback, sqlStatementContext, executionContext, prepareEngine, addCallback, replayCallback), connection, database.getProtocolType())
- : useDriverToExecuteUpdate(database, updateCallback, sqlStatementContext, executionContext, prepareEngine, addCallback, replayCallback);
+ private int executeJDBCUpdate(final ShardingSphereDatabase database, final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
+ final StatementExecuteUpdateCallback updateCallback, final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
+ boolean isImplicitCommitTransaction = globalRuleMetaData.getSingleRule(TransactionRule.class).isImplicitCommitTransaction(executionContext.getSqlStatementContext().getSqlStatement(),
+ connection.getDatabaseConnectionManager().getConnectionTransaction(), connection.getAutoCommit(), executionContext.getExecutionUnits().size() > 1);
+ return isImplicitCommitTransaction
+ ? executeWithImplicitCommitTransaction(database, () -> doExecuteJDBCUpdate(database, executionContext, prepareEngine, updateCallback, addCallback, replayCallback))
+ : doExecuteJDBCUpdate(database, executionContext, prepareEngine, updateCallback, addCallback, replayCallback);
}
- private boolean hasRawExecutionRule(final ShardingSphereDatabase database) {
- return !database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty();
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private int doExecuteJDBCUpdate(final ShardingSphereDatabase database, final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
+ final StatementExecuteUpdateCallback updateCallback, final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
+ ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext = prepareEngine.prepare(
+ executionContext.getRouteContext(), executionContext.getExecutionUnits(), new ExecutionGroupReportContext(processId, database.getName(), new Grantee("", "")));
+ for (ExecutionGroup<JDBCExecutionUnit> each : executionGroupContext.getInputGroups()) {
+ addCallback.add(getStatements(each), JDBCDriverType.PREPARED_STATEMENT.equals(prepareEngine.getType()) ? getParameterSets(each) : Collections.emptyList());
+ }
+ replayCallback.replay();
+ ProcessEngine processEngine = new ProcessEngine();
+ try {
+ processEngine.executeSQL(executionGroupContext, executionContext.getQueryContext());
+ List<Integer> updateCounts = jdbcExecutor.execute(
+ executionGroupContext, createExecuteUpdateCallback(database, updateCallback, executionContext.getSqlStatementContext(), prepareEngine.getType()));
+ new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, props)
+ .refresh(executionContext.getQueryContext().getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
+ return isNeedAccumulate(database.getRuleMetaData().getRules(), executionContext.getQueryContext().getSqlStatementContext()) ? accumulate(updateCounts) : updateCounts.get(0);
+ } finally {
+ processEngine.completeSQLExecution(executionGroupContext.getReportContext().getProcessId());
+ }
}
private Collection<Statement> getStatements(final ExecutionGroup<JDBCExecutionUnit> executionGroup) {
@@ -147,52 +161,20 @@ public final class DriverExecuteUpdateExecutor {
return result;
}
- private ExecutionGroupContext<RawSQLExecutionUnit> createRawExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext) throws SQLException {
- int maxConnectionsSizePerQuery = metaData.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
- return new RawExecutionPrepareEngine(maxConnectionsSizePerQuery, database.getRuleMetaData().getRules()).prepare(
- executionContext.getRouteContext(), executionContext.getExecutionUnits(), new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", "")));
- }
-
- private <T> T executeWithImplicitCommitTransaction(final ImplicitTransactionCallback<T> 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 StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext,
- final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
- final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
- ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext = createExecutionGroupContext(database, executionContext, prepareEngine);
- for (ExecutionGroup<JDBCExecutionUnit> each : executionGroupContext.getInputGroups()) {
- addCallback.add(getStatements(each), JDBCDriverType.PREPARED_STATEMENT.equals(prepareEngine.getType()) ? getParameterSets(each) : Collections.emptyList());
- }
- replayCallback.replay();
- JDBCExecutorCallback<Integer> callback = createExecuteUpdateCallback(database, updateCallback, sqlStatementContext, prepareEngine.getType());
- return useDriverToExecuteUpdate(database, executionGroupContext, executionContext.getQueryContext(), executionContext.getRouteContext().getRouteUnits(), callback);
- }
-
- private int useDriverToExecuteUpdate(final ShardingSphereDatabase database, final ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext,
- final QueryContext queryContext, final Collection<RouteUnit> routeUnits, final JDBCExecutorCallback<Integer> callback) throws SQLException {
- ProcessEngine processEngine = new ProcessEngine();
- try {
- processEngine.executeSQL(executionGroupContext, queryContext);
- List<Integer> results = doExecute(database, executionGroupContext, queryContext, routeUnits, callback);
- return isNeedAccumulate(database.getRuleMetaData().getRules(), queryContext.getSqlStatementContext()) ? accumulate(results) : results.get(0);
- } finally {
- processEngine.completeSQLExecution(executionGroupContext.getReportContext().getProcessId());
- }
+ private JDBCExecutorCallback<Integer> createExecuteUpdateCallback(final ShardingSphereDatabase database, final StatementExecuteUpdateCallback updateCallback,
+ final SQLStatementContext sqlStatementContext, final String jdbcDriverType) {
+ return new JDBCExecutorCallback<Integer>(database.getProtocolType(), database.getResourceMetaData(), sqlStatementContext.getSqlStatement(), SQLExecutorExceptionHandler.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<Integer> getSaneResult(final SQLStatement sqlStatement, final SQLException ex) {
+ return Optional.empty();
+ }
+ };
}
private boolean isNeedAccumulate(final Collection<ShardingSphereRule> rules, final SQLStatementContext sqlStatementContext) {
@@ -224,50 +206,29 @@ public final class DriverExecuteUpdateExecutor {
return result;
}
- private <T> List<T> doExecute(final ShardingSphereDatabase database, final ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext, final QueryContext queryContext,
- final Collection<RouteUnit> routeUnits, final JDBCExecutorCallback<T> callback) throws SQLException {
- List<T> results = jdbcExecutor.execute(executionGroupContext, callback);
- new MetaDataRefreshEngine(
- connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps()).refresh(queryContext.getSqlStatementContext(), routeUnits);
- return results;
- }
-
- private ExecutionGroupContext<JDBCExecutionUnit> createExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext,
- final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine) throws SQLException {
- return prepareEngine.prepare(executionContext.getRouteContext(), executionContext.getExecutionUnits(),
- new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", "")));
- }
-
- private JDBCExecutorCallback<Integer> createExecuteUpdateCallback(final ShardingSphereDatabase database,
- final StatementExecuteUpdateCallback updateCallback, final SQLStatementContext sqlStatementContext, final String jdbcDriverType) {
- boolean isExceptionThrown = SQLExecutorExceptionHandler.isExceptionThrown();
- return new JDBCExecutorCallback<Integer>(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<Integer> getSaneResult(final SQLStatement sqlStatement, final SQLException ex) {
- return Optional.empty();
- }
- };
+ private <T> T executeWithImplicitCommitTransaction(final ShardingSphereDatabase database, final ImplicitTransactionCallback<T> callback) throws SQLException {
+ try {
+ connection.setAutoCommit(false);
+ T result = callback.execute();
+ connection.commit();
+ return result;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ connection.rollback();
+ throw SQLExceptionTransformEngine.toSQLException(ex, database.getProtocolType());
+ } finally {
+ connection.setAutoCommit(true);
+ }
}
- private boolean isNeedImplicitCommitTransaction(final ShardingSphereConnection connection, final SQLStatement sqlStatement, final boolean multiExecutionUnits) {
- if (!connection.getAutoCommit()) {
- return false;
- }
- TransactionType transactionType = connection.getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(TransactionRule.class).getDefaultType();
- boolean isInTransaction = connection.getDatabaseConnectionManager().getConnectionTransaction().isInTransaction();
- if (!TransactionType.isDistributedTransaction(transactionType) || isInTransaction) {
- return false;
- }
- return isWriteDMLStatement(sqlStatement) && multiExecutionUnits;
+ private int executeRawUpdate(final ShardingSphereDatabase database, final ExecutionContext executionContext) throws SQLException {
+ return accumulate(rawExecutor.execute(createRawExecutionGroupContext(database, executionContext), executionContext.getQueryContext(), new RawSQLExecutorCallback()));
}
- private boolean isWriteDMLStatement(final SQLStatement sqlStatement) {
- return sqlStatement instanceof DMLStatement && !(sqlStatement instanceof SelectStatement);
+ private ExecutionGroupContext<RawSQLExecutionUnit> createRawExecutionGroupContext(final ShardingSphereDatabase database, final ExecutionContext executionContext) throws SQLException {
+ int maxConnectionsSizePerQuery = props.<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
+ return new RawExecutionPrepareEngine(maxConnectionsSizePerQuery, database.getRuleMetaData().getRules()).prepare(
+ executionContext.getRouteContext(), executionContext.getExecutionUnits(), new ExecutionGroupReportContext(processId, database.getName(), new Grantee("", "")));
}
}
diff --git a/kernel/transaction/core/src/main/java/org/apache/shardingsphere/transaction/rule/TransactionRule.java b/kernel/transaction/core/src/main/java/org/apache/shardingsphere/transaction/rule/TransactionRule.java
index d1761019c8c..571039f0442 100644
--- a/kernel/transaction/core/src/main/java/org/apache/shardingsphere/transaction/rule/TransactionRule.java
+++ b/kernel/transaction/core/src/main/java/org/apache/shardingsphere/transaction/rule/TransactionRule.java
@@ -23,6 +23,10 @@ import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.rule.attribute.RuleAttributes;
import org.apache.shardingsphere.infra.rule.scope.GlobalRule;
+import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
+import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.DMLStatement;
+import org.apache.shardingsphere.sql.parser.sql.common.statement.dml.SelectStatement;
+import org.apache.shardingsphere.transaction.ConnectionTransaction;
import org.apache.shardingsphere.transaction.ShardingSphereTransactionManagerEngine;
import org.apache.shardingsphere.transaction.api.TransactionType;
import org.apache.shardingsphere.transaction.config.TransactionRuleConfiguration;
@@ -89,6 +93,29 @@ public final class TransactionRule implements GlobalRule, AutoCloseable {
return resource.get();
}
+ /**
+ * Judge whether to implicit commit transaction.
+ *
+ * @param sqlStatement SQL statement
+ * @param connectionTransaction connection transaction
+ * @param isAutoCommit is auto commit
+ * @param multiExecutionUnits is multi execution units
+ * @return is implicit commit transaction or not
+ */
+ public boolean isImplicitCommitTransaction(final SQLStatement sqlStatement, final ConnectionTransaction connectionTransaction, final boolean isAutoCommit, final boolean multiExecutionUnits) {
+ if (!isAutoCommit) {
+ return false;
+ }
+ if (!TransactionType.isDistributedTransaction(defaultType) || connectionTransaction.isInTransaction()) {
+ return false;
+ }
+ return isWriteDMLStatement(sqlStatement) && multiExecutionUnits;
+ }
+
+ private boolean isWriteDMLStatement(final SQLStatement sqlStatement) {
+ return sqlStatement instanceof DMLStatement && !(sqlStatement instanceof SelectStatement);
+ }
+
@Override
public void refresh(final Map<String, ShardingSphereDatabase> databases, final GlobalRuleChangedType changedType) {
if (GlobalRuleChangedType.DATABASE_CHANGED != changedType) {