[Feature]Upgrade to flink 1.17 #103

itinycheng · Oct 15, 2023 · 98ee6ce · 98ee6ce
1 parent a43f9c3
commit 98ee6ce
Show file tree

Hide file tree

Showing 72 changed files with 1,357 additions and 618 deletions.
diff --git a/.mvn/wrapper/maven-wrapper.properties b/.mvn/wrapper/maven-wrapper.properties
diff --git a/docs/content.zh/docs/connectors/.keep b/docs/content.zh/docs/connectors/.keep
diff --git a/docs/content/docs/connectors/.keep b/docs/content/docs/connectors/.keep
diff --git a/flink-connector-clickhouse-e2e-test/pom.xml b/flink-connector-clickhouse-e2e-test/pom.xml
@@ -0,0 +1,24 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-connector-clickhouse-parent</artifactId>
+		<version>1.17.1-SNAPSHOT</version>
+	</parent>
+
+	<groupId>org.apache.flink</groupId>
+	<name>Flink : Connectors : Clickhouse : E2E Tests</name>
+	<artifactId>flink-connector-clickhouse-e2e-test</artifactId>
+	<url>https://flink.apache.org</url>
+	<packaging>jar</packaging>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+
+	<dependencies>
+
+	</dependencies>
+</project>
diff --git a/flink-connector-clickhouse/pom.xml b/flink-connector-clickhouse/pom.xml
@@ -0,0 +1,230 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-connector-clickhouse-parent</artifactId>
+		<version>1.17.1-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>flink-connector-clickhouse</artifactId>
+	<name>Flink : Connectors : Clickhouse</name>
+
+	<packaging>jar</packaging>
+
+	<dependencies>
+
+		<!-- Core -->
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java</artifactId>
+			<version>${flink.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
+		<!-- Connectors -->
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+		</dependency>
+
+		<!-- Table ecosystem -->
+
+		<!-- Projects depending on this project won't depend on flink-table-*. -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-api-java-bridge</artifactId>
+			<version>${flink.version}</version>
+			<scope>provided</scope>
+			<optional>true</optional>
+		</dependency>
+
+		<!-- Clickhouse -->
+
+		<dependency>
+			<groupId>ru.yandex.clickhouse</groupId>
+			<artifactId>clickhouse-jdbc</artifactId>
+			<version>${clickhouse-jdbc.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>com.fasterxml.jackson.core</groupId>
+					<artifactId>jackson-core</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>com.fasterxml.jackson.core</groupId>
+					<artifactId>jackson-databind</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+			<version>${jackson.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-lang3</artifactId>
+			<version>${commons-lang3.version}</version>
+		</dependency>
+
+		<!-- Tests -->
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-core</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-test-utils</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-test-utils</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-metrics-jmx</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- Clickhouse table factory testing -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-common</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-test-utils</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- IT test with formats -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-json</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-avro</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-avro-confluent-registry</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-csv</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<executions>
+					<execution>
+						<goals>
+							<goal>test-jar</goal>
+						</goals>
+						<configuration>
+							<includes>
+								<include>META-INF/LICENSE</include>
+								<include>META-INF/NOTICE</include>
+							</includes>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-source-plugin</artifactId>
+				<executions>
+					<execution>
+						<id>attach-test-sources</id>
+						<goals>
+							<goal>test-jar-no-fork</goal>
+						</goals>
+						<configuration>
+							<archive>
+								<!-- Globally exclude maven metadata, because it may accidentally bundle files we don't intend to -->
+								<addMavenDescriptor>false</addMavenDescriptor>
+							</archive>
+							<includes>
+								<include>META-INF/LICENSE</include>
+								<include>META-INF/NOTICE</include>
+							</includes>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>
diff --git a/...khouse/ClickHouseDynamicTableFactory.java → ...khouse/ClickHouseDynamicTableFactory.java b/...khouse/ClickHouseDynamicTableFactory.java → ...khouse/ClickHouseDynamicTableFactory.java
diff --git a/...lickhouse/ClickHouseDynamicTableSink.java → ...lickhouse/ClickHouseDynamicTableSink.java b/...lickhouse/ClickHouseDynamicTableSink.java → ...lickhouse/ClickHouseDynamicTableSink.java
diff --git a/...ckhouse/ClickHouseDynamicTableSource.java → ...ckhouse/ClickHouseDynamicTableSource.java b/...ckhouse/ClickHouseDynamicTableSource.java → ...ckhouse/ClickHouseDynamicTableSource.java
diff --git a/...clickhouse/catalog/ClickHouseCatalog.java → ...clickhouse/catalog/ClickHouseCatalog.java b/...clickhouse/catalog/ClickHouseCatalog.java → ...clickhouse/catalog/ClickHouseCatalog.java
diff --git a/...use/catalog/ClickHouseCatalogFactory.java → ...use/catalog/ClickHouseCatalogFactory.java b/...use/catalog/ClickHouseCatalogFactory.java → ...use/catalog/ClickHouseCatalogFactory.java
diff --git a/...r/clickhouse/config/ClickHouseConfig.java → ...r/clickhouse/config/ClickHouseConfig.java b/...r/clickhouse/config/ClickHouseConfig.java → ...r/clickhouse/config/ClickHouseConfig.java
diff --git a/...house/config/ClickHouseConfigOptions.java → ...house/config/ClickHouseConfigOptions.java b/...house/config/ClickHouseConfigOptions.java → ...house/config/ClickHouseConfigOptions.java
diff --git a/...ternal/AbstractClickHouseInputFormat.java → ...ternal/AbstractClickHouseInputFormat.java b/...ternal/AbstractClickHouseInputFormat.java → ...ternal/AbstractClickHouseInputFormat.java
diff --git a/...ernal/AbstractClickHouseOutputFormat.java → ...ernal/AbstractClickHouseOutputFormat.java b/...ernal/AbstractClickHouseOutputFormat.java → ...ernal/AbstractClickHouseOutputFormat.java
@@ -123,8 +123,7 @@ protected void checkFlushException() {
     /** Builder for {@link ClickHouseBatchOutputFormat} and {@link ClickHouseShardOutputFormat}. */
     public static class Builder {
 
-        private static final Logger LOG =
-                LoggerFactory.getLogger(AbstractClickHouseOutputFormat.Builder.class);
+        private static final Logger LOG = LoggerFactory.getLogger(Builder.class);
 
         private DataType[] fieldTypes;
 
@@ -142,7 +141,7 @@ public static class Builder {
 
         public Builder() {}
 
-        public AbstractClickHouseOutputFormat.Builder withOptions(ClickHouseDmlOptions options) {
+        public Builder withOptions(ClickHouseDmlOptions options) {
             this.options = options;
             return this;
         }
@@ -152,7 +151,7 @@ public Builder withConnectionProperties(Properties connectionProperties) {
             return this;
         }
 
-        public AbstractClickHouseOutputFormat.Builder withFieldTypes(DataType[] fieldTypes) {
+        public Builder withFieldTypes(DataType[] fieldTypes) {
             this.fieldTypes = fieldTypes;
             this.logicalTypes =
                     Arrays.stream(fieldTypes)
@@ -161,17 +160,17 @@ public AbstractClickHouseOutputFormat.Builder withFieldTypes(DataType[] fieldTyp
             return this;
         }
 
-        public AbstractClickHouseOutputFormat.Builder withFieldNames(String[] fieldNames) {
+        public Builder withFieldNames(String[] fieldNames) {
             this.fieldNames = fieldNames;
             return this;
         }
 
-        public AbstractClickHouseOutputFormat.Builder withPrimaryKey(String[] primaryKeys) {
+        public Builder withPrimaryKey(String[] primaryKeys) {
             this.primaryKeys = primaryKeys;
             return this;
         }
 
-        public AbstractClickHouseOutputFormat.Builder withPartitionKey(String[] partitionKeys) {
+        public Builder withPartitionKey(String[] partitionKeys) {
             this.partitionKeys = partitionKeys;
             return this;
         }

diff --git a/.../internal/ClickHouseBatchInputFormat.java → .../internal/ClickHouseBatchInputFormat.java b/.../internal/ClickHouseBatchInputFormat.java → .../internal/ClickHouseBatchInputFormat.java
diff --git a/...internal/ClickHouseBatchOutputFormat.java → ...internal/ClickHouseBatchOutputFormat.java b/...internal/ClickHouseBatchOutputFormat.java → ...internal/ClickHouseBatchOutputFormat.java
diff --git a/...rnal/ClickHouseRowDataLookupFunction.java → ...rnal/ClickHouseRowDataLookupFunction.java b/...rnal/ClickHouseRowDataLookupFunction.java → ...rnal/ClickHouseRowDataLookupFunction.java
diff --git a/.../internal/ClickHouseShardInputFormat.java → .../internal/ClickHouseShardInputFormat.java b/.../internal/ClickHouseShardInputFormat.java → .../internal/ClickHouseShardInputFormat.java
diff --git a/...internal/ClickHouseShardOutputFormat.java → ...internal/ClickHouseShardOutputFormat.java b/...internal/ClickHouseShardOutputFormat.java → ...internal/ClickHouseShardOutputFormat.java
diff --git a/.../internal/ClickHouseStatementFactory.java → .../internal/ClickHouseStatementFactory.java b/.../internal/ClickHouseStatementFactory.java → .../internal/ClickHouseStatementFactory.java
diff --git a/...nection/ClickHouseConnectionProvider.java → ...nection/ClickHouseConnectionProvider.java b/...nection/ClickHouseConnectionProvider.java → ...nection/ClickHouseConnectionProvider.java
diff --git a/...onnection/ClickHouseStatementWrapper.java → ...onnection/ClickHouseStatementWrapper.java b/...onnection/ClickHouseStatementWrapper.java → ...onnection/ClickHouseStatementWrapper.java
diff --git a/...l/converter/ClickHouseConverterUtils.java → ...l/converter/ClickHouseConverterUtils.java b/...l/converter/ClickHouseConverterUtils.java → ...l/converter/ClickHouseConverterUtils.java
diff --git a/...nal/converter/ClickHouseRowConverter.java → ...nal/converter/ClickHouseRowConverter.java b/...nal/converter/ClickHouseRowConverter.java → ...nal/converter/ClickHouseRowConverter.java
@@ -80,8 +80,7 @@ public void toExternal(RowData rowData, ClickHouseStatementWrapper statement)
         }
     }
 
-    protected ClickHouseRowConverter.DeserializationConverter createToInternalConverter(
-            LogicalType type) {
+    protected DeserializationConverter createToInternalConverter(LogicalType type) {
         switch (type.getTypeRoot()) {
             case NULL:
                 return val -> null;
@@ -134,8 +133,7 @@ protected ClickHouseRowConverter.DeserializationConverter createToInternalConver
         }
     }
 
-    protected ClickHouseRowConverter.SerializationConverter createToExternalConverter(
-            LogicalType type) {
+    protected SerializationConverter createToExternalConverter(LogicalType type) {
         switch (type.getTypeRoot()) {
             case BOOLEAN:
                 return (val, index, statement) ->

diff --git a/...nal/executor/ClickHouseBatchExecutor.java → ...nal/executor/ClickHouseBatchExecutor.java b/...nal/executor/ClickHouseBatchExecutor.java → ...nal/executor/ClickHouseBatchExecutor.java
diff --git a/...internal/executor/ClickHouseExecutor.java → ...internal/executor/ClickHouseExecutor.java b/...internal/executor/ClickHouseExecutor.java → ...internal/executor/ClickHouseExecutor.java
diff --git a/...al/executor/ClickHouseUpsertExecutor.java → ...al/executor/ClickHouseUpsertExecutor.java b/...al/executor/ClickHouseUpsertExecutor.java → ...al/executor/ClickHouseUpsertExecutor.java
diff --git a/.../options/ClickHouseConnectionOptions.java → .../options/ClickHouseConnectionOptions.java b/.../options/ClickHouseConnectionOptions.java → .../options/ClickHouseConnectionOptions.java