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/18 17:54:10 UTC
(shardingsphere) branch master updated: Refactor ProcessEngine (#31751)
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 d4e2edaadf8 Refactor ProcessEngine (#31751)
d4e2edaadf8 is described below
commit d4e2edaadf854e19b0ca399581bc5333d3311023
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Wed Jun 19 01:54:04 2024 +0800
Refactor ProcessEngine (#31751)
---
.../kernel/model/ExecutionGroupReportContext.java | 15 +++++++++++
.../infra/executor/sql/process/Process.java | 7 +++---
.../infra/executor/sql/process/ProcessEngine.java | 29 ++++++++++++++++------
.../BatchPreparedStatementExecutor.java | 3 +--
.../DriverExecuteBatchExecutor.java | 3 +--
.../core/connection/ShardingSphereConnection.java | 3 +--
.../BatchPreparedStatementExecutorTest.java | 5 ++--
.../enumerable/EnumerableScanExecutor.java | 7 +++---
.../traffic/executor/TrafficExecutor.java | 3 +--
.../netty/FrontendChannelInboundHandler.java | 2 +-
10 files changed, 51 insertions(+), 26 deletions(-)
diff --git a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/model/ExecutionGroupReportContext.java b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/model/ExecutionGroupReportContext.java
index 9f85bcc9990..615aabd16a3 100644
--- a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/model/ExecutionGroupReportContext.java
+++ b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/kernel/model/ExecutionGroupReportContext.java
@@ -21,6 +21,8 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.metadata.user.Grantee;
+import java.util.Optional;
+
/**
* Execution group report context.
*/
@@ -34,4 +36,17 @@ public final class ExecutionGroupReportContext {
private final String databaseName;
private final Grantee grantee;
+
+ public ExecutionGroupReportContext(final String processId, final String databaseName) {
+ this(processId, databaseName, null);
+ }
+
+ /**
+ * Get grantee.
+ *
+ * @return grantee
+ */
+ public Optional<Grantee> getGrantee() {
+ return Optional.ofNullable(grantee);
+ }
}
diff --git a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/Process.java b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/Process.java
index 343cb31256f..b07c1da2492 100644
--- a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/Process.java
+++ b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/Process.java
@@ -29,6 +29,7 @@ import org.apache.shardingsphere.infra.metadata.user.Grantee;
import java.sql.Statement;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -75,9 +76,9 @@ public final class Process {
startMillis = System.currentTimeMillis();
this.sql = sql;
databaseName = executionGroupContext.getReportContext().getDatabaseName();
- Grantee grantee = executionGroupContext.getReportContext().getGrantee();
- username = null == grantee ? "" : grantee.getUsername();
- hostname = null == grantee ? "" : grantee.getHostname();
+ Optional<Grantee> grantee = executionGroupContext.getReportContext().getGrantee();
+ username = grantee.map(Grantee::getUsername).orElse("");
+ hostname = grantee.map(Grantee::getHostname).orElse("");
totalUnitCount = new AtomicInteger(getTotalUnitCount(executionGroupContext));
processStatements.putAll(createProcessStatements(executionGroupContext));
completedUnitCount = new AtomicInteger(0);
diff --git a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/ProcessEngine.java b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/ProcessEngine.java
index 5044a92fff1..8c9dc56e6f8 100644
--- a/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/ProcessEngine.java
+++ b/infra/executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/process/ProcessEngine.java
@@ -40,19 +40,34 @@ public final class ProcessEngine {
/**
* Connect.
*
- * @param grantee grantee
* @param databaseName database name
* @return process ID
*/
- public String connect(final Grantee grantee, final String databaseName) {
- String processId = new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
- ExecutionGroupContext<? extends SQLExecutionUnit> executionGroupContext =
- new ExecutionGroupContext<>(Collections.emptyList(), new ExecutionGroupReportContext(processId, databaseName, grantee));
- Process process = new Process(executionGroupContext);
- ProcessRegistry.getInstance().add(process);
+ public String connect(final String databaseName) {
+ return connect(new ExecutionGroupReportContext(getProcessId(), databaseName));
+ }
+
+ /**
+ * Connect.
+ *
+ * @param databaseName database name
+ * @param grantee grantee
+ * @return process ID
+ */
+ public String connect(final String databaseName, final Grantee grantee) {
+ return connect(new ExecutionGroupReportContext(getProcessId(), databaseName, grantee));
+ }
+
+ private String connect(final ExecutionGroupReportContext reportContext) {
+ ExecutionGroupContext<? extends SQLExecutionUnit> executionGroupContext = new ExecutionGroupContext<>(Collections.emptyList(), reportContext);
+ ProcessRegistry.getInstance().add(new Process(executionGroupContext));
return executionGroupContext.getReportContext().getProcessId();
}
+ private String getProcessId() {
+ return new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
+ }
+
/**
* Disconnect.
*
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/BatchPreparedStatementExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/BatchPreparedStatementExecutor.java
index ec4095cd031..5735fe56d56 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/BatchPreparedStatementExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/BatchPreparedStatementExecutor.java
@@ -31,7 +31,6 @@ import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.J
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.metadata.database.ShardingSphereDatabase;
-import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.rule.attribute.datanode.DataNodeRuleAttribute;
import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
@@ -65,7 +64,7 @@ public final class BatchPreparedStatementExecutor {
public BatchPreparedStatementExecutor(final ShardingSphereDatabase database, final JDBCExecutor jdbcExecutor, final String processId) {
this.database = database;
this.jdbcExecutor = jdbcExecutor;
- executionGroupContext = new ExecutionGroupContext<>(new LinkedList<>(), new ExecutionGroupReportContext(processId, database.getName(), new Grantee("", "")));
+ executionGroupContext = new ExecutionGroupContext<>(new LinkedList<>(), new ExecutionGroupReportContext(processId, database.getName()));
batchExecutionUnits = new LinkedList<>();
}
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/DriverExecuteBatchExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/DriverExecuteBatchExecutor.java
index 7a06dd6dfa7..d6fe56fc8e7 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/DriverExecuteBatchExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/batch/preparedstatement/DriverExecuteBatchExecutor.java
@@ -38,7 +38,6 @@ import org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.Statemen
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.RouteContext;
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.traffic.rule.TrafficRule;
@@ -152,7 +151,7 @@ public final class DriverExecuteBatchExecutor {
executionUnits.add(executionUnit);
}
batchExecutor.init(prepareEngine
- .prepare(database.getName(), executionContext.getRouteContext(), executionUnits, new ExecutionGroupReportContext(connection.getProcessId(), database.getName(), new Grantee("", ""))));
+ .prepare(database.getName(), executionContext.getRouteContext(), executionUnits, new ExecutionGroupReportContext(connection.getProcessId(), database.getName())));
setBatchParameters(replayCallback);
}
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/connection/ShardingSphereConnection.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/connection/ShardingSphereConnection.java
index 457bf0e654e..8021b5e6435 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/connection/ShardingSphereConnection.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/jdbc/core/connection/ShardingSphereConnection.java
@@ -28,7 +28,6 @@ import org.apache.shardingsphere.driver.jdbc.core.statement.StatementManager;
import org.apache.shardingsphere.infra.annotation.HighFrequencyInvocation;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.executor.sql.process.ProcessEngine;
-import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.traffic.engine.TrafficEngine;
@@ -83,7 +82,7 @@ public final class ShardingSphereConnection extends AbstractConnectionAdapter {
this.databaseName = databaseName;
this.contextManager = contextManager;
databaseConnectionManager = new DriverDatabaseConnectionManager(databaseName, contextManager);
- processId = processEngine.connect(new Grantee("", ""), databaseName);
+ processId = processEngine.connect(databaseName);
}
/**
diff --git a/jdbc/src/test/java/org/apache/shardingsphere/driver/executor/batch/preparedstatement/BatchPreparedStatementExecutorTest.java b/jdbc/src/test/java/org/apache/shardingsphere/driver/executor/batch/preparedstatement/BatchPreparedStatementExecutorTest.java
index 9eec216ba7e..5706d591b8f 100644
--- a/jdbc/src/test/java/org/apache/shardingsphere/driver/executor/batch/preparedstatement/BatchPreparedStatementExecutorTest.java
+++ b/jdbc/src/test/java/org/apache/shardingsphere/driver/executor/batch/preparedstatement/BatchPreparedStatementExecutorTest.java
@@ -35,7 +35,6 @@ import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.J
import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutor;
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.rule.attribute.RuleAttributes;
import org.apache.shardingsphere.infra.rule.attribute.datanode.DataNodeRuleAttribute;
import org.apache.shardingsphere.infra.session.connection.ConnectionContext;
@@ -184,8 +183,8 @@ class BatchPreparedStatementExecutorTest {
@SneakyThrows(ReflectiveOperationException.class)
private void setFields(final Collection<ExecutionGroup<JDBCExecutionUnit>> executionGroups, final Collection<BatchExecutionUnit> batchExecutionUnits) {
String processId = new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
- Plugins.getMemberAccessor().set(BatchPreparedStatementExecutor.class.getDeclaredField("executionGroupContext"), executor, new ExecutionGroupContext<>(executionGroups,
- new ExecutionGroupReportContext(processId, "logic_db", new Grantee("", ""))));
+ Plugins.getMemberAccessor().set(BatchPreparedStatementExecutor.class.getDeclaredField("executionGroupContext"), executor,
+ new ExecutionGroupContext<>(executionGroups, new ExecutionGroupReportContext(processId, "logic_db")));
Plugins.getMemberAccessor().set(BatchPreparedStatementExecutor.class.getDeclaredField("batchExecutionUnits"), executor, batchExecutionUnits);
Plugins.getMemberAccessor().set(BatchPreparedStatementExecutor.class.getDeclaredField("batchCount"), executor, 2);
}
diff --git a/kernel/sql-federation/executor/src/main/java/org/apache/shardingsphere/sqlfederation/executor/enumerable/EnumerableScanExecutor.java b/kernel/sql-federation/executor/src/main/java/org/apache/shardingsphere/sqlfederation/executor/enumerable/EnumerableScanExecutor.java
index 15e9e1d976a..1322695ff11 100644
--- a/kernel/sql-federation/executor/src/main/java/org/apache/shardingsphere/sqlfederation/executor/enumerable/EnumerableScanExecutor.java
+++ b/kernel/sql-federation/executor/src/main/java/org/apache/shardingsphere/sqlfederation/executor/enumerable/EnumerableScanExecutor.java
@@ -52,7 +52,6 @@ import org.apache.shardingsphere.infra.metadata.database.rule.RuleMetaData;
import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereTable;
import org.apache.shardingsphere.infra.metadata.statistics.ShardingSphereStatistics;
import org.apache.shardingsphere.infra.metadata.statistics.ShardingSphereTableData;
-import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.parser.sql.SQLStatementParserEngine;
import org.apache.shardingsphere.infra.session.connection.ConnectionContext;
import org.apache.shardingsphere.infra.session.query.QueryContext;
@@ -131,9 +130,9 @@ public final class EnumerableScanExecutor implements ScanExecutor {
public Enumerator<Object> enumerator() {
computeConnectionOffsets(context);
// TODO pass grantee from proxy and jdbc adapter
- ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext =
- prepareEngine.prepare(database.getName(), context.getRouteContext(), executorContext.getConnectionOffsets(), context.getExecutionUnits(),
- new ExecutionGroupReportContext(federationContext.getProcessId(), database.getName(), new Grantee("", "")));
+ ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext = prepareEngine.prepare(
+ database.getName(), context.getRouteContext(), executorContext.getConnectionOffsets(),
+ context.getExecutionUnits(), new ExecutionGroupReportContext(federationContext.getProcessId(), database.getName()));
setParameters(executionGroupContext.getInputGroups());
ShardingSpherePreconditions.checkState(!ProcessRegistry.getInstance().get(federationContext.getProcessId()).isInterrupted(), SQLExecutionInterruptedException::new);
processEngine.executeSQL(executionGroupContext, federationContext.getQueryContext());
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 36c00d92d01..a5a840374ea 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
@@ -24,7 +24,6 @@ import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
import org.apache.shardingsphere.infra.executor.sql.context.SQLUnit;
import org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutionUnit;
import org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine;
-import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.route.context.RouteContext;
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.traffic.exception.EmptyTrafficExecutionUnitException;
@@ -74,7 +73,7 @@ public final class TrafficExecutor implements AutoCloseable {
final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine) throws SQLException {
ExecutionUnit executionUnit = new ExecutionUnit(trafficInstanceId, new SQLUnit(queryContext.getSql(), queryContext.getParameters()));
ExecutionGroupContext<JDBCExecutionUnit> context =
- prepareEngine.prepare(databaseName, new RouteContext(), Collections.singleton(executionUnit), new ExecutionGroupReportContext(processId, databaseName, new Grantee("", "")));
+ prepareEngine.prepare(databaseName, new RouteContext(), Collections.singleton(executionUnit), new ExecutionGroupReportContext(processId, databaseName));
return context.getInputGroups().stream().flatMap(each -> each.getInputs().stream()).findFirst().orElseThrow(EmptyTrafficExecutionUnitException::new);
}
diff --git a/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandler.java b/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandler.java
index c7c0680af15..8558d2e5a97 100644
--- a/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandler.java
+++ b/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandler.java
@@ -78,7 +78,7 @@ public final class FrontendChannelInboundHandler extends ChannelInboundHandlerAd
if (authResult.isFinished()) {
connectionSession.setGrantee(new Grantee(authResult.getUsername(), authResult.getHostname()));
connectionSession.setCurrentDatabase(authResult.getDatabase());
- connectionSession.setProcessId(processEngine.connect(connectionSession.getConnectionContext().getGrantee(), connectionSession.getDatabaseName()));
+ connectionSession.setProcessId(processEngine.connect(connectionSession.getDatabaseName(), connectionSession.getConnectionContext().getGrantee()));
}
return authResult.isFinished();
// CHECKSTYLE:OFF