You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zi...@apache.org on 2024/06/01 19:21:25 UTC
(shardingsphere) branch master updated: Refactor TrafficExecutorCallback (#31522)
This is an automated email from the ASF dual-hosted git repository.
zichaowang 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 c84a3d278ea Refactor TrafficExecutorCallback (#31522)
c84a3d278ea is described below
commit c84a3d278eaf51ee48ae16df0ae0aa9d1118cc21
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sun Jun 2 03:21:19 2024 +0800
Refactor TrafficExecutorCallback (#31522)
---
.../driver/executor/DriverExecutor.java | 12 +++++-----
.../statement/ShardingSpherePreparedStatement.java | 6 ++---
.../core/statement/ShardingSphereStatement.java | 26 ++++++++++------------
.../traffic/executor/TrafficExecutor.java | 2 +-
.../traffic/executor/TrafficExecutorCallback.java | 4 ++--
5 files changed, 23 insertions(+), 27 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 8d2430f191c..42f7ebc8691 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
@@ -164,7 +164,7 @@ public final class DriverExecutor implements AutoCloseable {
}
private TrafficExecutorCallback<ResultSet> getTrafficExecuteQueryCallback(final String jdbcDriverType) {
- return JDBCDriverType.STATEMENT.equals(jdbcDriverType) ? Statement::executeQuery : ((statement, sql) -> ((PreparedStatement) statement).executeQuery());
+ return JDBCDriverType.STATEMENT.equals(jdbcDriverType) ? ((sql, statement) -> statement.executeQuery(sql)) : ((sql, statement) -> ((PreparedStatement) statement).executeQuery());
}
private ExecuteQueryCallback getExecuteQueryCallback(final ShardingSphereDatabase database, final QueryContext queryContext, final String jdbcDriverType) {
@@ -257,7 +257,6 @@ public final class DriverExecutor implements AutoCloseable {
* @param database database
* @param queryContext query context
* @param prepareEngine prepare engine
- * @param trafficCallback traffic callback
* @param updateCallback update callback
* @param statementReplayCallback statement replay callback
* @return updated row count
@@ -265,11 +264,11 @@ public final class DriverExecutor implements AutoCloseable {
*/
@SuppressWarnings("rawtypes")
public int executeUpdate(final ShardingSphereDatabase database, final QueryContext queryContext,
- final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine, final TrafficExecutorCallback<Integer> trafficCallback,
+ final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
final ExecuteUpdateCallback updateCallback, final StatementReplayCallback statementReplayCallback) 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, trafficCallback);
+ return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, updateCallback::executeUpdate);
}
ExecutionContext executionContext = createExecutionContext(database, queryContext);
return database.getRuleMetaData().getAttributes(RawExecutionRuleAttribute.class).isEmpty()
@@ -360,7 +359,6 @@ public final class DriverExecutor implements AutoCloseable {
* @param database database
* @param queryContext query context
* @param prepareEngine prepare engine
- * @param trafficCallback traffic callback
* @param executeCallback execute callback
* @param statementReplayCallback statement replay callback
* @return execute result
@@ -368,12 +366,12 @@ public final class DriverExecutor implements AutoCloseable {
*/
@SuppressWarnings("rawtypes")
public boolean executeAdvance(final ShardingSphereDatabase database, final QueryContext queryContext,
- final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine, final TrafficExecutorCallback<Boolean> trafficCallback,
+ final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
final ExecuteCallback executeCallback, final StatementReplayCallback statementReplayCallback) throws SQLException {
Optional<String> trafficInstanceId = connection.getTrafficInstanceId(metaData.getGlobalRuleMetaData().getSingleRule(TrafficRule.class), queryContext);
if (trafficInstanceId.isPresent()) {
executeType = ExecuteType.TRAFFIC;
- return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, trafficCallback);
+ return trafficExecutor.execute(connection.getProcessId(), database.getName(), trafficInstanceId.get(), queryContext, prepareEngine, executeCallback::execute);
}
if (sqlFederationEngine.decide(queryContext.getSqlStatementContext(), queryContext.getParameters(), database, metaData.getGlobalRuleMetaData())) {
executeType = ExecuteType.FEDERATION;
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 198fe9bb813..a3dfffaa2fa 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
@@ -261,7 +261,7 @@ public final class ShardingSpherePreparedStatement extends AbstractPreparedState
handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement());
ShardingSphereDatabase database = metaData.getDatabase(databaseName);
final int result = executor.executeUpdate(database, queryContext, createDriverExecutionPrepareEngine(database),
- (statement, sql) -> ((PreparedStatement) statement).executeUpdate(), null, (StatementReplayCallback<PreparedStatement>) this::replay);
+ (sql, statement) -> ((PreparedStatement) statement).executeUpdate(), (StatementReplayCallback<PreparedStatement>) this::replay);
for (Statement each : executor.getStatements()) {
statements.add((PreparedStatement) each);
}
@@ -289,8 +289,8 @@ public final class ShardingSpherePreparedStatement extends AbstractPreparedState
QueryContext queryContext = createQueryContext();
handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement());
ShardingSphereDatabase database = metaData.getDatabase(databaseName);
- final boolean result = executor.executeAdvance(database, queryContext, createDriverExecutionPrepareEngine(database), (statement, sql) -> ((PreparedStatement) statement).execute(),
- null, (StatementReplayCallback<PreparedStatement>) this::replay);
+ final boolean result = executor.executeAdvance(database, queryContext, createDriverExecutionPrepareEngine(database), (sql, statement) -> ((PreparedStatement) statement).execute(),
+ (StatementReplayCallback<PreparedStatement>) this::replay);
for (Statement each : executor.getStatements()) {
statements.add((PreparedStatement) each);
}
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 7cb715a0b5f..2437e7a54dc 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
@@ -55,7 +55,6 @@ import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.parser.rule.SQLParserRule;
import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
import org.apache.shardingsphere.sql.parser.sql.common.statement.dal.DALStatement;
-import org.apache.shardingsphere.traffic.executor.TrafficExecutorCallback;
import org.apache.shardingsphere.transaction.util.AutoCommitUtils;
import java.sql.Connection;
@@ -151,7 +150,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
@Override
public int executeUpdate(final String sql) throws SQLException {
try {
- return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL), Statement::executeUpdate);
+ return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL));
// CHECKSTYLE:OFF
} catch (final RuntimeException ex) {
// CHECKSTYLE:ON
@@ -168,8 +167,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
returnGeneratedKeys = true;
}
try {
- return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, autoGeneratedKeys),
- (statement, actualSQL) -> statement.executeUpdate(actualSQL, autoGeneratedKeys));
+ return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, autoGeneratedKeys));
// CHECKSTYLE:OFF
} catch (final RuntimeException ex) {
// CHECKSTYLE:ON
@@ -184,7 +182,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
public int executeUpdate(final String sql, final int[] columnIndexes) throws SQLException {
returnGeneratedKeys = true;
try {
- return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, columnIndexes), (statement, actualSQL) -> statement.executeUpdate(actualSQL, columnIndexes));
+ return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, columnIndexes));
// CHECKSTYLE:OFF
} catch (final RuntimeException ex) {
// CHECKSTYLE:ON
@@ -199,7 +197,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
public int executeUpdate(final String sql, final String[] columnNames) throws SQLException {
returnGeneratedKeys = true;
try {
- return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, columnNames), (statement, actualSQL) -> statement.executeUpdate(actualSQL, columnNames));
+ return executeUpdate(sql, (actualSQL, statement) -> statement.executeUpdate(actualSQL, columnNames));
// CHECKSTYLE:OFF
} catch (final RuntimeException ex) {
// CHECKSTYLE:ON
@@ -210,7 +208,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
}
}
- private int executeUpdate(final String sql, final ExecuteUpdateCallback updateCallback, final TrafficExecutorCallback<Integer> trafficCallback) throws SQLException {
+ private int executeUpdate(final String sql, final ExecuteUpdateCallback updateCallback) throws SQLException {
QueryContext queryContext = createQueryContext(sql);
handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement());
databaseName = queryContext.getDatabaseNameFromSQLStatement().orElse(connection.getDatabaseName());
@@ -218,7 +216,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
ShardingSphereDatabase database = metaData.getDatabase(databaseName);
sqlStatementContext = queryContext.getSqlStatementContext();
clearStatements();
- int result = executor.executeUpdate(database, queryContext, createDriverExecutionPrepareEngine(database), trafficCallback, updateCallback,
+ int result = executor.executeUpdate(database, queryContext, createDriverExecutionPrepareEngine(database), updateCallback,
(StatementReplayCallback<Statement>) (statements, parameterSets) -> replay(statements));
statements.addAll(executor.getStatements());
replay(statements);
@@ -228,7 +226,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
@Override
public boolean execute(final String sql) throws SQLException {
try {
- return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL), Statement::execute);
+ return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL));
// CHECKSTYLE:OFF
} catch (final SQLException ex) {
// CHECKSTYLE:ON
@@ -243,7 +241,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
if (RETURN_GENERATED_KEYS == autoGeneratedKeys) {
returnGeneratedKeys = true;
}
- return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, autoGeneratedKeys), (statement, actualSQL) -> statement.execute(actualSQL, autoGeneratedKeys));
+ return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, autoGeneratedKeys));
// CHECKSTYLE:OFF
} catch (final SQLException ex) {
// CHECKSTYLE:ON
@@ -256,7 +254,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
public boolean execute(final String sql, final int[] columnIndexes) throws SQLException {
try {
returnGeneratedKeys = true;
- return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, columnIndexes), (statement, actualSQL) -> statement.execute(actualSQL, columnIndexes));
+ return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, columnIndexes));
// CHECKSTYLE:OFF
} catch (final SQLException ex) {
// CHECKSTYLE:ON
@@ -269,7 +267,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
public boolean execute(final String sql, final String[] columnNames) throws SQLException {
try {
returnGeneratedKeys = true;
- return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, columnNames), (statement, actualSQL) -> statement.execute(actualSQL, columnNames));
+ return execute0(sql, (actualSQL, statement) -> statement.execute(actualSQL, columnNames));
// CHECKSTYLE:OFF
} catch (final SQLException ex) {
// CHECKSTYLE:ON
@@ -278,7 +276,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
}
}
- private boolean execute0(final String sql, final ExecuteCallback executeCallback, final TrafficExecutorCallback<Boolean> trafficCallback) throws SQLException {
+ private boolean execute0(final String sql, final ExecuteCallback executeCallback) throws SQLException {
currentResultSet = null;
QueryContext queryContext = createQueryContext(sql);
handleAutoCommit(queryContext.getSqlStatementContext().getSqlStatement());
@@ -287,7 +285,7 @@ public final class ShardingSphereStatement extends AbstractStatementAdapter {
ShardingSphereDatabase database = metaData.getDatabase(databaseName);
sqlStatementContext = queryContext.getSqlStatementContext();
clearStatements();
- boolean result = executor.executeAdvance(database, queryContext, createDriverExecutionPrepareEngine(database), trafficCallback,
+ boolean result = executor.executeAdvance(database, queryContext, createDriverExecutionPrepareEngine(database),
executeCallback, (StatementReplayCallback<Statement>) (statements, parameterSets) -> replay(statements));
statements.addAll(executor.getStatements());
return result;
diff --git a/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutor.java b/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutor.java
index 7706ac23650..56e42edf713 100644
--- a/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutor.java
+++ b/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutor.java
@@ -65,7 +65,7 @@ public final class TrafficExecutor implements AutoCloseable {
JDBCExecutionUnit executionUnit = createTrafficExecutionUnit(processId, databaseName, trafficInstanceId, queryContext, prepareEngine);
SQLUnit sqlUnit = executionUnit.getExecutionUnit().getSqlUnit();
cacheStatement(sqlUnit.getParameters(), executionUnit.getStorageResource());
- T result = callback.execute(statement, sqlUnit.getSql());
+ T result = callback.execute(sqlUnit.getSql(), statement);
resultSet = statement.getResultSet();
return result;
}
diff --git a/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutorCallback.java b/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutorCallback.java
index bc6a5140d37..e6b25160ef0 100644
--- a/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutorCallback.java
+++ b/kernel/traffic/core/src/main/java/org/apache/shardingsphere/traffic/executor/TrafficExecutorCallback.java
@@ -30,10 +30,10 @@ public interface TrafficExecutorCallback<T> {
/**
* Execute.
*
- * @param statement statement
* @param sql SQL
+ * @param statement statement
* @return execution result
* @throws SQLException SQL exception
*/
- T execute(Statement statement, String sql) throws SQLException;
+ T execute(String sql, Statement statement) throws SQLException;
}