You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by re...@apache.org on 2024/04/24 04:11:10 UTC
(flink-cdc) branch master updated: [pipeline-connector][paimon] add paimon pipeline data sink connector. (#2916)
This is an automated email from the ASF dual-hosted git repository.
renqs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new ef2eece25 [pipeline-connector][paimon] add paimon pipeline data sink connector. (#2916)
ef2eece25 is described below
commit ef2eece256f35f3d7928f2c4e3407c049f895909
Author: Kunni <lv...@alibaba-inc.com>
AuthorDate: Wed Apr 24 12:11:05 2024 +0800
[pipeline-connector][paimon] add paimon pipeline data sink connector. (#2916)
---
docs/content.zh/docs/connectors/paimon.md | 232 +++++++++
docs/content/docs/connectors/paimon.md | 232 +++++++++
.../cdc/common/types/utils/DataTypeUtils.java | 92 +++-
.../cdc/common/types/utils/DataTypeUtilsTest.java | 5 +-
.../flink-cdc-pipeline-connector-paimon/pom.xml | 284 +++++++++++
.../cdc/connectors/paimon/sink/PaimonDataSink.java | 72 +++
.../paimon/sink/PaimonDataSinkFactory.java | 126 +++++
.../paimon/sink/PaimonDataSinkOptions.java | 67 +++
.../paimon/sink/PaimonMetadataApplier.java | 203 ++++++++
.../v2/MultiTableCommittableChannelComputer.java | 59 +++
.../connectors/paimon/sink/v2/PaimonCommitter.java | 68 +++
.../cdc/connectors/paimon/sink/v2/PaimonEvent.java | 70 +++
.../sink/v2/PaimonRecordEventSerializer.java | 87 ++++
.../paimon/sink/v2/PaimonRecordSerializer.java | 26 ++
.../cdc/connectors/paimon/sink/v2/PaimonSink.java | 98 ++++
.../connectors/paimon/sink/v2/PaimonWriter.java | 197 ++++++++
.../paimon/sink/v2/PaimonWriterHelper.java | 171 +++++++
.../paimon/sink/v2/PreCommitOperator.java | 97 ++++
.../paimon/sink/v2/StoreSinkWriteImpl.java | 223 +++++++++
.../connectors/paimon/sink/v2/TableSchemaInfo.java | 45 ++
.../org.apache.flink.cdc.common.factories.Factory | 16 +
.../paimon/sink/PaimonDataSinkFactoryTest.java | 63 +++
.../paimon/sink/PaimonMetadataApplierTest.java | 336 +++++++++++++
.../paimon/sink/v2/PaimonSinkITCase.java | 519 +++++++++++++++++++++
.../paimon/sink/v2/PaimonWriterHelperTest.java | 175 +++++++
.../TestMultiTableCommittableChannelComputer.java | 72 +++
.../test/resources/hadoop-conf-dir/core-site.xml | 36 ++
.../src/test/resources/hive-conf-dir/hive-site.xml | 47 ++
.../src/test/resources/log4j2-test.properties | 25 +
.../flink-cdc-pipeline-connectors/pom.xml | 1 +
30 files changed, 3720 insertions(+), 24 deletions(-)
diff --git a/docs/content.zh/docs/connectors/paimon.md b/docs/content.zh/docs/connectors/paimon.md
new file mode 100644
index 000000000..26cec4806
--- /dev/null
+++ b/docs/content.zh/docs/connectors/paimon.md
@@ -0,0 +1,232 @@
+---
+title: "Paimon"
+weight: 6
+type: docs
+aliases:
+- /connectors/paimon
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Paimon Pipeline 连接器
+
+Paimon Pipeline 连接器可以用作 Pipeline 的 *Data Sink*，将数据写入[Paimon](https://paimon.apache.org)。 本文档介绍如何设置 Paimon Pipeline 连接器。
+
+## 连接器的功能
+* 自动建表
+* 表结构变更同步
+* 数据实时同步
+
+如何创建 Pipeline
+----------------
+
+从 MySQL 读取数据同步到 Paimon 的 Pipeline 可以定义如下：
+
+```yaml
+source:
+ type: mysql
+ name: MySQL Source
+ hostname: 127.0.0.1
+ port: 3306
+ username: admin
+ password: pass
+ tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+ server-id: 5401-5404
+
+sink:
+ type: paimon
+ name: Paimon Sink
+ metastore: filesystem
+ warehouse: /path/warehouse
+
+pipeline:
+ name: MySQL to Paimon Pipeline
+ parallelism: 2
+```
+
+Pipeline 连接器配置项
+----------------
+<div class="highlight">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 7%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>type</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>指定要使用的连接器, 这里需要设置成 <code>'paimon'</code>.</td>
+ </tr>
+ <tr>
+ <td>name</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Sink 的名称.</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.metastore</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;"><code>"filesystem"</code></td>
+ <td>String</td>
+ <td>用于构建 Paimon Catalog 的类型。可选填值 filesystem 或者 hive。</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.warehouse</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Paimon 仓库存储数据的根目录。</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.uri</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Hive metastore 的 uri，在 metastore 设置为 hive 的时候需要。</td>
+ </tr>
+ <tr>
+ <td>commit.user</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;"><code>admin</code></td>
+ <td>String</td>
+ <td>提交数据文件时的用户名。</td>
+ </tr>
+ <tr>
+ <td>partition.key</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>设置每个分区表的分区字段，允许填写成多个分区表的多个分区字段。 不同的表使用 ';'分割， 而不同的字段则使用 ','分割。举个例子， 我们可以为两张表的不同分区字段作如下的设置 'testdb.table1:id1,id2;testdb.table2:name'。</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>将 Paimon catalog 支持的参数传递给 pipeline，参考 <a href="https://paimon.apache.org/docs/master/maintenance/configurations/#catalogoptions">Paimon catalog options</a>。 </td>
+ </tr>
+ <tr>
+ <td>table.properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>将 Paimon table 支持的参数传递给 pipeline，参考 <a href="https://paimon.apache.org/docs/master/maintenance/configurations/#coreoptions">Paimon table options</a>。 </td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+使用说明
+--------
+
+* 只支持主键表，因此源表必须有主键
+
+* 暂不支持 exactly-once，连接器 通过 at-least-once 和主键表实现幂等写
+
+数据类型映射
+----------------
+<div class="wy-table-responsive">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left">CDC type</th>
+ <th class="text-left">Paimon type</th>
+ <th class="text-left" style="width:60%;">NOTE</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>TINYINT</td>
+ <td>TINYINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>SMALLINT</td>
+ <td>SMALLINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>INT</td>
+ <td>INT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BIGINT</td>
+ <td>BIGINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>FLOAT</td>
+ <td>FLOAT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DOUBLE</td>
+ <td>DOUBLE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DECIMAL(p, s)</td>
+ <td>DECIMAL(p, s)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BOOLEAN</td>
+ <td>BOOLEAN</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DATE</td>
+ <td>DATE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP</td>
+ <td>DATETIME</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP_LTZ</td>
+ <td>TIMESTAMP_LTZ</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>CHAR(n)</td>
+ <td>CHAR(n)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>VARCHAR(n)</td>
+ <td>VARCHAR(n)</td>
+ <td></td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+{{< top >}}
\ No newline at end of file
diff --git a/docs/content/docs/connectors/paimon.md b/docs/content/docs/connectors/paimon.md
new file mode 100644
index 000000000..f8d030779
--- /dev/null
+++ b/docs/content/docs/connectors/paimon.md
@@ -0,0 +1,232 @@
+---
+title: "Paimon"
+weight: 6
+type: docs
+aliases:
+- /connectors/paimon
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Paimon Pipeline Connector
+
+The Paimon Pipeline connector can be used as the *Data Sink* of the pipeline, and write data to [Paimon](https://paimon.apache.org). This document describes how to set up the Paimon Pipeline connector.
+
+## What can the connector do?
+* Create table automatically if not exist
+* Schema change synchronization
+* Data synchronization
+
+How to create Pipeline
+----------------
+
+The pipeline for reading data from MySQL and sink to Paimon can be defined as follows:
+
+```yaml
+source:
+ type: mysql
+ name: MySQL Source
+ hostname: 127.0.0.1
+ port: 3306
+ username: admin
+ password: pass
+ tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+ server-id: 5401-5404
+
+sink:
+ type: paimon
+ name: Paimon Sink
+ metastore: filesystem
+ warehouse: /path/warehouse
+
+pipeline:
+ name: MySQL to Paimon Pipeline
+ parallelism: 2
+```
+
+Pipeline Connector Options
+----------------
+<div class="highlight">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 7%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>type</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, here should be <code>'paimon'</code>.</td>
+ </tr>
+ <tr>
+ <td>name</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The name of the sink.</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.metastore</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Metastore of paimon catalog, supports filesystem and hive.</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.warehouse</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The warehouse root path of catalog.</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.uri</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Uri of metastore server.</td>
+ </tr>
+ <tr>
+ <td>commit.user</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;"><code>"admin"</code></td>
+ <td>String</td>
+ <td>User name for committing data files.</td>
+ </tr>
+ <tr>
+ <td>partition.key</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Partition keys for each partitioned table, allow setting multiple primary keys for multiTables. Each table are separated by ';', and each partition key are separated by ','. For example, we can set partition.key of two tables by 'testdb.table1:id1,id2;testdb.table2:name'.</td>
+ </tr>
+ <tr>
+ <td>catalog.properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Pass options of Paimon catalog to pipeline，See <a href="https://paimon.apache.org/docs/master/maintenance/configurations/#catalogoptions">Paimon catalog options</a>。 </td>
+ </tr>
+ <tr>
+ <td>table.properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Pass options of Paimon table to pipeline，See <a href="https://paimon.apache.org/docs/master/maintenance/configurations/#coreoptions">Paimon table options</a>。 </td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+Usage Notes
+--------
+
+* Only support Paimon primary key table, so the source table must have primary keys.
+
+* Not support exactly-once. The connector uses at-least-once + primary key table for idempotent writing.
+
+Data Type Mapping
+----------------
+<div class="wy-table-responsive">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left">CDC type</th>
+ <th class="text-left">Paimon type</th>
+ <th class="text-left" style="width:60%;">NOTE</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>TINYINT</td>
+ <td>TINYINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>SMALLINT</td>
+ <td>SMALLINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>INT</td>
+ <td>INT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BIGINT</td>
+ <td>BIGINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>FLOAT</td>
+ <td>FLOAT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DOUBLE</td>
+ <td>DOUBLE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DECIMAL(p, s)</td>
+ <td>DECIMAL(p, s)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BOOLEAN</td>
+ <td>BOOLEAN</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DATE</td>
+ <td>DATE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP</td>
+ <td>TIMESTAMP</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP_LTZ</td>
+ <td>TIMESTAMP_LTZ</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>CHAR(n)</td>
+ <td>CHAR(n)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>VARCHAR(n)</td>
+ <td>VARCHAR(n)</td>
+ <td></td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+{{< top >}}
\ No newline at end of file
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/utils/DataTypeUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/utils/DataTypeUtils.java
index cc3405008..f54fba2ff 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/utils/DataTypeUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/utils/DataTypeUtils.java
@@ -95,46 +95,92 @@ public class DataTypeUtils {
int scale = DataTypes.getScale(type).orElse(0);
switch (type.getTypeRoot()) {
case CHAR:
- return org.apache.flink.table.api.DataTypes.CHAR(length);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.CHAR(length)
+ : org.apache.flink.table.api.DataTypes.CHAR(length).notNull();
case VARCHAR:
- return org.apache.flink.table.api.DataTypes.VARCHAR(length);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.VARCHAR(length)
+ : org.apache.flink.table.api.DataTypes.VARCHAR(length).notNull();
case BOOLEAN:
- return org.apache.flink.table.api.DataTypes.BOOLEAN();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.BOOLEAN()
+ : org.apache.flink.table.api.DataTypes.BOOLEAN().notNull();
case BINARY:
- return org.apache.flink.table.api.DataTypes.BINARY(length);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.BINARY(length)
+ : org.apache.flink.table.api.DataTypes.BINARY(length).notNull();
case VARBINARY:
- return org.apache.flink.table.api.DataTypes.VARBINARY(length);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.VARBINARY(length)
+ : org.apache.flink.table.api.DataTypes.VARBINARY(length).notNull();
case DECIMAL:
- return org.apache.flink.table.api.DataTypes.DECIMAL(precision, scale);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.DECIMAL(precision, scale)
+ : org.apache.flink.table.api.DataTypes.DECIMAL(precision, scale).notNull();
case TINYINT:
- return org.apache.flink.table.api.DataTypes.TINYINT();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.TINYINT()
+ : org.apache.flink.table.api.DataTypes.TINYINT().notNull();
case SMALLINT:
- return org.apache.flink.table.api.DataTypes.SMALLINT();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.SMALLINT()
+ : org.apache.flink.table.api.DataTypes.SMALLINT().notNull();
case INTEGER:
- return org.apache.flink.table.api.DataTypes.INT();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.INT()
+ : org.apache.flink.table.api.DataTypes.INT().notNull();
case DATE:
- return org.apache.flink.table.api.DataTypes.DATE();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.DATE()
+ : org.apache.flink.table.api.DataTypes.DATE().notNull();
case TIME_WITHOUT_TIME_ZONE:
- return org.apache.flink.table.api.DataTypes.TIME(precision);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.TIME(precision)
+ : org.apache.flink.table.api.DataTypes.TIME(precision).notNull();
case BIGINT:
- return org.apache.flink.table.api.DataTypes.BIGINT();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.BIGINT()
+ : org.apache.flink.table.api.DataTypes.BIGINT().notNull();
case FLOAT:
- return org.apache.flink.table.api.DataTypes.FLOAT();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.FLOAT()
+ : org.apache.flink.table.api.DataTypes.FLOAT().notNull();
case DOUBLE:
- return org.apache.flink.table.api.DataTypes.DOUBLE();
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.DOUBLE()
+ : org.apache.flink.table.api.DataTypes.DOUBLE().notNull();
case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_TIME_ZONE:
- return org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE(precision);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.TIMESTAMP(precision)
+ : org.apache.flink.table.api.DataTypes.TIMESTAMP(precision).notNull();
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(
- precision);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(
+ precision)
+ : org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(
+ precision)
+ .notNull();
+ case TIMESTAMP_WITH_TIME_ZONE:
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE(precision)
+ : org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE(precision)
+ .notNull();
case ARRAY:
Preconditions.checkState(children != null && !children.isEmpty());
- return org.apache.flink.table.api.DataTypes.ARRAY(toFlinkDataType(children.get(0)));
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.ARRAY(
+ toFlinkDataType(children.get(0)))
+ : org.apache.flink.table.api.DataTypes.ARRAY(
+ toFlinkDataType(children.get(0)).notNull());
case MAP:
Preconditions.checkState(children != null && children.size() > 1);
- return org.apache.flink.table.api.DataTypes.MAP(
- toFlinkDataType(children.get(0)), toFlinkDataType(children.get(1)));
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.MAP(
+ toFlinkDataType(children.get(0)), toFlinkDataType(children.get(1)))
+ : org.apache.flink.table.api.DataTypes.MAP(
+ toFlinkDataType(children.get(0)),
+ toFlinkDataType(children.get(1)).notNull());
case ROW:
Preconditions.checkState(!CollectionUtil.isNullOrEmpty(children));
RowType rowType = (RowType) type;
@@ -142,7 +188,9 @@ public class DataTypeUtils {
rowType.getFields().stream()
.map(DataField::toFlinkDataTypeField)
.collect(Collectors.toList());
- return org.apache.flink.table.api.DataTypes.ROW(fields);
+ return type.isNullable()
+ ? org.apache.flink.table.api.DataTypes.ROW(fields)
+ : org.apache.flink.table.api.DataTypes.ROW(fields).notNull();
default:
throw new IllegalArgumentException("Illegal type: " + type);
}
diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/utils/DataTypeUtilsTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/utils/DataTypeUtilsTest.java
index 5853f5769..2807cd326 100644
--- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/utils/DataTypeUtilsTest.java
+++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/utils/DataTypeUtilsTest.java
@@ -45,6 +45,7 @@ import static org.apache.flink.table.api.DataTypes.ROW;
import static org.apache.flink.table.api.DataTypes.SMALLINT;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.TIME;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE;
import static org.apache.flink.table.api.DataTypes.TINYINT;
@@ -116,8 +117,8 @@ class DataTypeUtilsTest {
FIELD("f14", DATE()),
FIELD("f15", TIME()),
FIELD("f16", TIME(6)),
- FIELD("f17", TIMESTAMP_WITH_TIME_ZONE()),
- FIELD("f18", TIMESTAMP_WITH_TIME_ZONE(6)),
+ FIELD("f17", TIMESTAMP()),
+ FIELD("f18", TIMESTAMP(6)),
FIELD("f19", TIMESTAMP_LTZ()),
FIELD("f20", TIMESTAMP_LTZ(6)),
FIELD("f21", TIMESTAMP_WITH_TIME_ZONE()),
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml
new file mode 100644
index 000000000..2c80c8aa9
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml
@@ -0,0 +1,284 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+<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">
+ <parent>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-cdc-pipeline-connector-paimon</artifactId>
+
+ <properties>
+ <paimon.version>0.7.0-incubating</paimon.version>
+ <hadoop.version>2.8.5</hadoop.version>
+ <hive.version>2.3.9</hive.version>
+ <mockito.version>3.4.6</mockito.version>
+ </properties>
+
+ <dependencies>
+ <!-- paimon dependency -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-flink-${flink.major.version}</artifactId>
+ <version>${paimon.version}</version>
+ </dependency>
+
+ <!-- test -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_2.12</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-inline</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito.version}</version>
+ <type>jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- hadoop dependency -->
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>jdk.tools</groupId>
+ <artifactId>jdk.tools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>jdk.tools</groupId>
+ <artifactId>jdk.tools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <!-- hive dependency -->
+
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-metastore</artifactId>
+ <version>${hive.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.orc</groupId>
+ <artifactId>orc-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>jdk.tools</groupId>
+ <artifactId>jdk.tools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <version>${hive.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.orc</groupId>
+ <artifactId>orc-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.pentaho</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-druid</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.calcite.avatica</groupId>
+ <artifactId>avatica</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-avatica</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.1.1</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>org.apache.paimon:*</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.kafka</pattern>
+ <shadedPattern>shaded.paimon.org.apache.kafka</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSink.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSink.java
new file mode 100644
index 000000000..6c94a3a74
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSink.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.EventSinkProvider;
+import org.apache.flink.cdc.common.sink.FlinkSinkProvider;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonRecordSerializer;
+import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonSink;
+
+import org.apache.paimon.options.Options;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/** A {@link DataSink} for Paimon connector that supports schema evolution. */
+public class PaimonDataSink implements DataSink, Serializable {
+
+ // options for creating Paimon catalog.
+ private final Options options;
+
+ // options for creating Paimon table.
+ private final Map<String, String> tableOptions;
+
+ private final String commitUser;
+
+ private final Map<TableId, List<String>> partitionMaps;
+
+ private final PaimonRecordSerializer<Event> serializer;
+
+ public PaimonDataSink(
+ Options options,
+ Map<String, String> tableOptions,
+ String commitUser,
+ Map<TableId, List<String>> partitionMaps,
+ PaimonRecordSerializer<Event> serializer) {
+ this.options = options;
+ this.tableOptions = tableOptions;
+ this.commitUser = commitUser;
+ this.partitionMaps = partitionMaps;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return FlinkSinkProvider.of(new PaimonSink<>(options, commitUser, serializer));
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return new PaimonMetadataApplier(options, tableOptions, partitionMaps);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java
new file mode 100644
index 000000000..3927fbc73
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonRecordEventSerializer;
+import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonRecordSerializer;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** A {@link DataSinkFactory} to create {@link PaimonDataSink}. */
+public class PaimonDataSinkFactory implements DataSinkFactory {
+
+ public static final String IDENTIFIER = "paimon";
+
+ @Override
+ public DataSink createDataSink(Context context) {
+ Map<String, String> allOptions = context.getFactoryConfiguration().toMap();
+ Map<String, String> catalogOptions = new HashMap<>();
+ Map<String, String> tableOptions = new HashMap<>();
+ allOptions.forEach(
+ (key, value) -> {
+ if (key.startsWith(PaimonDataSinkOptions.PREFIX_TABLE_PROPERTIES)) {
+ tableOptions.put(
+ key.substring(
+ PaimonDataSinkOptions.PREFIX_TABLE_PROPERTIES.length()),
+ value);
+ } else if (key.startsWith(PaimonDataSinkOptions.PREFIX_CATALOG_PROPERTIES)) {
+ catalogOptions.put(
+ key.substring(
+ PaimonDataSinkOptions.PREFIX_CATALOG_PROPERTIES.length()),
+ value);
+ }
+ });
+ Options options = Options.fromMap(catalogOptions);
+ try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options)) {
+ Preconditions.checkNotNull(
+ catalog.listDatabases(), "catalog option of Paimon is invalid.");
+ } catch (Exception e) {
+ throw new RuntimeException("failed to create or use paimon catalog", e);
+ }
+ ZoneId zoneId = ZoneId.systemDefault();
+ if (!Objects.equals(
+ context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+ PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
+ zoneId =
+ ZoneId.of(
+ context.getPipelineConfiguration()
+ .get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
+ }
+ String commitUser =
+ context.getFactoryConfiguration().get(PaimonDataSinkOptions.COMMIT_USER);
+ String partitionKey =
+ context.getFactoryConfiguration().get(PaimonDataSinkOptions.PARTITION_KEY);
+ Map<TableId, List<String>> partitionMaps = new HashMap<>();
+ if (!partitionKey.isEmpty()) {
+ for (String tables : partitionKey.split(";")) {
+ String[] splits = tables.split(":");
+ if (splits.length == 2) {
+ TableId tableId = TableId.parse(splits[0]);
+ List<String> partitions = Arrays.asList(splits[1].split(","));
+ partitionMaps.put(tableId, partitions);
+ } else {
+ throw new IllegalArgumentException(
+ PaimonDataSinkOptions.PARTITION_KEY.key()
+ + " is malformed, please refer to the documents");
+ }
+ }
+ }
+ PaimonRecordSerializer<Event> serializer = new PaimonRecordEventSerializer(zoneId);
+ return new PaimonDataSink(options, tableOptions, commitUser, partitionMaps, serializer);
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PaimonDataSinkOptions.METASTORE);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PaimonDataSinkOptions.URI);
+ options.add(PaimonDataSinkOptions.WAREHOUSE);
+ options.add(PaimonDataSinkOptions.COMMIT_USER);
+ return options;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkOptions.java
new file mode 100644
index 000000000..018920d09
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkOptions.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+
+import org.apache.paimon.options.CatalogOptions;
+
+import static org.apache.flink.cdc.common.configuration.ConfigOptions.key;
+
+/** copy from {@link CatalogOptions}. Options for {@link PaimonDataSink}. */
+public class PaimonDataSinkOptions {
+
+ // prefix for passing properties for table creation.
+ public static final String PREFIX_TABLE_PROPERTIES = "table.properties.";
+
+ // prefix for passing properties for catalog creation.
+ public static final String PREFIX_CATALOG_PROPERTIES = "catalog.properties.";
+
+ public static final ConfigOption<String> COMMIT_USER =
+ key("commit.user")
+ .stringType()
+ .defaultValue("admin")
+ .withDescription("User name for committing data files.");
+
+ public static final ConfigOption<String> WAREHOUSE =
+ key("catalog.properties.warehouse")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The warehouse root path of catalog.");
+
+ public static final ConfigOption<String> METASTORE =
+ key("catalog.properties.metastore")
+ .stringType()
+ .defaultValue("filesystem")
+ .withDescription("Metastore of paimon catalog, supports filesystem and hive.");
+
+ public static final ConfigOption<String> URI =
+ key("catalog.properties.uri")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Uri of metastore server.");
+
+ public static final ConfigOption<String> PARTITION_KEY =
+ key("partition.key")
+ .stringType()
+ .defaultValue("")
+ .withDescription(
+ "Partition keys for each partitioned table, allow setting multiple primary keys for multiTables. "
+ + "Tables are separated by ';', and partition keys are separated by ','. "
+ + "For example, we can set partition.key of two tables by 'testdb.table1:id1,id2;testdb.table2:name'.");
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java
new file mode 100644
index 000000000..9253108ca
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.RenameColumnEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.LogicalTypeConversion;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaChange;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A {@code MetadataApplier} that applies metadata changes to Paimon. Support primary key table
+ * only.
+ */
+public class PaimonMetadataApplier implements MetadataApplier {
+
+ // Catalog is unSerializable.
+ private transient Catalog catalog;
+
+ // currently, we set table options for all tables using the same options.
+ private final Map<String, String> tableOptions;
+
+ private final Options catalogOptions;
+
+ private final Map<TableId, List<String>> partitionMaps;
+
+ public PaimonMetadataApplier(Options catalogOptions) {
+ this.catalogOptions = catalogOptions;
+ this.tableOptions = new HashMap<>();
+ this.partitionMaps = new HashMap<>();
+ }
+
+ public PaimonMetadataApplier(
+ Options catalogOptions,
+ Map<String, String> tableOptions,
+ Map<TableId, List<String>> partitionMaps) {
+ this.catalogOptions = catalogOptions;
+ this.tableOptions = tableOptions;
+ this.partitionMaps = partitionMaps;
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ if (catalog == null) {
+ catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
+ }
+ try {
+ if (schemaChangeEvent instanceof CreateTableEvent) {
+ applyCreateTable((CreateTableEvent) schemaChangeEvent);
+ } else if (schemaChangeEvent instanceof AddColumnEvent) {
+ applyAddColumn((AddColumnEvent) schemaChangeEvent);
+ } else if (schemaChangeEvent instanceof DropColumnEvent) {
+ applyDropColumn((DropColumnEvent) schemaChangeEvent);
+ } else if (schemaChangeEvent instanceof RenameColumnEvent) {
+ applyRenameColumn((RenameColumnEvent) schemaChangeEvent);
+ } else if (schemaChangeEvent instanceof AlterColumnTypeEvent) {
+ applyAlterColumn((AlterColumnTypeEvent) schemaChangeEvent);
+ } else {
+ throw new UnsupportedOperationException(
+ "PaimonDataSink doesn't support schema change event " + schemaChangeEvent);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void applyCreateTable(CreateTableEvent event)
+ throws Catalog.DatabaseAlreadyExistException, Catalog.TableAlreadyExistException,
+ Catalog.DatabaseNotExistException {
+ if (!catalog.databaseExists(event.tableId().getSchemaName())) {
+ catalog.createDatabase(event.tableId().getSchemaName(), true);
+ }
+ Schema schema = event.getSchema();
+ org.apache.paimon.schema.Schema.Builder builder =
+ new org.apache.paimon.schema.Schema.Builder();
+ schema.getColumns()
+ .forEach(
+ (column) ->
+ builder.column(
+ column.getName(),
+ LogicalTypeConversion.toDataType(
+ DataTypeUtils.toFlinkDataType(column.getType())
+ .getLogicalType())));
+ builder.primaryKey(schema.primaryKeys().toArray(new String[0]));
+ if (partitionMaps.containsKey(event.tableId())) {
+ builder.partitionKeys(partitionMaps.get(event.tableId()));
+ } else if (schema.partitionKeys() != null && !schema.partitionKeys().isEmpty()) {
+ builder.partitionKeys(schema.partitionKeys());
+ }
+ builder.options(tableOptions);
+ catalog.createTable(
+ new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
+ builder.build(),
+ true);
+ }
+
+ private void applyAddColumn(AddColumnEvent event)
+ throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException,
+ Catalog.ColumnNotExistException {
+ List<SchemaChange> tableChangeList = new ArrayList<>();
+ event.getAddedColumns()
+ .forEach(
+ (column) -> {
+ SchemaChange tableChange =
+ SchemaChange.addColumn(
+ column.getAddColumn().getName(),
+ LogicalTypeConversion.toDataType(
+ DataTypeUtils.toFlinkDataType(
+ column.getAddColumn().getType())
+ .getLogicalType()));
+ tableChangeList.add(tableChange);
+ });
+ catalog.alterTable(
+ new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
+ tableChangeList,
+ true);
+ }
+
+ private void applyDropColumn(DropColumnEvent event)
+ throws Catalog.ColumnAlreadyExistException, Catalog.TableNotExistException,
+ Catalog.ColumnNotExistException {
+ List<SchemaChange> tableChangeList = new ArrayList<>();
+ event.getDroppedColumnNames()
+ .forEach(
+ (column) -> {
+ SchemaChange tableChange = SchemaChange.dropColumn(column);
+ tableChangeList.add(tableChange);
+ });
+ catalog.alterTable(
+ new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
+ tableChangeList,
+ true);
+ }
+
+ private void applyRenameColumn(RenameColumnEvent event)
+ throws Catalog.ColumnAlreadyExistException, Catalog.TableNotExistException,
+ Catalog.ColumnNotExistException {
+ List<SchemaChange> tableChangeList = new ArrayList<>();
+ event.getNameMapping()
+ .forEach(
+ (oldName, newName) -> {
+ SchemaChange tableChange = SchemaChange.renameColumn(oldName, newName);
+ tableChangeList.add(tableChange);
+ });
+ catalog.alterTable(
+ new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
+ tableChangeList,
+ true);
+ }
+
+ private void applyAlterColumn(AlterColumnTypeEvent event)
+ throws Catalog.ColumnAlreadyExistException, Catalog.TableNotExistException,
+ Catalog.ColumnNotExistException {
+ List<SchemaChange> tableChangeList = new ArrayList<>();
+ event.getTypeMapping()
+ .forEach(
+ (oldName, newType) -> {
+ SchemaChange tableChange =
+ SchemaChange.updateColumnType(
+ oldName,
+ LogicalTypeConversion.toDataType(
+ DataTypeUtils.toFlinkDataType(newType)
+ .getLogicalType()));
+ tableChangeList.add(tableChange);
+ });
+ catalog.alterTable(
+ new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
+ tableChangeList,
+ true);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/MultiTableCommittableChannelComputer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/MultiTableCommittableChannelComputer.java
new file mode 100644
index 000000000..6694b4074
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/MultiTableCommittableChannelComputer.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
+
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.table.sink.ChannelComputer;
+
+import java.util.Objects;
+
+/**
+ * {@link ChannelComputer} for {@link MultiTableCommittable}. Shuffle by database and table, make
+ * sure that MultiTableCommittable from the same table will be sent to the same channel.
+ */
+public class MultiTableCommittableChannelComputer
+ implements ChannelComputer<CommittableMessage<MultiTableCommittable>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient int numChannels;
+
+ @Override
+ public void setup(int numChannels) {
+ this.numChannels = numChannels;
+ }
+
+ @Override
+ public int channel(CommittableMessage<MultiTableCommittable> committableMessage) {
+ if (committableMessage instanceof CommittableWithLineage) {
+ MultiTableCommittable multiTableCommittable =
+ ((CommittableWithLineage<MultiTableCommittable>) committableMessage)
+ .getCommittable();
+ return Math.floorMod(
+ Objects.hash(
+ multiTableCommittable.getDatabase(), multiTableCommittable.getTable()),
+ numChannels);
+ } else {
+ // CommittableSummary is inaccurate after shuffling, need to be recreated.
+ return Math.floorMod(Objects.hash(committableMessage), numChannels);
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java
new file mode 100644
index 000000000..d0db819e1
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.api.connector.sink2.Committer;
+
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.flink.sink.StoreMultiCommitter;
+import org.apache.paimon.manifest.WrappedManifestCommittable;
+import org.apache.paimon.options.Options;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** A {@link Committer} to commit write results for multiple tables. */
+public class PaimonCommitter implements Committer<MultiTableCommittable> {
+
+ private final StoreMultiCommitter storeMultiCommitter;
+
+ public PaimonCommitter(Options catalogOptions, String commitUser) {
+ // flinkMetricGroup could be passed after FLIP-371.
+ storeMultiCommitter =
+ new StoreMultiCommitter(
+ () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions),
+ commitUser,
+ null);
+ }
+
+ @Override
+ public void commit(Collection<CommitRequest<MultiTableCommittable>> commitRequests)
+ throws IOException, InterruptedException {
+ if (commitRequests.isEmpty()) {
+ return;
+ }
+ List<MultiTableCommittable> committables =
+ commitRequests.stream()
+ .map(CommitRequest::getCommittable)
+ .collect(Collectors.toList());
+ long checkpointId = committables.get(0).checkpointId();
+ WrappedManifestCommittable wrappedManifestCommittable =
+ storeMultiCommitter.combine(checkpointId, 1L, committables);
+ storeMultiCommitter.commit(Collections.singletonList(wrappedManifestCommittable));
+ }
+
+ @Override
+ public void close() throws Exception {
+ storeMultiCommitter.close();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonEvent.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonEvent.java
new file mode 100644
index 000000000..eb136aae7
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonEvent.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+
+/** Contains the data to be written for {@link PaimonWriter}. */
+public class PaimonEvent {
+
+ // Identifier for the Paimon table to be written.
+ Identifier tableId;
+
+ // The actual record to be written to Paimon table.
+ GenericRow genericRow;
+
+ // if true, means that table schema has changed right before this genericRow.
+ boolean shouldRefreshSchema;
+
+ public PaimonEvent(Identifier tableId, GenericRow genericRow) {
+ this.tableId = tableId;
+ this.genericRow = genericRow;
+ this.shouldRefreshSchema = false;
+ }
+
+ public PaimonEvent(Identifier tableId, GenericRow genericRow, boolean shouldRefreshSchema) {
+ this.tableId = tableId;
+ this.genericRow = genericRow;
+ this.shouldRefreshSchema = shouldRefreshSchema;
+ }
+
+ public Identifier getTableId() {
+ return tableId;
+ }
+
+ public void setTableId(Identifier tableId) {
+ this.tableId = tableId;
+ }
+
+ public boolean isShouldRefreshSchema() {
+ return shouldRefreshSchema;
+ }
+
+ public void setShouldRefreshSchema(boolean shouldRefreshSchema) {
+ this.shouldRefreshSchema = shouldRefreshSchema;
+ }
+
+ public GenericRow getGenericRow() {
+ return genericRow;
+ }
+
+ public void setGenericRow(GenericRow genericRow) {
+ this.genericRow = genericRow;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordEventSerializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordEventSerializer.java
new file mode 100644
index 000000000..53b63f3b5
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordEventSerializer.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.cdc.common.event.ChangeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A {@link PaimonRecordSerializer} for converting {@link Event} into {@link PaimonEvent} for {@link
+ * PaimonWriter}.
+ */
+public class PaimonRecordEventSerializer implements PaimonRecordSerializer<Event> {
+
+ // maintain the latest schema of tableId.
+ private final Map<TableId, TableSchemaInfo> schemaMaps;
+
+ // ZoneId for converting relevant type.
+ private final ZoneId zoneId;
+
+ public PaimonRecordEventSerializer(ZoneId zoneId) {
+ schemaMaps = new HashMap<>();
+ this.zoneId = zoneId;
+ }
+
+ @Override
+ public PaimonEvent serialize(Event event) {
+ Identifier tableId =
+ Identifier.create(
+ ((ChangeEvent) event).tableId().getSchemaName(),
+ ((ChangeEvent) event).tableId().getTableName());
+ if (event instanceof SchemaChangeEvent) {
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ schemaMaps.put(
+ createTableEvent.tableId(),
+ new TableSchemaInfo(createTableEvent.getSchema(), zoneId));
+ } else {
+ SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
+ schemaMaps.put(
+ schemaChangeEvent.tableId(),
+ new TableSchemaInfo(
+ SchemaUtils.applySchemaChangeEvent(
+ schemaMaps.get(schemaChangeEvent.tableId()).getSchema(),
+ schemaChangeEvent),
+ zoneId));
+ }
+ return new PaimonEvent(tableId, null, true);
+ } else if (event instanceof DataChangeEvent) {
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+ GenericRow genericRow =
+ PaimonWriterHelper.convertEventToGenericRow(
+ dataChangeEvent,
+ schemaMaps.get(dataChangeEvent.tableId()).getFieldGetters());
+ return new PaimonEvent(tableId, genericRow);
+ } else {
+ throw new IllegalArgumentException(
+ "failed to convert Input into PaimonEvent, unsupported event: " + event);
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordSerializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordSerializer.java
new file mode 100644
index 000000000..b4f30cc31
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordSerializer.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/** Converting Input into {@link PaimonEvent} for {@link PaimonWriter}. */
+public interface PaimonRecordSerializer<Input> extends Serializable {
+ PaimonEvent serialize(Input t) throws IOException;
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSink.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSink.java
new file mode 100644
index 000000000..24002bd24
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSink.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.WithPreCommitTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.flink.sink.MultiTableCommittableSerializer;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+
+/**
+ * A {@link Sink} for Paimon. Maintain this package until Paimon has it own sinkV2 implementation.
+ */
+public class PaimonSink<InputT>
+ implements TwoPhaseCommittingSink<InputT, MultiTableCommittable>,
+ WithPreCommitTopology<InputT, MultiTableCommittable> {
+
+ // provided a default commit user.
+ public static final String DEFAULT_COMMIT_USER = "admin";
+
+ private final Options catalogOptions;
+
+ private final String commitUser;
+
+ private final PaimonRecordSerializer<InputT> serializer;
+
+ public PaimonSink(Options catalogOptions, PaimonRecordSerializer<InputT> serializer) {
+ this.catalogOptions = catalogOptions;
+ this.serializer = serializer;
+ commitUser = DEFAULT_COMMIT_USER;
+ }
+
+ public PaimonSink(
+ Options catalogOptions, String commitUser, PaimonRecordSerializer<InputT> serializer) {
+ this.catalogOptions = catalogOptions;
+ this.commitUser = commitUser;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public PaimonWriter<InputT> createWriter(InitContext context) {
+ return new PaimonWriter<>(catalogOptions, context.metricGroup(), commitUser, serializer);
+ }
+
+ @Override
+ public Committer<MultiTableCommittable> createCommitter() {
+ return new PaimonCommitter(catalogOptions, commitUser);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<MultiTableCommittable> getCommittableSerializer() {
+ CommitMessageSerializer fileSerializer = new CommitMessageSerializer();
+ return new MultiTableCommittableSerializer(fileSerializer);
+ }
+
+ @Override
+ public DataStream<CommittableMessage<MultiTableCommittable>> addPreCommitTopology(
+ DataStream<CommittableMessage<MultiTableCommittable>> committables) {
+ TypeInformation<CommittableMessage<MultiTableCommittable>> typeInformation =
+ CommittableMessageTypeInfo.of(this::getCommittableSerializer);
+ // shuffle MultiTableCommittable by tables
+ DataStream<CommittableMessage<MultiTableCommittable>> partitioned =
+ FlinkStreamPartitioner.partition(
+ committables,
+ new MultiTableCommittableChannelComputer(),
+ committables.getParallelism());
+
+ // add correct checkpointId to MultiTableCommittable and recreate CommittableSummary.
+ return partitioned
+ .transform("preCommit", typeInformation, new PreCommitOperator())
+ .setParallelism(committables.getParallelism());
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java
new file mode 100644
index 000000000..1b51535b2
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.flink.sink.StoreSinkWrite;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemoryPoolFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.ExecutorThreadFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+
+/** A {@link Sink} to write {@link DataChangeEvent} to Paimon storage. */
+public class PaimonWriter<InputT>
+ implements TwoPhaseCommittingSink.PrecommittingSinkWriter<InputT, MultiTableCommittable> {
+
+ // use `static` because Catalog is unSerializable.
+ private static Catalog catalog;
+ private final IOManager ioManager;
+
+ // Each job can only have one user name and this name must be consistent across restarts.
+ private final String commitUser;
+ // all table write should share one write buffer so that writers can preempt memory
+ // from those of other tables
+ private MemoryPoolFactory memoryPoolFactory;
+
+ // deserializer that converts Input into PaimonEvent.
+ private final PaimonRecordSerializer<InputT> serializer;
+ private final Map<Identifier, FileStoreTable> tables;
+ private final Map<Identifier, StoreSinkWrite> writes;
+ private final ExecutorService compactExecutor;
+ private final MetricGroup metricGroup;
+ private final List<MultiTableCommittable> committables;
+
+ public PaimonWriter(
+ Options catalogOptions,
+ MetricGroup metricGroup,
+ String commitUser,
+ PaimonRecordSerializer<InputT> serializer) {
+ catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
+ this.metricGroup = metricGroup;
+ this.commitUser = commitUser;
+ this.tables = new HashMap<>();
+ this.writes = new HashMap<>();
+ this.committables = new ArrayList<>();
+ this.ioManager = new IOManagerAsync();
+ this.compactExecutor =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory(
+ Thread.currentThread().getName() + "-CdcMultiWrite-Compaction"));
+ this.serializer = serializer;
+ }
+
+ @Override
+ public Collection<MultiTableCommittable> prepareCommit() {
+ Collection<MultiTableCommittable> allCommittables = new ArrayList<>(committables);
+ committables.clear();
+ return allCommittables;
+ }
+
+ @Override
+ public void write(InputT event, Context context) throws IOException {
+ PaimonEvent paimonEvent = serializer.serialize(event);
+ Identifier tableId = paimonEvent.getTableId();
+ if (paimonEvent.isShouldRefreshSchema()) {
+ // remove the table temporarily, then add the table with latest schema when received
+ // DataChangeEvent.
+ writes.remove(tableId);
+ tables.remove(tableId);
+ }
+ if (paimonEvent.getGenericRow() != null) {
+ FileStoreTable table;
+ table = getTable(tableId);
+ if (memoryPoolFactory == null) {
+ memoryPoolFactory =
+ new MemoryPoolFactory(
+ // currently, the options of all tables are the same in CDC
+ new HeapMemorySegmentPool(
+ table.coreOptions().writeBufferSize(),
+ table.coreOptions().pageSize()));
+ }
+ StoreSinkWrite write =
+ writes.computeIfAbsent(
+ tableId,
+ id -> {
+ StoreSinkWriteImpl storeSinkWrite =
+ new StoreSinkWriteImpl(
+ table,
+ commitUser,
+ ioManager,
+ false,
+ false,
+ true,
+ memoryPoolFactory,
+ metricGroup);
+ storeSinkWrite.withCompactExecutor(compactExecutor);
+ return storeSinkWrite;
+ });
+ try {
+ write.write(paimonEvent.getGenericRow());
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ }
+
+ private FileStoreTable getTable(Identifier tableId) {
+ FileStoreTable table =
+ tables.computeIfAbsent(
+ tableId,
+ id -> {
+ try {
+ return (FileStoreTable) catalog.getTable(tableId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ if (table.bucketMode() != BucketMode.FIXED) {
+ throw new UnsupportedOperationException(
+ "Unified Sink only supports FIXED bucket mode, but is " + table.bucketMode());
+ }
+ return table;
+ }
+
+ /**
+ * Called on checkpoint or end of input so that the writer to flush all pending data for
+ * at-least-once.
+ *
+ * <p>this method will also be called when receiving {@link FlushEvent}, but we don't need to
+ * commit the MultiTableCommittables immediately in this case, because {@link PaimonCommitter}
+ * support committing data of different schemas.
+ */
+ @Override
+ public void flush(boolean endOfInput) throws IOException {
+ for (Map.Entry<Identifier, StoreSinkWrite> entry : writes.entrySet()) {
+ Identifier key = entry.getKey();
+ StoreSinkWrite write = entry.getValue();
+ boolean waitCompaction = false;
+ // checkpointId will be updated correctly by PreCommitOperator.
+ long checkpointId = 1L;
+ committables.addAll(
+ write.prepareCommit(waitCompaction, checkpointId).stream()
+ .map(
+ committable ->
+ MultiTableCommittable.fromCommittable(key, committable))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ for (StoreSinkWrite write : writes.values()) {
+ write.close();
+ }
+ if (compactExecutor != null) {
+ compactExecutor.shutdownNow();
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java
new file mode 100644
index 000000000..a2df17455
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypeChecks;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.RowKind;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.cdc.common.types.DataTypeChecks.getFieldCount;
+
+/** A helper class for {@link PaimonWriter} to create FieldGetter and GenericRow. */
+public class PaimonWriterHelper {
+
+ /** create a list of {@link RecordData.FieldGetter} for {@link PaimonWriter}. */
+ public static List<RecordData.FieldGetter> createFieldGetters(Schema schema, ZoneId zoneId) {
+ List<Column> columns = schema.getColumns();
+ List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(columns.size());
+ for (int i = 0; i < columns.size(); i++) {
+ fieldGetters.add(createFieldGetter(columns.get(i).getType(), i, zoneId));
+ }
+ return fieldGetters;
+ }
+
+ private static RecordData.FieldGetter createFieldGetter(
+ DataType fieldType, int fieldPos, ZoneId zoneId) {
+ final RecordData.FieldGetter fieldGetter;
+ // ordered by type root definition
+ switch (fieldType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ fieldGetter = row -> BinaryString.fromString(row.getString(fieldPos).toString());
+ break;
+ case BOOLEAN:
+ fieldGetter = row -> row.getBoolean(fieldPos);
+ break;
+ case BINARY:
+ case VARBINARY:
+ fieldGetter = row -> row.getBinary(fieldPos);
+ break;
+ case DECIMAL:
+ final int decimalPrecision = DataTypeChecks.getPrecision(fieldType);
+ final int decimalScale = DataTypeChecks.getScale(fieldType);
+ fieldGetter =
+ row -> {
+ DecimalData decimalData =
+ row.getDecimal(fieldPos, decimalPrecision, decimalScale);
+ return Decimal.fromBigDecimal(
+ decimalData.toBigDecimal(), decimalPrecision, decimalScale);
+ };
+ break;
+ case TINYINT:
+ fieldGetter = row -> row.getByte(fieldPos);
+ break;
+ case SMALLINT:
+ fieldGetter = row -> row.getShort(fieldPos);
+ break;
+ case BIGINT:
+ fieldGetter = row -> row.getLong(fieldPos);
+ break;
+ case FLOAT:
+ fieldGetter = row -> row.getFloat(fieldPos);
+ break;
+ case DOUBLE:
+ fieldGetter = row -> row.getDouble(fieldPos);
+ break;
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ fieldGetter = row -> row.getInt(fieldPos);
+ break;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ fieldGetter =
+ row ->
+ Timestamp.fromSQLTimestamp(
+ row.getTimestamp(
+ fieldPos,
+ DataTypeChecks.getPrecision(fieldType))
+ .toTimestamp());
+ break;
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ fieldGetter =
+ row ->
+ Timestamp.fromLocalDateTime(
+ ZonedDateTime.ofInstant(
+ row.getLocalZonedTimestampData(
+ fieldPos,
+ DataTypeChecks.getPrecision(
+ fieldType))
+ .toInstant(),
+ zoneId)
+ .toLocalDateTime());
+ break;
+ case ROW:
+ final int rowFieldCount = getFieldCount(fieldType);
+ fieldGetter = row -> row.getRow(fieldPos, rowFieldCount);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "don't support type of " + fieldType.getTypeRoot());
+ }
+ if (!fieldType.isNullable()) {
+ return fieldGetter;
+ }
+ return row -> {
+ if (row.isNullAt(fieldPos)) {
+ return null;
+ }
+ return fieldGetter.getFieldOrNull(row);
+ };
+ }
+
+ /** create a {@link GenericRow} from a {@link DataChangeEvent} for {@link PaimonWriter}. */
+ public static GenericRow convertEventToGenericRow(
+ DataChangeEvent dataChangeEvent, List<RecordData.FieldGetter> fieldGetters) {
+ GenericRow genericRow;
+ RecordData recordData;
+ switch (dataChangeEvent.op()) {
+ case INSERT:
+ case UPDATE:
+ case REPLACE:
+ {
+ recordData = dataChangeEvent.after();
+ genericRow = new GenericRow(RowKind.INSERT, recordData.getArity());
+ break;
+ }
+ case DELETE:
+ {
+ recordData = dataChangeEvent.before();
+ genericRow = new GenericRow(RowKind.DELETE, recordData.getArity());
+ break;
+ }
+ default:
+ throw new IllegalArgumentException("don't support type of " + dataChangeEvent.op());
+ }
+ for (int i = 0; i < recordData.getArity(); i++) {
+ genericRow.setField(i, fieldGetters.get(i).getFieldOrNull(recordData));
+ }
+ return genericRow;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PreCommitOperator.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PreCommitOperator.java
new file mode 100644
index 000000000..07a7e0cd7
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PreCommitOperator.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
+import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** An Operator to add checkpointId to MultiTableCommittable and generate CommittableSummary. */
+public class PreCommitOperator
+ extends AbstractStreamOperator<CommittableMessage<MultiTableCommittable>>
+ implements OneInputStreamOperator<
+ CommittableMessage<MultiTableCommittable>,
+ CommittableMessage<MultiTableCommittable>> {
+
+ /** store a list of MultiTableCommittable in one checkpoint. */
+ private final List<MultiTableCommittable> results;
+
+ public PreCommitOperator() {
+ results = new ArrayList<>();
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ }
+
+ @Override
+ public void processElement(StreamRecord<CommittableMessage<MultiTableCommittable>> element) {
+ if (element.getValue() instanceof CommittableWithLineage) {
+ results.add(
+ ((CommittableWithLineage<MultiTableCommittable>) element.getValue())
+ .getCommittable());
+ }
+ }
+
+ @Override
+ public void finish() {
+ prepareSnapshotPreBarrier(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ // CommittableSummary should be sent before all CommittableWithLineage.
+ CommittableMessage<MultiTableCommittable> summary =
+ new CommittableSummary<>(
+ getRuntimeContext().getIndexOfThisSubtask(),
+ getRuntimeContext().getNumberOfParallelSubtasks(),
+ checkpointId,
+ results.size(),
+ results.size(),
+ 0);
+ output.collect(new StreamRecord<>(summary));
+
+ results.forEach(
+ committable -> {
+ // update the right checkpointId for MultiTableCommittable
+ MultiTableCommittable committableWithCheckPointId =
+ new MultiTableCommittable(
+ committable.getDatabase(),
+ committable.getTable(),
+ checkpointId,
+ committable.kind(),
+ committable.wrappedCommittable());
+ CommittableMessage<MultiTableCommittable> message =
+ new CommittableWithLineage<>(
+ committableWithCheckPointId,
+ checkpointId,
+ getRuntimeContext().getIndexOfThisSubtask());
+ output.collect(new StreamRecord<>(message));
+ });
+ results.clear();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java
new file mode 100644
index 000000000..3d49086b2
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.state.StateInitializationContext;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.StoreSinkWrite;
+import org.apache.paimon.flink.sink.StoreSinkWriteState;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemoryPoolFactory;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.SinkRecord;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * copy from {@link org.apache.paimon.flink.sink.StoreSinkWriteImpl}. remove {@link
+ * StoreSinkWriteState} because we can't create it using {@link StateInitializationContext}.
+ */
+public class StoreSinkWriteImpl implements StoreSinkWrite {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(org.apache.paimon.flink.sink.StoreSinkWriteImpl.class);
+
+ protected final String commitUser;
+ private final IOManagerImpl paimonIOManager;
+ private final boolean ignorePreviousFiles;
+ private final boolean waitCompaction;
+ private final boolean isStreamingMode;
+ @Nullable private final MemorySegmentPool memoryPool;
+ @Nullable private final MemoryPoolFactory memoryPoolFactory;
+
+ protected TableWriteImpl<?> write;
+
+ @Nullable private final MetricGroup metricGroup;
+
+ public StoreSinkWriteImpl(
+ FileStoreTable table,
+ String commitUser,
+ IOManager ioManager,
+ boolean ignorePreviousFiles,
+ boolean waitCompaction,
+ boolean isStreamingMode,
+ MemoryPoolFactory memoryPoolFactory,
+ @Nullable MetricGroup metricGroup) {
+ this(
+ table,
+ commitUser,
+ ioManager,
+ ignorePreviousFiles,
+ waitCompaction,
+ isStreamingMode,
+ null,
+ memoryPoolFactory,
+ metricGroup);
+ }
+
+ private StoreSinkWriteImpl(
+ FileStoreTable table,
+ String commitUser,
+ IOManager ioManager,
+ boolean ignorePreviousFiles,
+ boolean waitCompaction,
+ boolean isStreamingMode,
+ @Nullable MemorySegmentPool memoryPool,
+ @Nullable MemoryPoolFactory memoryPoolFactory,
+ @Nullable MetricGroup metricGroup) {
+ this.commitUser = commitUser;
+ this.paimonIOManager = new IOManagerImpl(ioManager.getSpillingDirectoriesPaths());
+ this.ignorePreviousFiles = ignorePreviousFiles;
+ this.waitCompaction = waitCompaction;
+ this.isStreamingMode = isStreamingMode;
+ this.memoryPool = memoryPool;
+ this.memoryPoolFactory = memoryPoolFactory;
+ this.metricGroup = metricGroup;
+ this.write = newTableWrite(table);
+ }
+
+ private TableWriteImpl<?> newTableWrite(FileStoreTable table) {
+ checkArgument(
+ !(memoryPool != null && memoryPoolFactory != null),
+ "memoryPool and memoryPoolFactory cannot be set at the same time.");
+
+ TableWriteImpl<?> tableWrite =
+ table.newWrite(commitUser, (part, bucket) -> true)
+ .withIOManager(paimonIOManager)
+ .withIgnorePreviousFiles(ignorePreviousFiles);
+
+ if (metricGroup != null) {
+ tableWrite.withMetricRegistry(new FlinkMetricRegistry(metricGroup));
+ }
+
+ if (memoryPoolFactory != null) {
+ return tableWrite.withMemoryPoolFactory(memoryPoolFactory);
+ } else {
+ return tableWrite.withMemoryPool(
+ memoryPool != null
+ ? memoryPool
+ : new HeapMemorySegmentPool(
+ table.coreOptions().writeBufferSize(),
+ table.coreOptions().pageSize()));
+ }
+ }
+
+ public void withCompactExecutor(ExecutorService compactExecutor) {
+ write.withCompactExecutor(compactExecutor);
+ }
+
+ @Override
+ public SinkRecord write(InternalRow internalRow) throws Exception {
+ return write.writeAndReturn(internalRow);
+ }
+
+ @Override
+ public SinkRecord write(InternalRow internalRow, int i) throws Exception {
+ return write.writeAndReturn(internalRow);
+ }
+
+ @Override
+ public SinkRecord toLogRecord(SinkRecord record) {
+ return write.toLogRecord(record);
+ }
+
+ @Override
+ public void compact(BinaryRow partition, int bucket, boolean fullCompaction) throws Exception {
+ write.compact(partition, bucket, fullCompaction);
+ }
+
+ @Override
+ public void notifyNewFiles(
+ long snapshotId, BinaryRow partition, int bucket, List<DataFileMeta> files) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Receive {} new files from snapshot {}, partition {}, bucket {}",
+ files.size(),
+ snapshotId,
+ partition,
+ bucket);
+ }
+ write.notifyNewFiles(snapshotId, partition, bucket, files);
+ }
+
+ @Override
+ public List<Committable> prepareCommit(boolean waitCompaction, long checkpointId)
+ throws IOException {
+ List<Committable> committables = new ArrayList<>();
+ if (write != null) {
+ try {
+ for (CommitMessage committable :
+ write.prepareCommit(this.waitCompaction || waitCompaction, checkpointId)) {
+ committables.add(
+ new Committable(checkpointId, Committable.Kind.FILE, committable));
+ }
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ return committables;
+ }
+
+ @Override
+ public void snapshotState() {
+ // do nothing
+ }
+
+ @Override
+ public boolean streamingMode() {
+ return isStreamingMode;
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (write != null) {
+ write.close();
+ }
+
+ paimonIOManager.close();
+ }
+
+ @Override
+ public void replace(FileStoreTable newTable) throws Exception {
+ if (commitUser == null) {
+ return;
+ }
+ write.close();
+ write = newTableWrite(newTable);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TableSchemaInfo.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TableSchemaInfo.java
new file mode 100644
index 000000000..26b4a1b12
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TableSchemaInfo.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.schema.Schema;
+
+import java.time.ZoneId;
+import java.util.List;
+
+/** Keep a list of {@link RecordData.FieldGetter} for a specific {@link Schema}. */
+public class TableSchemaInfo {
+
+ private final Schema schema;
+
+ private final List<RecordData.FieldGetter> fieldGetters;
+
+ public TableSchemaInfo(Schema schema, ZoneId zoneId) {
+ this.schema = schema;
+ this.fieldGetters = PaimonWriterHelper.createFieldGetters(schema, zoneId);
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public List<RecordData.FieldGetter> getFieldGetters() {
+ return fieldGetters;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 000000000..6ad058535
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.cdc.connectors.paimon.sink.PaimonDataSinkFactory
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactoryTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactoryTest.java
new file mode 100644
index 000000000..f2ab4a51b
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactoryTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.UUID;
+
+/** Tests for {@link PaimonDataSinkFactory}. */
+public class PaimonDataSinkFactoryTest {
+
+ @TempDir public static java.nio.file.Path temporaryFolder;
+
+ @Test
+ public void testCreateDataSink() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("paimon", DataSinkFactory.class);
+ Assertions.assertInstanceOf(PaimonDataSinkFactory.class, sinkFactory);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(PaimonDataSinkOptions.METASTORE.key(), "filesystem")
+ .put(
+ PaimonDataSinkOptions.WAREHOUSE.key(),
+ new File(
+ temporaryFolder.toFile(),
+ UUID.randomUUID().toString())
+ .toString())
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertInstanceOf(PaimonDataSink.class, dataSink);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java
new file mode 100644
index 000000000..dc24ff37f
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java
@@ -0,0 +1,336 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.RenameColumnEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.types.DataType;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/** Tests for {@link PaimonMetadataApplier}. */
+public class PaimonMetadataApplierTest {
+
+ @TempDir public static java.nio.file.Path temporaryFolder;
+
+ private Catalog catalog;
+
+ private Options catalogOptions;
+
+ public static final String TEST_DATABASE = "test";
+
+ private static final String HADOOP_CONF_DIR =
+ Objects.requireNonNull(
+ Thread.currentThread()
+ .getContextClassLoader()
+ .getResource("hadoop-conf-dir"))
+ .getPath();
+
+ private static final String HIVE_CONF_DIR =
+ Objects.requireNonNull(
+ Thread.currentThread()
+ .getContextClassLoader()
+ .getResource("hive-conf-dir"))
+ .getPath();
+
+ private void initialize(String metastore)
+ throws Catalog.DatabaseNotEmptyException, Catalog.DatabaseNotExistException {
+ catalogOptions = new Options();
+ String warehouse =
+ new File(temporaryFolder.toFile(), UUID.randomUUID().toString()).toString();
+ if ("hive".equals(metastore)) {
+ catalogOptions.setString("hadoop-conf-dir", HADOOP_CONF_DIR);
+ catalogOptions.setString("hive-conf-dir", HIVE_CONF_DIR);
+ }
+ catalogOptions.setString("metastore", metastore);
+ catalogOptions.setString("warehouse", warehouse);
+ this.catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
+ this.catalog.dropDatabase(TEST_DATABASE, true, true);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testApplySchemaChange(String metastore)
+ throws Catalog.TableNotExistException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ MetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(
+ TableId.parse("test.table1"),
+ org.apache.flink.cdc.common.schema.Schema.newBuilder()
+ .physicalColumn(
+ "col1",
+ org.apache.flink.cdc.common.types.DataTypes.STRING()
+ .notNull())
+ .physicalColumn(
+ "col2", org.apache.flink.cdc.common.types.DataTypes.INT())
+ .primaryKey("col1")
+ .build());
+ metadataApplier.applySchemaChange(createTableEvent);
+ RowType tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "col2", DataTypes.INT())));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+
+ List<AddColumnEvent.ColumnWithPosition> addedColumns = new ArrayList<>();
+ addedColumns.add(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "col3", org.apache.flink.cdc.common.types.DataTypes.STRING())));
+ AddColumnEvent addColumnEvent =
+ new AddColumnEvent(TableId.parse("test.table1"), addedColumns);
+ metadataApplier.applySchemaChange(addColumnEvent);
+ tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "col2", DataTypes.INT()),
+ new DataField(2, "col3", DataTypes.STRING())));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+
+ Map<String, String> nameMapping = new HashMap<>();
+ nameMapping.put("col2", "newcol2");
+ nameMapping.put("col3", "newcol3");
+ RenameColumnEvent renameColumnEvent =
+ new RenameColumnEvent(TableId.parse("test.table1"), nameMapping);
+ metadataApplier.applySchemaChange(renameColumnEvent);
+ tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "newcol2", DataTypes.INT()),
+ new DataField(2, "newcol3", DataTypes.STRING())));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+
+ Map<String, DataType> typeMapping = new HashMap<>();
+ typeMapping.put("newcol2", org.apache.flink.cdc.common.types.DataTypes.STRING());
+ AlterColumnTypeEvent alterColumnTypeEvent =
+ new AlterColumnTypeEvent(TableId.parse("test.table1"), typeMapping);
+ metadataApplier.applySchemaChange(alterColumnTypeEvent);
+ tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "newcol2", DataTypes.STRING()),
+ new DataField(2, "newcol3", DataTypes.STRING())));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+
+ DropColumnEvent dropColumnEvent =
+ new DropColumnEvent(
+ TableId.parse("test.table1"), Collections.singletonList("newcol2"));
+ metadataApplier.applySchemaChange(dropColumnEvent);
+ // id of DataField should keep the same as before dropping column
+ tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(2, "newcol3", DataTypes.STRING())));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testCreateTableWithOptions(String metastore)
+ throws Catalog.TableNotExistException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("bucket", "-1");
+ Map<TableId, List<String>> partitionMaps = new HashMap<>();
+ partitionMaps.put(TableId.parse("test.table1"), Arrays.asList("col3", "col4"));
+ MetadataApplier metadataApplier =
+ new PaimonMetadataApplier(catalogOptions, tableOptions, partitionMaps);
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(
+ TableId.parse("test.table1"),
+ org.apache.flink.cdc.common.schema.Schema.newBuilder()
+ .physicalColumn(
+ "col1",
+ org.apache.flink.cdc.common.types.DataTypes.STRING()
+ .notNull())
+ .physicalColumn(
+ "col2",
+ org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .physicalColumn(
+ "col3",
+ org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .physicalColumn(
+ "col4",
+ org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .primaryKey("col1")
+ .build());
+ metadataApplier.applySchemaChange(createTableEvent);
+ Table table = catalog.getTable(Identifier.fromString("test.table1"));
+ RowType tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "col2", DataTypes.STRING()),
+ new DataField(2, "col3", DataTypes.STRING()),
+ new DataField(3, "col4", DataTypes.STRING())));
+ Assertions.assertEquals(tableSchema, table.rowType());
+ Assertions.assertEquals(Collections.singletonList("col1"), table.primaryKeys());
+ Assertions.assertEquals(Arrays.asList("col3", "col4"), table.partitionKeys());
+ Assertions.assertEquals("-1", table.options().get("bucket"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testCreateTableWithAllDataTypes(String metastore)
+ throws Catalog.TableNotExistException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ MetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(
+ TableId.parse("test.table1"),
+ org.apache.flink.cdc.common.schema.Schema.newBuilder()
+ .physicalColumn(
+ "col1",
+ org.apache.flink.cdc.common.types.DataTypes.STRING()
+ .notNull())
+ .physicalColumn(
+ "boolean",
+ org.apache.flink.cdc.common.types.DataTypes.BOOLEAN())
+ .physicalColumn(
+ "binary",
+ org.apache.flink.cdc.common.types.DataTypes.BINARY(3))
+ .physicalColumn(
+ "varbinary",
+ org.apache.flink.cdc.common.types.DataTypes.VARBINARY(10))
+ .physicalColumn(
+ "bytes",
+ org.apache.flink.cdc.common.types.DataTypes.BYTES())
+ .physicalColumn(
+ "tinyint",
+ org.apache.flink.cdc.common.types.DataTypes.TINYINT())
+ .physicalColumn(
+ "smallint",
+ org.apache.flink.cdc.common.types.DataTypes.SMALLINT())
+ .physicalColumn(
+ "int", org.apache.flink.cdc.common.types.DataTypes.INT())
+ .physicalColumn(
+ "float",
+ org.apache.flink.cdc.common.types.DataTypes.FLOAT())
+ .physicalColumn(
+ "double",
+ org.apache.flink.cdc.common.types.DataTypes.DOUBLE())
+ .physicalColumn(
+ "decimal",
+ org.apache.flink.cdc.common.types.DataTypes.DECIMAL(6, 3))
+ .physicalColumn(
+ "char", org.apache.flink.cdc.common.types.DataTypes.CHAR(5))
+ .physicalColumn(
+ "varchar",
+ org.apache.flink.cdc.common.types.DataTypes.VARCHAR(10))
+ .physicalColumn(
+ "string",
+ org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .physicalColumn(
+ "date", org.apache.flink.cdc.common.types.DataTypes.DATE())
+ .physicalColumn(
+ "time", org.apache.flink.cdc.common.types.DataTypes.TIME())
+ .physicalColumn(
+ "time_with_precision",
+ org.apache.flink.cdc.common.types.DataTypes.TIME(6))
+ .physicalColumn(
+ "timestamp",
+ org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP())
+ .physicalColumn(
+ "timestamp_with_precision",
+ org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP(3))
+ .physicalColumn(
+ "timestamp_ltz",
+ org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ())
+ .physicalColumn(
+ "timestamp_ltz_with_precision",
+ org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ(
+ 3))
+ .primaryKey("col1")
+ .build());
+ metadataApplier.applySchemaChange(createTableEvent);
+ RowType tableSchema =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "col1", DataTypes.STRING().notNull()),
+ new DataField(1, "boolean", DataTypes.BOOLEAN()),
+ new DataField(2, "binary", DataTypes.BINARY(3)),
+ new DataField(3, "varbinary", DataTypes.VARBINARY(10)),
+ new DataField(4, "bytes", DataTypes.BYTES()),
+ new DataField(5, "tinyint", DataTypes.TINYINT()),
+ new DataField(6, "smallint", DataTypes.SMALLINT()),
+ new DataField(7, "int", DataTypes.INT()),
+ new DataField(8, "float", DataTypes.FLOAT()),
+ new DataField(9, "double", DataTypes.DOUBLE()),
+ new DataField(10, "decimal", DataTypes.DECIMAL(6, 3)),
+ new DataField(11, "char", DataTypes.CHAR(5)),
+ new DataField(12, "varchar", DataTypes.VARCHAR(10)),
+ new DataField(13, "string", DataTypes.STRING()),
+ new DataField(14, "date", DataTypes.DATE()),
+ new DataField(15, "time", DataTypes.TIME(0)),
+ new DataField(16, "time_with_precision", DataTypes.TIME(6)),
+ new DataField(17, "timestamp", DataTypes.TIMESTAMP(6)),
+ new DataField(
+ 18, "timestamp_with_precision", DataTypes.TIMESTAMP(3)),
+ new DataField(
+ 19,
+ "timestamp_ltz",
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)),
+ new DataField(
+ 20,
+ "timestamp_ltz_with_precision",
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3))));
+ Assertions.assertEquals(
+ tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
new file mode 100644
index 000000000..edc785e89
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
@@ -0,0 +1,519 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.operators.ProcessingTimeService;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.paimon.sink.PaimonMetadataApplier;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.options.Options;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.OptionalLong;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/** An ITCase for {@link PaimonWriter} and {@link PaimonCommitter}. */
+public class PaimonSinkITCase {
+
+ @TempDir public static java.nio.file.Path temporaryFolder;
+
+ private Options catalogOptions;
+
+ private TableEnvironment tEnv;
+
+ private String warehouse;
+
+ private TableId table1;
+
+ private BinaryRecordDataGenerator generator;
+
+ public static final String TEST_DATABASE = "test";
+ private static final String HADOOP_CONF_DIR =
+ Objects.requireNonNull(
+ Thread.currentThread()
+ .getContextClassLoader()
+ .getResource("hadoop-conf-dir"))
+ .getPath();
+
+ private static final String HIVE_CONF_DIR =
+ Objects.requireNonNull(
+ Thread.currentThread()
+ .getContextClassLoader()
+ .getResource("hive-conf-dir"))
+ .getPath();
+
+ private void initialize(String metastore)
+ throws Catalog.DatabaseNotEmptyException, Catalog.DatabaseNotExistException {
+ tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ warehouse = new File(temporaryFolder.toFile(), UUID.randomUUID().toString()).toString();
+ catalogOptions = new Options();
+ catalogOptions.setString("metastore", metastore);
+ catalogOptions.setString("warehouse", warehouse);
+ table1 = TableId.tableId("test", "table1");
+ if ("hive".equals(metastore)) {
+ catalogOptions.setString("hadoop-conf-dir", HADOOP_CONF_DIR);
+ catalogOptions.setString("hive-conf-dir", HIVE_CONF_DIR);
+ tEnv.executeSql(
+ String.format(
+ "CREATE CATALOG paimon_catalog WITH ("
+ + "'type'='paimon', "
+ + "'warehouse'='%s', "
+ + "'metastore'='hive', "
+ + "'hadoop-conf-dir'='%s', "
+ + "'hive-conf-dir'='%s' "
+ + ")",
+ warehouse, HADOOP_CONF_DIR, HIVE_CONF_DIR));
+ } else {
+ tEnv.executeSql(
+ String.format(
+ "CREATE CATALOG paimon_catalog WITH ('type'='paimon', 'warehouse'='%s')",
+ warehouse));
+ }
+ FlinkCatalogFactory.createPaimonCatalog(catalogOptions)
+ .dropDatabase(TEST_DATABASE, true, true);
+ }
+
+ private List<Event> createTestEvents() {
+ List<Event> testEvents = new ArrayList<>();
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(table1, schema);
+ testEvents.add(createTableEvent);
+ PaimonMetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
+ metadataApplier.applySchemaChange(createTableEvent);
+
+ generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ // insert
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ testEvents.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ testEvents.add(insertEvent2);
+ return testEvents;
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testSinkWithDataChange(String metastore)
+ throws IOException, InterruptedException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ PaimonSink<Event> paimonSink =
+ new PaimonSink<>(
+ catalogOptions, new PaimonRecordEventSerializer(ZoneId.systemDefault()));
+ PaimonWriter<Event> writer = paimonSink.createWriter(new MockInitContext());
+ Committer<MultiTableCommittable> committer = paimonSink.createCommitter();
+
+ // insert
+ for (Event event : createTestEvents()) {
+ writer.write(event, null);
+ }
+ writer.flush(false);
+ Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ List<Row> result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, "1", "1"), Row.ofKind(RowKind.INSERT, "2", "2")),
+ result);
+
+ // delete
+ Event event =
+ DataChangeEvent.deleteEvent(
+ TableId.tableId("test", "table1"),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ writer.write(event, null);
+ writer.flush(false);
+ commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Collections.singletonList(Row.ofKind(RowKind.INSERT, "2", "2")), result);
+
+ // update
+ event =
+ DataChangeEvent.updateEvent(
+ TableId.tableId("test", "table1"),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("x")
+ }));
+ writer.write(event, null);
+ writer.flush(false);
+ commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Collections.singletonList(Row.ofKind(RowKind.INSERT, "2", "x")), result);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testSinkWithSchemaChange(String metastore)
+ throws IOException, InterruptedException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ PaimonSink<Event> paimonSink =
+ new PaimonSink(
+ catalogOptions, new PaimonRecordEventSerializer(ZoneId.systemDefault()));
+ PaimonWriter<Event> writer = paimonSink.createWriter(new MockInitContext());
+ Committer<MultiTableCommittable> committer = paimonSink.createCommitter();
+
+ // 1. receive only DataChangeEvents during one checkpoint
+ for (Event event : createTestEvents()) {
+ writer.write(event, null);
+ }
+ writer.flush(false);
+ Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ List<Row> result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, "1", "1"), Row.ofKind(RowKind.INSERT, "2", "2")),
+ result);
+
+ // 2. receive DataChangeEvents and SchemaChangeEvents during one checkpoint
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ writer.write(insertEvent3, null);
+ writer.flush(false);
+
+ // add column
+ AddColumnEvent.ColumnWithPosition columnWithPosition =
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("col3", DataTypes.STRING()));
+ AddColumnEvent addColumnEvent =
+ new AddColumnEvent(table1, Collections.singletonList(columnWithPosition));
+ PaimonMetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
+ writer.write(addColumnEvent, null);
+ metadataApplier.applySchemaChange(addColumnEvent);
+ generator =
+ new BinaryRecordDataGenerator(
+ RowType.of(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent4 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("4"),
+ BinaryStringData.fromString("4"),
+ BinaryStringData.fromString("4")
+ }));
+ writer.write(insertEvent4, null);
+ writer.flush(false);
+ commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, "1", "1", null),
+ Row.ofKind(RowKind.INSERT, "2", "2", null),
+ Row.ofKind(RowKind.INSERT, "3", "3", null),
+ Row.ofKind(RowKind.INSERT, "4", "4", "4")),
+ result);
+
+ // 2. receive DataChangeEvents and SchemaChangeEvents during one checkpoint
+ DataChangeEvent insertEvent5 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("5"),
+ BinaryStringData.fromString("5"),
+ BinaryStringData.fromString("5")
+ }));
+ writer.write(insertEvent5, null);
+ writer.flush(false);
+ // drop column
+ DropColumnEvent dropColumnEvent =
+ new DropColumnEvent(table1, Collections.singletonList("col2"));
+ metadataApplier.applySchemaChange(dropColumnEvent);
+ writer.write(dropColumnEvent, null);
+ generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent6 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("6"),
+ BinaryStringData.fromString("6")
+ }));
+ writer.write(insertEvent6, null);
+ writer.flush(false);
+ commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, "1", null),
+ Row.ofKind(RowKind.INSERT, "2", null),
+ Row.ofKind(RowKind.INSERT, "3", null),
+ Row.ofKind(RowKind.INSERT, "4", "4"),
+ Row.ofKind(RowKind.INSERT, "5", "5"),
+ Row.ofKind(RowKind.INSERT, "6", "6")),
+ result);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"filesystem", "hive"})
+ public void testSinkWithMultiTables(String metastore)
+ throws IOException, InterruptedException, Catalog.DatabaseNotEmptyException,
+ Catalog.DatabaseNotExistException {
+ initialize(metastore);
+ PaimonSink<Event> paimonSink =
+ new PaimonSink<>(
+ catalogOptions, new PaimonRecordEventSerializer(ZoneId.systemDefault()));
+ PaimonWriter<Event> writer = paimonSink.createWriter(new MockInitContext());
+ Committer<MultiTableCommittable> committer = paimonSink.createCommitter();
+ List<Event> testEvents = createTestEvents();
+ // create table
+ TableId table2 = TableId.tableId("test", "table2");
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(table2, schema);
+ testEvents.add(createTableEvent);
+ PaimonMetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
+ metadataApplier.applySchemaChange(createTableEvent);
+ // insert
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ table2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ testEvents.add(insertEvent1);
+
+ // insert
+ for (Event event : testEvents) {
+ writer.write(event, null);
+ }
+ writer.flush(false);
+ Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList());
+ committer.commit(commitRequests);
+ List<Row> result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table1")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, "1", "1"), Row.ofKind(RowKind.INSERT, "2", "2")),
+ result);
+ result = new ArrayList<>();
+ tEnv.sqlQuery("select * from paimon_catalog.test.table2")
+ .execute()
+ .collect()
+ .forEachRemaining(result::add);
+ Assertions.assertEquals(
+ Collections.singletonList(Row.ofKind(RowKind.INSERT, "1", "1")), result);
+ }
+
+ private static class MockCommitRequestImpl<CommT> extends CommitRequestImpl<CommT> {
+
+ protected MockCommitRequestImpl(CommT committable) {
+ super(committable);
+ }
+ }
+
+ private static class MockInitContext
+ implements Sink.InitContext, SerializationSchema.InitializationContext {
+
+ private MockInitContext() {}
+
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return null;
+ }
+
+ public MailboxExecutor getMailboxExecutor() {
+ return null;
+ }
+
+ public ProcessingTimeService getProcessingTimeService() {
+ return null;
+ }
+
+ public int getSubtaskId() {
+ return 0;
+ }
+
+ public int getNumberOfParallelSubtasks() {
+ return 0;
+ }
+
+ public int getAttemptNumber() {
+ return 0;
+ }
+
+ public SinkWriterMetricGroup metricGroup() {
+ return null;
+ }
+
+ public MetricGroup getMetricGroup() {
+ return null;
+ }
+
+ public OptionalLong getRestoredCheckpointId() {
+ return OptionalLong.empty();
+ }
+
+ public SerializationSchema.InitializationContext
+ asSerializationSchemaInitializationContext() {
+ return this;
+ }
+
+ public boolean isObjectReuseEnabled() {
+ return false;
+ }
+
+ public <IN> TypeSerializer<IN> createInputSerializer() {
+ return null;
+ }
+
+ public JobID getJobId() {
+ return null;
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelperTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelperTest.java
new file mode 100644
index 000000000..d432e3fe2
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelperTest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.RowKind;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.List;
+
+/** Tests for {@link PaimonWriterHelper}. */
+public class PaimonWriterHelperTest {
+
+ @Test
+ public void testConvertEventToGenericRowOfAllDataTypes() {
+ RowType rowType =
+ RowType.of(
+ DataTypes.BOOLEAN(),
+ DataTypes.BINARY(3),
+ DataTypes.VARBINARY(10),
+ DataTypes.BYTES(),
+ DataTypes.TINYINT(),
+ DataTypes.SMALLINT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.FLOAT(),
+ DataTypes.DOUBLE(),
+ DataTypes.DECIMAL(6, 3),
+ DataTypes.CHAR(5),
+ DataTypes.VARCHAR(10),
+ DataTypes.STRING(),
+ DataTypes.DATE(),
+ DataTypes.TIME(),
+ DataTypes.TIME(6),
+ DataTypes.TIMESTAMP(),
+ DataTypes.TIMESTAMP(3),
+ DataTypes.TIMESTAMP_LTZ(),
+ DataTypes.TIMESTAMP_LTZ(3),
+ DataTypes.STRING());
+ Object[] testData =
+ new Object[] {
+ true,
+ new byte[] {1, 2},
+ new byte[] {3, 4},
+ new byte[] {5, 6, 7},
+ (byte) 1,
+ (short) 2,
+ 3,
+ 4L,
+ 5.1f,
+ 6.2,
+ DecimalData.fromBigDecimal(new BigDecimal("7.123"), 6, 3),
+ BinaryStringData.fromString("test1"),
+ BinaryStringData.fromString("test2"),
+ BinaryStringData.fromString("test3"),
+ 100,
+ 200,
+ 300,
+ TimestampData.fromTimestamp(
+ java.sql.Timestamp.valueOf("2023-01-01 00:00:00.000")),
+ TimestampData.fromTimestamp(java.sql.Timestamp.valueOf("2023-01-01 00:00:00")),
+ LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
+ LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
+ null
+ };
+ BinaryRecordData recordData = new BinaryRecordDataGenerator(rowType).generate(testData);
+ Schema schema = Schema.newBuilder().fromRowDataType(rowType).build();
+ List<RecordData.FieldGetter> fieldGetters =
+ PaimonWriterHelper.createFieldGetters(schema, ZoneId.of("UTC+8"));
+ DataChangeEvent dataChangeEvent =
+ DataChangeEvent.insertEvent(TableId.parse("database.table"), recordData);
+ GenericRow genericRow =
+ PaimonWriterHelper.convertEventToGenericRow(dataChangeEvent, fieldGetters);
+ Assertions.assertEquals(
+ GenericRow.ofKind(
+ RowKind.INSERT,
+ true,
+ new byte[] {1, 2},
+ new byte[] {3, 4},
+ new byte[] {5, 6, 7},
+ (byte) 1,
+ (short) 2,
+ 3,
+ 4L,
+ 5.1f,
+ 6.2,
+ Decimal.fromBigDecimal(new BigDecimal("7.123"), 6, 3),
+ BinaryString.fromString("test1"),
+ BinaryString.fromString("test2"),
+ BinaryString.fromString("test3"),
+ 100,
+ 200,
+ 300,
+ Timestamp.fromSQLTimestamp(
+ java.sql.Timestamp.valueOf("2023-01-01 00:00:00.000")),
+ Timestamp.fromSQLTimestamp(
+ java.sql.Timestamp.valueOf("2023-01-01 00:00:00")),
+ // plus 8 hours.
+ Timestamp.fromInstant(Instant.parse("2023-01-01T08:00:00.000Z")),
+ Timestamp.fromInstant(Instant.parse("2023-01-01T08:00:00.000Z")),
+ null),
+ genericRow);
+ }
+
+ @Test
+ public void testConvertEventToGenericRowOfDataChangeTypes() {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .build();
+ List<RecordData.FieldGetter> fieldGetters =
+ PaimonWriterHelper.createFieldGetters(schema, ZoneId.systemDefault());
+ TableId tableId = TableId.parse("database.table");
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ BinaryRecordData recordData =
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"), BinaryStringData.fromString("1")
+ });
+
+ DataChangeEvent dataChangeEvent = DataChangeEvent.insertEvent(tableId, recordData);
+ GenericRow genericRow =
+ PaimonWriterHelper.convertEventToGenericRow(dataChangeEvent, fieldGetters);
+ Assertions.assertEquals(genericRow.getRowKind(), RowKind.INSERT);
+
+ dataChangeEvent = DataChangeEvent.deleteEvent(tableId, recordData);
+ genericRow = PaimonWriterHelper.convertEventToGenericRow(dataChangeEvent, fieldGetters);
+ Assertions.assertEquals(genericRow.getRowKind(), RowKind.DELETE);
+
+ dataChangeEvent = DataChangeEvent.updateEvent(tableId, recordData, recordData);
+ genericRow = PaimonWriterHelper.convertEventToGenericRow(dataChangeEvent, fieldGetters);
+ Assertions.assertEquals(genericRow.getRowKind(), RowKind.INSERT);
+
+ dataChangeEvent = DataChangeEvent.replaceEvent(tableId, recordData, null);
+ genericRow = PaimonWriterHelper.convertEventToGenericRow(dataChangeEvent, fieldGetters);
+ Assertions.assertEquals(genericRow.getRowKind(), RowKind.INSERT);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TestMultiTableCommittableChannelComputer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TestMultiTableCommittableChannelComputer.java
new file mode 100644
index 000000000..2c0a9b4cb
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TestMultiTableCommittableChannelComputer.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.paimon.sink.v2;
+
+import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
+
+import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** A test for {@link MultiTableCommittableChannelComputer}. */
+public class TestMultiTableCommittableChannelComputer {
+
+ @Test
+ public void testChannel() {
+ MultiTableCommittableChannelComputer computer = new MultiTableCommittableChannelComputer();
+ computer.setup(4);
+ List<MultiTableCommittable> commits =
+ Arrays.asList(
+ new MultiTableCommittable("database", "table1", 1L, null, null),
+ new MultiTableCommittable("database", "table2", 1L, null, null),
+ new MultiTableCommittable("database", "table1", 1L, null, null),
+ new MultiTableCommittable("database", "table5", 1L, null, null),
+ new MultiTableCommittable("database", "table3", 1L, null, null),
+ new MultiTableCommittable("database", "table8", 1L, null, null),
+ new MultiTableCommittable("database", "table5", 1L, null, null),
+ new MultiTableCommittable("database", "table1", 1L, null, null),
+ new MultiTableCommittable("database", "table9", 1L, null, null),
+ new MultiTableCommittable("database", "table5", 1L, null, null),
+ new MultiTableCommittable("database", "table3", 1L, null, null),
+ new MultiTableCommittable("database", "table8", 1L, null, null));
+ Map<Integer, Set<String>> map = new HashMap<>();
+ commits.forEach(
+ (commit) -> {
+ int channel = computer.channel(new CommittableWithLineage<>(commit, 1L, 0));
+ Set<String> set = map.getOrDefault(channel, new HashSet<>());
+ set.add(commit.getTable());
+ map.put(channel, set);
+ });
+ Set<String> actualtables = new HashSet<>();
+ for (Map.Entry<Integer, Set<String>> entry : map.entrySet()) {
+ actualtables.addAll(entry.getValue());
+ }
+ Set<String> expectedTables =
+ new HashSet<>(
+ Arrays.asList("table1", "table2", "table3", "table5", "table8", "table9"));
+ // Not a table is appeared in more than one channel.
+ Assertions.assertEquals(actualtables, expectedTables);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hadoop-conf-dir/core-site.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hadoop-conf-dir/core-site.xml
new file mode 100644
index 000000000..38b1c4202
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hadoop-conf-dir/core-site.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<!--Autogenerated by Cloudera Manager-->
+<configuration>
+ <property>
+ <name>fs.defaultFS</name>
+ <value>file:///</value>
+ </property>
+ <property>
+ <name>fs.trash.interval</name>
+ <value>1</value>
+ </property>
+ <property>
+ <name>io.compression.codecs</name>
+ <value>org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.DeflateCodec,org.apache.hadoop.io.compress.SnappyCodec,org.apache.hadoop.io.compress.Lz4Codec</value>
+ </property>
+ <property>
+ <name>hadoop.security.authentication</name>
+ <value>none</value>
+ </property>
+</configuration>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hive-conf-dir/hive-site.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hive-conf-dir/hive-site.xml
new file mode 100644
index 000000000..845ff9f62
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/hive-conf-dir/hive-site.xml
@@ -0,0 +1,47 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+
+<configuration>
+
+ <property>
+ <name>hive.metastore.schema.verification</name>
+ <value>false</value>
+ </property>
+
+ <property>
+ <name>hive.metastore.client.capability.check</name>
+ <value>false</value>
+ </property>
+
+ <property>
+ <name>datanucleus.schema.autoCreateTables</name>
+ <value>true</value>
+ </property>
+
+ <property>
+ <name>datanucleus.schema.autoCreateAll</name>
+ <value>true</value>
+ </property>
+
+ <property>
+ <name>javax.jdo.option.ConnectionURL</name>
+ <value>jdbc:derby:memory:test;create=true</value>
+ </property>
+
+</configuration>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/log4j2-test.properties b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..8f6c07473
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/resources/log4j2-test.properties
@@ -0,0 +1,25 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+# http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level=OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c - %m%n
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
index c5265af9f..424372f4c 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
@@ -33,6 +33,7 @@ limitations under the License.
<module>flink-cdc-pipeline-connector-doris</module>
<module>flink-cdc-pipeline-connector-starrocks</module>
<module>flink-cdc-pipeline-connector-kafka</module>
+ <module>flink-cdc-pipeline-connector-paimon</module>
</modules>
<dependencies>