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 2025/01/23 14:03:31 UTC
(shardingsphere) branch master updated: Refactor MetaDataRefreshEngine.refreshFederation() (#34450)
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 61444f8e08d Refactor MetaDataRefreshEngine.refreshFederation() (#34450)
61444f8e08d is described below
commit 61444f8e08d587e2b8435035b054df6414de02cf
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Thu Jan 23 22:03:26 2025 +0800
Refactor MetaDataRefreshEngine.refreshFederation() (#34450)
* Move package of FederationMetaDataRefresher
* Refactor MetaDataRefreshEngine
* Refactor MetaDataRefreshEngine.refreshFederation()
* Refactor MetaDataRefreshEngine.refreshFederation()
---
.../executor/engine/DriverExecuteExecutor.java | 13 ++---
.../jdbc/DriverJDBCPushDownExecuteExecutor.java | 8 +--
.../DriverJDBCPushDownExecuteUpdateExecutor.java | 12 ++--
.../refresher/metadata/MetaDataRefreshEngine.java | 66 +++++++++-------------
.../connector/StandardDatabaseConnector.java | 17 +++---
5 files changed, 47 insertions(+), 69 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 7d7c20cbd36..4c780e46c27 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
@@ -101,23 +101,18 @@ public final class DriverExecuteExecutor {
new ExecuteQueryCallbackFactory(prepareEngine.getType()).newInstance(database, queryContext), new SQLFederationContext(false, queryContext, metaData, connection.getProcessId()));
return null != resultSet;
}
- MetaDataRefreshEngine metaDataRefreshEngine = getMetaDataRefreshEngine(database);
- if (sqlFederationEngine.enabled() && metaDataRefreshEngine.isFederation(queryContext.getSqlStatementContext())) {
- metaDataRefreshEngine.refresh(queryContext.getSqlStatementContext());
+ if (sqlFederationEngine.enabled()) {
+ new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps())
+ .refreshFederation(queryContext.getSqlStatementContext());
return true;
}
if (transactionExecutor.decide(queryContext)) {
return transactionExecutor.execute((TCLStatement) queryContext.getSqlStatementContext().getSqlStatement());
}
- ExecutionContext executionContext =
- new KernelProcessor().generateExecutionContext(queryContext, metaData.getGlobalRuleMetaData(), metaData.getProps());
+ ExecutionContext executionContext = new KernelProcessor().generateExecutionContext(queryContext, metaData.getGlobalRuleMetaData(), metaData.getProps());
return executePushDown(database, executionContext, prepareEngine, executeCallback, addCallback, replayCallback);
}
- private MetaDataRefreshEngine getMetaDataRefreshEngine(final ShardingSphereDatabase database) {
- return new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps());
- }
-
@SuppressWarnings("rawtypes")
private boolean executePushDown(final ShardingSphereDatabase database, final ExecutionContext executionContext, final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine,
final StatementExecuteCallback executeCallback, final StatementAddCallback addCallback, final StatementReplayCallback replayCallback) throws SQLException {
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteExecutor.java
index 23581b492e2..171ef59380e 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteExecutor.java
@@ -104,13 +104,11 @@ public final class DriverJDBCPushDownExecuteExecutor {
processEngine.executeSQL(executionGroupContext, executionContext.getQueryContext());
List<Boolean> results = jdbcExecutor.execute(executionGroupContext,
new ExecuteCallbackFactory(prepareEngine.getType()).newInstance(database, executeCallback, executionContext.getSqlStatementContext().getSqlStatement()));
- if (isNeedImplicitCommit(executionContext.getQueryContext().getSqlStatementContext())) {
+ if (isNeedImplicitCommit(executionContext.getSqlStatementContext())) {
connection.commit();
}
- if (MetaDataRefreshEngine.isRefreshMetaDataRequired(executionContext.getSqlStatementContext())) {
- new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps())
- .refresh(executionContext.getQueryContext().getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
- }
+ new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, metaData.getProps())
+ .refresh(executionContext.getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
return null != results && !results.isEmpty() && null != results.get(0) && results.get(0);
} finally {
processEngine.completeSQLExecution(executionGroupContext.getReportContext().getProcessId());
diff --git a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteUpdateExecutor.java b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteUpdateExecutor.java
index cd87cca94ca..db4d9671ed3 100644
--- a/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteUpdateExecutor.java
+++ b/jdbc/src/main/java/org/apache/shardingsphere/driver/executor/engine/pushdown/jdbc/DriverJDBCPushDownExecuteUpdateExecutor.java
@@ -106,16 +106,16 @@ public final class DriverJDBCPushDownExecuteUpdateExecutor {
try {
processEngine.executeSQL(executionGroupContext, executionContext.getQueryContext());
JDBCExecutorCallback<Integer> callback = new ExecuteUpdateCallbackFactory(prepareEngine.getType())
- .newInstance(database, executionContext.getQueryContext().getSqlStatementContext().getSqlStatement(), updateCallback);
+ .newInstance(database, executionContext.getSqlStatementContext().getSqlStatement(), updateCallback);
List<Integer> updateCounts = jdbcExecutor.execute(executionGroupContext, callback);
- if (MetaDataRefreshEngine.isRefreshMetaDataRequired(executionContext.getQueryContext().getSqlStatementContext())) {
- if (isNeedImplicitCommit(executionContext.getQueryContext().getSqlStatementContext())) {
+ MetaDataRefreshEngine metaDataRefreshEngine = new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, props);
+ if (metaDataRefreshEngine.isNeedRefreshMetaData(executionContext.getSqlStatementContext())) {
+ if (isNeedImplicitCommit(executionContext.getSqlStatementContext())) {
connection.commit();
}
- new MetaDataRefreshEngine(connection.getContextManager().getPersistServiceFacade().getMetaDataManagerPersistService(), database, props)
- .refresh(executionContext.getQueryContext().getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
+ metaDataRefreshEngine.refresh(executionContext.getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
}
- return isNeedAccumulate(database.getRuleMetaData().getRules(), executionContext.getQueryContext().getSqlStatementContext()) ? accumulate(updateCounts) : updateCounts.get(0);
+ return isNeedAccumulate(database.getRuleMetaData().getRules(), executionContext.getSqlStatementContext()) ? accumulate(updateCounts) : updateCounts.get(0);
} finally {
processEngine.completeSQLExecution(executionGroupContext.getReportContext().getProcessId());
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/refresher/metadata/MetaDataRefreshEngine.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/refresher/metadata/MetaDataRefreshEngine.java
index 4a75931c60c..30bde2fee73 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/refresher/metadata/MetaDataRefreshEngine.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/refresher/metadata/MetaDataRefreshEngine.java
@@ -67,6 +67,17 @@ public final class MetaDataRefreshEngine {
private final ConfigurationProperties props;
+ /**
+ * Whether to need refresh meta data.
+ *
+ * @param sqlStatementContext SQL statement context
+ * @return is need refresh meta data or not
+ */
+ public boolean isNeedRefreshMetaData(final SQLStatementContext sqlStatementContext) {
+ Class<?> sqlStatementClass = sqlStatementContext.getSqlStatement().getClass().getSuperclass();
+ return DDL_STATEMENT_CLASSES.contains(sqlStatementClass);
+ }
+
/**
* Refresh meta data.
*
@@ -76,29 +87,19 @@ public final class MetaDataRefreshEngine {
*/
@SuppressWarnings({"unchecked", "rawtypes"})
public void refresh(final SQLStatementContext sqlStatementContext, final Collection<RouteUnit> routeUnits) throws SQLException {
- Class sqlStatementClass = sqlStatementContext.getSqlStatement().getClass().getSuperclass();
- if (!DDL_STATEMENT_CLASSES.contains(sqlStatementClass)) {
+ if (!isNeedRefreshMetaData(sqlStatementContext)) {
return;
}
- Optional<MetaDataRefresher> schemaRefresher = TypedSPILoader.findService(MetaDataRefresher.class, sqlStatementClass);
- if (schemaRefresher.isPresent()) {
- Collection<String> logicDataSourceNames = routeUnits.stream().map(each -> each.getDataSourceMapper().getLogicName()).collect(Collectors.toList());
- String schemaName = sqlStatementContext instanceof TableAvailable ? getSchemaName(sqlStatementContext) : null;
- DatabaseType databaseType = routeUnits.stream().map(each -> database.getResourceMetaData().getStorageUnits().get(each.getDataSourceMapper().getActualName()))
- .filter(Objects::nonNull).findFirst().map(StorageUnit::getStorageType).orElseGet(sqlStatementContext::getDatabaseType);
- schemaRefresher.get().refresh(metaDataManagerPersistService, database, logicDataSourceNames, schemaName, databaseType, sqlStatementContext.getSqlStatement(), props);
+ Class<?> sqlStatementClass = sqlStatementContext.getSqlStatement().getClass().getSuperclass();
+ Optional<MetaDataRefresher> metaDataRefresher = TypedSPILoader.findService(MetaDataRefresher.class, sqlStatementClass);
+ if (!metaDataRefresher.isPresent()) {
+ return;
}
- }
-
- /**
- * Refresh meta data for federation.
- *
- * @param sqlStatementContext SQL statement context
- */
- @SuppressWarnings("unchecked")
- public void refresh(final SQLStatementContext sqlStatementContext) {
- getFederationMetaDataRefresher(sqlStatementContext).ifPresent(
- optional -> optional.refresh(metaDataManagerPersistService, database, getSchemaName(sqlStatementContext), sqlStatementContext.getSqlStatement()));
+ Collection<String> logicDataSourceNames = routeUnits.stream().map(each -> each.getDataSourceMapper().getLogicName()).collect(Collectors.toList());
+ String schemaName = sqlStatementContext instanceof TableAvailable ? getSchemaName(sqlStatementContext) : null;
+ DatabaseType databaseType = routeUnits.stream().map(each -> database.getResourceMetaData().getStorageUnits().get(each.getDataSourceMapper().getActualName()))
+ .filter(Objects::nonNull).findFirst().map(StorageUnit::getStorageType).orElseGet(sqlStatementContext::getDatabaseType);
+ metaDataRefresher.get().refresh(metaDataManagerPersistService, database, logicDataSourceNames, schemaName, databaseType, sqlStatementContext.getSqlStatement(), props);
}
private String getSchemaName(final SQLStatementContext sqlStatementContext) {
@@ -107,27 +108,14 @@ public final class MetaDataRefreshEngine {
}
/**
- * SQL statement is federation or not.
- *
- * @param sqlStatementContext SQL statement context
- * @return is federation or not
- */
- public boolean isFederation(final SQLStatementContext sqlStatementContext) {
- return getFederationMetaDataRefresher(sqlStatementContext).isPresent();
- }
-
- @SuppressWarnings("rawtypes")
- private Optional<FederationMetaDataRefresher> getFederationMetaDataRefresher(final SQLStatementContext sqlStatementContext) {
- return TypedSPILoader.findService(FederationMetaDataRefresher.class, sqlStatementContext.getSqlStatement().getClass().getSuperclass());
- }
-
- /**
- * Is refresh meta data required.
+ * Refresh meta data for federation.
*
* @param sqlStatementContext SQL statement context
- * @return is refresh meta data required or not
*/
- public static boolean isRefreshMetaDataRequired(final SQLStatementContext sqlStatementContext) {
- return DDL_STATEMENT_CLASSES.contains(sqlStatementContext.getSqlStatement().getClass().getSuperclass());
+ @SuppressWarnings("unchecked")
+ public void refreshFederation(final SQLStatementContext sqlStatementContext) {
+ Class<?> sqlStatementClass = sqlStatementContext.getSqlStatement().getClass().getSuperclass();
+ TypedSPILoader.findService(FederationMetaDataRefresher.class, sqlStatementClass).ifPresent(
+ optional -> optional.refresh(metaDataManagerPersistService, database, getSchemaName(sqlStatementContext), sqlStatementContext.getSqlStatement()));
}
}
diff --git a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseConnector.java b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseConnector.java
index 1fa6068d1c5..08fa0840757 100644
--- a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseConnector.java
+++ b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseConnector.java
@@ -110,6 +110,8 @@ public final class StandardDatabaseConnector implements DatabaseConnector {
private final ProxySQLExecutor proxySQLExecutor;
+ private final MetaDataRefreshEngine metaDataRefreshEngine;
+
private final Collection<Statement> cachedStatements = Collections.newSetFromMap(new ConcurrentHashMap<>());
private final Collection<ResultSet> cachedResultSets = Collections.newSetFromMap(new ConcurrentHashMap<>());
@@ -131,6 +133,8 @@ public final class StandardDatabaseConnector implements DatabaseConnector {
prepareCursorStatementContext((CursorAvailable) sqlStatementContext);
}
proxySQLExecutor = new ProxySQLExecutor(driverType, databaseConnectionManager, this, queryContext);
+ metaDataRefreshEngine = new MetaDataRefreshEngine(
+ contextManager.getPersistServiceFacade().getMetaDataManagerPersistService(), database, contextManager.getMetaDataContexts().getMetaData().getProps());
}
private void checkBackendReady(final SQLStatementContext sqlStatementContext) {
@@ -179,9 +183,8 @@ public final class StandardDatabaseConnector implements DatabaseConnector {
if (proxySQLExecutor.getSqlFederationEngine().decide(queryContext, contextManager.getMetaDataContexts().getMetaData().getGlobalRuleMetaData())) {
return processExecuteFederation(doExecuteFederation());
}
- MetaDataRefreshEngine metaDataRefreshEngine = getMetaDataRefreshEngine();
- if (proxySQLExecutor.getSqlFederationEngine().enabled() && metaDataRefreshEngine.isFederation(queryContext.getSqlStatementContext())) {
- metaDataRefreshEngine.refresh(queryContext.getSqlStatementContext());
+ if (proxySQLExecutor.getSqlFederationEngine().enabled()) {
+ metaDataRefreshEngine.refreshFederation(queryContext.getSqlStatementContext());
return new UpdateResponseHeader(queryContext.getSqlStatementContext().getSqlStatement());
}
ExecutionContext executionContext = generateExecutionContext();
@@ -233,9 +236,7 @@ public final class StandardDatabaseConnector implements DatabaseConnector {
List<ExecuteResult> executeResults = advancedExecutors.isEmpty()
? proxySQLExecutor.execute(executionContext)
: advancedExecutors.iterator().next().execute(executionContext, contextManager, database, this);
- if (MetaDataRefreshEngine.isRefreshMetaDataRequired(queryContext.getSqlStatementContext())) {
- getMetaDataRefreshEngine().refresh(queryContext.getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
- }
+ metaDataRefreshEngine.refresh(queryContext.getSqlStatementContext(), executionContext.getRouteContext().getRouteUnits());
Object executeResultSample = executeResults.iterator().next();
return executeResultSample instanceof QueryResult
? processExecuteQuery(queryContext.getSqlStatementContext(), executeResults.stream().map(QueryResult.class::cast).collect(Collectors.toList()), (QueryResult) executeResultSample)
@@ -271,10 +272,6 @@ public final class StandardDatabaseConnector implements DatabaseConnector {
return new QueryResponseHeader(queryHeaders);
}
- private MetaDataRefreshEngine getMetaDataRefreshEngine() {
- return new MetaDataRefreshEngine(contextManager.getPersistServiceFacade().getMetaDataManagerPersistService(), database, contextManager.getMetaDataContexts().getMetaData().getProps());
- }
-
private QueryResponseHeader processExecuteQuery(final SQLStatementContext sqlStatementContext, final List<QueryResult> queryResults, final QueryResult queryResultSample) throws SQLException {
queryHeaders = createQueryHeaders(sqlStatementContext, queryResultSample);
mergedResult = mergeQuery(sqlStatementContext, queryResults);