[refactor] Simplify connection creation by removing database parameter

Author: itinycheng

flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/ClickHouseShardInputFormat.java

@@ -109,8 +109,7 @@ public void open(InputSplit split) {
                 // PreparedStatement.
                 String shardUrl = shardMap.get((Integer) shardIds[i]);
                 ClickHouseConnection connection =
-                        connectionProvider.createAndStoreShardConnection(
-                                shardUrl, engineFullSchema.getDatabase());
+                        connectionProvider.createAndStoreShardConnection(shardUrl);
                 String query =
                         getQuery(engineFullSchema.getTable(), engineFullSchema.getDatabase());
                 PreparedStatement statement = connection.prepareStatement(query);

flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/ClickHouseShardOutputFormat.java

@@ -90,8 +90,7 @@ protected ClickHouseShardOutputFormat(
     public void open(int taskNumber, int numTasks) throws IOException {
         try {
             Map<Integer, ClickHouseConnection> connectionMap =
-                    connectionProvider.createShardConnections(
-                            clusterSpec, shardTableSchema.getDatabase());
+                    connectionProvider.createShardConnections(clusterSpec);
             for (Map.Entry<Integer, ClickHouseConnection> connectionEntry :
                     connectionMap.entrySet()) {
                 ClickHouseExecutor executor =

flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/connection/ClickHouseConnectionProvider.java

@@ -75,21 +75,20 @@ public synchronized ClickHouseConnection getOrCreateConnection() throws SQLExcep
     }
 
     public synchronized Map<Integer, ClickHouseConnection> createShardConnections(
-            ClusterSpec clusterSpec, String defaultDatabase) throws SQLException {
+            ClusterSpec clusterSpec) throws SQLException {
         Map<Integer, ClickHouseConnection> connectionMap = new HashMap<>();
         String urlSuffix = options.getUrlSuffix();
         for (ShardSpec shardSpec : clusterSpec.getShards()) {
             String shardUrl = shardSpec.getJdbcUrls() + urlSuffix;
-            ClickHouseConnection connection =
-                    createAndStoreShardConnection(shardUrl, defaultDatabase);
+            ClickHouseConnection connection = createAndStoreShardConnection(shardUrl);
             connectionMap.put(shardSpec.getNum(), connection);
         }
 
         return connectionMap;
     }
 
-    public synchronized ClickHouseConnection createAndStoreShardConnection(
-            String url, String database) throws SQLException {
+    public synchronized ClickHouseConnection createAndStoreShardConnection(String url)
+            throws SQLException {
         if (shardConnections == null) {
             shardConnections = new ArrayList<>();
         }