Skip to content

Commit f8c8d48

Browse files
czy006itinycheng
czy006
authored and
itinycheng
committed
[Feature] Upgrade Clickhouse JDBC Version to 0.6.0-patch3 #105 
1 parent 24e024b commit f8c8d48

22 files changed

+144
-158
lines changed

‎README.md

+4-5
Original file line numberDiff line numberDiff line change
@@ -206,8 +206,7 @@ tEnv.executeSql("insert into `clickhouse`.`default`.`t_table` select...");
206206

207207
## Roadmap
208208

209-
- [x] Implement the Flink SQL Sink function.
210-
- [x] Support array and Map types.
211-
- [x] Support ClickHouseCatalog.
212-
- [x] Implement the Flink SQL Source function.
213-
- [x] Implement the Flink SQL Lookup function.
209+
The main branch is currently unstable and should not be used in production
210+
211+
- [ ] Flink Clickhouse Connector donated to Apache Flink #102 @czy006
212+
- [ ] Perfect Junit Tests for Connector

‎flink-connector-clickhouse-e2e-test/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
<parent>
66
<groupId>org.apache.flink</groupId>
77
<artifactId>flink-connector-clickhouse-parent</artifactId>
8-
<version>1.17.1-SNAPSHOT</version>
8+
<version>1.0.0-SNAPSHOT</version>
99
</parent>
1010

1111
<groupId>org.apache.flink</groupId>

‎flink-connector-clickhouse/pom.xml

+10-12
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
<parent>
66
<groupId>org.apache.flink</groupId>
77
<artifactId>flink-connector-clickhouse-parent</artifactId>
8-
<version>1.17.1-SNAPSHOT</version>
8+
<version>1.0.0-SNAPSHOT</version>
99
</parent>
1010

1111
<artifactId>flink-connector-clickhouse</artifactId>
@@ -46,19 +46,9 @@
4646
<!-- Clickhouse -->
4747

4848
<dependency>
49-
<groupId>ru.yandex.clickhouse</groupId>
49+
<groupId>com.clickhouse</groupId>
5050
<artifactId>clickhouse-jdbc</artifactId>
5151
<version>${clickhouse-jdbc.version}</version>
52-
<exclusions>
53-
<exclusion>
54-
<groupId>com.fasterxml.jackson.core</groupId>
55-
<artifactId>jackson-core</artifactId>
56-
</exclusion>
57-
<exclusion>
58-
<groupId>com.fasterxml.jackson.core</groupId>
59-
<artifactId>jackson-databind</artifactId>
60-
</exclusion>
61-
</exclusions>
6252
</dependency>
6353
<dependency>
6454
<groupId>com.fasterxml.jackson.core</groupId>
@@ -71,6 +61,14 @@
7161
<version>${commons-lang3.version}</version>
7262
</dependency>
7363

64+
<!-- Http -->
65+
66+
<dependency>
67+
<groupId>org.apache.httpcomponents.client5</groupId>
68+
<artifactId>httpclient5</artifactId>
69+
<version>5.2.1</version>
70+
</dependency>
71+
7472
<!-- Tests -->
7573

7674
<dependency>

‎flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/catalog/ClickHouseCatalog.java

+11-16
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import org.apache.flink.connector.clickhouse.ClickHouseDynamicTableFactory;
44
import org.apache.flink.connector.clickhouse.internal.schema.DistributedEngineFull;
5-
import org.apache.flink.connector.clickhouse.util.ClickHouseUtil;
65
import org.apache.flink.connector.clickhouse.util.DataTypeUtil;
76
import org.apache.flink.table.api.TableSchema;
87
import org.apache.flink.table.catalog.AbstractCatalog;
@@ -33,13 +32,13 @@
3332
import org.apache.flink.table.factories.Factory;
3433
import org.apache.flink.table.types.DataType;
3534

