This is an automated email from the ASF dual-hosted git repository.
panjuan 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 7ba3f1d optimize calcite parameter pass (#10731)
7ba3f1d is described below
commit 7ba3f1dd7ed39dce8ae00521f9a9b484e359744a
Author: Zhengqiang Duan <[email protected]>
AuthorDate: Wed Jun 9 17:24:02 2021 +0800
optimize calcite parameter pass (#10731)
---
.../sql/federate/execute/FederateExecutor.java | 10 ++++---
.../sql/federate/execute/FederateJDBCExecutor.java | 33 ++++++++--------------
.../federate/execute/raw/FederateRawExecutor.java | 6 ++--
.../federate/schema/row/FederateRowExecutor.java | 23 ++-------------
.../statement/ShardingSpherePreparedStatement.java | 15 ++++++----
.../core/statement/ShardingSphereStatement.java | 15 ++++++----
.../backend/communication/ProxySQLExecutor.java | 14 ++++++---
7 files changed, 52 insertions(+), 64 deletions(-)
diff --git
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateExecutor.java
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateExecutor.java
index f646d34..6991346 100644
---
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateExecutor.java
+++
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateExecutor.java
@@ -18,11 +18,13 @@
package org.apache.shardingsphere.infra.executor.sql.federate.execute;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
+import
org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutionUnit;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutorCallback;
import
org.apache.shardingsphere.infra.executor.sql.execute.result.ExecuteResult;
import
org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryResult;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption;
+import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine;
+import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
@@ -37,12 +39,12 @@ public interface FederateExecutor {
*
* @param executionContext execution context
* @param callback callback
- * @param type JDBC driver type
- * @param statementOption statement option
+ * @param prepareEngine prepare engine
* @return execute result
* @throws SQLException SQL exception
*/
- List<QueryResult> executeQuery(ExecutionContext executionContext,
JDBCExecutorCallback<? extends ExecuteResult> callback, String type,
StatementOption statementOption) throws SQLException;
+ List<QueryResult> executeQuery(ExecutionContext executionContext,
JDBCExecutorCallback<? extends ExecuteResult> callback,
+
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException;
/**
* Close.
diff --git
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateJDBCExecutor.java
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateJDBCExecutor.java
index 84eed4d..9905ce3 100644
---
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateJDBCExecutor.java
+++
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/FederateJDBCExecutor.java
@@ -22,6 +22,7 @@ import org.apache.calcite.jdbc.CalciteConnection;
import
org.apache.shardingsphere.infra.config.properties.ConfigurationProperties;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
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.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.result.ExecuteResult;
@@ -29,10 +30,8 @@ import
org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryRe
import
org.apache.shardingsphere.infra.executor.sql.execute.result.query.impl.driver.jdbc.type.stream.JDBCStreamQueryResult;
import
org.apache.shardingsphere.infra.executor.sql.federate.schema.FederateLogicSchema;
import
org.apache.shardingsphere.infra.executor.sql.federate.schema.row.FederateRowExecutor;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.ExecutorJDBCManager;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption;
+import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine;
import org.apache.shardingsphere.infra.optimize.context.OptimizeContextFactory;
-import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
import org.apache.shardingsphere.sql.parser.sql.common.util.SQLUtil;
import java.sql.Connection;
@@ -41,7 +40,6 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
-import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -59,12 +57,8 @@ public final class FederateJDBCExecutor implements
FederateExecutor {
private final OptimizeContextFactory factory;
- private final Collection<ShardingSphereRule> rules;
-
private final ConfigurationProperties props;
- private final ExecutorJDBCManager jdbcManager;
-
private final JDBCExecutor jdbcExecutor;
private Statement statement;
@@ -77,20 +71,17 @@ public final class FederateJDBCExecutor implements
FederateExecutor {
}
}
- public FederateJDBCExecutor(final String schema, final
OptimizeContextFactory factory, final Collection<ShardingSphereRule> rules,
- final ConfigurationProperties props, final
ExecutorJDBCManager jdbcManager, final JDBCExecutor jdbcExecutor) {
+ public FederateJDBCExecutor(final String schema, final
OptimizeContextFactory factory, final ConfigurationProperties props, final
JDBCExecutor jdbcExecutor) {
this.schema = schema;
this.factory = factory;
- this.rules = rules;
this.props = props;
- this.jdbcManager = jdbcManager;
this.jdbcExecutor = jdbcExecutor;
}
@Override
public List<QueryResult> executeQuery(final ExecutionContext
executionContext, final JDBCExecutorCallback<? extends ExecuteResult> callback,
- final String type, final
StatementOption statementOption) throws SQLException {
- QueryResult result = new
JDBCStreamQueryResult(execute(executionContext, callback, type,
statementOption));
+ final
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException {
+ QueryResult result = new
JDBCStreamQueryResult(execute(executionContext, callback, prepareEngine));
return Collections.singletonList(result);
}
@@ -109,19 +100,19 @@ public final class FederateJDBCExecutor implements
FederateExecutor {
}
private ResultSet execute(final ExecutionContext executionContext, final
JDBCExecutorCallback<? extends ExecuteResult> callback,
- final String type, final StatementOption
statementOption) throws SQLException {
+ final
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException {
SQLUnit sqlUnit =
executionContext.getExecutionUnits().iterator().next().getSqlUnit();
- PreparedStatement statement = getConnection(executionContext,
callback, type,
statementOption).prepareStatement(SQLUtil.trimSemicolon(sqlUnit.getSql()));
+ PreparedStatement statement = getConnection(executionContext,
callback,
prepareEngine).prepareStatement(SQLUtil.trimSemicolon(sqlUnit.getSql()));
setParameters(statement, sqlUnit.getParameters());
this.statement = statement;
return statement.executeQuery();
}
private Connection getConnection(final ExecutionContext executionContext,
final JDBCExecutorCallback<? extends ExecuteResult> callback,
- final String type, final StatementOption
statementOption) throws SQLException {
+ final
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException {
Connection result = DriverManager.getConnection(CONNECTION_URL,
getProperties());
CalciteConnection calciteConnection =
result.unwrap(CalciteConnection.class);
- addSchema(calciteConnection, executionContext, callback, type,
statementOption);
+ addSchema(calciteConnection, executionContext, callback,
prepareEngine);
return result;
}
@@ -132,9 +123,9 @@ public final class FederateJDBCExecutor implements
FederateExecutor {
return result;
}
- private void addSchema(final CalciteConnection calciteConnection, final
ExecutionContext executionContext,
- final JDBCExecutorCallback<? extends ExecuteResult>
callback, final String type, final StatementOption statementOption) throws
SQLException {
- FederateRowExecutor executor = new FederateRowExecutor(rules, props,
jdbcManager, jdbcExecutor, executionContext, callback, type, statementOption);
+ private void addSchema(final CalciteConnection calciteConnection, final
ExecutionContext executionContext, final JDBCExecutorCallback<? extends
ExecuteResult> callback,
+ final
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException {
+ FederateRowExecutor executor = new FederateRowExecutor(props,
jdbcExecutor, executionContext, callback, prepareEngine);
FederateLogicSchema logicSchema = new
FederateLogicSchema(factory.getSchemaMetadatas().getSchemaMetadataBySchemaName(schema),
executor);
calciteConnection.getRootSchema().add(schema, logicSchema);
calciteConnection.setSchema(schema);
diff --git
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/raw/FederateRawExecutor.java
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/raw/FederateRawExecutor.java
index 33721cd..690db51 100644
---
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/raw/FederateRawExecutor.java
+++
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/execute/raw/FederateRawExecutor.java
@@ -25,14 +25,16 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
+import
org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutionUnit;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutorCallback;
import
org.apache.shardingsphere.infra.executor.sql.execute.result.ExecuteResult;
import
org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryResult;
import
org.apache.shardingsphere.infra.executor.sql.federate.execute.FederateExecutor;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption;
+import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine;
import org.apache.shardingsphere.infra.optimize.ShardingSphereOptimizer;
import org.apache.shardingsphere.infra.optimize.context.OptimizeContext;
+import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
@@ -52,7 +54,7 @@ public final class FederateRawExecutor implements
FederateExecutor {
@Override
public List<QueryResult> executeQuery(final ExecutionContext
executionContext, final JDBCExecutorCallback<? extends ExecuteResult> callback,
- final String type, final
StatementOption statementOption) throws SQLException {
+ final
DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection> prepareEngine)
throws SQLException {
// TODO
return Collections.emptyList();
}
diff --git
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/schema/row/FederateRowExecutor.java
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/schema/row/FederateRowExecutor.java
index efd50ef..81f3942 100644
---
a/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/schema/row/FederateRowExecutor.java
+++
b/shardingsphere-infra/shardingsphere-infra-executor/src/main/java/org/apache/shardingsphere/infra/executor/sql/federate/schema/row/FederateRowExecutor.java
@@ -21,11 +21,9 @@ import lombok.RequiredArgsConstructor;
import org.apache.calcite.DataContext;
import org.apache.calcite.rex.RexNode;
import
org.apache.shardingsphere.infra.config.properties.ConfigurationProperties;
-import
org.apache.shardingsphere.infra.config.properties.ConfigurationPropertyKey;
import org.apache.shardingsphere.infra.exception.ShardingSphereException;
import
org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroupContext;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
-import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
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;
@@ -34,10 +32,7 @@ import
org.apache.shardingsphere.infra.executor.sql.execute.result.query.QueryRe
import
org.apache.shardingsphere.infra.executor.sql.federate.schema.table.generator.FederateExecutionContextGenerator;
import
org.apache.shardingsphere.infra.executor.sql.federate.schema.table.generator.FederateExecutionSQLGenerator;
import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.ExecutorJDBCManager;
-import
org.apache.shardingsphere.infra.executor.sql.prepare.driver.jdbc.StatementOption;
import
org.apache.shardingsphere.infra.executor.sql.process.ExecuteProcessEngine;
-import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
import java.sql.Connection;
import java.sql.SQLException;
@@ -51,22 +46,15 @@ import java.util.stream.Collectors;
@RequiredArgsConstructor
public final class FederateRowExecutor {
- // TODO Consider use emptyList
- private final Collection<ShardingSphereRule> rules;
-
private final ConfigurationProperties props;
- private final ExecutorJDBCManager jdbcManager;
-
private final JDBCExecutor jdbcExecutor;
private final ExecutionContext routeExecutionContext;
private final JDBCExecutorCallback<? extends ExecuteResult> callback;
- private final String type;
-
- private final StatementOption statementOption;
+ private final DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine;
/**
* Execute.
@@ -84,7 +72,7 @@ public final class FederateRowExecutor {
private Collection<QueryResult> execute(final ExecutionContext context) {
try {
- ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext =
createExecutionGroupContext(context);
+ ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext =
prepareEngine.prepare(context.getRouteContext(), context.getExecutionUnits());
ExecuteProcessEngine.initialize(context.getSqlStatementContext(),
executionGroupContext, props);
Collection<QueryResult> result =
jdbcExecutor.execute(executionGroupContext, callback).stream().map(each ->
(QueryResult) each).collect(Collectors.toList());
ExecuteProcessEngine.finish(executionGroupContext.getExecutionID());
@@ -95,11 +83,4 @@ public final class FederateRowExecutor {
ExecuteProcessEngine.clean();
}
}
-
- private ExecutionGroupContext<JDBCExecutionUnit>
createExecutionGroupContext(final ExecutionContext executionContext) throws
SQLException {
- int maxConnectionsSizePerQuery =
props.getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
- Collection<ExecutionUnit> executionUnits =
executionContext.getExecutionUnits();
- DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = new DriverExecutionPrepareEngine<>(type,
maxConnectionsSizePerQuery, jdbcManager, statementOption, rules);
- return prepareEngine.prepare(executionContext.getRouteContext(),
executionUnits);
- }
}
diff --git
a/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java
b/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java
index ad98836..12d0f8b 100644
---
a/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java
+++
b/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSpherePreparedStatement.java
@@ -165,8 +165,7 @@ public final class ShardingSpherePreparedStatement extends
AbstractPreparedState
driverJDBCExecutor = new DriverJDBCExecutor(metaDataContexts,
jdbcExecutor);
rawExecutor = new RawExecutor(metaDataContexts.getExecutorEngine(),
connection.isHoldTransaction(), metaDataContexts.getProps());
// TODO Consider FederateRawExecutor
- federateExecutor = new FederateJDBCExecutor(DefaultSchema.LOGIC_NAME,
metaDataContexts.getOptimizeContextFactory(),
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules(),
- metaDataContexts.getProps(), connection, jdbcExecutor);
+ federateExecutor = new FederateJDBCExecutor(DefaultSchema.LOGIC_NAME,
metaDataContexts.getOptimizeContextFactory(), metaDataContexts.getProps(),
jdbcExecutor);
batchPreparedStatementExecutor = new
BatchPreparedStatementExecutor(metaDataContexts, jdbcExecutor);
kernelProcessor = new KernelProcessor();
}
@@ -214,7 +213,13 @@ public final class ShardingSpherePreparedStatement extends
AbstractPreparedState
}
PreparedStatementExecuteQueryCallback callback = new
PreparedStatementExecuteQueryCallback(metaDataContexts.getDefaultMetaData().getResource().getDatabaseType(),
sqlStatement,
SQLExecutorExceptionHandler.isExceptionThrown());
- return federateExecutor.executeQuery(executionContext, callback,
JDBCDriverType.PREPARED_STATEMENT, statementOption);
+ return federateExecutor.executeQuery(executionContext, callback,
createDriverExecutionPrepareEngine());
+ }
+
+ private DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
createDriverExecutionPrepareEngine() {
+ int maxConnectionsSizePerQuery =
metaDataContexts.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
+ return new
DriverExecutionPrepareEngine<>(JDBCDriverType.PREPARED_STATEMENT,
maxConnectionsSizePerQuery, connection,
+ statementOption,
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules());
}
@Override
@@ -285,9 +290,7 @@ public final class ShardingSpherePreparedStatement extends
AbstractPreparedState
}
private ExecutionGroupContext<JDBCExecutionUnit>
createExecutionGroupContext() throws SQLException {
- int maxConnectionsSizePerQuery =
metaDataContexts.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
- DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = new DriverExecutionPrepareEngine<>(
- JDBCDriverType.PREPARED_STATEMENT, maxConnectionsSizePerQuery,
connection, statementOption,
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules());
+ DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = createDriverExecutionPrepareEngine();
return prepareEngine.prepare(executionContext.getRouteContext(),
executionContext.getExecutionUnits());
}
diff --git
a/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java
b/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java
index 396ca52..aa14e2d 100644
---
a/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java
+++
b/shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/jdbc/core/statement/ShardingSphereStatement.java
@@ -134,8 +134,7 @@ public final class ShardingSphereStatement extends
AbstractStatementAdapter {
driverJDBCExecutor = new DriverJDBCExecutor(metaDataContexts,
jdbcExecutor);
rawExecutor = new RawExecutor(metaDataContexts.getExecutorEngine(),
connection.isHoldTransaction(), metaDataContexts.getProps());
// TODO Consider FederateRawExecutor
- federateExecutor = new FederateJDBCExecutor(DefaultSchema.LOGIC_NAME,
metaDataContexts.getOptimizeContextFactory(),
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules(),
- metaDataContexts.getProps(), connection, jdbcExecutor);
+ federateExecutor = new FederateJDBCExecutor(DefaultSchema.LOGIC_NAME,
metaDataContexts.getOptimizeContextFactory(), metaDataContexts.getProps(),
jdbcExecutor);
kernelProcessor = new KernelProcessor();
}
@@ -185,7 +184,13 @@ public final class ShardingSphereStatement extends
AbstractStatementAdapter {
}
StatementExecuteQueryCallback callback = new
StatementExecuteQueryCallback(metaDataContexts.getDefaultMetaData().getResource().getDatabaseType(),
executionContext.getSqlStatementContext().getSqlStatement(),
SQLExecutorExceptionHandler.isExceptionThrown());
- return federateExecutor.executeQuery(executionContext, callback,
JDBCDriverType.STATEMENT, statementOption);
+ return federateExecutor.executeQuery(executionContext, callback,
createDriverExecutionPrepareEngine());
+ }
+
+ private DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
createDriverExecutionPrepareEngine() {
+ int maxConnectionsSizePerQuery =
metaDataContexts.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
+ return new DriverExecutionPrepareEngine<>(JDBCDriverType.STATEMENT,
maxConnectionsSizePerQuery, connection,
+ statementOption,
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules());
}
@Override
@@ -405,9 +410,7 @@ public final class ShardingSphereStatement extends
AbstractStatementAdapter {
}
private ExecutionGroupContext<JDBCExecutionUnit> createExecutionContext()
throws SQLException {
- int maxConnectionsSizePerQuery =
metaDataContexts.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
- DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = new DriverExecutionPrepareEngine<>(
- JDBCDriverType.STATEMENT, maxConnectionsSizePerQuery,
connection, statementOption,
metaDataContexts.getDefaultMetaData().getRuleMetaData().getRules());
+ DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = createDriverExecutionPrepareEngine();
return prepareEngine.prepare(executionContext.getRouteContext(),
executionContext.getExecutionUnits());
}
diff --git
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/ProxySQLExecutor.java
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/ProxySQLExecutor.java
index 68fbf60..fba5049 100644
---
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/ProxySQLExecutor.java
+++
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/ProxySQLExecutor.java
@@ -82,9 +82,8 @@ public final class ProxySQLExecutor {
MetaDataContexts metaDataContexts =
ProxyContext.getInstance().getMetaDataContexts();
rawExecutor = new RawExecutor(executorEngine, isSerialExecute,
metaDataContexts.getProps());
// TODO Consider FederateRawExecutor
- federateExecutor = new
FederateJDBCExecutor(backendConnection.getSchemaName(),
metaDataContexts.getOptimizeContextFactory(),
-
metaDataContexts.getMetaData(backendConnection.getSchemaName()).getRuleMetaData().getRules(),
- metaDataContexts.getProps(), backendConnection, new
JDBCExecutor(executorEngine, isSerialExecute));
+ federateExecutor = new
FederateJDBCExecutor(backendConnection.getSchemaName(),
metaDataContexts.getOptimizeContextFactory(),
+ metaDataContexts.getProps(), new JDBCExecutor(executorEngine,
isSerialExecute));
}
/**
@@ -148,7 +147,14 @@ public final class ProxySQLExecutor {
ProxyJDBCExecutorCallback callback =
ProxyJDBCExecutorCallbackFactory.newInstance(type,
metaData.getMetaData(backendConnection.getSchemaName()).getResource().getDatabaseType(),
executionContext.getSqlStatementContext().getSqlStatement(),
backendConnection, isReturnGeneratedKeys, isExceptionThrown, true);
backendConnection.setFederateExecutor(federateExecutor);
- return federateExecutor.executeQuery(executionContext, callback, type,
new StatementOption(isReturnGeneratedKeys)).stream().map(each ->
(ExecuteResult) each).collect(Collectors.toList());
+ DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
prepareEngine = createDriverExecutionPrepareEngine(isReturnGeneratedKeys,
metaData);
+ return federateExecutor.executeQuery(executionContext, callback,
prepareEngine).stream().map(each -> (ExecuteResult)
each).collect(Collectors.toList());
+ }
+
+ private DriverExecutionPrepareEngine<JDBCExecutionUnit, Connection>
createDriverExecutionPrepareEngine(final boolean isReturnGeneratedKeys, final
MetaDataContexts metaData) {
+ int maxConnectionsSizePerQuery =
metaData.getProps().<Integer>getValue(ConfigurationPropertyKey.MAX_CONNECTIONS_SIZE_PER_QUERY);
+ return new DriverExecutionPrepareEngine<>(type,
maxConnectionsSizePerQuery, backendConnection, new
StatementOption(isReturnGeneratedKeys),
+
metaData.getMetaData(backendConnection.getSchemaName()).getRuleMetaData().getRules());
}
private Collection<ExecuteResult> useDriverToExecute(final
ExecutionContext executionContext, final Collection<ShardingSphereRule> rules,
