Skip to content

Commit

Permalink
[Feature]Upgrade to flink 1.17 #103
Browse files Browse the repository at this point in the history
  • Loading branch information
@czy006 @itinycheng
czy006 authored and itinycheng committed Oct 15, 2023
1 parent a43f9c3 commit a71a0ed
Show file tree
Hide file tree
Showing 72 changed files with 1,357 additions and 618 deletions.
18 changes: 0 additions & 18 deletions .mvn/wrapper/maven-wrapper.properties
View file

This file was deleted.

Empty file added docs/content.zh/docs/connectors/.keep
View file
Empty file.
Empty file added docs/content/docs/connectors/.keep
View file
Empty file.
24 changes: 24 additions & 0 deletions flink-connector-clickhouse-e2e-test/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -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>
230 changes: 230 additions & 0 deletions flink-connector-clickhouse/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -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>
0 ...khouse/ClickHouseDynamicTableFactory.java → ...khouse/ClickHouseDynamicTableFactory.java
View file
File renamed without changes.
0 ...lickhouse/ClickHouseDynamicTableSink.java → ...lickhouse/ClickHouseDynamicTableSink.java
View file
File renamed without changes.
0 ...ckhouse/ClickHouseDynamicTableSource.java → ...ckhouse/ClickHouseDynamicTableSource.java
View file
File renamed without changes.
0 ...clickhouse/catalog/ClickHouseCatalog.java → ...clickhouse/catalog/ClickHouseCatalog.java
View file
File renamed without changes.
0 ...use/catalog/ClickHouseCatalogFactory.java → ...use/catalog/ClickHouseCatalogFactory.java
View file
File renamed without changes.
0 ...r/clickhouse/config/ClickHouseConfig.java → ...r/clickhouse/config/ClickHouseConfig.java
View file
File renamed without changes.
0 ...house/config/ClickHouseConfigOptions.java → ...house/config/ClickHouseConfigOptions.java
View file
File renamed without changes.
0 ...ternal/AbstractClickHouseInputFormat.java → ...ternal/AbstractClickHouseInputFormat.java
View file
File renamed without changes.
13 changes: 6 additions & 7 deletions ...ernal/AbstractClickHouseOutputFormat.java → ...ernal/AbstractClickHouseOutputFormat.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;
}
Expand All @@ -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)
Expand All @@ -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;
}
Expand Down
0 .../internal/ClickHouseBatchInputFormat.java → .../internal/ClickHouseBatchInputFormat.java
View file
File renamed without changes.
0 ...internal/ClickHouseBatchOutputFormat.java → ...internal/ClickHouseBatchOutputFormat.java
View file
File renamed without changes.
0 ...rnal/ClickHouseRowDataLookupFunction.java → ...rnal/ClickHouseRowDataLookupFunction.java
View file
File renamed without changes.
0 .../internal/ClickHouseShardInputFormat.java → .../internal/ClickHouseShardInputFormat.java
View file
File renamed without changes.
0 ...internal/ClickHouseShardOutputFormat.java → ...internal/ClickHouseShardOutputFormat.java
View file
File renamed without changes.
0 .../internal/ClickHouseStatementFactory.java → .../internal/ClickHouseStatementFactory.java
View file
File renamed without changes.
0 ...nection/ClickHouseConnectionProvider.java → ...nection/ClickHouseConnectionProvider.java
View file
File renamed without changes.
0 ...onnection/ClickHouseStatementWrapper.java → ...onnection/ClickHouseStatementWrapper.java
View file
File renamed without changes.
0 ...l/converter/ClickHouseConverterUtils.java → ...l/converter/ClickHouseConverterUtils.java
View file
File renamed without changes.
6 changes: 2 additions & 4 deletions ...nal/converter/ClickHouseRowConverter.java → ...nal/converter/ClickHouseRowConverter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) ->
Expand Down
0 ...nal/executor/ClickHouseBatchExecutor.java → ...nal/executor/ClickHouseBatchExecutor.java
View file
File renamed without changes.
0 ...internal/executor/ClickHouseExecutor.java → ...internal/executor/ClickHouseExecutor.java
View file
File renamed without changes.
0 ...al/executor/ClickHouseUpsertExecutor.java → ...al/executor/ClickHouseUpsertExecutor.java
View file
File renamed without changes.
0 .../options/ClickHouseConnectionOptions.java → .../options/ClickHouseConnectionOptions.java
View file
File renamed without changes.
Loading

0 comments on commit a71a0ed

Please sign in to comment.