support Connection.setSchema for mysql, fix #3865

src/main/java/com/alibaba/druid/pool/DruidConnectionHolder.java

@@ -73,6 +73,7 @@ public final class DruidConnectionHolder {
     protected final Map<String, Object>           variables;
     protected final Map<String, Object>           globleVariables;
     final ReentrantLock                           lock                     = new ReentrantLock();
+    protected String                              initSchema;
 
     public DruidConnectionHolder(DruidAbstractDataSource dataSource, PhysicalConnectionInfo pyConnectInfo)
                                                                                                           throws SQLException{

src/main/java/com/alibaba/druid/pool/DruidDataSource.java

@@ -1953,6 +1953,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                     return;
                 }
             }
+            if (holder.initSchema != null) {
+                holder.conn.setSchema(holder.initSchema);
+                holder.initSchema = null;
+            }
 
             if (!enable) {
                 discardConnection(holder);

src/main/java/com/alibaba/druid/pool/DruidPooledConnection.java

@@ -48,6 +48,7 @@ import com.alibaba.druid.pool.PreparedStatementPool.MethodType;
 import com.alibaba.druid.proxy.jdbc.TransactionInfo;
 import com.alibaba.druid.support.logging.Log;
 import com.alibaba.druid.support.logging.LogFactory;
+import com.alibaba.druid.util.JdbcUtils;
 
 /**
  * @author wenshao [szujobs@hotmail.com]
@@ -1190,6 +1191,16 @@ public class DruidPooledConnection extends PoolableWrapper implements javax.sql.
     }
 
     public void setSchema(String schema) throws SQLException {
+        if (JdbcUtils.isMysqlDbType(holder.getDataSource().getDbType())) {
+            if (holder.initSchema == null) {
+                holder.initSchema = conn.getSchema();
+            }
+            conn.setSchema(schema);
+            if (holder.statementPool != null) {
+                holder.clearStatementCache();
+            }
+            return;
+        }
         throw new SQLFeatureNotSupportedException();
     }