Commit ec1da2fb authored by terrymanu's avatar terrymanu
Browse files

add ProxyStatementExecuteUnit

parent 8cacdd88

sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecuteUnit.java→sharding-core/src/main/java/io/shardingsphere/core/executor/StatementExecuteUnit.java

+0 −0

File moved.

View file

sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/execute/ProxyStatementExecuteUnit.java

0 → 100644
+39 −0
Original line number Diff line number Diff line 
/*

 * Copyright 2016-2018 shardingsphere.io.

 * <p>

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this file except in compliance with the License.

 * You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 * </p>

 */



package io.shardingsphere.proxy.backend.jdbc.execute;



import io.shardingsphere.core.executor.StatementExecuteUnit;

import io.shardingsphere.core.routing.SQLExecutionUnit;

import lombok.Getter;

import lombok.RequiredArgsConstructor;



import java.sql.Statement;



/**

 * Statement execute unit for Proxy.

 *

 * @author zhangliang

 */

@RequiredArgsConstructor

@Getter

public final class ProxyStatementExecuteUnit implements StatementExecuteUnit {

    

    private final SQLExecutionUnit sqlExecutionUnit;

    

    private final Statement statement;

}

sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/execute/memory/ConnectionStrictlyExecuteEngine.java

+16 −15
Original line number Diff line number Diff line
@@ -18,12 +18,15 @@ 
package io.shardingsphere.proxy.backend.jdbc.execute.memory;



import io.shardingsphere.core.executor.ShardingGroupExecuteCallback;

import io.shardingsphere.core.executor.StatementExecuteUnit;

import io.shardingsphere.core.merger.QueryResult;

import io.shardingsphere.core.routing.SQLExecutionUnit;

import io.shardingsphere.core.routing.SQLRouteResult;

import io.shardingsphere.core.routing.SQLUnit;

import io.shardingsphere.proxy.backend.BackendExecutorContext;

import io.shardingsphere.proxy.backend.jdbc.connection.BackendConnection;

import io.shardingsphere.proxy.backend.jdbc.execute.JDBCExecuteEngine;