35+
import com.clickhouse.client.config.ClickHouseDefaults;
36+
import com.clickhouse.data.ClickHouseColumn;
37+
import com.clickhouse.jdbc.ClickHouseConnection;
38+
import com.clickhouse.jdbc.ClickHouseDriver;
39+
import com.clickhouse.jdbc.ClickHouseResultSetMetaData;
3640
import org.slf4j.Logger;
3741
import org.slf4j.LoggerFactory;
38-
import ru.yandex.clickhouse.BalancedClickhouseDataSource;
39-
import ru.yandex.clickhouse.ClickHouseConnection;
40-
import ru.yandex.clickhouse.response.ClickHouseColumnInfo;
41-
import ru.yandex.clickhouse.response.ClickHouseResultSetMetaData;
42-
import ru.yandex.clickhouse.settings.ClickHouseQueryParam;
4342

4443
import javax.annotation.Nullable;
4544

@@ -129,14 +128,11 @@ public void open() throws CatalogException {
129128
try {
130129
Properties configuration = new Properties();
131130
configuration.putAll(properties);
132-
configuration.setProperty(ClickHouseQueryParam.USER.getKey(), username);
133-
configuration.setProperty(ClickHouseQueryParam.PASSWORD.getKey(), password);
134-
String jdbcUrl = ClickHouseUtil.getJdbcUrl(baseUrl, getDefaultDatabase());
135-
BalancedClickhouseDataSource dataSource =
136-
new BalancedClickhouseDataSource(jdbcUrl, configuration);
137-
dataSource.actualize();
138-
connection = dataSource.getConnection();
139-
LOG.info("Created catalog {}, established connection to {}", getName(), jdbcUrl);
131+
configuration.setProperty(ClickHouseDefaults.USER.getKey(), username);
132+
configuration.setProperty(ClickHouseDefaults.PASSWORD.getKey(), password);
133+
ClickHouseDriver driver = new ClickHouseDriver();
134+
connection = driver.connect(baseUrl, configuration);
135+
LOG.info("Created catalog {}, established connection to {}", getName(), baseUrl);
140136
} catch (Exception e) {
141137
throw new CatalogException(String.format("Opening catalog %s failed.", getName()), e);
142138
}
@@ -307,8 +303,7 @@ private synchronized TableSchema createTableSchema(String databaseName, String t
307303
List<String> primaryKeys = getPrimaryKeys(databaseName, tableName);
308304
TableSchema.Builder builder = TableSchema.builder();
309305
for (int idx = 1; idx <= metaData.getColumnCount(); idx++) {
310-
ClickHouseColumnInfo columnInfo =
311-
(ClickHouseColumnInfo) getColMethod.invoke(metaData, idx);
306+
ClickHouseColumn columnInfo = (ClickHouseColumn) getColMethod.invoke(metaData, idx);
312307
String columnName = columnInfo.getColumnName();
313308
DataType columnType = DataTypeUtil.toFlinkType(columnInfo);
314309
if (primaryKeys.contains(columnName)) {

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,9 @@
1818
import org.apache.flink.util.Preconditions;
1919
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
2020

21+
import com.clickhouse.jdbc.ClickHouseConnection;
2122
import org.slf4j.Logger;
2223
import org.slf4j.LoggerFactory;
23-
import ru.yandex.clickhouse.ClickHouseConnection;
2424

2525
import java.io.Flushable;
2626
import java.io.IOException;

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,9 @@
77
import org.apache.flink.core.io.InputSplit;
88
import org.apache.flink.table.data.RowData;
99

10+
import com.clickhouse.jdbc.ClickHouseConnection;
1011
import org.slf4j.Logger;
1112
import org.slf4j.LoggerFactory;
12-
import ru.yandex.clickhouse.ClickHouseConnection;
1313

1414
import java.io.IOException;
1515
import java.sql.PreparedStatement;

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,9 @@
1313
import org.apache.flink.table.types.logical.LogicalType;
1414
import org.apache.flink.table.types.logical.RowType;
1515

16+
import com.clickhouse.jdbc.ClickHousePreparedStatement;
1617
import org.slf4j.Logger;
1718
import org.slf4j.LoggerFactory;
18-
import ru.yandex.clickhouse.ClickHousePreparedStatement;
1919

2020
import java.sql.Connection;
2121
import java.sql.ResultSet;
@@ -146,7 +146,7 @@ private void establishConnectionAndStatement() throws SQLException {
146146
Connection dbConn = connectionProvider.getOrCreateConnection();
147147
statement =
148148
new ClickHouseStatementWrapper(
149-
(ClickHousePreparedStatement) dbConn.prepareStatement(query));
149+
dbConn, (ClickHousePreparedStatement) dbConn.prepareStatement(query));
150150
}
151151

152152
@Override

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,9 @@
1010
import org.apache.flink.core.io.InputSplit;
1111
import org.apache.flink.table.data.RowData;
1212

13+
import com.clickhouse.jdbc.ClickHouseConnection;
1314
import org.slf4j.Logger;
1415
import org.slf4j.LoggerFactory;
15-
import ru.yandex.clickhouse.ClickHouseConnection;
1616

1717
import java.io.IOException;
1818
import java.sql.PreparedStatement;

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
import org.apache.flink.table.types.logical.LogicalType;
1111
import org.apache.flink.util.Preconditions;
1212

13-
import ru.yandex.clickhouse.ClickHouseConnection;
13+
import com.clickhouse.jdbc.ClickHouseConnection;
1414

1515
import javax.annotation.Nonnull;
1616

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

+10-15
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,12 @@
33
import org.apache.flink.connector.clickhouse.internal.options.ClickHouseConnectionOptions;
44
import org.apache.flink.connector.clickhouse.internal.schema.ClusterSpec;
55
import org.apache.flink.connector.clickhouse.internal.schema.ShardSpec;
6-
import org.apache.flink.connector.clickhouse.util.ClickHouseUtil;
76

7+
import com.clickhouse.client.config.ClickHouseDefaults;
8+
import com.clickhouse.jdbc.ClickHouseConnection;
9+
import com.clickhouse.jdbc.ClickHouseDriver;
810
import org.slf4j.Logger;
911
import org.slf4j.LoggerFactory;
10-
import ru.yandex.clickhouse.BalancedClickhouseDataSource;
11-
import ru.yandex.clickhouse.ClickHouseConnection;
12-
import ru.yandex.clickhouse.settings.ClickHouseProperties;
1312

1413
import java.io.Serializable;
1514
import java.sql.PreparedStatement;
@@ -110,17 +109,13 @@ public List<String> getShardUrls(String remoteCluster) throws SQLException {
110109

111110
private ClickHouseConnection createConnection(String url, String database) throws SQLException {
112111
LOG.info("connecting to {}, database {}", url, database);
113-
114-
String jdbcUrl = ClickHouseUtil.getJdbcUrl(url, database);
115-
ClickHouseProperties properties = new ClickHouseProperties(connectionProperties);
116-
properties.setUser(options.getUsername().orElse(null));
117-
properties.setPassword(options.getPassword().orElse(null));
118-
BalancedClickhouseDataSource dataSource =
119-
new BalancedClickhouseDataSource(jdbcUrl, properties);
120-
if (dataSource.getAllClickhouseUrls().size() > 1) {
121-
dataSource.actualize();
122-
}
123-
return dataSource.getConnection();
112+
Properties configuration = new Properties();
113+
configuration.setProperty(
114+
ClickHouseDefaults.USER.getKey(), options.getUsername().orElse(null));
115+
configuration.setProperty(
116+
ClickHouseDefaults.PASSWORD.getKey(), options.getPassword().orElse(null));
117+
ClickHouseDriver driver = new ClickHouseDriver();
118+
return driver.connect(url, configuration);
124119
}
125120

126121
public void closeConnections() {

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

+8-3
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
package org.apache.flink.connector.clickhouse.internal.connection;
22

3-
import ru.yandex.clickhouse.ClickHousePreparedStatement;
3+
import com.clickhouse.jdbc.ClickHousePreparedStatement;
44

55
import java.math.BigDecimal;
6+
import java.sql.Connection;
67
import java.sql.Date;
78
import java.sql.ResultSet;
89
import java.sql.SQLException;
@@ -11,9 +12,12 @@
1112
/** Wrapper class for ClickHousePreparedStatement. */
1213
public class ClickHouseStatementWrapper {
1314
public final ClickHousePreparedStatement statement;
15+
public final Connection connection;
1416

15-
public ClickHouseStatementWrapper(ClickHousePreparedStatement statement) {
17+
public ClickHouseStatementWrapper(
18+
Connection connection, ClickHousePreparedStatement statement) {
1619
this.statement = statement;
20+
this.connection = connection;
1721
}
1822

1923
public void addBatch() throws SQLException {
@@ -77,7 +81,8 @@ public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
7781
}
7882

7983
public void setArray(int parameterIndex, Object[] array) throws SQLException {
80-
statement.setArray(parameterIndex, array);
84+
java.sql.Array sqlArray = connection.createArrayOf("VARCHAR", array);
85+
statement.setArray(parameterIndex, sqlArray);
8186
}
8287

8388
public void setObject(int parameterIndex, Object x) throws SQLException {

‎flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/converter/ClickHouseRowConverter.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@
1414
import org.apache.flink.table.types.logical.TimestampType;
1515
import org.apache.flink.util.Preconditions;
1616

17-
import ru.yandex.clickhouse.ClickHousePreparedStatement;
18-
import ru.yandex.clickhouse.response.ClickHouseResultSet;
17+
import com.clickhouse.jdbc.ClickHousePreparedStatement;
18+
import com.clickhouse.jdbc.ClickHouseResultSet;
1919

2020
import java.io.Serializable;
2121
import java.math.BigDecimal;

‎flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/executor/ClickHouseBatchExecutor.java

+3-2
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,10 @@
88
import org.apache.flink.connector.clickhouse.internal.options.ClickHouseDmlOptions;
99
import org.apache.flink.table.data.RowData;
1010

11+
import com.clickhouse.jdbc.ClickHouseConnection;
12+
import com.clickhouse.jdbc.ClickHousePreparedStatement;
1113
import org.slf4j.Logger;
1214
import org.slf4j.LoggerFactory;
13-
import ru.yandex.clickhouse.ClickHouseConnection;
14-
import ru.yandex.clickhouse.ClickHousePreparedStatement;
1515

1616
import java.sql.SQLException;
1717

@@ -43,6 +43,7 @@ public ClickHouseBatchExecutor(
4343
public void prepareStatement(ClickHouseConnection connection) throws SQLException {
4444
statement =
4545
new ClickHouseStatementWrapper(
46+
connection,
4647
(ClickHousePreparedStatement) connection.prepareStatement(insertSql));
4748
}
4849

‎flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/executor/ClickHouseExecutor.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,10 @@
1111
import org.apache.flink.table.types.logical.LogicalType;
1212
import org.apache.flink.table.types.logical.RowType;
1313

14+
import com.clickhouse.jdbc.ClickHouseConnection;
1415
import org.apache.commons.lang3.ArrayUtils;
1516
import org.slf4j.Logger;
1617
import org.slf4j.LoggerFactory;
17-
import ru.yandex.clickhouse.ClickHouseConnection;
1818

1919
import java.io.Serializable;
2020
import java.sql.SQLException;

‎flink-connector-clickhouse/src/main/java/org/apache/flink/connector/clickhouse/internal/executor/ClickHouseUpsertExecutor.java

+5-2
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,10 @@
99
import org.apache.flink.connector.clickhouse.internal.options.ClickHouseDmlOptions;
1010
import org.apache.flink.table.data.RowData;
1111

12+
import com.clickhouse.jdbc.ClickHouseConnection;
13+
import com.clickhouse.jdbc.ClickHousePreparedStatement;
1214
import org.slf4j.Logger;
1315
import org.slf4j.LoggerFactory;
14-
import ru.yandex.clickhouse.ClickHouseConnection;
15-
import ru.yandex.clickhouse.ClickHousePreparedStatement;
1616

1717
import java.sql.SQLException;
1818
import java.util.Arrays;
@@ -90,12 +90,15 @@ public ClickHouseUpsertExecutor(
9090
public void prepareStatement(ClickHouseConnection connection) throws SQLException {
9191
this.insertStatement =
9292
new ClickHouseStatementWrapper(
93+
connection,
9394
(ClickHousePreparedStatement) connection.prepareStatement(this.insertSql));
9495
this.updateStatement =
9596
new ClickHouseStatementWrapper(
97+
connection,
9698
(ClickHousePreparedStatement) connection.prepareStatement(this.updateSql));
9799
this.deleteStatement =
98100
new ClickHouseStatementWrapper(
101+
connection,
99102
(ClickHousePreparedStatement) connection.prepareStatement(this.deleteSql));
100103
}
101104

0 commit comments

Comments
 (0)