parameters, final DataSourceMetaData dataSourceMetaData) {
if (SQLType.DQL == sqlType) {
- return new DQLExecutionEvent(sqlExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
+ return new DQLExecutionEvent(statementExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
}
if (SQLType.DML == sqlType) {
- return new DMLExecutionEvent(sqlExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
+ return new DMLExecutionEvent(statementExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
}
- return new SQLExecutionEvent(sqlExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
+ return new SQLExecutionEvent(statementExecuteUnit.getRouteUnit(), parameters, dataSourceMetaData);
}
}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeFinishEvent.java b/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeFinishEvent.java
deleted file mode 100644
index 13a31b0c3173d..0000000000000
--- a/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeFinishEvent.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.event.root;
-
-/**
- * Root invoke finish event.
- *
- * @author zhangyonglun
- */
-public final class RootInvokeFinishEvent extends RootInvokeEvent {
-}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeStartEvent.java b/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeStartEvent.java
deleted file mode 100644
index 768ad90a1b63c..0000000000000
--- a/sharding-core/src/main/java/io/shardingsphere/core/event/root/RootInvokeStartEvent.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.event.root;
-
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
-/**
- * Root invoke start event.
- *
- * @author zhangyonglun
- */
-@RequiredArgsConstructor
-@Getter
-public final class RootInvokeStartEvent extends RootInvokeEvent {
-
- private final boolean parallelExecute;
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecuteUnit.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/StatementExecuteUnit.java
similarity index 65%
rename from sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecuteUnit.java
rename to sharding-core/src/main/java/io/shardingsphere/core/executor/StatementExecuteUnit.java
index 3cb49bfc7c3bf..c4944625c42ea 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecuteUnit.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/StatementExecuteUnit.java
@@ -15,12 +15,11 @@
*
*/
-package io.shardingsphere.core.executor.statement;
+package io.shardingsphere.core.executor;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.constant.ConnectionMode;
import io.shardingsphere.core.routing.RouteUnit;
import lombok.Getter;
-import lombok.RequiredArgsConstructor;
import java.sql.Statement;
@@ -28,12 +27,20 @@
* Execute unit for JDBC statement.
*
* @author zhangliang
+ * @author panjuan
*/
-@RequiredArgsConstructor
@Getter
-public final class StatementExecuteUnit implements SQLExecuteUnit {
+public final class StatementExecuteUnit {
private final RouteUnit routeUnit;
private final Statement statement;
+
+ private final ConnectionMode connectionMode;
+
+ public StatementExecuteUnit(final RouteUnit routeUnit, final Statement statement, final ConnectionMode connectionMode) {
+ this.routeUnit = routeUnit;
+ this.statement = statement;
+ this.connectionMode = connectionMode;
+ }
}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/SQLExecuteUnit.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/SQLExecuteUnit.java
deleted file mode 100644
index 18b2f35e0a27f..0000000000000
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/SQLExecuteUnit.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.sql;
-
-import io.shardingsphere.core.routing.RouteUnit;
-
-import java.sql.Statement;
-
-/**
- * SQL execute unit.
- *
- * @author zhangliang
- */
-public interface SQLExecuteUnit {
-
- /**
- * Get route unit.
- *
- * @return route unit
- */
- RouteUnit getRouteUnit();
-
- /**
- * Get statement.
- *
- * @return statement
- */
- Statement getStatement();
-}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java
index ae2fbc1e07370..8b0afad0f8fa9 100644
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java
@@ -25,7 +25,7 @@
import io.shardingsphere.core.event.executor.SQLExecutionEventFactory;
import io.shardingsphere.core.executor.ShardingExecuteCallback;
import io.shardingsphere.core.executor.ShardingGroupExecuteCallback;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.executor.StatementExecuteUnit;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
import io.shardingsphere.core.metadata.datasource.DataSourceMetaData;
@@ -47,7 +47,7 @@
* @param class type of return value
*/
@RequiredArgsConstructor
-public abstract class SQLExecuteCallback implements ShardingExecuteCallback, ShardingGroupExecuteCallback {
+public abstract class SQLExecuteCallback implements ShardingExecuteCallback, ShardingGroupExecuteCallback {
private final DatabaseType databaseType;
@@ -60,38 +60,38 @@ public abstract class SQLExecuteCallback implements ShardingExecuteCallback execute(final Collection sqlExecuteUnits) throws SQLException {
+ public final Collection execute(final Collection statementExecuteUnits) throws SQLException {
Collection result = new LinkedList<>();
- for (SQLExecuteUnit each : sqlExecuteUnits) {
+ for (StatementExecuteUnit each : statementExecuteUnits) {
result.add(execute0(each));
}
return result;
}
- private T execute0(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
+ private T execute0(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
ExecutorDataMap.setDataMap(dataMap);
- List> parameterSets = sqlExecuteUnit.getRouteUnit().getSqlUnit().getParameterSets();
- DataSourceMetaData dataSourceMetaData = DataSourceMetaDataFactory.newInstance(databaseType, sqlExecuteUnit.getStatement().getConnection().getMetaData().getURL());
+ List> parameterSets = statementExecuteUnit.getRouteUnit().getSqlUnit().getParameterSets();
+ DataSourceMetaData dataSourceMetaData = DataSourceMetaDataFactory.newInstance(databaseType, statementExecuteUnit.getStatement().getConnection().getMetaData().getURL());
for (List each : parameterSets) {
- shardingEventBus.post(SQLExecutionEventFactory.createEvent(sqlType, sqlExecuteUnit, each, dataSourceMetaData));
+ shardingEventBus.post(SQLExecutionEventFactory.createEvent(sqlType, statementExecuteUnit, each, dataSourceMetaData));
}
try {
- T result = executeSQL(sqlExecuteUnit);
+ T result = executeSQL(statementExecuteUnit);
for (List each : parameterSets) {
- SQLExecutionEvent finishEvent = SQLExecutionEventFactory.createEvent(sqlType, sqlExecuteUnit, each, dataSourceMetaData);
+ SQLExecutionEvent finishEvent = SQLExecutionEventFactory.createEvent(sqlType, statementExecuteUnit, each, dataSourceMetaData);
finishEvent.setExecuteSuccess();
shardingEventBus.post(finishEvent);
}
return result;
} catch (final SQLException ex) {
for (List each : parameterSets) {
- SQLExecutionEvent finishEvent = SQLExecutionEventFactory.createEvent(sqlType, sqlExecuteUnit, each, dataSourceMetaData);
+ SQLExecutionEvent finishEvent = SQLExecutionEventFactory.createEvent(sqlType, statementExecuteUnit, each, dataSourceMetaData);
finishEvent.setExecuteFailure(ex);
shardingEventBus.post(finishEvent);
}
@@ -100,5 +100,5 @@ private T execute0(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
}
}
- protected abstract T executeSQL(SQLExecuteUnit sqlExecuteUnit) throws SQLException;
+ protected abstract T executeSQL(StatementExecuteUnit statementExecuteUnit) throws SQLException;
}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteTemplate.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteTemplate.java
index c130e29adf944..8c930ea6a6a78 100644
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteTemplate.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteTemplate.java
@@ -19,7 +19,7 @@
import io.shardingsphere.core.executor.ShardingExecuteEngine;
import io.shardingsphere.core.executor.ShardingExecuteGroup;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.executor.StatementExecuteUnit;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
import lombok.RequiredArgsConstructor;
@@ -44,20 +44,20 @@ public final class SQLExecuteTemplate {
/**
* Execute.
*
- * @param sqlExecuteUnits SQL execute units
+ * @param statementExecuteUnits SQL execute units
* @param callback SQL execute callback
* @param class type of return value
* @return execute result
* @throws SQLException SQL exception
*/
- public List execute(final Collection sqlExecuteUnits, final SQLExecuteCallback callback) throws SQLException {
- return execute(sqlExecuteUnits, null, callback);
+ public List execute(final Collection statementExecuteUnits, final SQLExecuteCallback callback) throws SQLException {
+ return execute(statementExecuteUnits, null, callback);
}
/**
* Execute.
*
- * @param sqlExecuteUnits SQL execute units
+ * @param statementExecuteUnits SQL execute units
* @param firstExecuteCallback first SQL execute callback
* @param callback SQL execute callback
* @param class type of return value
@@ -65,16 +65,18 @@ public List execute(final Collection sqlExecute
* @throws SQLException SQL exception
*/
@SuppressWarnings("unchecked")
- public List execute(final Collection sqlExecuteUnits,
- final SQLExecuteCallback firstExecuteCallback, final SQLExecuteCallback callback) throws SQLException {
+ public List execute(
+ final Collection statementExecuteUnits, final SQLExecuteCallback firstExecuteCallback, final SQLExecuteCallback callback) throws SQLException {
try {
- return executeEngine.execute((Collection) sqlExecuteUnits, firstExecuteCallback, callback);
+ return executeEngine.execute((Collection) statementExecuteUnits, firstExecuteCallback, callback);
} catch (final SQLException ex) {
ExecutorExceptionHandler.handleException(ex);
return Collections.emptyList();
}
}
+
+
/**
* Execute group.
*
@@ -84,7 +86,7 @@ public List execute(final Collection sqlExecute
* @return execute result
* @throws SQLException SQL exception
*/
- public List executeGroup(final Collection> sqlExecuteGroups, final SQLExecuteCallback callback) throws SQLException {
+ public List executeGroup(final Collection> sqlExecuteGroups, final SQLExecuteCallback callback) throws SQLException {
return executeGroup(sqlExecuteGroups, null, callback);
}
@@ -99,7 +101,7 @@ public List executeGroup(final Collection List executeGroup(final Collection> sqlExecuteGroups,
+ public List executeGroup(final Collection> sqlExecuteGroups,
final SQLExecuteCallback firstCallback, final SQLExecuteCallback callback) throws SQLException {
try {
return executeEngine.groupExecute((Collection) sqlExecuteGroups, firstCallback, callback);
@@ -109,3 +111,4 @@ public List executeGroup(final Collection columnLabelAndIndexMap;
+ private final Multimap columnLabelAndIndexMap;
private final Iterator> resultData;
@@ -52,8 +52,8 @@ public MemoryQueryResult(final ResultSet resultSet) throws SQLException {
resultData = getResultData(resultSet);
}
- private Map getMetaData(final ResultSetMetaData resultSetMetaData) throws SQLException {
- Map result = new HashMap<>();
+ private Multimap getMetaData(final ResultSetMetaData resultSetMetaData) throws SQLException {
+ Multimap result = HashMultimap.create();
for (int columnIndex = 1; columnIndex <= resultSetMetaData.getColumnCount(); columnIndex++) {
result.put(resultSetMetaData.getColumnLabel(columnIndex), columnIndex);
}
@@ -89,7 +89,7 @@ public Object getValue(final int columnIndex, final Class type) {
@Override
public Object getValue(final String columnLabel, final Class type) {
- return currentRow.get(columnLabelAndIndexMap.get(columnLabel));
+ return currentRow.get(getIndexByColumnLabel(columnLabel));
}
@Override
@@ -99,7 +99,7 @@ public Object getCalendarValue(final int columnIndex, final Class type, final
@Override
public Object getCalendarValue(final String columnLabel, final Class type, final Calendar calendar) {
- return currentRow.get(columnLabelAndIndexMap.get(columnLabel));
+ return currentRow.get(getIndexByColumnLabel(columnLabel));
}
@Override
@@ -109,7 +109,7 @@ public InputStream getInputStream(final int columnIndex, final String type) {
@Override
public InputStream getInputStream(final String columnLabel, final String type) {
- return (InputStream) currentRow.get(columnLabelAndIndexMap.get(columnLabel));
+ return (InputStream) currentRow.get(getIndexByColumnLabel(columnLabel));
}
@Override
@@ -124,11 +124,15 @@ public int getColumnCount() {
@Override
public String getColumnLabel(final int columnIndex) throws SQLException {
- for (Entry entry : columnLabelAndIndexMap.entrySet()) {
+ for (Entry entry : columnLabelAndIndexMap.entries()) {
if (columnIndex == entry.getValue()) {
return entry.getKey();
}
}
throw new SQLException("Column index out of range", "9999");
}
+
+ private Integer getIndexByColumnLabel(final String columnLabel) {
+ return new ArrayList<>(columnLabelAndIndexMap.get(columnLabel)).get(0) - 1;
+ }
}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/result/StreamQueryResult.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/result/StreamQueryResult.java
index 47ebbe87344ee..38354dfc0f5c1 100644
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/result/StreamQueryResult.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/execute/result/StreamQueryResult.java
@@ -172,6 +172,7 @@ public Object getCalendarValue(final String columnLabel, final Class type, fi
throw new SQLException(String.format("Unsupported type: %s", type));
}
+ @SuppressWarnings("deprecation")
@Override
public InputStream getInputStream(final int columnIndex, final String type) throws SQLException {
switch (type) {
@@ -186,6 +187,7 @@ public InputStream getInputStream(final int columnIndex, final String type) thro
}
}
+ @SuppressWarnings("deprecation")
@Override
public InputStream getInputStream(final String columnLabel, final String type) throws SQLException {
switch (type) {
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareCallback.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareCallback.java
index f9f4f77bfd597..ba60a15d13be0 100644
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareCallback.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareCallback.java
@@ -17,16 +17,19 @@
package io.shardingsphere.core.executor.sql.prepare;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.constant.ConnectionMode;
+import io.shardingsphere.core.executor.StatementExecuteUnit;
import io.shardingsphere.core.routing.RouteUnit;
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.List;
/**
* SQL execute prepare callback.
*
* @author zhangliang
+ * @author panjuan
*/
public interface SQLExecutePrepareCallback {
@@ -34,18 +37,20 @@ public interface SQLExecutePrepareCallback {
* Get connection.
*
* @param dataSourceName data source name
+ * @param connectionSize connection size
* @return connection
* @throws SQLException SQL exception
*/
- Connection getConnection(String dataSourceName) throws SQLException;
+ List getConnections(String dataSourceName, int connectionSize) throws SQLException;
/**
* Create SQL execute unit.
*
* @param connection connection
* @param routeUnit route unit
+ * @param connectionMode connection mode
* @return SQL execute unit
* @throws SQLException SQL exception
*/
- SQLExecuteUnit createSQLExecuteUnit(Connection connection, RouteUnit routeUnit) throws SQLException;
+ StatementExecuteUnit createStatementExecuteUnit(Connection connection, RouteUnit routeUnit, ConnectionMode connectionMode) throws SQLException;
}
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareTemplate.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareTemplate.java
index e41ca93b21f83..48a10529b3e51 100644
--- a/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareTemplate.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/executor/sql/prepare/SQLExecutePrepareTemplate.java
@@ -18,8 +18,9 @@
package io.shardingsphere.core.executor.sql.prepare;
import com.google.common.collect.Lists;
+import io.shardingsphere.core.constant.ConnectionMode;
import io.shardingsphere.core.executor.ShardingExecuteGroup;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.executor.StatementExecuteUnit;
import io.shardingsphere.core.routing.RouteUnit;
import io.shardingsphere.core.routing.SQLUnit;
import lombok.RequiredArgsConstructor;
@@ -38,6 +39,7 @@
*
* @author zhaojun
* @author zhangliang
+ * @author panjuan
*/
@RequiredArgsConstructor
public final class SQLExecutePrepareTemplate {
@@ -52,9 +54,9 @@ public final class SQLExecutePrepareTemplate {
* @return statement execute unit groups
* @throws SQLException SQL exception
*/
- public Collection> getExecuteUnitGroups(final Collection routeUnits, final SQLExecutePrepareCallback callback) throws SQLException {
+ public Collection> getExecuteUnitGroups(final Collection routeUnits, final SQLExecutePrepareCallback callback) throws SQLException {
Map> sqlUnitGroups = getSQLUnitGroups(routeUnits);
- Collection> result = new LinkedList<>();
+ Collection> result = new LinkedList<>();
for (Entry> entry : sqlUnitGroups.entrySet()) {
result.addAll(getSQLExecuteGroups(entry.getKey(), entry.getValue(), callback));
}
@@ -72,22 +74,28 @@ private Map> getSQLUnitGroups(final Collection
return result;
}
- private List> getSQLExecuteGroups(final String dataSourceName, final List sqlUnits, final SQLExecutePrepareCallback callback) throws SQLException {
- List> result = new LinkedList<>();
+ private List> getSQLExecuteGroups(
+ final String dataSourceName, final List sqlUnits, final SQLExecutePrepareCallback callback) throws SQLException {
+ List> result = new LinkedList<>();
int desiredPartitionSize = Math.max(sqlUnits.size() / maxConnectionsSizePerQuery, 1);
- for (List each : Lists.partition(sqlUnits, desiredPartitionSize)) {
- // TODO get connection sync to prevent dead lock
- result.add(getSQLExecuteGroup(callback.getConnection(dataSourceName), dataSourceName, each, callback));
+ List> sqlUnitGroups = Lists.partition(sqlUnits, desiredPartitionSize);
+ List connections = callback.getConnections(dataSourceName, sqlUnitGroups.size());
+ int count = 0;
+ for (List each : sqlUnitGroups) {
+ result.add(getSQLExecuteGroup(connections.get(count++), dataSourceName, each, callback));
}
return result;
}
- private ShardingExecuteGroup getSQLExecuteGroup(
+ private ShardingExecuteGroup getSQLExecuteGroup(
final Connection connection, final String dataSourceName, final List sqlUnitGroup, final SQLExecutePrepareCallback callback) throws SQLException {
- List result = new LinkedList<>();
+ List result = new LinkedList<>();
+ ConnectionMode connectionMode = 1 == sqlUnitGroup.size() ? ConnectionMode.MEMORY_STRICTLY : ConnectionMode.CONNECTION_STRICTLY;
for (SQLUnit each : sqlUnitGroup) {
- result.add(callback.createSQLExecuteUnit(connection, new RouteUnit(dataSourceName, each)));
+ result.add(callback.createStatementExecuteUnit(connection, new RouteUnit(dataSourceName, each), connectionMode));
}
return new ShardingExecuteGroup<>(result);
}
}
+
+
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecuteUnit.java b/sharding-core/src/main/java/io/shardingsphere/core/routing/BatchRouteUnit.java
similarity index 73%
rename from sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecuteUnit.java
rename to sharding-core/src/main/java/io/shardingsphere/core/routing/BatchRouteUnit.java
index f9726a605b94a..01fe802c4a139 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecuteUnit.java
+++ b/sharding-core/src/main/java/io/shardingsphere/core/routing/BatchRouteUnit.java
@@ -15,42 +15,44 @@
*
*/
-package io.shardingsphere.core.executor.batch;
+package io.shardingsphere.core.routing;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
-import io.shardingsphere.core.routing.RouteUnit;
import lombok.AccessLevel;
+import lombok.EqualsAndHashCode;
import lombok.Getter;
-import lombok.RequiredArgsConstructor;
+import lombok.ToString;
-import java.sql.PreparedStatement;
import java.util.LinkedHashMap;
import java.util.Map;
/**
- * Execute unit for JDBC prepared statement add batch.
+ * Batch route unit.
*
- * @author zhangliang
+ * @author panjuan
*/
-@RequiredArgsConstructor
@Getter
-public final class BatchPreparedStatementExecuteUnit implements SQLExecuteUnit {
+@EqualsAndHashCode(of = { "routeUnit" })
+@ToString
+public final class BatchRouteUnit {
private final RouteUnit routeUnit;
- private final PreparedStatement statement;
-
private final Map jdbcAndActualAddBatchCallTimesMap = new LinkedHashMap<>();
@Getter(AccessLevel.NONE)
private int actualCallAddBatchTimes;
+ public BatchRouteUnit(final RouteUnit routeUnit) {
+ this.routeUnit = routeUnit;
+ }
+
/**
* Map times of use JDBC API call addBatch and times of actual call addBatch after route.
- *
+ *
* @param jdbcAddBatchTimes times of use JDBC API call addBatch
*/
public void mapAddBatchCount(final int jdbcAddBatchTimes) {
jdbcAndActualAddBatchCallTimesMap.put(jdbcAddBatchTimes, actualCallAddBatchTimes++);
}
}
+
diff --git a/sharding-core/src/test/java/io/shardingsphere/core/executor/fixture/ExecutorTestUtil.java b/sharding-core/src/test/java/io/shardingsphere/core/executor/fixture/ExecutorTestUtil.java
index 50e4cafa81504..eda7b37998eef 100644
--- a/sharding-core/src/test/java/io/shardingsphere/core/executor/fixture/ExecutorTestUtil.java
+++ b/sharding-core/src/test/java/io/shardingsphere/core/executor/fixture/ExecutorTestUtil.java
@@ -20,7 +20,6 @@
import io.shardingsphere.core.event.ShardingEventType;
import io.shardingsphere.core.event.executor.SQLExecutionEvent;
import io.shardingsphere.core.event.root.RootInvokeEvent;
-import io.shardingsphere.core.event.root.RootInvokeStartEvent;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
@@ -53,7 +52,6 @@ public static void listen(final EventCaller eventCaller, final SQLExecutionEvent
* @param event overall execution event
*/
public static void listen(final EventCaller eventCaller, final RootInvokeEvent event) {
- eventCaller.verifyIsParallelExecute(((RootInvokeStartEvent) event).isParallelExecute());
if (ShardingEventType.EXECUTE_FAILURE == event.getEventType()) {
eventCaller.verifyException(event.getException());
}
diff --git a/sharding-core/src/test/java/io/shardingsphere/core/merger/fixture/TestQueryResult.java b/sharding-core/src/test/java/io/shardingsphere/core/merger/fixture/TestQueryResult.java
index e3a6925fdc173..2f4f2538af0e1 100644
--- a/sharding-core/src/test/java/io/shardingsphere/core/merger/fixture/TestQueryResult.java
+++ b/sharding-core/src/test/java/io/shardingsphere/core/merger/fixture/TestQueryResult.java
@@ -166,6 +166,7 @@ public Object getCalendarValue(final String columnLabel, final Class type, fi
throw new SQLException(String.format("Unsupported type: %s", type));
}
+ @SuppressWarnings("deprecation")
@Override
public InputStream getInputStream(final int columnIndex, final String type) throws SQLException {
switch (type) {
@@ -180,6 +181,7 @@ public InputStream getInputStream(final int columnIndex, final String type) thro
}
}
+ @SuppressWarnings("deprecation")
@Override
public InputStream getInputStream(final String columnLabel, final String type) throws SQLException {
switch (type) {
diff --git a/sharding-core/src/test/java/io/shardingsphere/core/parsing/parser/sql/UpdateStatementParserTest.java b/sharding-core/src/test/java/io/shardingsphere/core/parsing/parser/sql/UpdateStatementParserTest.java
index 22c5ebe005573..1ae55b7aee87c 100644
--- a/sharding-core/src/test/java/io/shardingsphere/core/parsing/parser/sql/UpdateStatementParserTest.java
+++ b/sharding-core/src/test/java/io/shardingsphere/core/parsing/parser/sql/UpdateStatementParserTest.java
@@ -17,57 +17,19 @@
package io.shardingsphere.core.parsing.parser.sql;
-import com.google.common.collect.Range;
-import io.shardingsphere.core.api.algorithm.sharding.ListShardingValue;
-import io.shardingsphere.core.api.algorithm.sharding.RangeShardingValue;
import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.ShardingOperator;
import io.shardingsphere.core.parsing.SQLParsingEngine;
-import io.shardingsphere.core.parsing.parser.context.condition.Column;
-import io.shardingsphere.core.parsing.parser.context.condition.Condition;
import io.shardingsphere.core.parsing.parser.sql.dml.DMLStatement;
import io.shardingsphere.core.rule.ShardingRule;
-import org.hamcrest.CoreMatchers;
import org.junit.Test;
-import java.util.Collections;
-import java.util.Iterator;
import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
public final class UpdateStatementParserTest extends AbstractStatementParserTest {
-
- @Test
- public void parseWithoutParameter() {
- ShardingRule shardingRule = createShardingRule();
- SQLParsingEngine statementParser = new SQLParsingEngine(DatabaseType.MySQL, "UPDATE TABLE_XXX xxx SET TABLE_XXX.field1=field1+1,xxx.field2=2 WHERE TABLE_XXX.field4<10 AND"
- + " TABLE_XXX.field1=1 AND xxx.field5>10 AND TABLE_XXX.field2 IN (1,3) AND xxx.field6<=10 AND TABLE_XXX.field3 BETWEEN 5 AND 20 AND xxx.field7>=10", shardingRule, null);
- DMLStatement updateStatement = (DMLStatement) statementParser.parse(false);
- assertUpdateStatementWithoutParameter(updateStatement);
- }
-
- private void assertUpdateStatementWithoutParameter(final DMLStatement updateStatement) {
- assertThat(updateStatement.getTables().find("TABLE_XXX").get().getName(), is("TABLE_XXX"));
- assertThat(updateStatement.getTables().find("TABLE_XXX").get().getAlias().get(), is("xxx"));
- Condition condition1 = updateStatement.getConditions().find(new Column("field1", "TABLE_XXX")).get();
- assertThat(condition1.getOperator(), CoreMatchers.is(ShardingOperator.EQUAL));
- assertThat(((ListShardingValue) condition1.getShardingValue(Collections.emptyList())).getValues().iterator().next(), is((Comparable) 1));
- Condition condition2 = updateStatement.getConditions().find(new Column("field2", "TABLE_XXX")).get();
- assertThat(condition2.getOperator(), is(ShardingOperator.IN));
- Iterator shardingValues2 = ((ListShardingValue) condition2.getShardingValue(Collections.emptyList())).getValues().iterator();
- assertThat(shardingValues2.next(), is((Object) 1));
- assertThat(shardingValues2.next(), is((Object) 3));
- assertFalse(shardingValues2.hasNext());
- Condition condition3 = updateStatement.getConditions().find(new Column("field3", "TABLE_XXX")).get();
- Range shardingValues3 = ((RangeShardingValue) condition3.getShardingValue(Collections.emptyList())).getValueRange();
- assertThat(condition3.getOperator(), is(ShardingOperator.BETWEEN));
- assertThat(shardingValues3.lowerEndpoint(), is((Comparable) 5));
- assertThat(shardingValues3.upperEndpoint(), is((Comparable) 20));
- }
-
+
@Test
public void parseWithOr() {
ShardingRule shardingRule = createShardingRule();
@@ -82,22 +44,4 @@ private void assertUpdateStatementWitOr(final DMLStatement updateStatement) {
assertTrue(updateStatement.getConditions().getOrCondition().getAndConditions().isEmpty());
}
-
- @Test
- public void parseWithSpecialSyntax() {
- parseWithSpecialSyntax(DatabaseType.MySQL, "UPDATE `TABLE_XXX` SET `field1`=1 WHERE `field1`=1");
- parseWithSpecialSyntax(DatabaseType.Oracle, "UPDATE /*+ index(field1) */ ONLY TABLE_XXX SET field1=1 WHERE field1=1 RETURN * LOG ERRORS INTO TABLE_LOG");
- parseWithSpecialSyntax(DatabaseType.Oracle, "UPDATE /*+ index(field1) */ ONLY TABLE_XXX SET field1=1 WHERE field1=1 RETURNING *");
- parseWithSpecialSyntax(DatabaseType.Oracle, "UPDATE /*+ index(field1) */ ONLY TABLE_XXX SET field1=1 WHERE field1=1 LOG ERRORS INTO TABLE_LOG");
- }
-
- private void parseWithSpecialSyntax(final DatabaseType dbType, final String actualSQL) {
- ShardingRule shardingRule = createShardingRule();
- DMLStatement updateStatement = (DMLStatement) new SQLParsingEngine(dbType, actualSQL, shardingRule, null).parse(false);
- assertThat(updateStatement.getTables().find("TABLE_XXX").get().getName(), is("TABLE_XXX"));
- assertFalse(updateStatement.getTables().find("TABLE_XXX").get().getAlias().isPresent());
- Condition condition = updateStatement.getConditions().find(new Column("field1", "TABLE_XXX")).get();
- assertThat(condition.getOperator(), is(ShardingOperator.EQUAL));
- assertThat(((ListShardingValue) condition.getShardingValue(Collections.emptyList())).getValues().iterator().next(), is((Object) 1));
- }
}
diff --git a/sharding-core/src/test/resources/parser/update.xml b/sharding-core/src/test/resources/parser/update.xml
index 44705f42bb519..041fce44f2e17 100644
--- a/sharding-core/src/test/resources/parser/update.xml
+++ b/sharding-core/src/test/resources/parser/update.xml
@@ -84,4 +84,62 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/AbstractStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/AbstractStatementExecutor.java
new file mode 100644
index 0000000000000..5cdbc6170331f
--- /dev/null
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/AbstractStatementExecutor.java
@@ -0,0 +1,115 @@
+/*
+ * Copyright 2016-2018 shardingsphere.io.
+ *
+ * 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.
+ *
+ */
+
+package io.shardingsphere.core.executor;
+
+import io.shardingsphere.core.constant.DatabaseType;
+import io.shardingsphere.core.constant.SQLType;
+import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
+import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
+import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareTemplate;
+import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Abstract statement executor.
+ *
+ * @author panjuan
+ */
+@Getter(AccessLevel.PROTECTED)
+public class AbstractStatementExecutor {
+
+ private final DatabaseType databaseType;
+
+ @Getter
+ private final int resultSetType;
+
+ @Getter
+ private final int resultSetConcurrency;
+
+ @Getter
+ private final int resultSetHoldability;
+
+ private final ShardingConnection connection;
+
+ private final SQLExecutePrepareTemplate sqlExecutePrepareTemplate;
+
+ private final SQLExecuteTemplate sqlExecuteTemplate;
+
+ private final Collection connections = new LinkedList<>();
+
+ @Getter
+ private final List> parameterSets = new LinkedList<>();
+
+ @Getter
+ private final List statements = new LinkedList<>();
+
+ @Getter
+ private final List resultSets = new CopyOnWriteArrayList<>();
+
+ private final Collection> executeGroups = new LinkedList<>();
+
+ @Getter
+ @Setter
+ private SQLType sqlType;
+
+ public AbstractStatementExecutor(final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability, final ShardingConnection shardingConnection) {
+ this.databaseType = shardingConnection.getShardingContext().getDatabaseType();
+ this.resultSetType = resultSetType;
+ this.resultSetConcurrency = resultSetConcurrency;
+ this.resultSetHoldability = resultSetHoldability;
+ this.connection = shardingConnection;
+ sqlExecutePrepareTemplate = new SQLExecutePrepareTemplate(connection.getShardingContext().getMaxConnectionsSizePerQuery());
+ sqlExecuteTemplate = new SQLExecuteTemplate(connection.getShardingContext().getExecuteEngine());
+ }
+
+ @SuppressWarnings("unchecked")
+ protected final List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
+ return sqlExecuteTemplate.executeGroup((Collection) executeGroups, executeCallback);
+ }
+
+ /**
+ * Clear data.
+ *
+ * @throws SQLException sql exception
+ */
+ public void clear() throws SQLException {
+ clearStatements();
+ statements.clear();
+ parameterSets.clear();
+ connections.clear();
+ resultSets.clear();
+ executeGroups.clear();
+ }
+
+ private void clearStatements() throws SQLException {
+ for (Statement each : getStatements()) {
+ each.close();
+ }
+ }
+}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/BatchPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/BatchPreparedStatementExecutor.java
new file mode 100644
index 0000000000000..fd96f1c1f34a4
--- /dev/null
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/BatchPreparedStatementExecutor.java
@@ -0,0 +1,242 @@
+/*
+ * Copyright 2016-2018 shardingsphere.io.
+ *
+ * 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.
+ *
+ */
+
+package io.shardingsphere.core.executor;
+
+import com.google.common.base.Function;
+import com.google.common.base.Optional;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import io.shardingsphere.core.constant.ConnectionMode;
+import io.shardingsphere.core.constant.DatabaseType;
+import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
+import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
+import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
+import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
+import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
+import io.shardingsphere.core.routing.BatchRouteUnit;
+import io.shardingsphere.core.routing.RouteUnit;
+import io.shardingsphere.core.routing.SQLRouteResult;
+import lombok.Getter;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * Prepared statement executor to process add batch.
+ *
+ * @author zhangliang
+ * @author maxiaoguang
+ * @author panjuan
+ */
+public final class BatchPreparedStatementExecutor extends AbstractStatementExecutor {
+
+ private final Collection routeUnits = new LinkedList<>();
+
+ @Getter
+ private final boolean returnGeneratedKeys;
+
+ private int batchCount;
+
+ public BatchPreparedStatementExecutor(final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability, final boolean returnGeneratedKeys,
+ final ShardingConnection shardingConnection) {
+ super(resultSetType, resultSetConcurrency, resultSetHoldability, shardingConnection);
+ this.returnGeneratedKeys = returnGeneratedKeys;
+ }
+
+ /**
+ * Initialize executor.
+ *
+ * @throws SQLException SQL exception
+ */
+ public void init() throws SQLException {
+ getExecuteGroups().addAll(obtainExecuteGroups(routeUnits));
+ }
+
+ private Collection> obtainExecuteGroups(final Collection routeUnits) throws SQLException {
+ return getSqlExecutePrepareTemplate().getExecuteUnitGroups(Lists.transform(new ArrayList<>(routeUnits), new Function() {
+
+ @Override
+ public RouteUnit apply(final BatchRouteUnit input) {
+ return input.getRouteUnit();
+ }
+ }), new SQLExecutePrepareCallback() {
+
+ @Override
+ public List getConnections(final String dataSourceName, final int connectionSize) throws SQLException {
+ return BatchPreparedStatementExecutor.super.getConnection().getConnections(dataSourceName, connectionSize);
+ }
+
+ @Override
+ public StatementExecuteUnit createStatementExecuteUnit(final Connection connection, final RouteUnit routeUnit, final ConnectionMode connectionMode) throws SQLException {
+ PreparedStatement preparedStatement = createPreparedStatement(connection, routeUnit.getSqlUnit().getSql());
+ return new StatementExecuteUnit(routeUnit, preparedStatement, connectionMode);
+ }
+ });
+ }
+
+ @SuppressWarnings("MagicConstant")
+ private PreparedStatement createPreparedStatement(final Connection connection, final String sql) throws SQLException {
+ return returnGeneratedKeys ? connection.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS)
+ : connection.prepareStatement(sql, getResultSetType(), getResultSetConcurrency(), getResultSetHoldability());
+ }
+
+ /**
+ * Add batch for route units.
+ *
+ * @param routeResult route result
+ */
+ public void addBatchForRouteUnits(final SQLRouteResult routeResult) {
+ setSqlType(routeResult.getSqlStatement().getType());
+ handleOldRouteUnits(createBatchRouteUnits(routeResult.getRouteUnits()));
+ handleNewRouteUnits(createBatchRouteUnits(routeResult.getRouteUnits()));
+ batchCount++;
+ }
+
+ private Collection createBatchRouteUnits(final Collection routeUnits) {
+ Collection result = new LinkedList<>();
+ for (RouteUnit each : routeUnits) {
+ result.add(new BatchRouteUnit(each));
+ }
+ return result;
+ }
+
+ private void handleOldRouteUnits(final Collection newRouteUnits) {
+ for (final BatchRouteUnit each : newRouteUnits) {
+ Optional batchRouteUnitOptional = Iterators.tryFind(routeUnits.iterator(), new Predicate() {
+ @Override
+ public boolean apply(final BatchRouteUnit input) {
+ return input.equals(each);
+ }
+ });
+ if (batchRouteUnitOptional.isPresent()) {
+ reviseBatchRouteUnit(batchRouteUnitOptional.get(), each);
+ }
+ }
+ }
+
+ private void reviseBatchRouteUnit(final BatchRouteUnit oldBatchRouteUnit, final BatchRouteUnit newBatchRouteUnit) {
+ oldBatchRouteUnit.getRouteUnit().getSqlUnit().getParameterSets().add(newBatchRouteUnit.getRouteUnit().getSqlUnit().getParameterSets().get(0));
+ oldBatchRouteUnit.mapAddBatchCount(batchCount);
+ }
+
+ private void handleNewRouteUnits(final Collection newRouteUnits) {
+ newRouteUnits.removeAll(routeUnits);
+ for (BatchRouteUnit each : newRouteUnits) {
+ each.mapAddBatchCount(batchCount);
+ }
+ routeUnits.addAll(newRouteUnits);
+ }
+
+ /**
+ * Execute batch.
+ *
+ * @return execute results
+ * @throws SQLException SQL exception
+ */
+ public int[] executeBatch() throws SQLException {
+ final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
+ final Map dataMap = ExecutorDataMap.getDataMap();
+ SQLExecuteCallback callback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+
+ @Override
+ protected int[] executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return statementExecuteUnit.getStatement().executeBatch();
+ }
+ };
+ return accumulate(executeCallback(callback));
+ }
+
+ private int[] accumulate(final List results) {
+ int[] result = new int[batchCount];
+ int count = 0;
+ for (BatchRouteUnit each : routeUnits) {
+ for (Entry entry : each.getJdbcAndActualAddBatchCallTimesMap().entrySet()) {
+ int value = null == results.get(count) ? 0 : results.get(count)[entry.getValue()];
+ if (DatabaseType.Oracle == getDatabaseType()) {
+ result[entry.getKey()] = value;
+ } else {
+ result[entry.getKey()] += value;
+ }
+ }
+ count++;
+ }
+ return result;
+ }
+
+ /**
+ * Get statements.
+ *
+ * @return statements
+ */
+ @Override
+ public List getStatements() {
+ List result = new LinkedList<>();
+ for (ShardingExecuteGroup each : getExecuteGroups()) {
+ result.addAll(Lists.transform(each.getInputs(), new Function() {
+
+ @Override
+ public Statement apply(final StatementExecuteUnit input) {
+ return input.getStatement();
+ }
+ }));
+ }
+ return result;
+ }
+
+ /**
+ * Get parameter sets.
+ *
+ * @param statement statement
+ * @return parameter sets
+ */
+ public List> getParameterSet(final Statement statement) {
+ Optional target;
+ List> result = new LinkedList<>();
+ for (ShardingExecuteGroup each : getExecuteGroups()) {
+ target = Iterators.tryFind(each.getInputs().iterator(), new Predicate() {
+ @Override
+ public boolean apply(final StatementExecuteUnit input) {
+ return input.getStatement().equals(statement);
+ }
+ });
+ if (target.isPresent()) {
+ result.addAll(target.get().getRouteUnit().getSqlUnit().getParameterSets());
+ break;
+ }
+ }
+ return result;
+ }
+
+ @Override
+ public void clear() throws SQLException {
+ super.clear();
+ batchCount = 0;
+ routeUnits.clear();
+ }
+}
+
+
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/PreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/PreparedStatementExecutor.java
new file mode 100644
index 0000000000000..f9e55ea62f36f
--- /dev/null
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/PreparedStatementExecutor.java
@@ -0,0 +1,172 @@
+/*
+ * Copyright 2016-2018 shardingsphere.io.
+ *
+ * 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.
+ *
+ */
+
+package io.shardingsphere.core.executor;
+
+import io.shardingsphere.core.constant.ConnectionMode;
+import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
+import io.shardingsphere.core.executor.sql.execute.result.MemoryQueryResult;
+import io.shardingsphere.core.executor.sql.execute.result.StreamQueryResult;
+import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
+import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
+import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
+import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
+import io.shardingsphere.core.merger.QueryResult;
+import io.shardingsphere.core.routing.RouteUnit;
+import io.shardingsphere.core.routing.SQLRouteResult;
+import lombok.Getter;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Prepared statement executor.
+ *
+ * @author zhangliang
+ * @author caohao
+ * @author maxiaoguang
+ * @author panjuan
+ */
+public final class PreparedStatementExecutor extends AbstractStatementExecutor {
+
+ @Getter
+ private final boolean returnGeneratedKeys;
+
+ public PreparedStatementExecutor(
+ final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability, final boolean returnGeneratedKeys, final ShardingConnection shardingConnection) {
+ super(resultSetType, resultSetConcurrency, resultSetHoldability, shardingConnection);
+ this.returnGeneratedKeys = returnGeneratedKeys;
+ }
+
+ /**
+ * Initialize executor.
+ *
+ * @param routeResult route result
+ * @throws SQLException SQL exception
+ */
+ public void init(final SQLRouteResult routeResult) throws SQLException {
+ setSqlType(routeResult.getSqlStatement().getType());
+ getExecuteGroups().addAll(obtainExecuteGroups(routeResult.getRouteUnits()));
+ }
+
+ private Collection> obtainExecuteGroups(final Collection routeUnits) throws SQLException {
+ return getSqlExecutePrepareTemplate().getExecuteUnitGroups(routeUnits, new SQLExecutePrepareCallback() {
+
+ @Override
+ public List getConnections(final String dataSourceName, final int connectionSize) throws SQLException {
+ return PreparedStatementExecutor.super.getConnection().getConnections(dataSourceName, connectionSize);
+ }
+
+ @Override
+ public StatementExecuteUnit createStatementExecuteUnit(final Connection connection, final RouteUnit routeUnit, final ConnectionMode connectionMode) throws SQLException {
+ PreparedStatement preparedStatement = createPreparedStatement(connection, routeUnit.getSqlUnit().getSql());
+ getStatements().add(preparedStatement);
+ getParameterSets().add(routeUnit.getSqlUnit().getParameterSets().get(0));
+ return new StatementExecuteUnit(routeUnit, preparedStatement, connectionMode);
+ }
+ });
+ }
+
+ @SuppressWarnings("MagicConstant")
+ private PreparedStatement createPreparedStatement(final Connection connection, final String sql) throws SQLException {
+ return returnGeneratedKeys ? connection.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS)
+ : connection.prepareStatement(sql, getResultSetType(), getResultSetConcurrency(), getResultSetHoldability());
+ }
+
+ /**
+ * Execute query.
+ *
+ * @return result set list
+ * @throws SQLException SQL exception
+ */
+ public List executeQuery() throws SQLException {
+ final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
+ final Map dataMap = ExecutorDataMap.getDataMap();
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+
+ @Override
+ protected QueryResult executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return getQueryResult(statementExecuteUnit);
+ }
+ };
+ return executeCallback(executeCallback);
+ }
+
+ private QueryResult getQueryResult(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ PreparedStatement preparedStatement = (PreparedStatement) statementExecuteUnit.getStatement();
+ ResultSet resultSet = preparedStatement.executeQuery();
+ getResultSets().add(resultSet);
+ return ConnectionMode.MEMORY_STRICTLY == statementExecuteUnit.getConnectionMode() ? new StreamQueryResult(resultSet) : new MemoryQueryResult(resultSet);
+ }
+
+ /**
+ * Execute update.
+ *
+ * @return effected records count
+ * @throws SQLException SQL exception
+ */
+ public int executeUpdate() throws SQLException {
+ final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
+ final Map dataMap = ExecutorDataMap.getDataMap();
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+
+ @Override
+ protected Integer executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return ((PreparedStatement) statementExecuteUnit.getStatement()).executeUpdate();
+ }
+ };
+ List results = executeCallback(executeCallback);
+ return accumulate(results);
+ }
+
+ private int accumulate(final List results) {
+ int result = 0;
+ for (Integer each : results) {
+ result += null == each ? 0 : each;
+ }
+ return result;
+ }
+
+ /**
+ * Execute SQL.
+ *
+ * @return return true if is DQL, false if is DML
+ * @throws SQLException SQL exception
+ */
+ public boolean execute() throws SQLException {
+ boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
+ Map dataMap = ExecutorDataMap.getDataMap();
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+
+ @Override
+ protected Boolean executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return ((PreparedStatement) statementExecuteUnit.getStatement()).execute();
+ }
+ };
+ List result = executeCallback(executeCallback);
+ if (null == result || result.isEmpty() || null == result.get(0)) {
+ return false;
+ }
+ return result.get(0);
+ }
+}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecutor.java
similarity index 64%
rename from sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecutor.java
rename to sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecutor.java
index bc2d6f65c6476..78c560299c004 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/StatementExecutor.java
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecutor.java
@@ -15,19 +15,25 @@
*
*/
-package io.shardingsphere.core.executor.statement;
+package io.shardingsphere.core.executor;
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
+import io.shardingsphere.core.constant.ConnectionMode;
import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
+import io.shardingsphere.core.executor.sql.execute.result.MemoryQueryResult;
+import io.shardingsphere.core.executor.sql.execute.result.StreamQueryResult;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
-import lombok.RequiredArgsConstructor;
+import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
+import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
+import io.shardingsphere.core.merger.QueryResult;
+import io.shardingsphere.core.routing.RouteUnit;
+import io.shardingsphere.core.routing.SQLRouteResult;
+import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -38,13 +44,43 @@
* @author caohao
* @author zhangliang
* @author maxiaoguang
+ * @author panjuan
*/
-@RequiredArgsConstructor
-public abstract class StatementExecutor {
+public final class StatementExecutor extends AbstractStatementExecutor {
- private final DatabaseType databaseType;
+ public StatementExecutor(final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability, final ShardingConnection shardingConnection) {
+ super(resultSetType, resultSetConcurrency, resultSetHoldability, shardingConnection);
+ }
+
+ /**
+ * Initialize executor.
+ *
+ * @param routeResult route result
+ * @throws SQLException SQL exception
+ */
+ public void init(final SQLRouteResult routeResult) throws SQLException {
+ setSqlType(routeResult.getSqlStatement().getType());
+ getExecuteGroups().addAll(obtainExecuteGroups(routeResult.getRouteUnits()));
+ }
- private final SQLType sqlType;
+ private Collection> obtainExecuteGroups(final Collection routeUnits) throws SQLException {
+ return getSqlExecutePrepareTemplate().getExecuteUnitGroups(routeUnits, new SQLExecutePrepareCallback() {
+
+ @Override
+ public List getConnections(final String dataSourceName, final int connectionSize) throws SQLException {
+ return StatementExecutor.super.getConnection().getConnections(dataSourceName, connectionSize);
+ }
+
+ @SuppressWarnings("MagicConstant")
+ @Override
+ public StatementExecuteUnit createStatementExecuteUnit(final Connection connection, final RouteUnit routeUnit, final ConnectionMode connectionMode) throws SQLException {
+ Statement statement = connection.createStatement(getResultSetType(), getResultSetConcurrency(), getResultSetHoldability());
+ getStatements().add(statement);
+ getParameterSets().add(routeUnit.getSqlUnit().getParameterSets().get(0));
+ return new StatementExecuteUnit(routeUnit, statement, connectionMode);
+ }
+ });
+ }
/**
* Execute query.
@@ -52,19 +88,25 @@ public abstract class StatementExecutor {
* @return result set list
* @throws SQLException SQL exception
*/
- public List executeQuery() throws SQLException {
+ public List executeQuery() throws SQLException {
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
@Override
- protected ResultSet executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return sqlExecuteUnit.getStatement().executeQuery(sqlExecuteUnit.getRouteUnit().getSqlUnit().getSql());
+ protected QueryResult executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return getQueryResult(statementExecuteUnit);
}
};
return executeCallback(executeCallback);
}
+ private QueryResult getQueryResult(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ ResultSet resultSet = statementExecuteUnit.getStatement().executeQuery(statementExecuteUnit.getRouteUnit().getSqlUnit().getSql());
+ getResultSets().add(resultSet);
+ return ConnectionMode.MEMORY_STRICTLY == statementExecuteUnit.getConnectionMode() ? new StreamQueryResult(resultSet) : new MemoryQueryResult(resultSet);
+ }
+
/**
* Execute update.
*
@@ -135,11 +177,11 @@ public int executeUpdate(final Statement statement, final String sql) throws SQL
private int executeUpdate(final Updater updater) throws SQLException {
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
@Override
- protected Integer executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return updater.executeUpdate(sqlExecuteUnit.getStatement(), sqlExecuteUnit.getRouteUnit().getSqlUnit().getSql());
+ protected Integer executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return updater.executeUpdate(statementExecuteUnit.getStatement(), statementExecuteUnit.getRouteUnit().getSqlUnit().getSql());
}
};
List results = executeCallback(executeCallback);
@@ -224,11 +266,11 @@ public boolean execute(final Statement statement, final String sql) throws SQLEx
private boolean execute(final Executor executor) throws SQLException {
final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
+ SQLExecuteCallback executeCallback = new SQLExecuteCallback(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
@Override
- protected Boolean executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return executor.execute(sqlExecuteUnit.getStatement(), sqlExecuteUnit.getRouteUnit().getSqlUnit().getSql());
+ protected Boolean executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
+ return executor.execute(statementExecuteUnit.getStatement(), statementExecuteUnit.getRouteUnit().getSqlUnit().getSql());
}
};
List result = executeCallback(executeCallback);
@@ -238,8 +280,6 @@ protected Boolean executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLExce
return result.get(0);
}
- protected abstract List executeCallback(SQLExecuteCallback executeCallback) throws SQLException;
-
private interface Updater {
int executeUpdate(Statement statement, String sql) throws SQLException;
@@ -250,3 +290,4 @@ private interface Executor {
boolean execute(Statement statement, String sql) throws SQLException;
}
}
+
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecutor.java
deleted file mode 100644
index fe41d1f45cd79..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/BatchPreparedStatementExecutor.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.batch;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
-import lombok.RequiredArgsConstructor;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-/**
- * Prepared statement executor to process add batch.
- *
- * @author zhangliang
- * @author maxiaoguang
- */
-@RequiredArgsConstructor
-public abstract class BatchPreparedStatementExecutor {
-
- private final DatabaseType databaseType;
-
- private final SQLType sqlType;
-
- private final int batchCount;
-
- /**
- * Execute batch.
- *
- * @return execute results
- * @throws SQLException SQL exception
- */
- public int[] executeBatch() throws SQLException {
- final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
- final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback callback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
-
- @Override
- protected int[] executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return sqlExecuteUnit.getStatement().executeBatch();
- }
- };
- return accumulate(executeCallback(callback));
- }
-
- private int[] accumulate(final List results) {
- int[] result = new int[batchCount];
- int count = 0;
- for (BatchPreparedStatementExecuteUnit each : getBatchPreparedStatementUnitGroups()) {
- for (Entry entry : each.getJdbcAndActualAddBatchCallTimesMap().entrySet()) {
- int value = null == results.get(count) ? 0 : results.get(count)[entry.getValue()];
- if (DatabaseType.Oracle == databaseType) {
- result[entry.getKey()] = value;
- } else {
- result[entry.getKey()] += value;
- }
- }
- count++;
- }
- return result;
- }
-
- protected abstract List executeCallback(SQLExecuteCallback executeCallback) throws SQLException;
-
- protected abstract Collection getBatchPreparedStatementUnitGroups();
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/ConnectionStrictlyBatchPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/ConnectionStrictlyBatchPreparedStatementExecutor.java
deleted file mode 100644
index a0955f49e2bb8..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/ConnectionStrictlyBatchPreparedStatementExecutor.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.batch;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Prepared statement executor to process add batch for connection strictly mode.
- *
- * @author zhangliang
- */
-public final class ConnectionStrictlyBatchPreparedStatementExecutor extends BatchPreparedStatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final List> batchPreparedStatementUnitGroups;
-
- public ConnectionStrictlyBatchPreparedStatementExecutor(final DatabaseType dbType, final SQLType sqlType, final int batchCount,
- final SQLExecuteTemplate executeTemplate, final List> batchPreparedStatementUnitGroups) {
- super(dbType, sqlType, batchCount);
- this.executeTemplate = executeTemplate;
- this.batchPreparedStatementUnitGroups = batchPreparedStatementUnitGroups;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.executeGroup((Collection) batchPreparedStatementUnitGroups, executeCallback);
- }
-
- @Override
- protected Collection getBatchPreparedStatementUnitGroups() {
- Collection result = new LinkedList<>();
- for (List each : batchPreparedStatementUnitGroups) {
- result.addAll(each);
- }
- return result;
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/MemoryStrictlyBatchPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/MemoryStrictlyBatchPreparedStatementExecutor.java
deleted file mode 100644
index c88b274118f54..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/batch/MemoryStrictlyBatchPreparedStatementExecutor.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.batch;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Prepared statement executor to process add batch for memory strictly mode.
- *
- * @author zhangliang
- */
-public final class MemoryStrictlyBatchPreparedStatementExecutor extends BatchPreparedStatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final Collection batchPreparedStatementUnits;
-
- public MemoryStrictlyBatchPreparedStatementExecutor(final DatabaseType dbType, final SQLType sqlType, final int batchCount,
- final SQLExecuteTemplate executeTemplate, final Collection batchPreparedStatementUnits) {
- super(dbType, sqlType, batchCount);
- this.executeTemplate = executeTemplate;
- this.batchPreparedStatementUnits = batchPreparedStatementUnits;
- }
-
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.execute(batchPreparedStatementUnits, executeCallback);
- }
-
- @Override
- protected Collection getBatchPreparedStatementUnitGroups() {
- return batchPreparedStatementUnits;
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/ConnectionStrictlyPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/ConnectionStrictlyPreparedStatementExecutor.java
deleted file mode 100644
index 8a825f75e064c..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/ConnectionStrictlyPreparedStatementExecutor.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.prepared;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.ShardingExecuteGroup;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Prepared statement executor for connection strictly mode.
- *
- * @author zhangliang
- */
-public final class ConnectionStrictlyPreparedStatementExecutor extends PreparedStatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final Collection> executeGroups;
-
- public ConnectionStrictlyPreparedStatementExecutor(
- final DatabaseType databaseType, final SQLType sqlType, final SQLExecuteTemplate executeTemplate, final Collection> executeGroups) {
- super(databaseType, sqlType);
- this.executeTemplate = executeTemplate;
- this.executeGroups = executeGroups;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.executeGroup((Collection) executeGroups, executeCallback);
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/MemoryStrictlyPreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/MemoryStrictlyPreparedStatementExecutor.java
deleted file mode 100644
index edf6029975fc5..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/MemoryStrictlyPreparedStatementExecutor.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.prepared;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Prepared statement executor for memory strictly mode.
- *
- * @author zhangliang
- */
-public final class MemoryStrictlyPreparedStatementExecutor extends PreparedStatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final Collection preparedStatementExecuteUnits;
-
- public MemoryStrictlyPreparedStatementExecutor(
- final DatabaseType databaseType, final SQLType sqlType, final SQLExecuteTemplate executeTemplate, final Collection preparedStatementExecuteUnits) {
- super(databaseType, sqlType);
- this.executeTemplate = executeTemplate;
- this.preparedStatementExecuteUnits = preparedStatementExecuteUnits;
- }
-
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.execute(preparedStatementExecuteUnits, executeCallback);
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecuteUnit.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecuteUnit.java
deleted file mode 100644
index aec1be1445b4e..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecuteUnit.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.prepared;
-
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
-import io.shardingsphere.core.routing.RouteUnit;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
-import java.sql.PreparedStatement;
-
-/**
- * Execute unit for JDBC prepared statement.
- *
- * @author zhangliang
- */
-@RequiredArgsConstructor
-@Getter
-public final class PreparedStatementExecuteUnit implements SQLExecuteUnit {
-
- private final RouteUnit routeUnit;
-
- private final PreparedStatement statement;
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecutor.java
deleted file mode 100644
index a1ff80edc6264..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/prepared/PreparedStatementExecutor.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.prepared;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.SQLExecuteUnit;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
-import lombok.RequiredArgsConstructor;
-
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Prepared statement executor.
- *
- * @author zhangliang
- * @author caohao
- * @author maxiaoguang
- */
-@RequiredArgsConstructor
-public abstract class PreparedStatementExecutor {
-
- private final DatabaseType databaseType;
-
- private final SQLType sqlType;
-
- /**
- * Execute query.
- *
- * @return result set list
- * @throws SQLException SQL exception
- */
- public List executeQuery() throws SQLException {
- final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
- final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
-
- @Override
- protected ResultSet executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return ((PreparedStatement) sqlExecuteUnit.getStatement()).executeQuery();
- }
- };
- return executeCallback(executeCallback);
- }
-
- /**
- * Execute update.
- *
- * @return effected records count
- * @throws SQLException SQL exception
- */
- public int executeUpdate() throws SQLException {
- final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
- final Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
-
- @Override
- protected Integer executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return ((PreparedStatement) sqlExecuteUnit.getStatement()).executeUpdate();
- }
- };
- List results = executeCallback(executeCallback);
- return accumulate(results);
- }
-
- private int accumulate(final List results) {
- int result = 0;
- for (Integer each : results) {
- result += null == each ? 0 : each;
- }
- return result;
- }
-
- /**
- * Execute SQL.
- *
- * @return return true if is DQL, false if is DML
- * @throws SQLException SQL exception
- */
- public boolean execute() throws SQLException {
- boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
- Map dataMap = ExecutorDataMap.getDataMap();
- SQLExecuteCallback executeCallback = new SQLExecuteCallback(databaseType, sqlType, isExceptionThrown, dataMap) {
-
- @Override
- protected Boolean executeSQL(final SQLExecuteUnit sqlExecuteUnit) throws SQLException {
- return ((PreparedStatement) sqlExecuteUnit.getStatement()).execute();
- }
- };
- List result = executeCallback(executeCallback);
- if (null == result || result.isEmpty() || null == result.get(0)) {
- return false;
- }
- return result.get(0);
- }
-
- protected abstract List executeCallback(SQLExecuteCallback executeCallback) throws SQLException;
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/ConnectionStrictlyStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/ConnectionStrictlyStatementExecutor.java
deleted file mode 100644
index cbfef654e8fe4..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/ConnectionStrictlyStatementExecutor.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.statement;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.ShardingExecuteGroup;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Statement executor for connection strictly mode.
- *
- * @author zhangliang
- */
-public final class ConnectionStrictlyStatementExecutor extends StatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final Collection> executeGroups;
-
- public ConnectionStrictlyStatementExecutor(
- final DatabaseType databaseType, final SQLType sqlType, final SQLExecuteTemplate executeTemplate, final Collection> executeGroups) {
- super(databaseType, sqlType);
- this.executeTemplate = executeTemplate;
- this.executeGroups = executeGroups;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.executeGroup((Collection) executeGroups, executeCallback);
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/MemoryStrictlyStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/MemoryStrictlyStatementExecutor.java
deleted file mode 100644
index a5412f3856fd3..0000000000000
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/statement/MemoryStrictlyStatementExecutor.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Copyright 2016-2018 shardingsphere.io.
- *
- * 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.
- *
- */
-
-package io.shardingsphere.core.executor.statement;
-
-import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.constant.SQLType;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
-
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * Statement executor for memory strictly mode.
- *
- * @author zhangliang
- */
-public final class MemoryStrictlyStatementExecutor extends StatementExecutor {
-
- private final SQLExecuteTemplate executeTemplate;
-
- private final Collection statementExecuteUnits;
-
- public MemoryStrictlyStatementExecutor(
- final DatabaseType databaseType, final SQLType sqlType, final SQLExecuteTemplate executeTemplate, final Collection statementExecuteUnits) {
- super(databaseType, sqlType);
- this.executeTemplate = executeTemplate;
- this.statementExecuteUnits = statementExecuteUnits;
- }
-
- @Override
- protected List executeCallback(final SQLExecuteCallback executeCallback) throws SQLException {
- return executeTemplate.execute(statementExecuteUnits, executeCallback);
- }
-}
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractConnectionAdapter.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractConnectionAdapter.java
index c716a865526ff..c7a41a1f0dce2 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractConnectionAdapter.java
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractConnectionAdapter.java
@@ -18,6 +18,8 @@
package io.shardingsphere.core.jdbc.adapter;
import com.google.common.base.Preconditions;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
import io.shardingsphere.core.constant.DatabaseType;
import io.shardingsphere.core.constant.transaction.TransactionOperationType;
import io.shardingsphere.core.constant.transaction.TransactionType;
@@ -29,7 +31,6 @@
import io.shardingsphere.core.event.connection.GetConnectionFinishEvent;
import io.shardingsphere.core.event.connection.GetConnectionStartEvent;
import io.shardingsphere.core.event.root.RootInvokeEvent;
-import io.shardingsphere.core.event.root.RootInvokeFinishEvent;
import io.shardingsphere.core.event.transaction.xa.XATransactionEvent;
import io.shardingsphere.core.hint.HintManagerHolder;
import io.shardingsphere.core.jdbc.adapter.executor.ForceExecuteCallback;
@@ -45,20 +46,23 @@
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import java.util.Map;
/**
* Adapter for {@code Connection}.
*
* @author zhangliang
+ * @author panjuan
*/
@RequiredArgsConstructor
public abstract class AbstractConnectionAdapter extends AbstractUnsupportedOperationConnection {
private final DatabaseType databaseType;
- private final Map cachedConnections = new HashMap<>();
+ private final Multimap cachedConnections = HashMultimap.create();
private boolean autoCommit = true;
@@ -80,31 +84,56 @@ public abstract class AbstractConnectionAdapter extends AbstractUnsupportedOpera
* @throws SQLException SQL exception
*/
public final Connection getConnection(final String dataSourceName) throws SQLException {
+ return getConnections(dataSourceName, 1).get(0);
+ }
+
+ /**
+ * Get database connections.
+ *
+ * @param dataSourceName data source name
+ * @param connectionSize size of connection list to be get
+ * @return database connections
+ * @throws SQLException SQL exception
+ */
+ public final List getConnections(final String dataSourceName, final int connectionSize) throws SQLException {
ShardingEventBusInstance.getInstance().post(new GetConnectionStartEvent(dataSourceName));
- try {
- if (cachedConnections.containsKey(dataSourceName)) {
- GetConnectionEvent finishEvent = new GetConnectionFinishEvent(DataSourceMetaDataFactory.newInstance(databaseType, cachedConnections.get(dataSourceName).getMetaData().getURL()));
- finishEvent.setExecuteSuccess();
- ShardingEventBusInstance.getInstance().post(finishEvent);
- return cachedConnections.get(dataSourceName);
+ DataSource dataSource = getDataSourceMap().get(dataSourceName);
+ Preconditions.checkState(null != dataSource, "Missing the data source name: '%s'", dataSourceName);
+ Collection connections = cachedConnections.get(dataSourceName);
+ List result;
+ if (connections.size() >= connectionSize) {
+ result = new ArrayList<>(cachedConnections.get(dataSourceName)).subList(0, connectionSize);
+ } else if (!connections.isEmpty()) {
+ result = new ArrayList<>(connectionSize);
+ result.addAll(connections);
+ List newConnections = createConnections(dataSource, connectionSize - connections.size());
+ result.addAll(newConnections);
+ cachedConnections.putAll(dataSourceName, newConnections);
+ } else {
+ result = new ArrayList<>(createConnections(dataSource, connectionSize));
+ cachedConnections.putAll(dataSourceName, result);
+ }
+ postGetConnectionEvent(result);
+ return result;
+ }
+
+ @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
+ private synchronized List createConnections(final DataSource dataSource, final int connectionSize) throws SQLException {
+ List result = new ArrayList<>(connectionSize);
+ synchronized (dataSource) {
+ for (int i = 0; i < connectionSize; i++) {
+ Connection connection = dataSource.getConnection();
+ replayMethodsInvocation(connection);
+ result.add(connection);
}
- DataSource dataSource = getDataSourceMap().get(dataSourceName);
- Preconditions.checkState(null != dataSource, "Missing the data source name: '%s'", dataSourceName);
- Connection result = dataSource.getConnection();
- cachedConnections.put(dataSourceName, result);
- replayMethodsInvocation(result);
- GetConnectionEvent finishEvent = new GetConnectionFinishEvent(DataSourceMetaDataFactory.newInstance(databaseType, cachedConnections.get(dataSourceName).getMetaData().getURL()));
- finishEvent.setExecuteSuccess();
- ShardingEventBusInstance.getInstance().post(finishEvent);
- return result;
- // CHECKSTYLE:OFF
- } catch (final Exception ex) {
- // CHECKSTYLE:ON
- GetConnectionEvent finishEvent = new GetConnectionFinishEvent(null);
- finishEvent.setExecuteFailure(ex);
- ShardingEventBusInstance.getInstance().post(finishEvent);
- throw ex;
}
+ return result;
+ }
+
+ private void postGetConnectionEvent(final List connections) throws SQLException {
+ GetConnectionEvent finishEvent = new GetConnectionFinishEvent(DataSourceMetaDataFactory.newInstance(databaseType, connections.get(0).getMetaData().getURL()));
+ finishEvent.setExecuteSuccess();
+ ShardingEventBusInstance.getInstance().post(finishEvent);
}
protected abstract Map getDataSourceMap();
@@ -174,7 +203,7 @@ public final void close() throws SQLException {
HintManagerHolder.clear();
MasterVisitedManager.clear();
TransactionTypeHolder.clear();
- forceExecuteTemplateForClose.execute(cachedConnections.entrySet(), new ForceExecuteCallback>() {
+ forceExecuteTemplateForClose.execute(cachedConnections.entries(), new ForceExecuteCallback>() {
@Override
public void execute(final Map.Entry cachedConnectionsEntrySet) throws SQLException {
@@ -195,11 +224,11 @@ public void execute(final Map.Entry cachedConnectionsEntrySe
}
}
});
- RootInvokeEvent finishEvent = new RootInvokeFinishEvent();
+ RootInvokeEvent finishEvent = new RootInvokeEvent();
finishEvent.setExecuteSuccess();
ShardingEventBusInstance.getInstance().post(finishEvent);
}
-
+
@Override
public final boolean isClosed() {
return closed;
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractDataSourceAdapter.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractDataSourceAdapter.java
index b1774f045bf15..fd2ae92eeca18 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractDataSourceAdapter.java
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/adapter/AbstractDataSourceAdapter.java
@@ -22,6 +22,8 @@
import io.shardingsphere.core.event.ShardingEventListenerRegistrySPILoader;
import io.shardingsphere.core.jdbc.unsupported.AbstractUnsupportedOperationDataSource;
import lombok.Getter;
+import lombok.Setter;
+
import javax.sql.DataSource;
import java.io.PrintWriter;
import java.sql.Connection;
@@ -35,14 +37,15 @@
* @author zhangliang
* @author panjuan
*/
+@Getter
+@Setter
public abstract class AbstractDataSourceAdapter extends AbstractUnsupportedOperationDataSource {
static {
ShardingEventListenerRegistrySPILoader.registerListeners();
}
- @Getter
- private DatabaseType databaseType;
+ private final DatabaseType databaseType;
private PrintWriter logWriter = new PrintWriter(System.out);
@@ -73,16 +76,6 @@ private DatabaseType getDatabaseType(final DataSource dataSource) throws SQLExce
}
}
- @Override
- public final PrintWriter getLogWriter() {
- return logWriter;
- }
-
- @Override
- public final void setLogWriter(final PrintWriter out) {
- this.logWriter = out;
- }
-
@Override
public final Logger getParentLogger() {
return Logger.getLogger(Logger.GLOBAL_LOGGER_NAME);
diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/core/ShardingContext.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/core/ShardingContext.java
index 339e37c7b97cf..5db5a4374bfe4 100644
--- a/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/core/ShardingContext.java
+++ b/sharding-jdbc/src/main/java/io/shardingsphere/core/jdbc/core/ShardingContext.java
@@ -17,9 +17,7 @@
package io.shardingsphere.core.jdbc.core;
-import io.shardingsphere.core.constant.ConnectionMode;
import io.shardingsphere.core.constant.DatabaseType;
-import io.shardingsphere.core.exception.ShardingException;
import io.shardingsphere.core.executor.ShardingExecuteEngine;
import io.shardingsphere.core.jdbc.metadata.JDBCTableMetaDataConnectionManager;
import io.shardingsphere.core.metadata.ShardingMetaData;
@@ -42,37 +40,30 @@
@Getter
public final class ShardingContext implements AutoCloseable {
- private ShardingRule shardingRule;
+ private final ShardingRule shardingRule;
- private DatabaseType databaseType;
+ private final DatabaseType databaseType;
- private ShardingExecuteEngine executeEngine;
+ private final ShardingExecuteEngine executeEngine;
- private ShardingMetaData metaData;
+ private final int maxConnectionsSizePerQuery;
- private ConnectionMode connectionMode;
+ private final boolean showSQL;
- private int maxConnectionsSizePerQuery;
+ private final ShardingMetaData metaData;
- private boolean showSQL;
-
- public ShardingContext(final Map dataSourceMap, final ShardingRule shardingRule, final DatabaseType databaseType,
- final ShardingExecuteEngine executeEngine, final ConnectionMode connectionMode, final int maxConnectionsSizePerQuery, final boolean showSQL) {
- init(dataSourceMap, shardingRule, databaseType, executeEngine, connectionMode, maxConnectionsSizePerQuery, showSQL);
- }
-
- private void init(final Map dataSourceMap, final ShardingRule shardingRule, final DatabaseType databaseType,
- final ShardingExecuteEngine executeEngine, final ConnectionMode connectionMode, final int maxConnectionsSizePerQuery, final boolean showSQL) {
+ public ShardingContext(final Map dataSourceMap, final ShardingRule shardingRule,
+ final DatabaseType databaseType, final ShardingExecuteEngine executeEngine, final int maxConnectionsSizePerQuery, final boolean showSQL) throws SQLException {
this.shardingRule = shardingRule;
- this.executeEngine = executeEngine;
this.databaseType = databaseType;
- this.connectionMode = connectionMode;
+ this.executeEngine = executeEngine;
this.maxConnectionsSizePerQuery = maxConnectionsSizePerQuery;
this.showSQL = showSQL;
- metaData = new ShardingMetaData(getDataSourceURLs(dataSourceMap), shardingRule, databaseType, executeEngine, new JDBCTableMetaDataConnectionManager(dataSourceMap), maxConnectionsSizePerQuery);
+ metaData = new ShardingMetaData(
+ getDataSourceURLs(dataSourceMap), shardingRule, databaseType, executeEngine, new JDBCTableMetaDataConnectionManager(dataSourceMap), maxConnectionsSizePerQuery);
}
- private static Map getDataSourceURLs(final Map dataSourceMap) {
+ private Map getDataSourceURLs(final Map dataSourceMap) throws SQLException {
Map result = new LinkedHashMap<>(dataSourceMap.size(), 1);
for (Entry entry : dataSourceMap.entrySet()) {
result.put(entry.getKey(), getDataSourceURL(entry.getValue()));
@@ -80,11 +71,9 @@ private static Map getDataSourceURLs(final Map