import io.shardingsphere.proxy.backend.jdbc.execute.ProxyStatementExecuteUnit;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteQueryResponse;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteResponse;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteUpdateResponse;
@@ -57,20 +60,20 @@ public final class ConnectionStrictlyExecuteEngine extends JDBCExecuteEngine { 
    @Override

    public ExecuteResponse execute(final SQLRouteResult routeResult, final boolean isReturnGeneratedKeys) throws SQLException {

        Map<String, Collection<SQLUnit>> sqlUnitGroups = routeResult.getSQLUnitGroups();

        Map<String, Map<SQLUnit, Statement>> sqlUnitStatements = new HashMap<>(sqlUnitGroups.size(), 1);

        Map<String, Collection<StatementExecuteUnit>> sqlUnitStatements = new HashMap<>(sqlUnitGroups.size(), 1);

        for (Entry<String, Collection<SQLUnit>> entry : sqlUnitGroups.entrySet()) {

            sqlUnitStatements.put(entry.getKey(), createSQLUnitStatement(entry.getKey(), entry.getValue(), isReturnGeneratedKeys));

        }

        Collection<ExecuteResponseUnit> executeResponseUnits = BackendExecutorContext.getInstance().getExecuteEngine().groupExecute(

                sqlUnitGroups, new FirstTransactionGroupExecuteCallback(isReturnGeneratedKeys), new TransactionGroupExecuteCallback(isReturnGeneratedKeys, sqlUnitStatements));

                sqlUnitStatements, new FirstTransactionGroupExecuteCallback(isReturnGeneratedKeys), new TransactionGroupExecuteCallback(isReturnGeneratedKeys));

        return getExecuteQueryResponse(executeResponseUnits);

    }

    

    private Map<SQLUnit, Statement> createSQLUnitStatement(final String dataSourceName, final Collection<SQLUnit> sqlUnits, final boolean isReturnGeneratedKeys) throws SQLException {

        Map<SQLUnit, Statement> result = new HashMap<>(sqlUnits.size(), 1);

    private Collection<StatementExecuteUnit> createSQLUnitStatement(final String dataSourceName, final Collection<SQLUnit> sqlUnits, final boolean isReturnGeneratedKeys) throws SQLException {

        Collection<StatementExecuteUnit> result = new LinkedList<>();

        Connection connection = getBackendConnection().getConnection(dataSourceName);

        for (SQLUnit each : sqlUnits) {

            result.put(each, getJdbcExecutorWrapper().createStatement(connection, each.getSql(), isReturnGeneratedKeys));

            result.add(new ProxyStatementExecuteUnit(new SQLExecutionUnit(dataSourceName, each), getJdbcExecutorWrapper().createStatement(connection, each.getSql(), isReturnGeneratedKeys)));

        }

        return result;

    }
@@ -103,17 +106,17 @@ public final class ConnectionStrictlyExecuteEngine extends JDBCExecuteEngine { 
    }

    

    @RequiredArgsConstructor

    class FirstTransactionGroupExecuteCallback implements ShardingGroupExecuteCallback<SQLUnit, ExecuteResponseUnit> {

    class FirstTransactionGroupExecuteCallback implements ShardingGroupExecuteCallback<StatementExecuteUnit, ExecuteResponseUnit> {

        

        private final boolean isReturnGeneratedKeys;

        

        @Override

        public Collection<ExecuteResponseUnit> execute(final String dataSourceName, final Collection<SQLUnit> sqlUnits) throws SQLException {

        public Collection<ExecuteResponseUnit> execute(final String dataSourceName, final Collection<StatementExecuteUnit> statementExecuteUnits) throws SQLException {

            Collection<ExecuteResponseUnit> result = new LinkedList<>();

            boolean hasMetaData = false;

            Connection connection = getBackendConnection().getConnection(dataSourceName);

            for (SQLUnit each : sqlUnits) {

                String actualSQL = each.getSql();

            for (StatementExecuteUnit each : statementExecuteUnits) {

                String actualSQL = each.getSqlExecutionUnit().getSqlUnit().getSql();

                Statement statement = getJdbcExecutorWrapper().createStatement(connection, actualSQL, isReturnGeneratedKeys);

                ExecuteResponseUnit response;

                if (hasMetaData) {
@@ -129,17 +132,15 @@ public final class ConnectionStrictlyExecuteEngine extends JDBCExecuteEngine { 
    }

    

    @RequiredArgsConstructor

    class TransactionGroupExecuteCallback implements ShardingGroupExecuteCallback<SQLUnit, ExecuteResponseUnit> {

    class TransactionGroupExecuteCallback implements ShardingGroupExecuteCallback<StatementExecuteUnit, ExecuteResponseUnit> {

    

        private final boolean isReturnGeneratedKeys;

    

        private final Map<String, Map<SQLUnit, Statement>> sqlUnitStatements;

        

        @Override

        public Collection<ExecuteResponseUnit> execute(final String dataSourceName, final Collection<SQLUnit> sqlUnits) throws SQLException {

        public Collection<ExecuteResponseUnit> execute(final String dataSourceName, final Collection<StatementExecuteUnit> statementExecuteUnits) throws SQLException {

            Collection<ExecuteResponseUnit> result = new LinkedList<>();

            for (Entry<SQLUnit, Statement> each : sqlUnitStatements.get(dataSourceName).entrySet()) {

                result.add(executeWithoutMetadata(each.getValue(), each.getKey().getSql(), isReturnGeneratedKeys));

            for (StatementExecuteUnit each : statementExecuteUnits) {

                result.add(executeWithoutMetadata(each.getStatement(), each.getSqlExecutionUnit().getSqlUnit().getSql(), isReturnGeneratedKeys));

            }

            return result;

        }

sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/execute/stream/MemoryStrictlyExecuteEngine.java

+16 −16
Original line number Diff line number Diff line
@@ -18,12 +18,14 @@ 
package io.shardingsphere.proxy.backend.jdbc.execute.stream;



import io.shardingsphere.core.executor.ShardingExecuteCallback;

import io.shardingsphere.core.executor.StatementExecuteUnit;

import io.shardingsphere.core.merger.QueryResult;

import io.shardingsphere.core.routing.SQLExecutionUnit;

import io.shardingsphere.core.routing.SQLRouteResult;

import io.shardingsphere.proxy.backend.BackendExecutorContext;

import io.shardingsphere.proxy.backend.jdbc.connection.BackendConnection;

import io.shardingsphere.proxy.backend.jdbc.execute.JDBCExecuteEngine;

import io.shardingsphere.proxy.backend.jdbc.execute.ProxyStatementExecuteUnit;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteQueryResponse;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteResponse;

import io.shardingsphere.proxy.backend.jdbc.execute.response.ExecuteUpdateResponse;
@@ -36,8 +38,7 @@ import java.sql.ResultSet; 
import java.sql.SQLException;

import java.sql.Statement;

import java.util.Collection;

import java.util.HashMap;

import java.util.Map;

import java.util.LinkedList;



/**

 * Memory strictly execute engine.
@@ -55,12 +56,13 @@ public final class MemoryStrictlyExecuteEngine extends JDBCExecuteEngine { 
    

    @Override

    public ExecuteResponse execute(final SQLRouteResult routeResult, final boolean isReturnGeneratedKeys) throws SQLException {

        Map<SQLExecutionUnit, Statement> statements = new HashMap<>(routeResult.getExecutionUnits().size(), 1);

        Collection<StatementExecuteUnit> statementExecuteUnits = new LinkedList<>();

        for (SQLExecutionUnit each : routeResult.getExecutionUnits()) {

            statements.put(each, getJdbcExecutorWrapper().createStatement(getBackendConnection().getConnection(each.getDataSource()), each.getSqlUnit().getSql(), isReturnGeneratedKeys));

            Statement statement = getJdbcExecutorWrapper().createStatement(getBackendConnection().getConnection(each.getDataSource()), each.getSqlUnit().getSql(), isReturnGeneratedKeys);

            statementExecuteUnits.add(new ProxyStatementExecuteUnit(each, statement));

        }

        Collection<ExecuteResponseUnit> executeResponseUnits = BackendExecutorContext.getInstance().getExecuteEngine().execute(

                routeResult.getExecutionUnits(), new FirstTransactionExecuteCallback(isReturnGeneratedKeys), new TransactionExecuteCallback(isReturnGeneratedKeys, statements));

                statementExecuteUnits, new FirstTransactionExecuteCallback(isReturnGeneratedKeys), new TransactionExecuteCallback(isReturnGeneratedKeys));

        ExecuteResponseUnit firstExecuteResponseUnit = executeResponseUnits.iterator().next();

        return firstExecuteResponseUnit instanceof ExecuteQueryResponseUnit

                ? getExecuteQueryResponse((ExecuteQueryResponseUnit) firstExecuteResponseUnit, executeResponseUnits) : getExecuteUpdateResponse(executeResponseUnits);
@@ -89,29 +91,27 @@ public final class MemoryStrictlyExecuteEngine extends JDBCExecuteEngine { 
    }

    

    @RequiredArgsConstructor

    class FirstTransactionExecuteCallback implements ShardingExecuteCallback<SQLExecutionUnit, ExecuteResponseUnit> {

    class FirstTransactionExecuteCallback implements ShardingExecuteCallback<StatementExecuteUnit, ExecuteResponseUnit> {

        

        private final boolean isReturnGeneratedKeys;

        

        @Override

        public ExecuteResponseUnit execute(final SQLExecutionUnit sqlExecutionUnit) throws SQLException {

            Statement statement = getJdbcExecutorWrapper().createStatement(

                    getBackendConnection().getConnection(sqlExecutionUnit.getDataSource()), sqlExecutionUnit.getSqlUnit().getSql(), isReturnGeneratedKeys);

            return executeWithMetadata(statement, sqlExecutionUnit.getSqlUnit().getSql(), isReturnGeneratedKeys);

        public ExecuteResponseUnit execute(final StatementExecuteUnit statementExecuteUnit) throws SQLException {

            Statement statement = getJdbcExecutorWrapper().createStatement(getBackendConnection().getConnection(statementExecuteUnit.getSqlExecutionUnit().getDataSource()), 

                    statementExecuteUnit.getSqlExecutionUnit().getSqlUnit().getSql(), isReturnGeneratedKeys);

            return executeWithMetadata(statement, statementExecuteUnit.getSqlExecutionUnit().getSqlUnit().getSql(), isReturnGeneratedKeys);

        }

    }

    

    @RequiredArgsConstructor

    class TransactionExecuteCallback implements ShardingExecuteCallback<SQLExecutionUnit, ExecuteResponseUnit> {

    class TransactionExecuteCallback implements ShardingExecuteCallback<StatementExecuteUnit, ExecuteResponseUnit> {

        

        private final boolean isReturnGeneratedKeys;

        

        private final Map<SQLExecutionUnit, Statement> statements;

        

        @Override

        public ExecuteResponseUnit execute(final SQLExecutionUnit sqlExecutionUnit) throws SQLException {

            String actualSQL = sqlExecutionUnit.getSqlUnit().getSql();

            return executeWithoutMetadata(statements.get(sqlExecutionUnit), actualSQL, isReturnGeneratedKeys);

        public ExecuteResponseUnit execute(final StatementExecuteUnit statementExecuteUnit) throws SQLException {

            String actualSQL = statementExecuteUnit.getSqlExecutionUnit().getSqlUnit().getSql();

            return executeWithoutMetadata(statementExecuteUnit.getStatement(), actualSQL, isReturnGeneratedKeys);

        }

    }

}