diff --git a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceMBean.java b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceMBean.java index 807851abe8..d594b0d0d5 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceMBean.java +++ b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceMBean.java @@ -17,33 +17,5 @@ package com.alibaba.druid.pool.ha; public interface MultiDataSourceMBean { - void failureDetect(); - - String[] getDataSourceNames(); - - boolean restartDataSource(String name); - - boolean startFailureDetectScheduleTask(); - - boolean stopFailureDetectScheduleTask(); - - boolean startConfigLoadScheduleTask(); - - boolean stopConfigLoadScheduleTask(); - - long getConfigLoadCount(); - - long getFailureDetectCount(); - - long getRetryGetConnectionCount(); - - long getBusySkipCount(); - - long getFailureDetectPeriodMillis(); - - long getConfigLoadPeriodMillis(); - long getActiveCount(); - - int getMaxPoolSize(); } diff --git a/src/main/java/com/alibaba/druid/proxy/jdbc/JdbcParameter.java b/src/main/java/com/alibaba/druid/proxy/jdbc/JdbcParameter.java index c00241af23..1a77f9556a 100644 --- a/src/main/java/com/alibaba/druid/proxy/jdbc/JdbcParameter.java +++ b/src/main/java/com/alibaba/druid/proxy/jdbc/JdbcParameter.java @@ -32,12 +32,18 @@ public class JdbcParameter { private final Object value; private final long length; private final Calendar calendar; + private final int scaleOrLength; - public JdbcParameter(int sqlType, Object value, long length, Calendar calendar){ + public JdbcParameter(int sqlType, Object value, long length, Calendar calendar, int scaleOrLength){ this.sqlType = sqlType; this.value = value; this.length = length; this.calendar = calendar; + this.scaleOrLength = scaleOrLength; + } + + public JdbcParameter(int sqlType, Object value, long length, Calendar calendar){ + this(sqlType, value, -1, null, -1); } public JdbcParameter(int sqlType, Object value){ @@ -52,6 +58,10 @@ public class JdbcParameter { this(sqlType, value, -1, calendar); } + public int getScaleOrLength() { + return scaleOrLength; + } + public Object getValue() { return value; } diff --git a/src/main/java/com/alibaba/druid/proxy/jdbc/PreparedStatementProxyImpl.java b/src/main/java/com/alibaba/druid/proxy/jdbc/PreparedStatementProxyImpl.java index ad26d7b631..3461b5cd2f 100644 --- a/src/main/java/com/alibaba/druid/proxy/jdbc/PreparedStatementProxyImpl.java +++ b/src/main/java/com/alibaba/druid/proxy/jdbc/PreparedStatementProxyImpl.java @@ -368,7 +368,7 @@ public class PreparedStatementProxyImpl extends StatementProxyImpl implements Pr @Override public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException { - setParameter(parameterIndex, new JdbcParameter(targetSqlType, x)); + setParameter(parameterIndex, new JdbcParameter(targetSqlType, x, -1, null, scaleOrLength)); createChain().preparedStatement_setObject(this, parameterIndex, x, targetSqlType, scaleOrLength); } diff --git a/src/main/java/com/alibaba/druid/sharding/ShardingConnection.java b/src/main/java/com/alibaba/druid/sharding/ShardingConnection.java new file mode 100644 index 0000000000..2fb7d8a264 --- /dev/null +++ b/src/main/java/com/alibaba/druid/sharding/ShardingConnection.java @@ -0,0 +1,683 @@ +package com.alibaba.druid.sharding; + +import java.sql.Array; +import java.sql.Blob; +import java.sql.CallableStatement; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.NClob; +import java.sql.PreparedStatement; +import java.sql.SQLClientInfoException; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLWarning; +import java.sql.SQLXML; +import java.sql.Savepoint; +import java.sql.Statement; +import java.sql.Struct; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.Executor; + +import com.alibaba.druid.pool.WrapperAdapter; +import com.alibaba.druid.pool.ha.DataSourceHolder; +import com.alibaba.druid.proxy.jdbc.CallableStatementProxyImpl; +import com.alibaba.druid.proxy.jdbc.ConnectionProxy; +import com.alibaba.druid.proxy.jdbc.DataSourceProxy; +import com.alibaba.druid.proxy.jdbc.TransactionInfo; +import com.alibaba.druid.util.JdbcUtils; + +public class ShardingConnection extends WrapperAdapter implements Connection, ConnectionProxy { + + private final ShardingDataSource dataSource; + + private Connection conn; + private DataSourceHolder dataSourceHolder; + + private final long id; + + private Boolean autoCommit = null; + private Boolean readOnly = null; + private String catalog = null; + private Integer transactionLeval = null; + private Map> typeMap = null; + private Integer holdability = null; + + private Properties clientInfo = null; + + private Map attributes = null; + + private Date connectedTime = null; + + private int closeCount = 0; + + private TransactionInfo transcationInfo; + + private String database; + + public ShardingConnection(ShardingDataSource dataSource, long id){ + this.dataSource = dataSource; + this.id = id; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public DataSourceHolder getDataSourceHolder() { + return dataSourceHolder; + } + + public void checkConnection(String sql) throws SQLException { + if (conn == null) { + conn = dataSource.getConnectionBySql(sql); + } + + initConnection(conn); + + connectedTime = new Date(); + } + + private void initConnection(Connection conn) throws SQLException, SQLClientInfoException { + if (autoCommit != null) { + conn.setAutoCommit(autoCommit); + } + + if (readOnly != null) { + conn.setReadOnly(readOnly); + } + + if (catalog != null) { + conn.setCatalog(catalog); + } + + if (transactionLeval != null) { + conn.setTransactionIsolation(transactionLeval); + } + + if (typeMap != null) { + conn.setTypeMap(typeMap); + } + + if (holdability != null) { + conn.setHoldability(holdability); + } + + if (clientInfo != null) { + conn.setClientInfo(clientInfo); + } + } + + public ShardingDataSource getDataSource() { + return dataSource; + } + + @Override + public String nativeSQL(String sql) throws SQLException { + checkConnection(sql); + + return conn.nativeSQL(sql); + } + + public Connection createRealConnectionByDb(String db) throws SQLException { + if (conn != null) { + JdbcUtils.close(conn); + conn = null; + } + + conn = dataSource.getConnectionByDb(db); + initConnection(conn); + return conn; + } + + public Connection getRealConnection() { + return conn; + } + + public void closeRealConnection() { + if (conn != null) { + JdbcUtils.close(conn); + conn = null; + } + } + + @Override + public void setAutoCommit(boolean autoCommit) throws SQLException { + if (conn != null) { + conn.setAutoCommit(autoCommit); + } else { + this.autoCommit = Boolean.valueOf(autoCommit); + } + + if (!autoCommit) { + if (transcationInfo == null) { + long transactionId = this.getDirectDataSource().createTransactionId(); + transcationInfo = new TransactionInfo(transactionId); + } + } else { + transcationInfo = null; + } + } + + @Override + public boolean getAutoCommit() throws SQLException { + if (conn != null) { + return conn.getAutoCommit(); + } + + if (this.autoCommit != null) { + return this.autoCommit.booleanValue(); + } + + return false; + } + + @Override + public void commit() throws SQLException { + if (conn != null) { + conn.commit(); + } + + if (transcationInfo != null) { + transcationInfo.setEndTimeMillis(); + } + } + + @Override + public void rollback() throws SQLException { + if (conn != null) { + conn.rollback(); + } + + if (transcationInfo != null) { + transcationInfo.setEndTimeMillis(); + } + } + + @Override + public void close() throws SQLException { + if (closeCount > 0) { + closeCount++; + return; + } + + if (conn != null) { + conn.close(); + } + this.closeCount++; + + dataSource.afterConnectionClosed(this); + } + + @Override + public boolean isClosed() throws SQLException { + return closeCount > 0; + } + + @Override + public DatabaseMetaData getMetaData() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.getMetaData(); + } + + @Override + public void setReadOnly(boolean readOnly) throws SQLException { + if (conn != null) { + conn.setReadOnly(readOnly); + } else { + this.readOnly = Boolean.valueOf(readOnly); + } + } + + @Override + public boolean isReadOnly() throws SQLException { + if (conn != null) { + return conn.isReadOnly(); + } + + if (this.readOnly != null) { + return this.readOnly.booleanValue(); + } + + return false; + } + + @Override + public void setCatalog(String catalog) throws SQLException { + if (conn != null) { + conn.setCatalog(catalog); + } else { + this.catalog = catalog; + } + } + + @Override + public String getCatalog() throws SQLException { + if (conn != null) { + return conn.getCatalog(); + } + + return this.catalog; + } + + @Override + public void setTransactionIsolation(int level) throws SQLException { + if (conn != null) { + conn.setTransactionIsolation(level); + } else { + this.transactionLeval = level; + } + } + + @Override + public int getTransactionIsolation() throws SQLException { + if (conn != null) { + return conn.getTransactionIsolation(); + } + + if (this.transactionLeval != null) { + return this.transactionLeval; + } + + return TRANSACTION_NONE; + } + + @Override + public SQLWarning getWarnings() throws SQLException { + if (conn != null) { + return conn.getWarnings(); + } + + return null; + } + + @Override + public void clearWarnings() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + if (conn != null) { + conn.clearWarnings(); + } + } + + @Override + public Map> getTypeMap() throws SQLException { + if (conn != null) { + return conn.getTypeMap(); + } + + return typeMap; + } + + @Override + public void setTypeMap(Map> map) throws SQLException { + if (conn != null) { + conn.setTypeMap(map); + } else { + this.typeMap = map; + } + } + + @Override + public void setHoldability(int holdability) throws SQLException { + if (conn != null) { + conn.setHoldability(holdability); + } else { + this.holdability = holdability; + } + } + + @Override + public int getHoldability() throws SQLException { + if (conn != null) { + return conn.getHoldability(); + } + + if (this.holdability != null) { + return this.holdability; + } + + return 0; + } + + @Override + public Savepoint setSavepoint() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.setSavepoint(); + } + + @Override + public Savepoint setSavepoint(String name) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.setSavepoint(name); + } + + @Override + public void rollback(Savepoint savepoint) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + conn.rollback(savepoint); + + if (transcationInfo != null) { + transcationInfo.setEndTimeMillis(); + } + } + + @Override + public void releaseSavepoint(Savepoint savepoint) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + conn.releaseSavepoint(savepoint); + } + + @Override + public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException { + long stmtId = dataSource.createStatementId(); + ShardingStatement stmt = new ShardingStatement(this, stmtId, resultSetType, resultSetConcurrency); + return stmt; + } + + @Override + public Statement createStatement() throws SQLException { + long stmtId = dataSource.createStatementId(); + ShardingStatement stmt = new ShardingStatement(this, stmtId); + return stmt; + } + + @Override + public PreparedStatement prepareStatement(String sql) throws SQLException { + long stmtId = dataSource.createStatementId(); + return new ShardingPreparedStatement(this, sql, stmtId); + } + + @Override + public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) + throws SQLException { + long stmtId = dataSource.createStatementId(); + + return new ShardingPreparedStatement(this, sql, stmtId); + } + + @Override + public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException { + // TODO + throw new UnsupportedOperationException(); + } + + @Override + public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException { + // TODO + throw new UnsupportedOperationException(); + } + + @Override + public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException { + // TODO + throw new UnsupportedOperationException(); + } + + @Override + public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, + int resultSetHoldability) throws SQLException { + + long stmtId = dataSource.createStatementId(); + + return new ShardingPreparedStatement(this, sql, stmtId, resultSetType, resultSetConcurrency, + resultSetHoldability); + } + + @Override + public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException { + throw new UnsupportedOperationException(); + } + + @Override + public CallableStatement prepareCall(String sql) throws SQLException { + checkConnection(sql); + + CallableStatement stmt = conn.prepareCall(sql); + long stmtId = dataSource.createStatementId(); + + return new CallableStatementProxyImpl(this, stmt, sql, stmtId); + } + + @Override + public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) + throws SQLException { + long stmtId = dataSource.createStatementId(); + ShardingStatement stmt = new ShardingStatement(this, stmtId, resultSetType, resultSetConcurrency, + resultSetHoldability); + return stmt; + } + + @Override + public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, + int resultSetHoldability) throws SQLException { + checkConnection(sql); + + CallableStatement stmt = conn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability); + long stmtId = dataSource.createStatementId(); + + return new CallableStatementProxyImpl(this, stmt, sql, stmtId); + } + + @Override + public Clob createClob() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createClob(); + } + + @Override + public Blob createBlob() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createBlob(); + } + + @Override + public NClob createNClob() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createNClob(); + } + + @Override + public SQLXML createSQLXML() throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createSQLXML(); + } + + @Override + public boolean isValid(int timeout) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.isValid(timeout); + } + + @Override + public void setClientInfo(String name, String value) throws SQLClientInfoException { + if (conn != null) { + conn.setClientInfo(name, value); + return; + } + + if (clientInfo == null) { + clientInfo = new Properties(); + } + clientInfo.setProperty(name, value); + } + + @Override + public void setClientInfo(Properties properties) throws SQLClientInfoException { + if (conn != null) { + conn.setClientInfo(properties); + return; + } + + this.clientInfo = properties; + } + + @Override + public String getClientInfo(String name) throws SQLException { + if (conn != null) { + return conn.getClientInfo(name); + } + + if (clientInfo == null) { + return null; + } + + return clientInfo.getProperty(name); + } + + @Override + public Properties getClientInfo() throws SQLException { + if (conn != null) { + return conn.getClientInfo(); + } + + return clientInfo; + } + + @Override + public Array createArrayOf(String typeName, Object[] elements) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createArrayOf(typeName, elements); + } + + @Override + public Struct createStruct(String typeName, Object[] attributes) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + return conn.createStruct(typeName, attributes); + } + + // ///// + + @Override + public long getId() { + return id; + } + + @Override + public Connection getRawObject() { + return conn; + } + + @Override + public Map getAttributes() { + if (attributes == null) { + attributes = new HashMap(); + } + return attributes; + } + + @Override + public Properties getProperties() { + return dataSource.getProperties(); + } + + @Override + public DataSourceProxy getDirectDataSource() { + return dataSource; + } + + @Override + public Date getConnectedTime() { + return connectedTime; + } + + public boolean isWrapperFor(Class iface) throws SQLException { + boolean result = super.isWrapperFor(iface); + + if ((!result) && conn != null) { + result = conn.isWrapperFor(iface); + } + + return result; + } + + public T unwrap(Class iface) throws SQLException { + T object = super.unwrap(iface); + + if (object == null && conn != null) { + object = conn.unwrap(iface); + } + + return object; + } + + public void setSchema(String schema) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + public String getSchema() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + public void abort(Executor executor) throws SQLException { + if (conn == null) { + throw new SQLException("connection not init"); + } + + throw new SQLFeatureNotSupportedException(); + } + + public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + public int getNetworkTimeout() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public TransactionInfo getTransactionInfo() { + return null; + } + + @Override + public int getCloseCount() { + return closeCount; + } + +} diff --git a/src/main/java/com/alibaba/druid/sharding/ShardingDataSource.java b/src/main/java/com/alibaba/druid/sharding/ShardingDataSource.java index 7933cf643a..f5846b49a1 100644 --- a/src/main/java/com/alibaba/druid/sharding/ShardingDataSource.java +++ b/src/main/java/com/alibaba/druid/sharding/ShardingDataSource.java @@ -15,10 +15,148 @@ */ package com.alibaba.druid.sharding; +import java.sql.Connection; +import java.sql.Driver; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicLong; + import javax.sql.DataSource; +import com.alibaba.druid.filter.Filter; import com.alibaba.druid.pool.DataSourceAdapter; +import com.alibaba.druid.pool.DruidPooledConnection; +import com.alibaba.druid.proxy.jdbc.DataSourceProxy; +import com.alibaba.druid.sharding.config.RouteConfig; +import com.alibaba.druid.sharding.sql.MySqlShardingVisitor; +import com.alibaba.druid.sharding.sql.ShardingVisitor; +import com.alibaba.druid.sql.ast.SQLStatement; +import com.alibaba.druid.sql.parser.SQLParserUtils; +import com.alibaba.druid.sql.parser.SQLStatementParser; +import com.alibaba.druid.stat.JdbcDataSourceStat; -public class ShardingDataSource extends DataSourceAdapter implements DataSource { +public class ShardingDataSource extends DataSourceAdapter implements DataSource, DataSourceProxy { + + private final AtomicLong connectionIdSeed = new AtomicLong(); + private final AtomicLong statementIdSeed = new AtomicLong(); + private final AtomicLong resultSetIdSeed = new AtomicLong(); + private final AtomicLong transactionIdSeed = new AtomicLong(); + private final Properties properties = new Properties(); + private String dbType; + private List filters = new CopyOnWriteArrayList(); + + private RouteConfig routeConfig = new RouteConfig(); + + public Connection getConnectionByPartition(String partition) { + throw new UnsupportedOperationException(); + } + + public Connection getConnectionByDb(String partition) { + throw new UnsupportedOperationException(); + } + + public DruidPooledConnection getConnectionBySql(String partition) { + throw new UnsupportedOperationException(); + } + + protected void afterConnectionClosed(ShardingConnection conn) { + + } + + public ShardingVisitor createShardingVisitor() { + return createShardingVisitor(new ArrayList()); + } + + public SQLStatement parseStatement(String sql) throws SQLException { + SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(sql, dbType); + List stmtList = parser.parseStatementList(); + if (stmtList.size() > 0) { + throw new SQLException("not support multi-statement"); + } + + return stmtList.get(0); + } + + public ShardingVisitor createShardingVisitor(List parameters) { + return new MySqlShardingVisitor(routeConfig, parameters); + } + + public RouteConfig getRouteConfig() { + return routeConfig; + } + + public void setRouteConfig(RouteConfig routeConfig) { + this.routeConfig = routeConfig; + } + + public void setDbType(String dbType) { + this.dbType = dbType; + } + + public long createStatementId() { + return statementIdSeed.incrementAndGet(); + } + + public long createConnectionId() { + return connectionIdSeed.getAndIncrement(); + } + + public Properties getProperties() { + return properties; + } + + @Override + public JdbcDataSourceStat getDataSourceStat() { + return null; + } + + @Override + public String getName() { + // TODO Auto-generated method stub + return null; + } + + @Override + public String getDbType() { + return dbType; + } + + @Override + public Driver getRawDriver() { + return null; + } + + @Override + public String getUrl() { + return null; + } + + @Override + public String getRawJdbcUrl() { + return null; + } + + @Override + public List getProxyFilters() { + return filters; + } + + @Override + public long createResultSetId() { + return resultSetIdSeed.incrementAndGet(); + } + + @Override + public long createTransactionId() { + return transactionIdSeed.incrementAndGet(); + } + + @Override + public Properties getConnectProperties() { + return properties; + } } diff --git a/src/main/java/com/alibaba/druid/sharding/ShardingPreparedStatement.java b/src/main/java/com/alibaba/druid/sharding/ShardingPreparedStatement.java new file mode 100644 index 0000000000..6043e44ef5 --- /dev/null +++ b/src/main/java/com/alibaba/druid/sharding/ShardingPreparedStatement.java @@ -0,0 +1,701 @@ +package com.alibaba.druid.sharding; + +import java.io.InputStream; +import java.io.Reader; +import java.math.BigDecimal; +import java.net.URL; +import java.sql.Array; +import java.sql.Blob; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.Date; +import java.sql.NClob; +import java.sql.ParameterMetaData; +import java.sql.PreparedStatement; +import java.sql.Ref; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.RowId; +import java.sql.SQLException; +import java.sql.SQLXML; +import java.sql.Time; +import java.sql.Timestamp; +import java.sql.Types; +import java.util.ArrayList; +import java.util.Calendar; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +import org.h2.util.StringUtils; + +import com.alibaba.druid.proxy.jdbc.JdbcParameter; +import com.alibaba.druid.proxy.jdbc.JdbcParameter.TYPE; +import com.alibaba.druid.sharding.sql.ShardingVisitor; +import com.alibaba.druid.sql.SQLUtils; +import com.alibaba.druid.sql.ast.SQLStatement; +import com.alibaba.druid.sql.ast.statement.SQLSelectStatement; +import com.alibaba.druid.util.JdbcUtils; + +public class ShardingPreparedStatement extends ShardingStatement implements PreparedStatement { + + private final String sql; + private PreparedStatement pstmt; + + private List tables; + protected final Map parameters = new TreeMap(); + + public ShardingPreparedStatement(ShardingConnection conn, String sql, long id){ + super(conn, id); + this.sql = sql; + } + + public ShardingPreparedStatement(ShardingConnection conn, String sql, long id, int resultSetType, + int resultSetConcurrency, int resultSetHoldability){ + super(conn, id, resultSetType, resultSetConcurrency, resultSetHoldability); + this.sql = sql; + } + + public ShardingPreparedStatement(ShardingConnection conn, String sql, long id, int resultSetType, + int resultSetConcurrency){ + super(conn, id, resultSetType, resultSetConcurrency); + this.sql = sql; + } + + @Override + public void setArray(int parameterIndex, Array x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.ARRAY, x)); + } + + @Override + public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.AsciiInputStream, x)); + } + + @Override + public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.AsciiInputStream, x, length)); + } + + @Override + public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.AsciiInputStream, x, length)); + } + + @Override + public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.DECIMAL, x)); + } + + @Override + public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.BinaryInputStream, x)); + + } + + @Override + public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.BinaryInputStream, x, length)); + + } + + @Override + public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(JdbcParameter.TYPE.BinaryInputStream, x, length)); + + } + + @Override + public void setBlob(int parameterIndex, Blob x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.BLOB, x)); + + } + + @Override + public void setBlob(int parameterIndex, InputStream x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.BLOB, x)); + + } + + @Override + public void setBlob(int parameterIndex, InputStream x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.BLOB, x, length)); + } + + @Override + public void setBoolean(int parameterIndex, boolean x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.BOOLEAN, x)); + } + + @Override + public void setByte(int parameterIndex, byte x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.TINYINT, x)); + + } + + @Override + public void setBytes(int parameterIndex, byte[] x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.BYTES, x)); + + } + + @Override + public void setCharacterStream(int parameterIndex, Reader x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.CharacterInputStream, x)); + + } + + @Override + public void setCharacterStream(int parameterIndex, Reader x, int length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.CharacterInputStream, x, length)); + + } + + @Override + public void setCharacterStream(int parameterIndex, Reader x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.CharacterInputStream, x, length)); + + } + + @Override + public void setClob(int parameterIndex, Clob x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.CLOB, x)); + + } + + @Override + public void setClob(int parameterIndex, Reader x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.CLOB, x)); + + } + + @Override + public void setClob(int parameterIndex, Reader x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.CLOB, x, length)); + + } + + @Override + public void setDate(int parameterIndex, Date x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.DATE, x)); + + } + + @Override + public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.DATE, x, cal)); + + } + + @Override + public void setDouble(int parameterIndex, double x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.DOUBLE, x)); + + } + + @Override + public void setFloat(int parameterIndex, float x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.FLOAT, x)); + + } + + @Override + public void setInt(int parameterIndex, int x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.INTEGER, x)); + + } + + @Override + public void setLong(int parameterIndex, long x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.BIGINT, x)); + + } + + @Override + public void setNCharacterStream(int parameterIndex, Reader x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.NCharacterInputStream, x)); + + } + + @Override + public void setNCharacterStream(int parameterIndex, Reader x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.NCharacterInputStream, x, length)); + + } + + @Override + public void setNClob(int parameterIndex, NClob x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.NCLOB, x)); + + } + + @Override + public void setNClob(int parameterIndex, Reader x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.NCLOB, x)); + + } + + @Override + public void setNClob(int parameterIndex, Reader x, long length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.NCLOB, x, length)); + + } + + @Override + public void setNString(int parameterIndex, String x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.NVARCHAR, x)); + + } + + @Override + public void setNull(int parameterIndex, int sqlType) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(sqlType, null)); + + } + + @Override + public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(sqlType, null)); + + } + + @Override + public void setObject(int parameterIndex, Object x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.OTHER, null)); + + } + + @Override + public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(targetSqlType, x)); + + } + + @Override + public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(targetSqlType, x, -1, null, scaleOrLength)); + } + + @Override + public void setRef(int parameterIndex, Ref x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.REF, x)); + + } + + @Override + public void setRowId(int parameterIndex, RowId x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.ROWID, x)); + + } + + @Override + public void setSQLXML(int parameterIndex, SQLXML x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.SQLXML, x)); + + } + + @Override + public void setShort(int parameterIndex, short x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.SMALLINT, x)); + + } + + @Override + public void setString(int parameterIndex, String x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.VARCHAR, x)); + + } + + @Override + public void setTime(int parameterIndex, Time x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.TIME, x)); + + } + + @Override + public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.TIME, x, cal)); + + } + + @Override + public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.TIMESTAMP, x)); + + } + + @Override + public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(Types.TIMESTAMP, x)); + + } + + @Override + public void setURL(int parameterIndex, URL x) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.URL, x)); + + } + + @Override + public void setUnicodeStream(int parameterIndex, InputStream x, int length) throws SQLException { + setParameter(parameterIndex, new JdbcParameter(TYPE.UnicodeStream, x, length)); + + } + + public void setParameter(int parameterIndex, JdbcParameter parameter) { + this.getParameters().put(parameterIndex, parameter); + } + + public Map getParameters() { + return parameters; + } + + @Override + public void clearParameters() throws SQLException { + this.parameters.clear(); + } + + @Override + public ResultSetMetaData getMetaData() throws SQLException { + if (pstmt != null) { + return pstmt.getMetaData(); + } + return null; + } + + @Override + public ParameterMetaData getParameterMetaData() throws SQLException { + if (pstmt != null) { + return pstmt.getParameterMetaData(); + } + return null; + } + + @Override + public ResultSet executeQuery() throws SQLException { + List values = getParameterValues(); + + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(values); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + checkParameterValues(values, resultSqlStmtList); + + if (resultSqlStmtList.size() > 1) { + throw new SQLException("executeQuery not support multi-statement"); + } + + SQLStatement resultSqlStmt = resultSqlStmtList.get(0); + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + checkPreparedStatement(resultSqlStmt, resultSql, database); + + return pstmt.executeQuery(); + } + + @SuppressWarnings("unchecked") + private void checkPreparedStatement(SQLStatement resultSqlStmt, String resultSql, String database) + throws SQLException { + if (conn.getRealConnection() != null // + && StringUtils.equals(database, conn.getDatabase()) // + ) { + if (pstmt == null) { + pstmt = createRealPreparedStatement(conn.getRealConnection(), resultSql); + tables = (List) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_TABLES); + } else if (tables != null // + && tables.equals(resultSqlStmt.getAttribute(ShardingVisitor.ATTR_TABLES))) { + if (pstmt != null) { + JdbcUtils.close(pstmt); + pstmt = null; + tables = null; + } + pstmt = createRealPreparedStatement(conn.getRealConnection(), resultSql); + tables = (List) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_TABLES); + } + } else { + conn.closeRealConnection(); + if (pstmt != null) { + JdbcUtils.close(pstmt); + pstmt = null; + tables = null; + } + conn.createRealConnectionByDb(database); + pstmt = createRealPreparedStatement(conn.getRealConnection(), resultSql); + } + } + + @Override + public int executeUpdate() throws SQLException { + List values = getParameterValues(); + + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(values); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + checkParameterValues(values, resultSqlStmtList); + + int updateCount = 0; + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + checkPreparedStatement(resultSqlStmt, resultSql, database); + updateCount += pstmt.executeUpdate(); + } + + return updateCount; + } + + private void checkParameterValues(List values, List resultSqlStmtList) throws SQLException { + if (resultSqlStmtList.size() > 1) { + for (Object value : values) { + if (value instanceof InputStream // + || value instanceof Reader // + || value instanceof Blob // + || value instanceof Clob // + || value instanceof NClob // + ) { + throw new SQLException(value.getClass().getName() + " not support multi-statement"); + } + } + } + } + + private List getParameterValues() { + List values = new ArrayList(); + for (Map.Entry entry : this.parameters.entrySet()) { + values.add(entry.getValue().getValue()); + } + return values; + } + + @Override + public boolean execute() throws SQLException { + List values = getParameterValues(); + + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(values); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + checkParameterValues(values, resultSqlStmtList); + + boolean isSelelct = sqlStmt instanceof SQLSelectStatement; + + if (resultSqlStmtList.size() > 1 && isSelelct) { + throw new SQLException("select not support multi-statement"); + } + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + checkPreparedStatement(resultSqlStmt, resultSql, database); + pstmt.execute(); + + if (!isSelelct) { + if (updateCount == null) { + updateCount = pstmt.getUpdateCount(); + } else { + updateCount = updateCount.intValue() + pstmt.getUpdateCount(); + } + } + } + + return isSelelct; + } + + @Override + public void addBatch() throws SQLException { + throw new UnsupportedOperationException(); + } + + @SuppressWarnings("deprecation") + private PreparedStatement createRealPreparedStatement(Connection conn, String sql) throws SQLException { + PreparedStatement stmt; + if (resultSetType != null && resultSetConcurrency != null && resultSetHoldability != null) { + stmt = conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability); + } else if (resultSetType != null && resultSetConcurrency != null) { + stmt = conn.prepareStatement(sql, resultSetType, resultSetConcurrency); + } else { + stmt = conn.prepareStatement(sql); + } + + initStatement(stmt); + + for (Map.Entry entry : this.parameters.entrySet()) { + int index = entry.getKey(); + JdbcParameter param = entry.getValue(); + Object value = param.getValue(); + + if (param.getScaleOrLength() != -1) { + stmt.setObject(index, value, param.getSqlType(), param.getScaleOrLength()); + } else if (value == null) { + stmt.setNull(index, param.getSqlType()); + } else { + switch (param.getSqlType()) { + case Types.ARRAY: + stmt.setArray(index, (Array) value); + break; + case TYPE.AsciiInputStream: + if (param.getLength() >= 0) { + stmt.setAsciiStream(index, (InputStream) value, param.getLength()); + } else { + stmt.setAsciiStream(index, (InputStream) value); + } + break; + case Types.DECIMAL: + stmt.setBigDecimal(index, (BigDecimal) value); + break; + case TYPE.BinaryInputStream: + if (param.getLength() >= 0) { + stmt.setBinaryStream(index, (InputStream) value, param.getLength()); + } else { + stmt.setBinaryStream(index, (InputStream) value); + } + break; + case Types.BLOB: + if (value instanceof Blob) { + stmt.setBlob(index, (Blob) value); + } else { + if (param.getLength() >= 0) { + stmt.setBlob(index, (InputStream) value, param.getLength()); + } else { + stmt.setBlob(index, (InputStream) value); + } + } + break; + case Types.BOOLEAN: + stmt.setBoolean(index, (Boolean) value); + break; + case Types.TINYINT: + stmt.setByte(index, (Byte) value); + break; + case TYPE.BYTES: + stmt.setBytes(index, (byte[]) value); + break; + case TYPE.CharacterInputStream: + if (param.getLength() >= 0) { + stmt.setCharacterStream(index, (Reader) value, param.getLength()); + } else { + stmt.setCharacterStream(index, (Reader) value); + } + break; + case Types.CLOB: + if (value instanceof Clob) { + stmt.setClob(index, (Clob) value); + } else { + if (param.getLength() >= 0) { + stmt.setClob(index, (Reader) value, param.getLength()); + } else { + stmt.setClob(index, (Reader) value); + } + } + break; + case Types.DATE: + if (param.getCalendar() != null) { + stmt.setDate(index, (Date) value, param.getCalendar()); + } else { + stmt.setDate(index, (Date) value); + } + break; + case Types.DOUBLE: + stmt.setDouble(index, (Double) value); + break; + case Types.FLOAT: + stmt.setFloat(index, (Float) value); + break; + case Types.INTEGER: + stmt.setInt(index, (Integer) value); + break; + case Types.BIGINT: + stmt.setLong(index, (Long) value); + break; + case TYPE.NCharacterInputStream: + if (param.getLength() >= 0) { + stmt.setNCharacterStream(index, (Reader) value, param.getLength()); + } else { + stmt.setNCharacterStream(index, (Reader) value); + } + break; + case Types.NCLOB: + if (value instanceof Clob) { + stmt.setNClob(index, (NClob) value); + } else { + if (param.getLength() >= 0) { + stmt.setNClob(index, (Reader) value, param.getLength()); + } else { + stmt.setNClob(index, (Reader) value); + } + } + break; + case Types.NCHAR: + case Types.NVARCHAR: + stmt.setNString(index, (String) value); + break; + case Types.OTHER: + stmt.setObject(index, value); + break; + case Types.REF: + stmt.setRef(index, (Ref) value); + break; + case Types.ROWID: + stmt.setRowId(index, (RowId) value); + break; + case Types.SQLXML: + stmt.setSQLXML(index, (SQLXML) value); + break; + case Types.SMALLINT: + stmt.setShort(index, (Short) value); + break; + case Types.VARCHAR: + case Types.CHAR: + stmt.setString(index, (String) value); + break; + case Types.TIME: + if (param.getCalendar() != null) { + stmt.setTime(index, (Time) value, param.getCalendar()); + } else { + stmt.setTime(index, (Time) value); + } + break; + case Types.TIMESTAMP: + if (param.getCalendar() != null) { + stmt.setTimestamp(index, (Timestamp) value, param.getCalendar()); + } else { + stmt.setTimestamp(index, (Timestamp) value); + } + break; + case TYPE.URL: + stmt.setURL(index, (URL) value); + break; + case TYPE.UnicodeStream: + stmt.setUnicodeStream(index, (InputStream) value, (int) param.getLength()); + break; + default: + stmt.setObject(index, value); + break; + } + } + } + + return stmt; + } + +} diff --git a/src/main/java/com/alibaba/druid/sharding/ShardingStatement.java b/src/main/java/com/alibaba/druid/sharding/ShardingStatement.java new file mode 100644 index 0000000000..10233d47ec --- /dev/null +++ b/src/main/java/com/alibaba/druid/sharding/ShardingStatement.java @@ -0,0 +1,798 @@ +package com.alibaba.druid.sharding; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLWarning; +import java.sql.Statement; +import java.util.List; + +import org.h2.util.StringUtils; + +import com.alibaba.druid.pool.WrapperAdapter; +import com.alibaba.druid.sharding.sql.ShardingVisitor; +import com.alibaba.druid.sql.SQLUtils; +import com.alibaba.druid.sql.ast.SQLStatement; +import com.alibaba.druid.sql.ast.statement.SQLInsertStatement; +import com.alibaba.druid.sql.ast.statement.SQLSelectStatement; +import com.alibaba.druid.util.JdbcUtils; + +public class ShardingStatement extends WrapperAdapter implements Statement { + + protected final ShardingConnection conn; + private final long id; + protected Integer resultSetType; + protected Integer resultSetConcurrency; + protected Integer resultSetHoldability; + + private Statement stmt; + protected Integer updateCount; + + private Integer maxFieldSize; + private Integer maxRows; + private Boolean escapeProcessing; + private Integer queryTimeout; + private String cursorName; + private Integer fetchDirection; + private Integer fetchSize; + + protected void initStatement(Statement stmt) throws SQLException { + if (maxFieldSize != null) { + stmt.setMaxFieldSize(maxFieldSize); + } + if (maxRows != null) { + stmt.setMaxRows(maxRows); + } + if (escapeProcessing != null) { + stmt.setEscapeProcessing(escapeProcessing); + } + if (queryTimeout != null) { + stmt.setQueryTimeout(queryTimeout); + } + if (cursorName != null) { + stmt.setCursorName(cursorName); + } + if (fetchDirection != null) { + stmt.setFetchDirection(fetchDirection); + } + if (fetchSize != null) { + stmt.setFetchSize(fetchSize); + } + } + + private Statement createRealStatement(Connection conn) throws SQLException { + Statement stmt; + if (resultSetType != null && resultSetConcurrency != null && resultSetHoldability != null) { + stmt = conn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability); + } else if (resultSetType != null && resultSetConcurrency != null) { + stmt = conn.createStatement(resultSetType, resultSetConcurrency); + } else { + stmt = conn.createStatement(); + } + + initStatement(stmt); + + return stmt; + } + + public ShardingStatement(ShardingConnection conn, long id){ + this.conn = conn; + this.id = id; + } + + public ShardingStatement(ShardingConnection conn, long id, int resultSetType, int resultSetConcurrency){ + this.conn = conn; + this.id = id; + this.resultSetType = resultSetType; + this.resultSetConcurrency = resultSetConcurrency; + } + + public ShardingStatement(ShardingConnection conn, long id, int resultSetType, int resultSetConcurrency, + int resultSetHoldability){ + this.conn = conn; + this.id = id; + this.resultSetType = resultSetType; + this.resultSetConcurrency = resultSetConcurrency; + this.resultSetHoldability = resultSetHoldability; + } + + public long getId() { + return id; + } + + @Override + public ResultSet executeQuery(String sql) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + if (resultSqlStmtList.size() > 1) { + throw new SQLException("executeQuery not support multi-statement"); + } + + SQLStatement resultSqlStmt = resultSqlStmtList.get(0); + + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + return stmt.executeQuery(resultSql); + } + + @Override + public void close() throws SQLException { + if (stmt != null) { + stmt.close(); + } + } + + @Override + public int getMaxFieldSize() throws SQLException { + if (stmt != null) { + return stmt.getMaxFieldSize(); + } + + if (maxFieldSize == null) { + return 0; + } + + return maxFieldSize; + } + + @Override + public void setMaxFieldSize(int max) throws SQLException { + if (stmt != null) { + stmt.setMaxFieldSize(max); + return; + } + + this.maxFieldSize = max; + } + + @Override + public int getMaxRows() throws SQLException { + if (stmt != null) { + return stmt.getMaxRows(); + } + + if (maxRows == null) { + return 0; + } + + return maxRows; + } + + @Override + public void setMaxRows(int max) throws SQLException { + if (stmt != null) { + stmt.setMaxRows(max); + return; + } + + this.maxRows = max; + } + + @Override + public void setEscapeProcessing(boolean enable) throws SQLException { + if (stmt != null) { + stmt.setEscapeProcessing(enable); + return; + } + + this.escapeProcessing = enable; + } + + @Override + public int getQueryTimeout() throws SQLException { + if (stmt != null) { + return stmt.getQueryTimeout(); + } + + if (queryTimeout == null) { + return 0; + } + + return queryTimeout; + } + + @Override + public void setQueryTimeout(int seconds) throws SQLException { + if (stmt != null) { + stmt.setQueryTimeout(seconds); + return; + } + + this.queryTimeout = seconds; + } + + @Override + public void cancel() throws SQLException { + if (stmt != null) { + stmt.cancel(); + } + } + + @Override + public SQLWarning getWarnings() throws SQLException { + if (stmt != null) { + return stmt.getWarnings(); + } + + return null; + } + + @Override + public void clearWarnings() throws SQLException { + if (stmt != null) { + stmt.clearWarnings(); + } + } + + @Override + public void setCursorName(String name) throws SQLException { + if (stmt != null) { + stmt.setCursorName(name); + return; + } + + this.cursorName = name; + } + + @Override + public boolean execute(String sql) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + boolean isSelelct = sqlStmt instanceof SQLSelectStatement; + + if (resultSqlStmtList.size() > 1 && isSelelct) { + throw new SQLException("select not support multi-statement"); + } + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + stmt.execute(resultSql); + + if (!isSelelct) { + if (updateCount == null) { + updateCount = stmt.getUpdateCount(); + } else { + updateCount = updateCount.intValue() + stmt.getUpdateCount(); + } + } + } + + return isSelelct; + } + + @Override + public ResultSet getResultSet() throws SQLException { + if (stmt != null) { + return stmt.getResultSet(); + } + + return null; + } + + @Override + public int getUpdateCount() throws SQLException { + if (updateCount != null) { + return updateCount.intValue(); + } + + return 0; + } + + @Override + public boolean getMoreResults() throws SQLException { + if (stmt != null) { + return stmt.getMoreResults(); + } + + return false; + } + + @Override + public void setFetchDirection(int direction) throws SQLException { + if (stmt != null) { + stmt.setFetchDirection(direction); + return; + } + + this.fetchDirection = direction; + } + + @Override + public int getFetchDirection() throws SQLException { + if (stmt != null) { + return stmt.getFetchDirection(); + } + + return this.fetchDirection; + } + + @Override + public void setFetchSize(int rows) throws SQLException { + if (stmt != null) { + stmt.setFetchSize(rows); + return; + } + + this.fetchSize = rows; + } + + @Override + public int getFetchSize() throws SQLException { + if (stmt != null) { + return stmt.getFetchSize(); + } + + return this.fetchSize; + } + + @Override + public int getResultSetConcurrency() throws SQLException { + if (stmt != null) { + return stmt.getResultSetConcurrency(); + } + + if (resultSetConcurrency != null) { + return resultSetConcurrency; + } + + return 0; + } + + @Override + public int getResultSetType() throws SQLException { + if (stmt != null) { + return stmt.getResultSetType(); + } + + return resultSetType; + } + + @Override + public void addBatch(String sql) throws SQLException { + throw new UnsupportedOperationException(); + } + + @Override + public void clearBatch() throws SQLException { + if (stmt != null) { + stmt.clearBatch(); + } + } + + @Override + public int[] executeBatch() throws SQLException { + if (stmt != null) { + return stmt.executeBatch(); + } + return new int[0]; + } + + @Override + public ShardingConnection getConnection() throws SQLException { + return conn; + } + + @Override + public boolean getMoreResults(int current) throws SQLException { + if (stmt != null) { + return stmt.getMoreResults(current); + } + + return false; + } + + @Override + public ResultSet getGeneratedKeys() throws SQLException { + if (stmt != null) { + return stmt.getGeneratedKeys(); + } + + return null; + } + + @Override + public int executeUpdate(String sql) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + int updateCount = 0; + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + updateCount += stmt.executeUpdate(resultSql); + } + + return updateCount; + } + + @Override + public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + if (autoGeneratedKeys == Statement.RETURN_GENERATED_KEYS && resultSqlStmtList.size() > 1 + && sqlStmt instanceof SQLInsertStatement) { + throw new SQLException("RETURN_GENERATED_KEYS not support multi-statement"); + } + + int updateCount = 0; + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + updateCount += stmt.executeUpdate(resultSql, autoGeneratedKeys); + } + + return updateCount; + } + + @Override + public int executeUpdate(String sql, int[] columnIndexes) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + int updateCount = 0; + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + updateCount += stmt.executeUpdate(resultSql, columnIndexes); + } + + return updateCount; + } + + @Override + public int executeUpdate(String sql, String[] columnNames) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + int updateCount = 0; + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + updateCount += stmt.executeUpdate(resultSql, columnNames); + } + + return updateCount; + } + + @Override + public boolean execute(String sql, int autoGeneratedKeys) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + boolean isSelelct = sqlStmt instanceof SQLSelectStatement; + + if (resultSqlStmtList.size() > 1) { + if (autoGeneratedKeys == Statement.RETURN_GENERATED_KEYS && sqlStmt instanceof SQLInsertStatement) { + throw new SQLException("RETURN_GENERATED_KEYS not support multi-statement"); + } else if (isSelelct) { + throw new SQLException("select not support multi-statement"); + } + } + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + stmt.execute(resultSql, autoGeneratedKeys); + + if (!isSelelct) { + if (updateCount == null) { + updateCount = stmt.getUpdateCount(); + } else { + updateCount = updateCount.intValue() + stmt.getUpdateCount(); + } + } + } + + return isSelelct; + } + + @Override + public boolean execute(String sql, int[] columnIndexes) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + boolean isSelelct = sqlStmt instanceof SQLSelectStatement; + + if (resultSqlStmtList.size() > 1 && isSelelct) { + throw new SQLException("select not support multi-statement"); + } + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + stmt.execute(resultSql, columnIndexes); + + if (!isSelelct) { + if (updateCount == null) { + updateCount = stmt.getUpdateCount(); + } else { + updateCount = updateCount.intValue() + stmt.getUpdateCount(); + } + } + } + + return isSelelct; + } + + @Override + public boolean execute(String sql, String[] columnNames) throws SQLException { + ShardingDataSource dataSource = conn.getDataSource(); + ShardingVisitor visitor = dataSource.createShardingVisitor(); + SQLStatement sqlStmt = dataSource.parseStatement(sql); + + sqlStmt.accept(visitor); + + List resultSqlStmtList = visitor.getResult(); + + boolean isSelelct = sqlStmt instanceof SQLSelectStatement; + + if (resultSqlStmtList.size() > 1 && isSelelct) { + throw new SQLException("select not support multi-statement"); + } + + for (SQLStatement resultSqlStmt : resultSqlStmtList) { + String resultSql = SQLUtils.toSQLString(resultSqlStmt, dataSource.getDbType()); + + String database = (String) resultSqlStmt.getAttribute(ShardingVisitor.ATTR_DB); + + if (conn.getRealConnection() != null && StringUtils.equals(database, conn.getDatabase())) { + if (stmt == null) { + stmt = createRealStatement(conn.getRealConnection()); + } + } else { + conn.closeRealConnection(); + if (stmt != null) { + JdbcUtils.close(stmt); + stmt = null; + } + conn.createRealConnectionByDb(database); + stmt = createRealStatement(conn.getRealConnection()); + } + stmt.execute(resultSql, columnNames); + + if (!isSelelct) { + if (updateCount == null) { + updateCount = stmt.getUpdateCount(); + } else { + updateCount = updateCount.intValue() + stmt.getUpdateCount(); + } + } + } + + return isSelelct; + } + + @Override + public int getResultSetHoldability() throws SQLException { + if (stmt != null) { + return stmt.getResultSetHoldability(); + } + + if (resultSetHoldability != null) { + return resultSetHoldability; + } + + return 0; + } + + @Override + public boolean isClosed() throws SQLException { + if (stmt != null) { + return stmt.isClosed(); + } + return false; + } + + @Override + public void setPoolable(boolean poolable) throws SQLException { + throw new SQLException("not supported"); + } + + @Override + public boolean isPoolable() throws SQLException { + return false; + } + + @SuppressWarnings("unchecked") + @Override + public T unwrap(Class iface) throws SQLException { + if (iface == null) { + return null; + } + + if (iface.isInstance(this)) { + return (T) this; + } + + if (stmt != null) { + return stmt.unwrap(iface); + } + + return null; + } + + public Boolean isEscapeProcessing() { + return escapeProcessing; + } + + public String getCursorName() { + return cursorName; + } + + public void closeOnCompletion() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + public boolean isCloseOnCompletion() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/src/main/java/com/alibaba/druid/sharding/config/RouteConfig.java b/src/main/java/com/alibaba/druid/sharding/config/RouteConfig.java index e8f34d3851..380e7f747e 100644 --- a/src/main/java/com/alibaba/druid/sharding/config/RouteConfig.java +++ b/src/main/java/com/alibaba/druid/sharding/config/RouteConfig.java @@ -20,8 +20,17 @@ import java.util.Map; public class RouteConfig { - private Map logicTables = new HashMap(); - private Map mappingRules = new HashMap(); + private String defaultDatabase; + private Map logicTables = new HashMap(); + private Map mappingRules = new HashMap(); + + public String getDefaultDatabase() { + return defaultDatabase; + } + + public void setDefaultDatabase(String defaultDatabase) { + this.defaultDatabase = defaultDatabase; + } public Map getLogicTables() { return logicTables; @@ -38,7 +47,7 @@ public class RouteConfig { public void setMappingRules(Map mappingRules) { this.mappingRules = mappingRules; } - + public MappingRule getMappingRule(String table) { if (table == null) { return null; diff --git a/src/main/java/com/alibaba/druid/sharding/sql/MySqlShardingVisitor.java b/src/main/java/com/alibaba/druid/sharding/sql/MySqlShardingVisitor.java index ff7e00e4dc..b657f08487 100644 --- a/src/main/java/com/alibaba/druid/sharding/sql/MySqlShardingVisitor.java +++ b/src/main/java/com/alibaba/druid/sharding/sql/MySqlShardingVisitor.java @@ -67,6 +67,10 @@ public class MySqlShardingVisitor extends MySqlASTVisitorAdapter implements Shar this(routeConfig, Arrays.asList(parameters)); } + public List getResult() { + return result; + } + public RouteConfig getRouteConfig() { return routeConfig; } @@ -106,17 +110,17 @@ public class MySqlShardingVisitor extends MySqlASTVisitorAdapter implements Shar @Override public boolean visit(MySqlUpdateStatement x) { input = x; - + if (x.getTableSource() != null) { x.getTableSource().setParent(x); x.getTableSource().accept(this); } - + if (x.getWhere() != null) { x.getWhere().setParent(x); x.getWhere().accept(this); } - + result.add(x); return false; } diff --git a/src/main/java/com/alibaba/druid/sharding/sql/ShardingVisitor.java b/src/main/java/com/alibaba/druid/sharding/sql/ShardingVisitor.java index a0f4f7a008..78ba7b2be8 100644 --- a/src/main/java/com/alibaba/druid/sharding/sql/ShardingVisitor.java +++ b/src/main/java/com/alibaba/druid/sharding/sql/ShardingVisitor.java @@ -15,10 +15,18 @@ */ package com.alibaba.druid.sharding.sql; -public interface ShardingVisitor { +import java.util.List; + +import com.alibaba.druid.sql.ast.SQLStatement; +import com.alibaba.druid.sql.visitor.SQLASTVisitor; + +public interface ShardingVisitor extends SQLASTVisitor { public static final String ATTR_DB = "sharding.db"; public static final String ATTR_PARTITION = "sharding.partition"; public static final String ATTR_TABLE_SOURCE = "sharding.tableSource"; public static final String ATTR_ALIAS = "sharding.alias"; + public static final String ATTR_TABLES = "sharding.tables"; + + List getResult(); }