You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2025/12/02 02:43:54 UTC
(paimon) branch master updated: [flink] Support cdc source (#6606)
This is an automated email from the ASF dual-hosted git repository.
yunfengzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 64e90d605a [flink] Support cdc source (#6606)
64e90d605a is described below
commit 64e90d605a099699df6df7823fbe31f079fe9805
Author: yunfengzhou-hub <yu...@outlook.com>
AuthorDate: Tue Dec 2 10:43:49 2025 +0800
[flink] Support cdc source (#6606)
---
.../flink-cdc-paimon-pipeline-connector.md | 172 ++++++
.../shortcodes/generated/cdc_configuration.html | 48 ++
.../configuration/ConfigOptionsDocGenerator.java | 2 +
paimon-flink/paimon-flink-cdc/pom.xml | 54 +-
.../paimon/flink/pipeline/cdc/CDCOptions.java | 105 ++++
.../pipeline/cdc/schema/CDCMetadataAccessor.java | 133 +++++
.../flink/pipeline/cdc/source/CDCDataSource.java | 60 +++
.../pipeline/cdc/source/CDCDataSourceFactory.java | 119 +++++
.../flink/pipeline/cdc/source/CDCSource.java | 210 ++++++++
.../pipeline/cdc/source/CDCSourceProvider.java | 47 ++
.../cdc/source/TableAwareFileStoreSourceSplit.java | 159 ++++++
.../cdc/source/enumerator/CDCCheckpoint.java | 133 +++++
.../cdc/source/enumerator/CDCSourceEnumerator.java | 455 ++++++++++++++++
.../cdc/source/enumerator/CDCSplitAssigner.java | 114 ++++
.../cdc/source/reader/CDCRecordsWithSplitIds.java | 140 +++++
.../cdc/source/reader/CDCSourceReader.java | 99 ++++
.../cdc/source/reader/CDCSourceSplitReader.java} | 234 +++++----
.../paimon/flink/pipeline/cdc/util/CDCUtils.java | 47 ++
.../cdc/util/PaimonToFlinkCDCTypeConverter.java | 15 +
.../org.apache.flink.cdc.common.factories.Factory | 16 +
.../cdc/schema/CDCMetadataAccessorTest.java | 74 +++
.../cdc/source/CDCDataSourceFactoryTest.java | 77 +++
.../flink/pipeline/cdc/source/CDCSourceITCase.java | 149 ++++++
.../pipeline/cdc/source/CDCSourceITCaseBase.java | 172 ++++++
.../cdc/source/CDCSourceSavepointITCase.java | 217 ++++++++
...bleAwareFileStoreSourceSplitSerializerTest.java | 60 +++
.../enumerator/CDCCheckpointSerializerTest.java | 66 +++
.../source/enumerator/CDCSourceEnumeratorTest.java | 530 +++++++++++++++++++
.../source/reader/CDCSourceSplitReaderTest.java | 575 +++++++++++++++++++++
.../flink/source/FileStoreSourceSplitReader.java | 2 +-
.../flink/source/FileStoreSourceSplitState.java | 13 +-
.../source/ContinuousFileSplitEnumeratorTest.java | 159 +-----
.../flink/source/FileSplitEnumeratorTestBase.java | 204 +++++++-
.../StaticFileStoreSplitEnumeratorTestBase.java | 23 +-
.../flink/source/TestChangelogDataReadWrite.java | 6 +-
.../AlignedContinuousFileSplitEnumeratorTest.java | 23 +-
36 files changed, 4426 insertions(+), 286 deletions(-)
diff --git a/docs/content/cdc-ingestion/flink-cdc-paimon-pipeline-connector.md b/docs/content/cdc-ingestion/flink-cdc-paimon-pipeline-connector.md
new file mode 100644
index 0000000000..d7a2584a48
--- /dev/null
+++ b/docs/content/cdc-ingestion/flink-cdc-paimon-pipeline-connector.md
@@ -0,0 +1,172 @@
+---
+title: "Flink CDC Paimon Pipeline Connector"
+weight: 10
+type: docs
+aliases:
+ - /cdc-ingestion/flink-cdc-paimon-pipeline-connector.html
+---
+<!--
+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.
+-->
+
+# Flink CDC Paimon Pipeline Connector
+
+Flink CDC is a streaming data integration tool for the Flink engine. It allows users to describe their ETL pipeline
+logic via YAML elegantly and help users automatically generating customized Flink operators and submitting job.
+
+The Paimon Pipeline connector can be used as both the Data Source or the Data Sink of the Flink CDC pipeline. This
+document describes how to set up the Paimon Pipeline connector as the Data Source. If you are interested in using
+Paimon as the Data Sink, please refer to Flink CDC's
+[Paimon Pipeline Connector](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/docs/connectors/pipeline-connectors/paimon/)
+document.
+
+## What can the connector do?
+
+* Synchronizes data from a Paimon warehouse, database or table to an external system supported by Flink CDC
+* Synchronizes schema changes
+* Automatically discovers newly created tables in the source Paimon warehouse.
+
+How to create Pipeline
+----------------
+
+The pipeline for reading data from Paimon and sink to Doris can be defined as follows:
+
+```yaml
+source:
+ type: paimon
+ name: Paimon Source
+ database: default
+ table: test_table
+ catalog.properties.metastore: filesystem
+ catalog.properties.warehouse: /path/warehouse
+
+sink:
+ type: doris
+ name: Doris Sink
+ fenodes: 127.0.0.1:8030
+ username: root
+ password: pass
+
+pipeline:
+ name: Paimon to Doris Pipeline
+ parallelism: 2
+```
+
+Pipeline Connector Options
+----------------
+
+{{< generated/cdc_configuration >}}
+
+Catalog Options
+----------------
+
+Apart from the pipeline connector options described above, in the CDC yaml file you can also configure options that
+starts with `catalog.properties.`. For example, `catalog.properties.warehouse` or `catalog.properties.metastore`. Such
+options will have their prefix removed and the rest be regarded as catalog options. Please refer to the
+[Configurations]({{< ref "maintenance/configurations" >}}) section for catalog options available.
+
+Usage Notes
+--------
+
+* Data updates for primary key tables (-U, +U) will be replaced with -D and +I.
+* Does not support dropping tables. If you need to drop a table from the Paimon warehouse, please restart the Flink CDC job after performing the drop operation. When the job restarts, it will stop reading data from the dropped table, and the target table in the external system will remain unchanged from its state before the job was stopped.
+* Data from the same table will be consumed by the same Flink source subtask. If the amount of data varies significantly across different tables, performance bottlenecks caused by data skew may be observed in Flink CDC jobs.
+* If the CDC job has consumed up to the latest snapshot of a table and the next snapshot is not available yet, the monitoring and consumption of this table may be temporarily paused until `continuous.discovery-interval` has passed.
+
+Data Type Mapping
+----------------
+<div class="wy-table-responsive">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left">Paimon type</th>
+ <th class="text-left">CDC 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/docs/layouts/shortcodes/generated/cdc_configuration.html b/docs/layouts/shortcodes/generated/cdc_configuration.html
new file mode 100644
index 0000000000..8c625f9223
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/cdc_configuration.html
@@ -0,0 +1,48 @@
+{{/*
+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.
+*/}}
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>database</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Name of the database to be scanned. By default, all databases will be scanned.</td>
+ </tr>
+ <tr>
+ <td><h5>table</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Name of the table to be scanned. By default, all tables will be scanned.</td>
+ </tr>
+ <tr>
+ <td><h5>table.discovery-interval</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The discovery interval of new tables. Only effective when database or table is not set.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
index 02a18f375c..3710d14a37 100644
--- a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
@@ -83,6 +83,8 @@ public class ConfigOptionsDocGenerator {
"paimon-flink/paimon-flink-common", "org.apache.paimon.flink"),
new OptionsClassLocation(
"paimon-flink/paimon-flink-cdc", "org.apache.paimon.flink.kafka"),
+ new OptionsClassLocation(
+ "paimon-flink/paimon-flink-cdc", "org.apache.paimon.flink.pipeline.cdc"),
new OptionsClassLocation(
"paimon-hive/paimon-hive-catalog", "org.apache.paimon.hive"),
new OptionsClassLocation(
diff --git a/paimon-flink/paimon-flink-cdc/pom.xml b/paimon-flink/paimon-flink-cdc/pom.xml
index 9499b39740..5bb119b498 100644
--- a/paimon-flink/paimon-flink-cdc/pom.xml
+++ b/paimon-flink/paimon-flink-cdc/pom.xml
@@ -92,6 +92,31 @@ under the License.
<!-- CDC dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-common</artifactId>
+ <version>${flink.cdc.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-runtime</artifactId>
+ <version>${flink.cdc.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-postgres-cdc</artifactId>
@@ -220,13 +245,6 @@ under the License.
<type>test-jar</type>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-files</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
@@ -307,8 +325,30 @@ under the License.
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${flink.cdc.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-values</artifactId>
+ <version>${flink.cdc.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
+<!-- <dependencyManagement>-->
+<!-- <dependencies>-->
+<!-- <dependency>-->
+<!-- <groupId>com.google.guava</groupId>-->
+<!-- <artifactId>guava</artifactId>-->
+<!-- <version>32.1.2-jre</version>-->
+<!-- </dependency>-->
+<!-- </dependencies>-->
+<!-- </dependencyManagement>-->
+
<build>
<plugins>
<plugin>
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/CDCOptions.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/CDCOptions.java
new file mode 100644
index 0000000000..74e23e8177
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/CDCOptions.java
@@ -0,0 +1,105 @@
+/*
+ * 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.paimon.flink.pipeline.cdc;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
+import org.apache.paimon.options.FallbackKey;
+
+import org.apache.flink.util.IterableUtils;
+
+import java.time.Duration;
+
+/** Options for the cdc source. */
+public class CDCOptions {
+ /** prefix for passing properties for catalog creation. */
+ public static final String PREFIX_CATALOG_PROPERTIES = "catalog.properties.";
+
+ public static final ConfigOption<String> DATABASE =
+ ConfigOptions.key("database")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Name of the database to be scanned. By default, all databases will be scanned.");
+
+ public static final ConfigOption<String> TABLE =
+ ConfigOptions.key("table")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Name of the table to be scanned. By default, all tables will be scanned.");
+
+ public static final ConfigOption<Duration> TABLE_DISCOVERY_INTERVAL =
+ ConfigOptions.key("table.discovery-interval")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(1))
+ .withDescription(
+ "The discovery interval of new tables. Only effective when database or table is not set.");
+
+ public static <T> org.apache.flink.cdc.common.configuration.ConfigOption<T> toCDCOption(
+ org.apache.paimon.options.ConfigOption<T> option) {
+ org.apache.flink.cdc.common.configuration.ConfigOptions.OptionBuilder builder =
+ org.apache.flink.cdc.common.configuration.ConfigOptions.key(option.key());
+ org.apache.flink.cdc.common.configuration.ConfigOption<T> cdcOption =
+ configTypeAndDefaultValue(builder, option.defaultValue());
+
+ return cdcOption
+ .withDescription(option.description().toString())
+ .withFallbackKeys(
+ IterableUtils.toStream(option.fallbackKeys())
+ .map(FallbackKey::getKey)
+ .toArray(String[]::new))
+ .withDeprecatedKeys(
+ IterableUtils.toStream(option.deprecatedKeys()).toArray(String[]::new));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T>
+ org.apache.flink.cdc.common.configuration.ConfigOption<T> configTypeAndDefaultValue(
+ org.apache.flink.cdc.common.configuration.ConfigOptions.OptionBuilder builder,
+ T defaultValue) {
+ if (defaultValue instanceof String) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.stringType().defaultValue((String) defaultValue);
+ } else if (defaultValue instanceof Integer) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.intType().defaultValue((Integer) defaultValue);
+ } else if (defaultValue instanceof Long) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.longType().defaultValue((Long) defaultValue);
+ } else if (defaultValue instanceof Boolean) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.booleanType().defaultValue((Boolean) defaultValue);
+ } else if (defaultValue instanceof Double) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.doubleType().defaultValue((Double) defaultValue);
+ } else if (defaultValue instanceof Float) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.floatType().defaultValue((Float) defaultValue);
+ } else if (defaultValue instanceof Duration) {
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.durationType().defaultValue((Duration) defaultValue);
+ } else {
+ // Default to string type for other types
+ return (org.apache.flink.cdc.common.configuration.ConfigOption<T>)
+ builder.stringType()
+ .defaultValue(defaultValue != null ? defaultValue.toString() : null);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessor.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessor.java
new file mode 100644
index 0000000000..05ffa91679
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessor.java
@@ -0,0 +1,133 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.schema;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.flink.pipeline.cdc.util.PaimonToFlinkCDCTypeConverter.convertPaimonSchemaToFlinkCDCSchema;
+
+/** The {@link MetadataAccessor} for cdc source. */
+public class CDCMetadataAccessor implements MetadataAccessor {
+ private static final Logger LOG = LoggerFactory.getLogger(CDCMetadataAccessor.class);
+
+ private final Catalog catalog;
+
+ public CDCMetadataAccessor(Catalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public List<String> listNamespaces() {
+ throw new UnsupportedOperationException("Paimon does not support namespaces");
+ }
+
+ @Override
+ public List<String> listSchemas(@Nullable String namespace) {
+ if (namespace != null) {
+ throw new UnsupportedOperationException("Paimon does not support namespaces");
+ }
+
+ return catalog.listDatabases();
+ }
+
+ @Override
+ public List<TableId> listTables(@Nullable String namespace, @Nullable String schemaName) {
+ if (namespace != null) {
+ throw new UnsupportedOperationException("Paimon does not support namespaces");
+ }
+
+ List<String> databaseNames;
+ if (schemaName == null) {
+ databaseNames = catalog.listDatabases();
+ } else {
+ databaseNames = Collections.singletonList(schemaName);
+ }
+
+ List<TableId> tableIds = new ArrayList<>();
+ for (String databaseName : databaseNames) {
+ try {
+ for (String tableName : catalog.listTables(databaseName)) {
+ Identifier identifier = Identifier.create(databaseName, tableName);
+ try {
+ Table table = catalog.getTable(identifier);
+ if (!(table instanceof FileStoreTable)) {
+ LOG.info(
+ "listTables found table {}, but it is a {} instead of a FileStoreTable. Skipping this table.",
+ identifier,
+ table.getClass().getSimpleName());
+ continue;
+ }
+ } catch (Catalog.TableNotExistException e) {
+ LOG.warn(
+ "Table {} does not exist. Perhaps it is dropped in the middle of this method. Skipping this table.",
+ identifier,
+ e);
+ continue;
+ }
+
+ tableIds.add(TableId.tableId(databaseName, tableName));
+ }
+ } catch (Catalog.DatabaseNotExistException e) {
+ LOG.warn(
+ "Database {} does not exist. Perhaps it is dropped in the middle of this method",
+ databaseName,
+ e);
+ }
+ }
+ return tableIds;
+ }
+
+ @Override
+ public Schema getTableSchema(TableId tableId) {
+ if (tableId.getNamespace() != null) {
+ throw new UnsupportedOperationException("Paimon does not support namespaces");
+ }
+
+ try {
+ Identifier identifier =
+ Identifier.create(tableId.getSchemaName(), tableId.getTableName());
+ Table table = catalog.getTable(identifier);
+ if (!(table instanceof FileStoreTable)) {
+ throw new RuntimeException(
+ String.format(
+ "Table %s is not a file store table, but a %s, which is not supported by CDC",
+ identifier, table.getClass().getName()));
+ }
+
+ return convertPaimonSchemaToFlinkCDCSchema(((FileStoreTable) table).schema());
+ } catch (Catalog.TableNotExistException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSource.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSource.java
new file mode 100644
index 0000000000..6cb4b6e7b9
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSource.java
@@ -0,0 +1,60 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.flink.pipeline.cdc.schema.CDCMetadataAccessor;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.common.source.EventSourceProvider;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+
+import static org.apache.paimon.flink.pipeline.cdc.util.CDCUtils.createCatalog;
+
+/** The {@link DataSource} for cdc source. */
+public class CDCDataSource implements DataSource {
+ private final CatalogContext catalogContext;
+ private final Configuration cdcConfig;
+ private final org.apache.flink.configuration.Configuration flinkConfig;
+ private Catalog catalog;
+
+ public CDCDataSource(
+ CatalogContext catalogContext,
+ Configuration cdcConfig,
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ this.catalogContext = catalogContext;
+ this.cdcConfig = cdcConfig;
+ this.flinkConfig = flinkConfig;
+ }
+
+ @Override
+ public EventSourceProvider getEventSourceProvider() {
+ return new CDCSourceProvider(catalogContext, cdcConfig, flinkConfig);
+ }
+
+ @Override
+ public MetadataAccessor getMetadataAccessor() {
+ if (catalog == null) {
+ catalog = createCatalog(catalogContext, flinkConfig);
+ }
+ return new CDCMetadataAccessor(catalog);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactory.java
new file mode 100644
index 0000000000..f006d0b87f
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactory.java
@@ -0,0 +1,119 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.ReflectionUtils;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.configuration.ReadableConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.DATABASE;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.PREFIX_CATALOG_PROPERTIES;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.TABLE;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.TABLE_DISCOVERY_INTERVAL;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.toCDCOption;
+
+/** The {@link DataSourceFactory} for cdc source. */
+public class CDCDataSourceFactory implements DataSourceFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(CDCDataSourceFactory.class);
+
+ public static final String IDENTIFIER = "paimon";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public DataSource createDataSource(Context context) {
+ FactoryHelper.createFactoryHelper(this, context).validateExcept(PREFIX_CATALOG_PROPERTIES);
+
+ Map<String, String> catalogOptions = new HashMap<>();
+ Map<String, String> cdcConfig = new HashMap<>();
+ context.getFactoryConfiguration()
+ .toMap()
+ .forEach(
+ (key, value) -> {
+ if (key.startsWith(PREFIX_CATALOG_PROPERTIES)) {
+ catalogOptions.put(
+ key.substring(PREFIX_CATALOG_PROPERTIES.length()), value);
+ } else {
+ cdcConfig.put(key, value);
+ }
+ });
+
+ org.apache.flink.configuration.Configuration flinkConfig;
+ boolean isReadableConfigAcquired = false;
+ try {
+ Method getFlinkConfMethod =
+ ReflectionUtils.getMethod(context.getClass(), "getFlinkConf", 0);
+ ReadableConfig readableConfig = (ReadableConfig) getFlinkConfMethod.invoke(context);
+ isReadableConfigAcquired = true;
+ if (readableConfig instanceof org.apache.flink.configuration.Configuration) {
+ flinkConfig = (org.apache.flink.configuration.Configuration) readableConfig;
+ } else {
+ Method toMapMethod =
+ ReflectionUtils.getMethod(readableConfig.getClass(), "toMap", 0);
+ Map<String, String> configMap =
+ (Map<String, String>) toMapMethod.invoke(readableConfig);
+ flinkConfig = org.apache.flink.configuration.Configuration.fromMap(configMap);
+ }
+ } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
+ String failReason = isReadableConfigAcquired ? "Flink CDC 3.5-" : "Flink 1.18-";
+ LOG.info(
+ "Cannot get Flink configuration from context. It is possibly due to compatibility with {}. Using empty Flink configuration to create catalog instead.",
+ failReason);
+ flinkConfig = new org.apache.flink.configuration.Configuration();
+ }
+
+ return new CDCDataSource(
+ CatalogContext.create(Options.fromMap(catalogOptions)),
+ Configuration.fromMap(cdcConfig),
+ flinkConfig);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> set = new HashSet<>();
+ set.add(toCDCOption(DATABASE));
+ set.add(toCDCOption(TABLE));
+ set.add(toCDCOption(TABLE_DISCOVERY_INTERVAL));
+ return set;
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSource.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSource.java
new file mode 100644
index 0000000000..0b85383133
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSource.java
@@ -0,0 +1,210 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
+import org.apache.paimon.flink.pipeline.cdc.source.enumerator.CDCCheckpoint;
+import org.apache.paimon.flink.pipeline.cdc.source.enumerator.CDCSourceEnumerator;
+import org.apache.paimon.flink.pipeline.cdc.source.reader.CDCSourceReader;
+import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.TableRead;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cdc.common.configuration.Configuration;
+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.configuration.CoreOptions;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.cdc.common.utils.SchemaMergingUtils.getSchemaDifference;
+import static org.apache.paimon.disk.IOManagerImpl.splitPaths;
+import static org.apache.paimon.flink.pipeline.cdc.util.CDCUtils.createCatalog;
+import static org.apache.paimon.flink.pipeline.cdc.util.PaimonToFlinkCDCTypeConverter.convertPaimonSchemaToFlinkCDCSchema;
+
+/** The {@link Source} that integrates with Flink CDC framework. */
+public class CDCSource implements Source<Event, TableAwareFileStoreSourceSplit, CDCCheckpoint> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final CatalogContext catalogContext;
+ private final Configuration cdcConfig;
+ private final org.apache.flink.configuration.Configuration flinkConfig;
+
+ public CDCSource(
+ CatalogContext catalogContext,
+ Configuration cdcConfig,
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ this.catalogContext = catalogContext;
+ this.cdcConfig = cdcConfig;
+ this.flinkConfig = flinkConfig;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SplitEnumerator<TableAwareFileStoreSourceSplit, CDCCheckpoint> createEnumerator(
+ SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context) {
+ return restoreEnumerator(context, null);
+ }
+
+ @Override
+ public SplitEnumerator<TableAwareFileStoreSourceSplit, CDCCheckpoint> restoreEnumerator(
+ SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context,
+ @Nullable CDCCheckpoint checkpoint) {
+ return new CDCSourceEnumerator(
+ context,
+ flinkConfig,
+ catalogContext
+ .options()
+ .get(org.apache.paimon.CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL)
+ .toMillis(),
+ catalogContext,
+ cdcConfig,
+ checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<TableAwareFileStoreSourceSplit> getSplitSerializer() {
+ return new TableAwareFileStoreSourceSplit.Serializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<CDCCheckpoint> getEnumeratorCheckpointSerializer() {
+ return new CDCCheckpoint.Serializer();
+ }
+
+ @Override
+ public SourceReader<Event, TableAwareFileStoreSourceSplit> createReader(
+ SourceReaderContext context) {
+ IOManager ioManager =
+ IOManager.create(splitPaths(context.getConfiguration().get(CoreOptions.TMP_DIRS)));
+ SourceReaderMetricGroup metricGroup = context.metricGroup();
+ FileStoreSourceReaderMetrics sourceReaderMetrics =
+ new FileStoreSourceReaderMetrics(metricGroup);
+
+ Catalog catalog = createCatalog(catalogContext, flinkConfig);
+ TableManager manager = new TableManager(catalog, ioManager, metricGroup);
+ return new CDCSourceReader(context, sourceReaderMetrics, ioManager, manager);
+ }
+
+ /** A manager for information related to the tables. */
+ public static class TableManager {
+ private final Map<Identifier, FileStoreTable> tableMap = new HashMap<>();
+ private final Map<Tuple2<Identifier, Long>, TableSchema> tableSchemaMap = new HashMap<>();
+ private final Map<Tuple2<Identifier, Long>, TableRead> tableReadMap = new HashMap<>();
+ private final Catalog catalog;
+ private final IOManager ioManager;
+ private final SourceReaderMetricGroup metricGroup;
+
+ protected TableManager(
+ Catalog catalog, IOManager ioManager, SourceReaderMetricGroup metricGroup) {
+ this.catalog = catalog;
+ this.ioManager = ioManager;
+ this.metricGroup = metricGroup;
+ }
+
+ public @Nullable TableSchema getTableSchema(
+ Identifier identifier, @Nullable Long schemaId) {
+ if (schemaId == null) {
+ return null;
+ }
+
+ Tuple2<Identifier, Long> cacheKey = Tuple2.of(identifier, schemaId);
+ if (tableSchemaMap.containsKey(cacheKey)) {
+ return tableSchemaMap.get(cacheKey);
+ }
+
+ FileStoreTable table = getTable(identifier);
+ TableSchema tableSchema = table.schemaManager().schema(schemaId);
+ tableSchemaMap.put(cacheKey, tableSchema);
+ return tableSchema;
+ }
+
+ public TableRead getTableRead(Identifier identifier, TableSchema schema) {
+ Tuple2<Identifier, Long> cacheKey = Tuple2.of(identifier, schema.id());
+ if (tableReadMap.containsKey(cacheKey)) {
+ return tableReadMap.get(cacheKey);
+ }
+
+ FileStoreTable table = getTable(identifier).copy(schema);
+ TableRead tableRead =
+ table.newReadBuilder()
+ .newRead()
+ .withIOManager(ioManager)
+ .withMetricRegistry(new FlinkMetricRegistry(metricGroup));
+ tableReadMap.put(cacheKey, tableRead);
+ return tableRead;
+ }
+
+ public List<SchemaChangeEvent> generateSchemaChangeEventList(
+ Identifier identifier, @Nullable Long lastSchemaId, long schemaId) {
+ if (lastSchemaId != null && lastSchemaId.equals(schemaId)) {
+ return Collections.emptyList();
+ }
+
+ return getSchemaDifference(
+ TableId.tableId(identifier.getDatabaseName(), identifier.getTableName()),
+ convertPaimonSchemaToFlinkCDCSchema(getTableSchema(identifier, lastSchemaId)),
+ convertPaimonSchemaToFlinkCDCSchema(getTableSchema(identifier, schemaId)));
+ }
+
+ private FileStoreTable getTable(Identifier identifier) {
+ if (tableMap.containsKey(identifier)) {
+ return tableMap.get(identifier);
+ }
+
+ Table table;
+ try {
+ table = catalog.getTable(identifier);
+ } catch (Catalog.TableNotExistException e) {
+ throw new RuntimeException(e);
+ }
+
+ // As the table should have just been checked by the enumerator, it is supposed to exist
+ // and be a FileStoreTable.
+ tableMap.put(identifier, (FileStoreTable) table);
+ return (FileStoreTable) table;
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceProvider.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceProvider.java
new file mode 100644
index 0000000000..7ef5a2f81b
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceProvider.java
@@ -0,0 +1,47 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.CatalogContext;
+
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+
+/** The {@link FlinkSourceProvider} for paimon cdc source. */
+public class CDCSourceProvider implements FlinkSourceProvider {
+ private final CatalogContext catalogContext;
+ private final Configuration cdcConfig;
+ private final org.apache.flink.configuration.Configuration flinkConfig;
+
+ public CDCSourceProvider(
+ CatalogContext catalogContext,
+ Configuration cdcConfig,
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ this.catalogContext = catalogContext;
+ this.cdcConfig = cdcConfig;
+ this.flinkConfig = flinkConfig;
+ }
+
+ @Override
+ public Source<Event, ?, ?> getSource() {
+ return new CDCSource(catalogContext, cdcConfig, flinkConfig);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplit.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplit.java
new file mode 100644
index 0000000000..ed84e14133
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplit.java
@@ -0,0 +1,159 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.source.FileStoreSourceSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.InstantiationUtil;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * A {@link FileStoreSourceSplit} that contains information about the table that corresponds to the
+ * {@link Split}.
+ */
+public class TableAwareFileStoreSourceSplit extends FileStoreSourceSplit {
+ private final Identifier identifier;
+ private final @Nullable Long lastSchemaId;
+ private final long schemaId;
+
+ public TableAwareFileStoreSourceSplit(
+ String id,
+ Split split,
+ long recordsToSkip,
+ Identifier identifier,
+ @Nullable Long lastSchemaId,
+ long schemaId) {
+ super(id, split, recordsToSkip);
+ this.identifier = identifier;
+ this.lastSchemaId = lastSchemaId;
+ this.schemaId = schemaId;
+ }
+
+ public Identifier getIdentifier() {
+ return identifier;
+ }
+
+ public @Nullable Long getLastSchemaId() {
+ return lastSchemaId;
+ }
+
+ public long getSchemaId() {
+ return schemaId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof TableAwareFileStoreSourceSplit)) {
+ return false;
+ }
+
+ TableAwareFileStoreSourceSplit other = (TableAwareFileStoreSourceSplit) o;
+ return Objects.equals(splitId(), other.splitId())
+ && Objects.equals(split(), other.split())
+ && recordsToSkip() == other.recordsToSkip()
+ && identifier.equals(other.identifier)
+ && Objects.equals(lastSchemaId, other.lastSchemaId)
+ && schemaId == other.schemaId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ splitId(), split(), recordsToSkip(), identifier, lastSchemaId, schemaId);
+ }
+
+ @Override
+ public String toString() {
+ return "TableAwareFileStoreSourceSplit{"
+ + "splitId='"
+ + splitId()
+ + "'"
+ + ", split="
+ + split()
+ + ", recordsToSkip="
+ + recordsToSkip()
+ + ", identifier="
+ + identifier
+ + ", lastSchemaId="
+ + lastSchemaId
+ + ", schemaId="
+ + schemaId
+ + '}';
+ }
+
+ /** The serializer for {@link TableAwareFileStoreSourceSplit}. */
+ public static class Serializer
+ implements SimpleVersionedSerializer<TableAwareFileStoreSourceSplit> {
+ private static final Long NULL_SCHEMA_ID = -1L;
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public byte[] serialize(TableAwareFileStoreSourceSplit split) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+ view.writeUTF(split.splitId());
+ InstantiationUtil.serializeObject(view, split.split());
+ view.writeLong(split.recordsToSkip());
+ view.writeUTF(JsonSerdeUtil.toJson(split.getIdentifier()));
+ view.writeLong(
+ split.getLastSchemaId() == null ? NULL_SCHEMA_ID : split.getLastSchemaId());
+ view.writeLong(split.getSchemaId());
+ return out.toByteArray();
+ }
+
+ @Override
+ public TableAwareFileStoreSourceSplit deserialize(int version, byte[] serialized)
+ throws IOException {
+ ByteArrayInputStream in = new ByteArrayInputStream(serialized);
+ DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
+ String splitId = view.readUTF();
+ Split split;
+ try {
+ split = InstantiationUtil.deserializeObject(in, getClass().getClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new IOException(e);
+ }
+ long recordsToSkip = view.readLong();
+ Identifier identifier = JsonSerdeUtil.fromJson(view.readUTF(), Identifier.class);
+ Long lastSchemaId = view.readLong();
+ if (lastSchemaId.equals(NULL_SCHEMA_ID)) {
+ lastSchemaId = null;
+ }
+ long schemaId = view.readLong();
+ return new TableAwareFileStoreSourceSplit(
+ splitId, split, recordsToSkip, identifier, lastSchemaId, schemaId);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpoint.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpoint.java
new file mode 100644
index 0000000000..b0e41fd922
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpoint.java
@@ -0,0 +1,133 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.enumerator;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+
+import java.io.ByteArrayOutputStream;
+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.Objects;
+
+/**
+ * A checkpoint of the current state of the containing the currently pending splits that are not yet
+ * assigned, and the consumption progress of each table.
+ */
+public class CDCCheckpoint {
+ private final Collection<TableAwareFileStoreSourceSplit> splits;
+
+ private final Map<Identifier, Long> currentSnapshotIdMap;
+
+ public CDCCheckpoint(
+ Collection<TableAwareFileStoreSourceSplit> splits,
+ Map<Identifier, Long> currentSnapshotIdMap) {
+ this.splits = splits;
+ this.currentSnapshotIdMap = currentSnapshotIdMap;
+ }
+
+ public Collection<TableAwareFileStoreSourceSplit> getSplits() {
+ return splits;
+ }
+
+ public Map<Identifier, Long> getCurrentSnapshotIdMap() {
+ return currentSnapshotIdMap;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(splits, currentSnapshotIdMap);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof CDCCheckpoint)) {
+ return false;
+ }
+
+ CDCCheckpoint cdcCheckpoint = (CDCCheckpoint) obj;
+ return Objects.equals(cdcCheckpoint.splits, splits)
+ && Objects.equals(cdcCheckpoint.currentSnapshotIdMap, currentSnapshotIdMap);
+ }
+
+ /** {@link SimpleVersionedSerializer} for {@link CDCCheckpoint}. */
+ public static class Serializer implements SimpleVersionedSerializer<CDCCheckpoint> {
+ private final SimpleVersionedSerializer<TableAwareFileStoreSourceSplit> splitSerializer =
+ new TableAwareFileStoreSourceSplit.Serializer();
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public byte[] serialize(CDCCheckpoint checkpoint) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+
+ view.writeInt(checkpoint.splits.size());
+ for (TableAwareFileStoreSourceSplit split : checkpoint.splits) {
+ byte[] bytes = splitSerializer.serialize(split);
+ view.writeInt(bytes.length);
+ view.write(bytes);
+ }
+
+ view.writeInt(checkpoint.currentSnapshotIdMap.size());
+ for (Map.Entry<Identifier, Long> entry : checkpoint.currentSnapshotIdMap.entrySet()) {
+ view.writeUTF(JsonSerdeUtil.toJson(entry.getKey()));
+ view.writeLong(entry.getValue());
+ }
+
+ return out.toByteArray();
+ }
+
+ @Override
+ public CDCCheckpoint deserialize(int version, byte[] serialized) throws IOException {
+ DataInputDeserializer view = new DataInputDeserializer(serialized);
+
+ int splitNumber = view.readInt();
+ List<TableAwareFileStoreSourceSplit> splits = new ArrayList<>(splitNumber);
+ for (int i = 0; i < splitNumber; i++) {
+ int byteNumber = view.readInt();
+ byte[] bytes = new byte[byteNumber];
+ view.readFully(bytes);
+ splits.add(splitSerializer.deserialize(version, bytes));
+ }
+
+ int currentSnapshotIdMapSize = view.readInt();
+ Map<Identifier, Long> currentSnapshotIdMap = new HashMap<>(currentSnapshotIdMapSize);
+ for (int i = 0; i < currentSnapshotIdMapSize; i++) {
+ Identifier identifier = JsonSerdeUtil.fromJson(view.readUTF(), Identifier.class);
+ long currentSnapshotId = view.readLong();
+ currentSnapshotIdMap.put(identifier, currentSnapshotId);
+ }
+
+ return new CDCCheckpoint(splits, currentSnapshotIdMap);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumerator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumerator.java
new file mode 100644
index 0000000000..c8789bf30e
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumerator.java
@@ -0,0 +1,455 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.enumerator;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.table.source.SnapshotNotExistPlan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.connector.source.SplitsAssignment;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.DATABASE;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.TABLE;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.TABLE_DISCOVERY_INTERVAL;
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.toCDCOption;
+import static org.apache.paimon.flink.pipeline.cdc.util.CDCUtils.createCatalog;
+
+/** {@link SplitEnumerator} for CDC source. */
+public class CDCSourceEnumerator
+ implements SplitEnumerator<TableAwareFileStoreSourceSplit, CDCCheckpoint> {
+ private static final Logger LOG = LoggerFactory.getLogger(CDCSourceEnumerator.class);
+
+ private final SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context;
+ private final long discoveryInterval;
+ private final long tableDiscoveryInterval;
+ private long lastTableDiscoveryTime = 0L;
+ private final AtomicInteger currentTableIndex = new AtomicInteger(0);
+ private final String database;
+ @Nullable private final String table;
+ private final Catalog catalog;
+
+ private final int splitMaxNum;
+ private final CDCSplitAssigner splitAssigner;
+ private final Set<Integer> readersAwaitingSplit;
+ private final SplitIdGenerator splitIdGenerator;
+ private final Map<Identifier, TableStatus> tableStatusMap = new HashMap<>();
+ private final Set<Identifier> discoveredNonFileStoreTables = new HashSet<>();
+ private int numTablesWithSubtaskAssigned;
+
+ private boolean stopTriggerScan = false;
+
+ public CDCSourceEnumerator(
+ SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context,
+ org.apache.flink.configuration.Configuration flinkConfig,
+ long discoveryInterval,
+ CatalogContext catalogContext,
+ Configuration cdcConfig,
+ @Nullable CDCCheckpoint checkpoint) {
+ this.context = context;
+ this.discoveryInterval = discoveryInterval;
+ this.tableDiscoveryInterval =
+ cdcConfig.get(toCDCOption(TABLE_DISCOVERY_INTERVAL)).toMillis();
+
+ this.database = cdcConfig.get(toCDCOption(DATABASE));
+ this.table = cdcConfig.get(toCDCOption(TABLE));
+ this.catalog = createCatalog(catalogContext, flinkConfig);
+
+ this.numTablesWithSubtaskAssigned = 0;
+
+ int splitMaxPerTask = catalogContext.options().get(CoreOptions.SCAN_MAX_SPLITS_PER_TASK);
+ this.splitMaxNum = context.currentParallelism() * splitMaxPerTask;
+
+ this.splitAssigner = new CDCSplitAssigner(splitMaxPerTask, context.currentParallelism());
+ this.readersAwaitingSplit = new LinkedHashSet<>();
+ this.splitIdGenerator = new SplitIdGenerator();
+
+ if (checkpoint != null) {
+ for (Identifier identifier : checkpoint.getCurrentSnapshotIdMap().keySet()) {
+ Table table;
+ try {
+ table = catalog.getTable(identifier);
+ } catch (Catalog.TableNotExistException e) {
+ LOG.warn(
+ "Table {} not found after recovery. The most possible cause is it has been deleted. Skipping this table.",
+ identifier,
+ e);
+ continue;
+ }
+
+ if (!(table instanceof FileStoreTable)) {
+ LOG.warn(
+ "Table {} used to be a FileStoreTable in the last checkpoint, but after recovery it turned out to be a {}. Skipping this table.",
+ identifier,
+ table.getClass().getSimpleName());
+ continue;
+ }
+
+ TableStatus tableStatus = new TableStatus(context, (FileStoreTable) table);
+ long currentSnapshotId = checkpoint.getCurrentSnapshotIdMap().get(identifier);
+ tableStatus.restore(currentSnapshotId);
+ tableStatusMap.put(identifier, tableStatus);
+ LOG.info(
+ "Restoring state for table {}. Next snapshot id: {}",
+ identifier,
+ currentSnapshotId);
+ }
+
+ addSplits(checkpoint.getSplits());
+ }
+ }
+
+ protected void addSplits(Collection<TableAwareFileStoreSourceSplit> splits) {
+ splits.forEach(this::addSplit);
+ }
+
+ private void addSplit(TableAwareFileStoreSourceSplit split) {
+ splitAssigner.addSplit(assignSuggestedTask(split), split);
+ }
+
+ @Override
+ public void start() {
+ context.callAsync(
+ this::scanNextSnapshot, this::processDiscoveredSplits, 0, discoveryInterval);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no resources to close
+ }
+
+ @Override
+ public void addReader(int subtaskId) {
+ // this source is purely lazy-pull-based, nothing to do upon registration
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
+ readersAwaitingSplit.add(subtaskId);
+ assignSplits();
+ // if current task assigned no split, we check conditions to scan one more time
+ if (readersAwaitingSplit.contains(subtaskId)) {
+ if (stopTriggerScan) {
+ return;
+ }
+ stopTriggerScan = true;
+ context.callAsync(this::scanNextSnapshot, this::processDiscoveredSplits);
+ }
+ }
+
+ @Override
+ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
+ LOG.error("Received unrecognized event: {}", sourceEvent);
+ }
+
+ @Override
+ public void addSplitsBack(List<TableAwareFileStoreSourceSplit> splits, int subtaskId) {
+ LOG.debug("File Source Enumerator adds splits back: {}", splits);
+ splitAssigner.addSplitsBack(subtaskId, new ArrayList<>(splits));
+ }
+
+ @Override
+ public CDCCheckpoint snapshotState(long checkpointId) {
+ Collection<TableAwareFileStoreSourceSplit> splits = splitAssigner.remainingSplits();
+
+ Map<Identifier, Long> nextSnapshotIdMap = new HashMap<>();
+ for (Map.Entry<Identifier, TableStatus> entry : tableStatusMap.entrySet()) {
+ nextSnapshotIdMap.put(entry.getKey(), entry.getValue().nextSnapshotId);
+ }
+ final CDCCheckpoint checkpoint = new CDCCheckpoint(splits, nextSnapshotIdMap);
+
+ LOG.debug("Source Checkpoint is {}", checkpoint);
+ return checkpoint;
+ }
+
+ private synchronized Optional<TableAwarePlan> scanNextSnapshot() throws Exception {
+ if (splitAssigner.numberOfRemainingSplits() >= splitMaxNum) {
+ return Optional.empty();
+ }
+
+ discoverTables();
+
+ if (tableStatusMap.isEmpty()) {
+ return Optional.empty();
+ }
+
+ // Round-robin selection of tables
+ List<Identifier> tableIdentifiers = new ArrayList<>(tableStatusMap.keySet());
+ int index = currentTableIndex.getAndUpdate(i -> (i + 1) % tableIdentifiers.size());
+ Identifier currentIdentifier = tableIdentifiers.get(index);
+ StreamTableScan currentScan = tableStatusMap.get(currentIdentifier).scan;
+
+ TableScan.Plan plan = currentScan.plan();
+ return Optional.of(new TableAwarePlan(plan, currentIdentifier, currentScan.checkpoint()));
+ }
+
+ // this method could not be synchronized, because it runs in coordinatorThread, which will make
+ // it serialize.
+ private void processDiscoveredSplits(Optional<TableAwarePlan> planOptional, Throwable error) {
+ if (error != null) {
+ if (error instanceof EndOfScanException) {
+ // finished
+ LOG.debug("Catching EndOfStreamException, the stream is finished.");
+ assignSplits();
+ } else {
+ LOG.error("Failed to enumerate files", error);
+ throw new RuntimeException(error);
+ }
+ return;
+ }
+
+ if (!planOptional.isPresent()) {
+ return;
+ }
+ TableAwarePlan tableAwarePlan = planOptional.get();
+ TableScan.Plan plan = tableAwarePlan.plan;
+ tableStatusMap.get(tableAwarePlan.identifier).nextSnapshotId =
+ tableAwarePlan.nextSnapshotId;
+ if (plan.equals(SnapshotNotExistPlan.INSTANCE)) {
+ stopTriggerScan = true;
+ return;
+ }
+
+ stopTriggerScan = false;
+ if (plan.splits().isEmpty()) {
+ return;
+ }
+
+ FileStoreTable table = tableStatusMap.get(tableAwarePlan.identifier).table;
+ List<TableAwareFileStoreSourceSplit> splits = new ArrayList<>();
+ for (Split split : plan.splits()) {
+ Long lastSchemaId = tableStatusMap.get(tableAwarePlan.identifier).schemaId;
+ TableAwareFileStoreSourceSplit tableAwareFileStoreSourceSplit =
+ toTableAwareSplit(
+ splitIdGenerator.getNextId(),
+ split,
+ table,
+ tableAwarePlan.identifier,
+ lastSchemaId);
+ tableStatusMap.get(tableAwarePlan.identifier).schemaId =
+ tableAwareFileStoreSourceSplit.getSchemaId();
+ splits.add(tableAwareFileStoreSourceSplit);
+ }
+
+ addSplits(splits);
+ assignSplits();
+ }
+
+ @VisibleForTesting
+ protected TableAwareFileStoreSourceSplit toTableAwareSplit(
+ String splitId,
+ Split split,
+ FileStoreTable table,
+ Identifier identifier,
+ @Nullable Long lastSchemaId) {
+ Preconditions.checkState(split instanceof DataSplit);
+ long snapshotId = ((DataSplit) split).snapshotId();
+ long schemaId = table.snapshot(snapshotId).schemaId();
+ return new TableAwareFileStoreSourceSplit(
+ splitId, split, 0, identifier, lastSchemaId, schemaId);
+ }
+
+ /**
+ * Method should be synchronized because {@link #handleSplitRequest} and {@link
+ * #processDiscoveredSplits} have thread conflicts.
+ */
+ protected synchronized void assignSplits() {
+ // create assignment
+ Map<Integer, List<TableAwareFileStoreSourceSplit>> assignment = new HashMap<>();
+ Iterator<Integer> readersAwait = readersAwaitingSplit.iterator();
+ Set<Integer> subtaskIds = context.registeredReaders().keySet();
+ while (readersAwait.hasNext()) {
+ Integer task = readersAwait.next();
+ if (!subtaskIds.contains(task)) {
+ readersAwait.remove();
+ continue;
+ }
+ List<TableAwareFileStoreSourceSplit> splits = splitAssigner.getNext(task);
+ if (!splits.isEmpty()) {
+ assignment.put(task, splits);
+ }
+ }
+
+ assignment.keySet().forEach(readersAwaitingSplit::remove);
+ context.assignSplits(new SplitsAssignment<>(assignment));
+ }
+
+ protected int assignSuggestedTask(TableAwareFileStoreSourceSplit split) {
+ if (tableStatusMap.get(split.getIdentifier()).subtaskId != null) {
+ return tableStatusMap.get(split.getIdentifier()).subtaskId;
+ }
+
+ int subtaskId = numTablesWithSubtaskAssigned % context.currentParallelism();
+ this.tableStatusMap.get(split.getIdentifier()).subtaskId = subtaskId;
+ this.numTablesWithSubtaskAssigned++;
+ LOG.info("Assigning table {} to subtask {}", split.getIdentifier(), subtaskId);
+ return subtaskId;
+ }
+
+ private void discoverTables() throws Exception {
+ if (System.currentTimeMillis() - lastTableDiscoveryTime < tableDiscoveryInterval) {
+ return;
+ }
+ lastTableDiscoveryTime = System.currentTimeMillis();
+
+ List<String> databaseNames;
+ if (database == null || database.isEmpty()) {
+ Preconditions.checkArgument(
+ table == null || table.isEmpty(),
+ "Tables should not be specified when databases is null. But tables is specified as "
+ + table);
+ // If database parameter is not specified, dynamically scan all databases
+ databaseNames = catalog.listDatabases();
+ } else {
+ databaseNames = Collections.singletonList(database);
+ }
+
+ for (String dbName : databaseNames) {
+ List<String> tableNames;
+ if (table == null || table.isEmpty()) {
+ // If tables parameter is not specified but database is specified, dynamically scan
+ // all tables in the database
+ tableNames = catalog.listTables(dbName);
+ } else {
+ tableNames = Collections.singletonList(table);
+ }
+
+ for (String tableName : tableNames) {
+ Identifier identifier = Identifier.create(dbName, tableName);
+ if (!tableStatusMap.containsKey(identifier)
+ && !discoveredNonFileStoreTables.contains(identifier)) {
+ // New table found, add it to scans
+ Table table = catalog.getTable(identifier);
+ if (table instanceof FileStoreTable) {
+ tableStatusMap.put(
+ identifier, new TableStatus(context, (FileStoreTable) table));
+ LOG.info("Discovered new table {} and added to scan list", identifier);
+ } else if (!discoveredNonFileStoreTables.contains(identifier)) {
+ LOG.info(
+ "Discovered new table {}, but it is a {} instead of a FileStoreTable. Skipping this table.",
+ identifier,
+ table.getClass().getSimpleName());
+ discoveredNonFileStoreTables.add(identifier);
+ }
+ }
+ }
+ }
+ }
+
+ @VisibleForTesting
+ void setScan(Identifier identifier, FileStoreTable table, StreamDataTableScan scan) {
+ tableStatusMap.put(identifier, new TableStatus(table, scan));
+ }
+
+ private static class TableAwarePlan {
+ private final TableScan.Plan plan;
+ private final Identifier identifier;
+ private final Long nextSnapshotId;
+
+ private TableAwarePlan(TableScan.Plan plan, Identifier identifier, Long nextSnapshotId) {
+ this.plan = plan;
+ this.identifier = identifier;
+ this.nextSnapshotId = nextSnapshotId;
+ }
+ }
+
+ private static class TableStatus {
+ private final FileStoreTable table;
+ private final StreamDataTableScan scan;
+ private Long schemaId;
+ private Integer subtaskId;
+ private Long nextSnapshotId;
+
+ private TableStatus(SplitEnumeratorContext<?> context, FileStoreTable table) {
+ this.table = table;
+ this.scan = table.newStreamScan();
+ if (metricGroup(context) != null) {
+ this.scan.withMetricRegistry(new FlinkMetricRegistry(context.metricGroup()));
+ }
+ }
+
+ @VisibleForTesting
+ private TableStatus(FileStoreTable table, StreamDataTableScan scan) {
+ this.table = table;
+ this.scan = scan;
+ }
+
+ private void restore(Long nextSnapshotId) {
+ this.subtaskId = null;
+ this.nextSnapshotId = nextSnapshotId;
+ this.scan.restore(nextSnapshotId);
+ }
+
+ @Nullable
+ private SplitEnumeratorMetricGroup metricGroup(SplitEnumeratorContext<?> context) {
+ try {
+ return context.metricGroup();
+ } catch (NullPointerException ignore) {
+ // ignore NPE for some Flink versions
+ return null;
+ }
+ }
+ }
+
+ private static class SplitIdGenerator {
+ private final String uuid = UUID.randomUUID().toString();
+ private final AtomicInteger idCounter = new AtomicInteger(1);
+
+ private String getNextId() {
+ return uuid + "-" + idCounter.getAndIncrement();
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSplitAssigner.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSplitAssigner.java
new file mode 100644
index 0000000000..52bcb8394d
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSplitAssigner.java
@@ -0,0 +1,114 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.enumerator;
+
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.flink.source.FileStoreSourceSplit;
+import org.apache.paimon.utils.BinPacking;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Pre-calculate which splits each task should process according to the weight, and then distribute
+ * the splits fairly.
+ *
+ * <p>This class is similar to {@link
+ * org.apache.paimon.flink.source.assigners.PreAssignSplitAssigner} but is integrated with {@link
+ * TableAwareFileStoreSourceSplit} instead of {@link FileStoreSourceSplit}.
+ */
+public class CDCSplitAssigner {
+
+ /** Default batch splits size to avoid exceed `akka.framesize`. */
+ private final int splitBatchSize;
+
+ private final Map<Integer, LinkedList<TableAwareFileStoreSourceSplit>> pendingSplitAssignment;
+
+ private final AtomicInteger numberOfPendingSplits;
+
+ public CDCSplitAssigner(int splitBatchSize, int parallelism) {
+ this.splitBatchSize = splitBatchSize;
+ this.pendingSplitAssignment =
+ createBatchFairSplitAssignment(Collections.emptyList(), parallelism);
+ this.numberOfPendingSplits = new AtomicInteger(0);
+ }
+
+ public List<TableAwareFileStoreSourceSplit> getNext(int subtask) {
+ // The following batch assignment operation is for two purposes:
+ // To distribute splits evenly when batch reading to prevent a few tasks from reading all
+ // the data (for example, the current resource can only schedule part of the tasks).
+ Queue<TableAwareFileStoreSourceSplit> taskSplits = pendingSplitAssignment.get(subtask);
+ List<TableAwareFileStoreSourceSplit> assignment = new ArrayList<>();
+ while (taskSplits != null && !taskSplits.isEmpty() && assignment.size() < splitBatchSize) {
+ assignment.add(taskSplits.poll());
+ }
+ numberOfPendingSplits.getAndAdd(-assignment.size());
+ return assignment;
+ }
+
+ public void addSplit(int suggestedTask, TableAwareFileStoreSourceSplit split) {
+ pendingSplitAssignment.computeIfAbsent(suggestedTask, k -> new LinkedList<>()).add(split);
+ numberOfPendingSplits.incrementAndGet();
+ }
+
+ public void addSplitsBack(int subtask, List<TableAwareFileStoreSourceSplit> splits) {
+ LinkedList<TableAwareFileStoreSourceSplit> remainingSplits =
+ pendingSplitAssignment.computeIfAbsent(subtask, k -> new LinkedList<>());
+ ListIterator<TableAwareFileStoreSourceSplit> iterator = splits.listIterator(splits.size());
+ while (iterator.hasPrevious()) {
+ remainingSplits.addFirst(iterator.previous());
+ }
+ numberOfPendingSplits.getAndAdd(splits.size());
+ }
+
+ public Collection<TableAwareFileStoreSourceSplit> remainingSplits() {
+ List<TableAwareFileStoreSourceSplit> splits = new ArrayList<>();
+ pendingSplitAssignment.values().forEach(splits::addAll);
+ return splits;
+ }
+
+ /**
+ * this method only reload restore for batch execute, because in streaming mode, we need to
+ * assign certain bucket to certain task.
+ */
+ private static Map<Integer, LinkedList<TableAwareFileStoreSourceSplit>>
+ createBatchFairSplitAssignment(
+ Collection<TableAwareFileStoreSourceSplit> splits, int numReaders) {
+ List<List<TableAwareFileStoreSourceSplit>> assignmentList =
+ BinPacking.packForFixedBinNumber(
+ splits, split -> split.split().rowCount(), numReaders);
+ Map<Integer, LinkedList<TableAwareFileStoreSourceSplit>> assignment = new HashMap<>();
+ for (int i = 0; i < assignmentList.size(); i++) {
+ assignment.put(i, new LinkedList<>(assignmentList.get(i)));
+ }
+ return assignment;
+ }
+
+ public int numberOfRemainingSplits() {
+ return numberOfPendingSplits.get();
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCRecordsWithSplitIds.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCRecordsWithSplitIds.java
new file mode 100644
index 0000000000..5c95c3279f
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCRecordsWithSplitIds.java
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.reader;
+
+import org.apache.paimon.flink.source.FileStoreSourceSplitState;
+import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
+import org.apache.paimon.utils.Reference;
+
+import org.apache.flink.api.common.eventtime.TimestampAssigner;
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.file.src.reader.BulkFormat.RecordIterator;
+import org.apache.flink.connector.file.src.util.RecordAndPosition;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * A {@link RecordsWithSplitIds} which contains only one iterator record. This can ensure that there
+ * will be no checkpoint segmentation in iterator consumption.
+ */
+public class CDCRecordsWithSplitIds implements RecordsWithSplitIds<RecordIterator<Event>> {
+
+ @Nullable private String splitId;
+
+ @Nullable private Reference<RecordIterator<Event>> recordsForSplitCurrent;
+
+ @Nullable private final RecordIterator<Event> recordsForSplit;
+
+ private final Set<String> finishedSplits;
+
+ private CDCRecordsWithSplitIds(
+ @Nullable String splitId,
+ @Nullable RecordIterator<Event> recordsForSplit,
+ Set<String> finishedSplits) {
+ this.splitId = splitId;
+ this.recordsForSplit = recordsForSplit;
+ this.finishedSplits = finishedSplits;
+ }
+
+ @Nullable
+ @Override
+ public String nextSplit() {
+ // move the split one (from current value to null)
+ final String nextSplit = this.splitId;
+ this.splitId = null;
+
+ // move the iterator, from null to value (if first move) or to null (if second move)
+ this.recordsForSplitCurrent =
+ nextSplit != null ? new Reference<>(this.recordsForSplit) : null;
+
+ return nextSplit;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<Event> nextRecordFromSplit() {
+ if (this.recordsForSplitCurrent == null) {
+ throw new IllegalStateException();
+ }
+
+ RecordIterator<Event> recordsForSplit = this.recordsForSplitCurrent.get();
+ this.recordsForSplitCurrent.set(null);
+ return recordsForSplit;
+ }
+
+ @Override
+ public Set<String> finishedSplits() {
+ return finishedSplits;
+ }
+
+ @Override
+ public void recycle() {
+ if (recordsForSplit != null) {
+ recordsForSplit.releaseBatch();
+ }
+ }
+
+ public static CDCRecordsWithSplitIds forRecords(
+ String splitId, RecordIterator<Event> recordsForSplit) {
+ return new CDCRecordsWithSplitIds(splitId, recordsForSplit, Collections.emptySet());
+ }
+
+ public static CDCRecordsWithSplitIds finishedSplit(String splitId) {
+ return new CDCRecordsWithSplitIds(null, null, Collections.singleton(splitId));
+ }
+
+ public static void emitRecord(
+ SourceReaderContext context,
+ RecordIterator<Event> element,
+ SourceOutput<Event> output,
+ FileStoreSourceSplitState state,
+ FileStoreSourceReaderMetrics metrics) {
+ long timestamp = TimestampAssigner.NO_TIMESTAMP;
+ if (metrics.getLatestFileCreationTime() != FileStoreSourceReaderMetrics.UNDEFINED) {
+ timestamp = metrics.getLatestFileCreationTime();
+ }
+
+ // This metric only counts the number of RecordIterator<Event> emitted,
+ // however what we really want is to count the number of Events emitted,
+ // so we replenish the missing record count here.
+ org.apache.flink.metrics.Counter numRecordsIn =
+ context.metricGroup().getIOMetricGroup().getNumRecordsInCounter();
+ boolean firstRecord = true;
+
+ RecordAndPosition<Event> record;
+ while ((record = element.next()) != null) {
+ // First record in the iterator is already counted by SourceReaderBase#pollNext.
+ if (firstRecord) {
+ firstRecord = false;
+ } else {
+ numRecordsIn.inc();
+ }
+
+ Event event = record.getRecord();
+ output.collect(event, timestamp);
+ state.setPosition(record);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceReader.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceReader.java
new file mode 100644
index 0000000000..05cf7880cd
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceReader.java
@@ -0,0 +1,99 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.reader;
+
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.flink.pipeline.cdc.source.CDCSource;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.flink.source.FileStoreSourceSplitState;
+import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
+
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
+import org.apache.flink.connector.file.src.reader.BulkFormat.RecordIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/** A {@link SourceReader} that read records from {@link TableAwareFileStoreSourceSplit}. */
+public class CDCSourceReader
+ extends SingleThreadMultiplexSourceReaderBase<
+ RecordIterator<Event>,
+ Event,
+ TableAwareFileStoreSourceSplit,
+ FileStoreSourceSplitState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CDCSourceReader.class);
+
+ private final IOManager ioManager;
+
+ public CDCSourceReader(
+ SourceReaderContext readerContext,
+ FileStoreSourceReaderMetrics metrics,
+ IOManager ioManager,
+ CDCSource.TableManager tableManager) {
+ super(
+ () -> new CDCSourceSplitReader(metrics, tableManager),
+ (element, output, state) ->
+ CDCRecordsWithSplitIds.emitRecord(
+ readerContext, element, output, state, metrics),
+ readerContext.getConfiguration(),
+ readerContext);
+ this.ioManager = ioManager;
+ }
+
+ @Override
+ public void start() {
+ // we request a split only if we did not get splits during the checkpoint restore
+ if (getNumberOfCurrentlyAssignedSplits() == 0) {
+ context.sendSplitRequest();
+ }
+ }
+
+ @Override
+ protected void onSplitFinished(Map<String, FileStoreSourceSplitState> finishedSplitIds) {
+ // this method is called each time when we consume one split
+ // it is possible that one response from the coordinator contains multiple splits
+ // we should only require for more splits after we've consumed all given splits
+ if (getNumberOfCurrentlyAssignedSplits() == 0) {
+ context.sendSplitRequest();
+ }
+ }
+
+ @Override
+ protected FileStoreSourceSplitState initializedState(TableAwareFileStoreSourceSplit split) {
+ LOG.info("Initializing split {}", split);
+ return new FileStoreSourceSplitState(split);
+ }
+
+ @Override
+ protected TableAwareFileStoreSourceSplit toSplitType(
+ String splitId, FileStoreSourceSplitState splitState) {
+ LOG.info("Converting split state {} with id {} to split", splitState, splitId);
+ return (TableAwareFileStoreSourceSplit) splitState.toSourceSplit();
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ ioManager.close();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
similarity index 54%
copy from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
copy to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
index c3d7710097..48c0ef82f2 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
@@ -16,21 +16,28 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.source;
+package org.apache.paimon.flink.pipeline.cdc.source.reader;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.flink.FlinkRowData;
-import org.apache.paimon.flink.FlinkRowDataWithBlob;
+import org.apache.paimon.flink.pipeline.cdc.source.CDCSource;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.flink.pipeline.cdc.util.PaimonToFlinkCDCDataConverter;
+import org.apache.paimon.flink.source.FileStoreSourceSplitReader;
import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.reader.RecordReader.RecordIterator;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.types.DataTypeRoot;
-import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pool;
+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.types.DataType;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
@@ -38,7 +45,6 @@ import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.util.MutableRecordAndPosition;
import org.apache.flink.connector.file.src.util.RecordAndPosition;
-import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,27 +53,27 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.time.Duration;
import java.util.Collection;
-import java.util.Collections;
import java.util.LinkedList;
+import java.util.List;
import java.util.Objects;
import java.util.Queue;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
-/** The {@link SplitReader} implementation for the file store source. */
-public class FileStoreSourceSplitReader
- implements SplitReader<BulkFormat.RecordIterator<RowData>, FileStoreSourceSplit> {
+import static org.apache.paimon.flink.pipeline.cdc.util.PaimonToFlinkCDCDataConverter.convertRowToDataChangeEvent;
+import static org.apache.paimon.flink.pipeline.cdc.util.PaimonToFlinkCDCTypeConverter.convertPaimonSchemaToFlinkCDCSchema;
- private static final Logger LOG = LoggerFactory.getLogger(FileStoreSourceSplitReader.class);
+/** The {@link SplitReader} implementation for the {@link CDCSource}. */
+public class CDCSourceSplitReader
+ implements SplitReader<BulkFormat.RecordIterator<Event>, TableAwareFileStoreSourceSplit> {
- private final TableRead tableRead;
+ private static final Logger LOG = LoggerFactory.getLogger(CDCSourceSplitReader.class);
- @Nullable private final RecordLimiter limiter;
-
- private final Queue<FileStoreSourceSplit> splits;
+ private final Queue<TableAwareFileStoreSourceSplit> splits;
private final Pool<FileStoreRecordIterator> pool;
+ private final CDCSource.TableManager tableManager;
+ private TableReaderInfo currentTableReaderInfo;
@Nullable private LazyRecordReader currentReader;
@Nullable private String currentSplitId;
private long currentNumRead;
@@ -76,29 +82,22 @@ public class FileStoreSourceSplitReader
private boolean paused;
private final AtomicBoolean wakeup;
private final FileStoreSourceReaderMetrics metrics;
- private final boolean blobAsDescriptor;
-
- public FileStoreSourceSplitReader(
- TableRead tableRead,
- @Nullable RecordLimiter limiter,
- FileStoreSourceReaderMetrics metrics,
- @Nullable RowType readType,
- boolean blobAsDescriptor) {
- this.tableRead = tableRead;
- this.limiter = limiter;
+
+ public CDCSourceSplitReader(
+ FileStoreSourceReaderMetrics metrics, CDCSource.TableManager tableManager) {
this.splits = new LinkedList<>();
this.pool = new Pool<>(1);
- this.pool.add(new FileStoreRecordIterator(readType));
+ this.pool.add(new FileStoreRecordIterator());
this.paused = false;
this.metrics = metrics;
this.wakeup = new AtomicBoolean(false);
- this.blobAsDescriptor = blobAsDescriptor;
+ this.tableManager = tableManager;
}
@Override
- public RecordsWithSplitIds<BulkFormat.RecordIterator<RowData>> fetch() throws IOException {
+ public RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> fetch() throws IOException {
if (paused) {
- return new EmptyRecordsWithSplitIds<>();
+ return new FileStoreSourceSplitReader.EmptyRecordsWithSplitIds<>();
}
checkSplitOrStartNext();
@@ -108,7 +107,7 @@ public class FileStoreSourceSplitReader
FileStoreRecordIterator iterator = poll();
if (iterator == null) {
LOG.info("Skip waiting for object pool due to wakeup");
- return new EmptyRecordsWithSplitIds<>();
+ return new FileStoreSourceSplitReader.EmptyRecordsWithSplitIds<>();
}
RecordIterator<InternalRow> nextBatch;
@@ -116,20 +115,14 @@ public class FileStoreSourceSplitReader
nextBatch = currentFirstBatch;
currentFirstBatch = null;
} else {
- nextBatch =
- reachLimit()
- ? null
- : Objects.requireNonNull(currentReader).recordReader().readBatch();
+ nextBatch = Objects.requireNonNull(currentReader).recordReader().readBatch();
}
if (nextBatch == null) {
pool.recycler().recycle(iterator);
return finishSplit();
}
- return FlinkRecordsWithSplitIds.forRecords(currentSplitId, iterator.replace(nextBatch));
- }
-
- private boolean reachLimit() {
- return limiter != null && limiter.reachLimit();
+ return CDCRecordsWithSplitIds.forRecords(
+ currentSplitId, iterator.replace(nextBatch, currentTableReaderInfo));
}
@Nullable
@@ -150,7 +143,7 @@ public class FileStoreSourceSplitReader
}
@Override
- public void handleSplitsChanges(SplitsChange<FileStoreSourceSplit> splitsChange) {
+ public void handleSplitsChanges(SplitsChange<TableAwareFileStoreSourceSplit> splitsChange) {
if (!(splitsChange instanceof SplitsAddition)) {
throw new UnsupportedOperationException(
String.format(
@@ -165,16 +158,16 @@ public class FileStoreSourceSplitReader
* Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.7-.
*/
public void pauseOrResumeSplits(
- Collection<FileStoreSourceSplit> splitsToPause,
- Collection<FileStoreSourceSplit> splitsToResume) {
- for (FileStoreSourceSplit split : splitsToPause) {
+ Collection<TableAwareFileStoreSourceSplit> splitsToPause,
+ Collection<TableAwareFileStoreSourceSplit> splitsToResume) {
+ for (TableAwareFileStoreSourceSplit split : splitsToPause) {
if (split.splitId().equals(currentSplitId)) {
paused = true;
break;
}
}
- for (FileStoreSourceSplit split : splitsToResume) {
+ for (TableAwareFileStoreSourceSplit split : splitsToResume) {
if (split.splitId().equals(currentSplitId)) {
paused = false;
break;
@@ -202,9 +195,9 @@ public class FileStoreSourceSplitReader
return;
}
- final FileStoreSourceSplit nextSplit = splits.poll();
+ final TableAwareFileStoreSourceSplit nextSplit = splits.poll();
if (nextSplit == null) {
- throw new IOException("Cannot fetch from another split - no split remaining");
+ return;
}
// update metric when split changes
@@ -216,17 +209,26 @@ public class FileStoreSourceSplitReader
metrics.recordSnapshotUpdate(eventTime);
}
+ Identifier identifier = nextSplit.getIdentifier();
+ TableSchema tableSchema = tableManager.getTableSchema(identifier, nextSplit.getSchemaId());
+ List<SchemaChangeEvent> schemaChangeEvents =
+ tableManager.generateSchemaChangeEventList(
+ identifier, nextSplit.getLastSchemaId(), nextSplit.getSchemaId());
+ currentTableReaderInfo = new TableReaderInfo(identifier, tableSchema, schemaChangeEvents);
currentSplitId = nextSplit.splitId();
- currentReader = new LazyRecordReader(nextSplit.split());
+ currentReader = createLazyRecordReader(nextSplit.split());
currentNumRead = nextSplit.recordsToSkip();
- if (limiter != null) {
- limiter.add(currentNumRead);
- }
+
if (currentNumRead > 0) {
seek(currentNumRead);
}
}
+ @VisibleForTesting
+ protected LazyRecordReader createLazyRecordReader(Split split) {
+ return new LazyRecordReader(split, currentTableReaderInfo, tableManager);
+ }
+
private void seek(long toSkip) throws IOException {
while (true) {
RecordIterator<InternalRow> nextBatch =
@@ -247,7 +249,7 @@ public class FileStoreSourceSplitReader
}
}
- private FlinkRecordsWithSplitIds finishSplit() throws IOException {
+ private CDCRecordsWithSplitIds finishSplit() throws IOException {
if (currentReader != null) {
if (currentReader.lazyRecordReader != null) {
currentReader.lazyRecordReader.close();
@@ -255,45 +257,49 @@ public class FileStoreSourceSplitReader
currentReader = null;
}
- final FlinkRecordsWithSplitIds finishRecords =
- FlinkRecordsWithSplitIds.finishedSplit(currentSplitId);
+ final CDCRecordsWithSplitIds finishRecords =
+ CDCRecordsWithSplitIds.finishedSplit(currentSplitId);
currentSplitId = null;
return finishRecords;
}
- private class FileStoreRecordIterator implements BulkFormat.RecordIterator<RowData> {
+ private class FileStoreRecordIterator implements BulkFormat.RecordIterator<Event> {
private RecordIterator<InternalRow> iterator;
- private final MutableRecordAndPosition<RowData> recordAndPosition =
+ private final MutableRecordAndPosition<Event> recordAndPosition =
new MutableRecordAndPosition<>();
- @Nullable private final Integer blobField;
- private FileStoreRecordIterator(@Nullable RowType rowType) {
- this.blobField = rowType == null ? null : blobFieldIndex(rowType);
- }
+ private TableReaderInfo tableReaderInfo;
+ private final Queue<SchemaChangeEvent> schemaChangeEventList = new LinkedList<>();
- private Integer blobFieldIndex(RowType rowType) {
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- if (rowType.getTypeAt(i).getTypeRoot() == DataTypeRoot.BLOB) {
- return i;
- }
- }
- return null;
- }
-
- public FileStoreRecordIterator replace(RecordIterator<InternalRow> iterator) {
+ public FileStoreRecordIterator replace(
+ RecordIterator<InternalRow> iterator, TableReaderInfo tableReaderInfo) {
this.iterator = iterator;
this.recordAndPosition.set(null, RecordAndPosition.NO_OFFSET, currentNumRead);
+ this.tableReaderInfo = tableReaderInfo;
+ this.schemaChangeEventList.addAll(tableReaderInfo.schemaChangeEvents);
return this;
}
@Nullable
@Override
- public RecordAndPosition<RowData> next() {
- if (reachLimit()) {
+ public RecordAndPosition<Event> next() {
+ Event event = nextEvent();
+ if (event == null) {
return null;
}
+
+ recordAndPosition.setNext(event);
+ currentNumRead++;
+ return recordAndPosition;
+ }
+
+ private Event nextEvent() {
+ if (!schemaChangeEventList.isEmpty()) {
+ return schemaChangeEventList.poll();
+ }
+
InternalRow row;
try {
row = iterator.next();
@@ -304,15 +310,11 @@ public class FileStoreSourceSplitReader
return null;
}
- recordAndPosition.setNext(
- blobField == null
- ? new FlinkRowData(row)
- : new FlinkRowDataWithBlob(row, blobField, blobAsDescriptor));
- currentNumRead++;
- if (limiter != null) {
- limiter.increment();
- }
- return recordAndPosition;
+ return convertRowToDataChangeEvent(
+ tableReaderInfo.tableId,
+ row,
+ tableReaderInfo.fieldGetters,
+ tableReaderInfo.generator);
}
@Override
@@ -323,45 +325,61 @@ public class FileStoreSourceSplitReader
}
/** Lazy to create {@link RecordReader} to improve performance for limit. */
- private class LazyRecordReader {
-
- private final Split split;
+ protected static class LazyRecordReader {
+ protected final Split split;
+ private final TableReaderInfo currentTableReaderInfo;
+ private final CDCSource.TableManager tableManager;
private RecordReader<InternalRow> lazyRecordReader;
- private LazyRecordReader(Split split) {
+ protected LazyRecordReader(
+ Split split,
+ TableReaderInfo currentTableReaderInfo,
+ CDCSource.TableManager tableManager) {
this.split = split;
+ this.currentTableReaderInfo = currentTableReaderInfo;
+ this.tableManager = tableManager;
}
public RecordReader<InternalRow> recordReader() throws IOException {
if (lazyRecordReader == null) {
- lazyRecordReader = tableRead.createReader(split);
+ lazyRecordReader =
+ tableManager
+ .getTableRead(
+ currentTableReaderInfo.identifier,
+ currentTableReaderInfo.currentSchema)
+ .createReader(split);
}
return lazyRecordReader;
}
}
- /**
- * An empty implementation of {@link RecordsWithSplitIds}. It is used to indicate that the
- * {@link FileStoreSourceSplitReader} is paused or wakeup.
- */
- private static class EmptyRecordsWithSplitIds<T> implements RecordsWithSplitIds<T> {
-
- @Nullable
- @Override
- public String nextSplit() {
- return null;
- }
-
- @Nullable
- @Override
- public T nextRecordFromSplit() {
- return null;
- }
-
- @Override
- public Set<String> finishedSplits() {
- return Collections.emptySet();
+ private static class TableReaderInfo {
+ private final Identifier identifier;
+ private final TableId tableId;
+ private final TableSchema currentSchema;
+ private final List<SchemaChangeEvent> schemaChangeEvents;
+ private final BinaryRecordDataGenerator generator;
+ private final List<InternalRow.FieldGetter> fieldGetters;
+
+ private TableReaderInfo(
+ Identifier identifier,
+ TableSchema currentSchema,
+ List<SchemaChangeEvent> schemaChangeEvents) {
+ this.identifier = identifier;
+ this.tableId = TableId.tableId(identifier.getDatabaseName(), identifier.getTableName());
+
+ this.currentSchema = currentSchema;
+ this.schemaChangeEvents = schemaChangeEvents;
+
+ org.apache.flink.cdc.common.schema.Schema currentCDCSchema =
+ convertPaimonSchemaToFlinkCDCSchema(currentSchema);
+ this.generator =
+ new BinaryRecordDataGenerator(
+ currentCDCSchema.getColumnDataTypes().toArray(new DataType[0]));
+ this.fieldGetters =
+ PaimonToFlinkCDCDataConverter.createFieldGetters(
+ currentCDCSchema.getColumnDataTypes());
}
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/CDCUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/CDCUtils.java
new file mode 100644
index 0000000000..393a067f5a
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/CDCUtils.java
@@ -0,0 +1,47 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.util;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.flink.FlinkCatalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.FactoryUtil;
+
+/** Utility methods for CDC class. */
+public class CDCUtils {
+ public static Catalog createCatalog(CatalogContext catalogContext, ReadableConfig flinkConfig) {
+
+ FlinkCatalogFactory flinkCatalogFactory =
+ FactoryUtil.discoverFactory(
+ FlinkCatalogFactory.class.getClassLoader(),
+ FlinkCatalogFactory.class,
+ FlinkCatalogFactory.IDENTIFIER);
+ FlinkCatalog flinkCatalog =
+ flinkCatalogFactory.createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "flink-catalog",
+ catalogContext.options().toMap(),
+ flinkConfig,
+ FlinkCatalog.class.getClassLoader()));
+ return flinkCatalog.catalog();
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCTypeConverter.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCTypeConverter.java
index 2c369ff9a0..ad677eaf25 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCTypeConverter.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCTypeConverter.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.pipeline.cdc.util;
import org.apache.paimon.flink.LogicalTypeConversion;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataType;
import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
@@ -28,9 +29,23 @@ import org.apache.flink.table.types.utils.TypeConversions;
/** Type converter from Paimon to Flink CDC. */
public class PaimonToFlinkCDCTypeConverter {
+ /** Convert Paimon schema to Flink CDC schema. */
+ public static org.apache.flink.cdc.common.schema.Schema convertPaimonSchemaToFlinkCDCSchema(
+ TableSchema schema) {
+ if (schema == null) {
+ return null;
+ }
+
+ return convertPaimonSchemaToFlinkCDCSchema(schema.toSchema());
+ }
+
/** Convert Paimon schema to Flink CDC schema. */
public static org.apache.flink.cdc.common.schema.Schema convertPaimonSchemaToFlinkCDCSchema(
Schema schema) {
+ if (schema == null) {
+ return null;
+ }
+
org.apache.flink.cdc.common.schema.Schema.Builder builder =
new org.apache.flink.cdc.common.schema.Schema.Builder();
schema.fields()
diff --git a/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory b/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 0000000000..323f4df957
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/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.paimon.flink.pipeline.cdc.source.CDCDataSourceFactory
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessorTest.java
new file mode 100644
index 0000000000..44469a6cc7
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/schema/CDCMetadataAccessorTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.schema;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.cdc.common.event.TableId;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link CDCMetadataAccessor}. */
+public class CDCMetadataAccessorTest {
+ @TempDir protected static Path temporaryFolder;
+
+ @Test
+ public void test() throws Exception {
+ String warehouse = temporaryFolder.toAbsolutePath().toString();
+
+ Options options = new Options();
+ options.set(CatalogOptions.WAREHOUSE, warehouse);
+ Catalog catalog = CatalogFactory.createCatalog(CatalogContext.create(options));
+ catalog.createDatabase("default", false);
+ Identifier identifier = Identifier.create("default", "test");
+ Schema schema = Schema.newBuilder().column("a", DataTypes.INT()).build();
+ catalog.createTable(identifier, schema, false);
+
+ TableId tableId = TableId.tableId("default", "test");
+ CDCMetadataAccessor metadataAccessor = new CDCMetadataAccessor(catalog);
+ assertThatThrownBy(metadataAccessor::listNamespaces)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .message()
+ .isEqualTo("Paimon does not support namespaces");
+ assertThat(metadataAccessor.listSchemas(null)).containsExactly("default");
+ assertThat(metadataAccessor.listTables(null, "default")).containsExactly(tableId);
+ assertThat(metadataAccessor.listTables(null, null)).containsExactly(tableId);
+
+ org.apache.flink.cdc.common.schema.Schema expectedSchema =
+ org.apache.flink.cdc.common.schema.Schema.newBuilder()
+ .physicalColumn("a", org.apache.flink.cdc.common.types.DataTypes.INT())
+ .option("path", warehouse + "/default.db/test")
+ .build();
+ org.apache.flink.cdc.common.schema.Schema actualSchema =
+ metadataAccessor.getTableSchema(tableId);
+ assertThat(actualSchema).isEqualTo(expectedSchema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactoryTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactoryTest.java
new file mode 100644
index 0000000000..35ca772dce
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCDataSourceFactoryTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
+import org.apache.flink.table.api.ValidationException;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CDCDataSourceFactory}. */
+public class CDCDataSourceFactoryTest {
+ @TempDir public static java.nio.file.Path temporaryFolder;
+
+ @Test
+ public void testCreateDataSink() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("paimon", DataSourceFactory.class);
+ assertThat(sourceFactory).isInstanceOf(CDCDataSourceFactory.class);
+
+ Map<String, String> catalogOptionsMap = new HashMap<>();
+ catalogOptionsMap.put("catalog.properties.warehouse", temporaryFolder.toUri().toString());
+ Configuration conf = Configuration.fromMap(catalogOptionsMap);
+ DataSource dataSource =
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ assertThat(dataSource).isInstanceOf(CDCDataSource.class);
+ }
+
+ @Test
+ void testUnsupportedOption() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("paimon", DataSourceFactory.class);
+ assertThat(sourceFactory).isInstanceOf(CDCDataSourceFactory.class);
+
+ Map<String, String> catalogOptionsMap = new HashMap<>();
+ catalogOptionsMap.put("warehouse", temporaryFolder.toUri().toString());
+ Configuration conf = Configuration.fromMap(catalogOptionsMap);
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported options found for 'paimon'.")
+ .hasMessageContaining("Unsupported options:")
+ .hasMessageContaining("warehouse");
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCase.java
new file mode 100644
index 0000000000..290ba3ae97
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCase.java
@@ -0,0 +1,149 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.DeleteAction;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.connectors.values.ValuesDatabase;
+import org.apache.flink.cdc.connectors.values.sink.ValuesDataSinkOptions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** The IT Case for the cdc source. */
+public class CDCSourceITCase extends CDCSourceITCaseBase {
+
+ private final TableId table1 = TableId.tableId("test", "table1");
+ private final TableId table2 = TableId.tableId("test2", "table2");
+
+ private static final Schema SCHEMA =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("varchar_value", DataTypes.VARCHAR(20))
+ .column("char_value", DataTypes.CHAR(20))
+ .column("tiny_int_value", DataTypes.TINYINT())
+ .column("small_int_value", DataTypes.SMALLINT())
+ .column("big_int_value", DataTypes.BIGINT())
+ .column("float_value", DataTypes.FLOAT())
+ .column("double_value", DataTypes.DOUBLE())
+ .column("boolean_value", DataTypes.BOOLEAN())
+ .column("decimal_value", DataTypes.DECIMAL(10, 2))
+ .primaryKey("id")
+ .option("parquet.use-native", "false")
+ .build();
+
+ private static final String[] DATA_SQLS =
+ new String[] {
+ "(1, 'a', 'a', CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(1 AS BIGINT), 1.0, 1.0, true, CAST('123.45' AS DECIMAL(10, 2)))",
+ "(2, 'b', 'b', CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(2 AS BIGINT), 2.0, 2.0, true, CAST('123.45' AS DECIMAL(10, 2)))",
+ "(1, 'c', 'c', CAST(3 AS TINYINT), CAST(3 AS SMALLINT), CAST(3 AS BIGINT), 3.0, 3.0, false)",
+ "(1, 'd', 'd', CAST(4 AS TINYINT), CAST(4 AS SMALLINT), CAST(4 AS BIGINT), 4.0, 4.0, false)",
+ };
+
+ private static final String[] DATA_VALUES =
+ new String[] {
+ "id=1;varchar_value=a;char_value=a ;tiny_int_value=1;small_int_value=1;big_int_value=1;float_value=1.0;double_value=1.0;boolean_value=true;decimal_value=123.45",
+ "id=2;varchar_value=b;char_value=b ;tiny_int_value=2;small_int_value=2;big_int_value=2;float_value=2.0;double_value=2.0;boolean_value=true;decimal_value=123.45",
+ "id=1;varchar_value=c;char_value=c ;tiny_int_value=3;small_int_value=3;big_int_value=3;float_value=3.0;double_value=3.0;boolean_value=false",
+ "id=2;varchar_value=b;char_value=b ;tiny_int_value=2;small_int_value=2;big_int_value=2;float_value=2.0;double_value=2.0;boolean_value=true",
+ "id=1;varchar_value=d;char_value=d ;tiny_int_value=4;small_int_value=4;big_int_value=4;float_value=4.0;double_value=4.0;boolean_value=false",
+ };
+
+ @AfterEach
+ @Override
+ public void afterEach() throws Exception {
+ super.afterEach();
+ catalog.dropDatabase(table1.getSchemaName(), true, true);
+ catalog.dropDatabase(table2.getSchemaName(), true, true);
+ }
+
+ @Test
+ public void test() throws Exception {
+ catalog.createDatabase(table1.getSchemaName(), false);
+ catalog.createTable(
+ Identifier.create(table1.getSchemaName(), table1.getTableName()), SCHEMA, false);
+
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ sinkConfig.set(ValuesDataSinkOptions.PRINT_ENABLED, false);
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+
+ // Execute the pipeline
+ jobClient = createAndExecutePipeline(env, sinkConfig, pipelineConfig);
+
+ insertInto(table1, DATA_SQLS[0], DATA_SQLS[1]);
+ waitUtilExpectedResult(table1, DATA_VALUES[0], DATA_VALUES[1]);
+
+ catalog.createDatabase(table2.getSchemaName(), false);
+ catalog.createTable(
+ Identifier.create(table2.getSchemaName(), table2.getTableName()), SCHEMA, false);
+
+ insertInto(table2, DATA_SQLS[0], DATA_SQLS[1]);
+ waitUtilExpectedResult(table2, DATA_VALUES[0], DATA_VALUES[1]);
+
+ DeleteAction action =
+ new DeleteAction(
+ table1.getSchemaName(),
+ table1.getTableName(),
+ "id = 1",
+ getCatalogOptions().toMap());
+
+ action.run();
+ waitUtilExpectedResult(table1, DATA_VALUES[1]);
+
+ catalog.alterTable(
+ Identifier.create(table1.getSchemaName(), table1.getTableName()),
+ SchemaChange.dropColumn("decimal_value"),
+ false);
+ insertInto(table1, DATA_SQLS[2]);
+ waitUtilExpectedResult(table1, DATA_VALUES[2], DATA_VALUES[3]);
+ }
+
+ private void waitUtilExpectedResult(TableId tableId, String... expectedResultArray)
+ throws InterruptedException {
+ List<String> expectedResult =
+ Arrays.stream(expectedResultArray)
+ .map(x -> String.format("%s:%s", tableId, x))
+ .sorted()
+ .collect(Collectors.toList());
+ waitUtilExpectedResultInternal(tableId, expectedResult);
+ }
+
+ @Override
+ protected List<String> getActualResult(TableId tableId) {
+ try {
+ return ValuesDatabase.getResults(tableId);
+ } catch (NullPointerException e) {
+ return Collections.singletonList("Uninitialized result");
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCaseBase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCaseBase.java
new file mode 100644
index 0000000000..ad5d7239f6
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceITCaseBase.java
@@ -0,0 +1,172 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.cdc.composer.definition.SinkDef;
+import org.apache.flink.cdc.composer.definition.SourceDef;
+import org.apache.flink.cdc.composer.flink.FlinkPipelineComposer;
+import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.PREFIX_CATALOG_PROPERTIES;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Abstract class containing common methods for CDC source IT cases. */
+public abstract class CDCSourceITCaseBase {
+
+ private static final String CATALOG_NAME = "paimon_catalog";
+ private static final long TIMEOUT_MILLIS = Duration.ofSeconds(30).toMillis();
+ private static final long CHECK_INTERVAL_MILLIS = Duration.ofSeconds(1).toMillis();
+
+ @TempDir public static java.nio.file.Path warehouseFolder;
+
+ protected StreamExecutionEnvironment env;
+ protected StreamTableEnvironment tEnv;
+ protected Catalog catalog;
+ protected JobClient jobClient;
+
+ @BeforeEach
+ public void initialize() throws Exception {
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env);
+
+ FlinkCatalogFactory flinkCatalogFactory =
+ FactoryUtil.discoverFactory(
+ FlinkCatalogFactory.class.getClassLoader(),
+ FlinkCatalogFactory.class,
+ FlinkCatalogFactory.IDENTIFIER);
+ FlinkCatalog flinkCatalog =
+ flinkCatalogFactory.createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "flink-catalog",
+ getCatalogOptions().toMap(),
+ null,
+ FlinkCatalog.class.getClassLoader()));
+ catalog = flinkCatalog.catalog();
+
+ tEnv.registerCatalog(CATALOG_NAME, flinkCatalog);
+ tEnv.useCatalog(CATALOG_NAME);
+ }
+
+ @AfterEach
+ public void afterEach() throws Exception {
+ if (jobClient != null) {
+ jobClient.cancel().get(1, TimeUnit.MINUTES);
+ jobClient = null;
+ }
+ }
+
+ protected Options getCatalogOptions() {
+ String warehouse = warehouseFolder.toFile().toString();
+ Options catalogOptions = new Options();
+ catalogOptions.setString("metastore", "filesystem");
+ catalogOptions.setString("warehouse", warehouse);
+ catalogOptions.setString("cache-enabled", "false");
+ catalogOptions.setString("continuous.discovery-interval", "1s");
+ return catalogOptions;
+ }
+
+ protected JobClient createAndExecutePipeline(
+ StreamExecutionEnvironment env,
+ Configuration additionalSinkConfig,
+ Configuration additionalPipelineConfig)
+ throws Exception {
+ Map<String, String> sourceConfigMap = new HashMap<>();
+ for (Map.Entry<String, String> catalogOption : getCatalogOptions().toMap().entrySet()) {
+ sourceConfigMap.put(
+ PREFIX_CATALOG_PROPERTIES + catalogOption.getKey(), catalogOption.getValue());
+ }
+ sourceConfigMap.put("table.discovery-interval", "3s");
+ Configuration sourceConfig = Configuration.fromMap(sourceConfigMap);
+ SourceDef sourceDef =
+ new SourceDef(CDCDataSourceFactory.IDENTIFIER, "Paimon Source", sourceConfig);
+
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.addAll(additionalSinkConfig);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
+ pipelineConfig.addAll(additionalPipelineConfig);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofApplicationCluster(env);
+ composer.compose(pipelineDef);
+ return env.executeAsync();
+ }
+
+ protected void insertInto(TableId tableId, String... rows) throws Exception {
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s.%s VALUES " + String.join(", ", rows),
+ tableId.getSchemaName(),
+ tableId.getTableName()))
+ .await();
+ }
+
+ protected void waitUtilExpectedResultInternal(TableId tableId, List<String> expectedResult)
+ throws InterruptedException {
+ long stopTime = System.currentTimeMillis() + TIMEOUT_MILLIS;
+ List<String> actualResult;
+ do {
+ actualResult = getActualResult(tableId);
+ Collections.sort(actualResult);
+ if (expectedResult.equals(actualResult)) {
+ return;
+ }
+
+ Thread.sleep(CHECK_INTERVAL_MILLIS);
+ } while (System.currentTimeMillis() < stopTime);
+ assertThat(actualResult).containsExactlyInAnyOrderElementsOf(expectedResult);
+ }
+
+ protected abstract List<String> getActualResult(TableId tableId);
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceSavepointITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceSavepointITCase.java
new file mode 100644
index 0000000000..08e322f0a3
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/CDCSourceSavepointITCase.java
@@ -0,0 +1,217 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The IT Case that verifies the checkpoint-restore functionalities for the cdc source. */
+public class CDCSourceSavepointITCase extends CDCSourceITCaseBase {
+ private static final TestTable[] TEST_TABLES =
+ new TestTable[] {
+ new TestTable(
+ "test1",
+ "test_table1",
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("string_value", DataTypes.STRING())
+ .primaryKey("id")
+ .build(),
+ id -> String.format("(%d, 'test_string_%s')", id, id),
+ id -> String.format("%d, test_string_%s", id, id)),
+ new TestTable(
+ "test2",
+ "test_table2",
+ Schema.newBuilder()
+ .column("append_index", DataTypes.INT())
+ .column("int_value", DataTypes.INT())
+ .build(),
+ id -> String.format("(%d, %d)", id, id),
+ id -> String.format("%d, %d", id, id)),
+ };
+
+ @TempDir private java.nio.file.Path savepointFolder;
+
+ private final ByteArrayOutputStream outCaptor = new ByteArrayOutputStream();
+ private final PrintStream standardOut = System.out;
+
+ @BeforeEach
+ public void initialize() throws Exception {
+ super.initialize();
+ for (TestTable testTable : TEST_TABLES) {
+ catalog.createDatabase(testTable.database, true);
+ catalog.createTable(testTable.identifier(), testTable.schema, true);
+ }
+
+ // Take over STDOUT as we need to check the output of values sink
+ System.setOut(new PrintStream(outCaptor));
+ }
+
+ @AfterEach
+ @Override
+ public void afterEach() throws Exception {
+ System.setOut(standardOut);
+ super.afterEach();
+
+ for (TestTable testTable : TEST_TABLES) {
+ catalog.dropTable(testTable.identifier(), true);
+ }
+
+ for (TestTable testTable : TEST_TABLES) {
+ catalog.dropDatabase(testTable.database, true, false);
+ }
+ }
+
+ @Test
+ public void testRescaleAndRecoverFromSavepoint() throws Exception {
+ Configuration sinkConfig = new Configuration();
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 2);
+
+ jobClient = createAndExecutePipeline(env, sinkConfig, pipelineConfig);
+
+ insertInto(TEST_TABLES[0], 0, 10);
+ insertInto(TEST_TABLES[1], 0, 10);
+
+ waitUtilExpectedResult(TEST_TABLES[0], 10);
+ waitUtilExpectedResult(TEST_TABLES[1], 10);
+
+ String savepointPath =
+ jobClient
+ .stopWithSavepoint(
+ false,
+ savepointFolder.toAbsolutePath().toString(),
+ SavepointFormatType.DEFAULT)
+ .get(1, TimeUnit.MINUTES);
+
+ // Before recovering from the savepoint, the following cases are constructed to be verified:
+ // 1. Change the parallelism of the source
+ // 2. Delete a table that was once synchronized
+
+ catalog.dropTable(TEST_TABLES[0].identifier(), false);
+ insertInto(TEST_TABLES[1], 10, 20);
+
+ org.apache.flink.configuration.Configuration conf =
+ new org.apache.flink.configuration.Configuration();
+ SavepointRestoreSettings savepointRestoreSettings =
+ SavepointRestoreSettings.forPath(savepointPath, false);
+ SavepointRestoreSettings.toConfiguration(savepointRestoreSettings, conf);
+ StreamExecutionEnvironment restoredEnv =
+ StreamExecutionEnvironment.getExecutionEnvironment(conf);
+
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ jobClient = createAndExecutePipeline(restoredEnv, sinkConfig, pipelineConfig);
+
+ waitUtilExpectedResult(TEST_TABLES[1], 20);
+
+ insertInto(TEST_TABLES[1], 20, 30);
+
+ waitUtilExpectedResult(TEST_TABLES[1], 30);
+
+ waitUtilExpectedResult(TEST_TABLES[0], 10);
+ // The deleted Paimon source table does not result in deleting the target table.
+ assertThat(outCaptor.toString().split("\n")).noneMatch(x -> x.contains("DropTableEvent"));
+ }
+
+ private void insertInto(TestTable testTable, int from, int to) throws Exception {
+ List<String> rows = new ArrayList<>();
+ for (int i = from; i < to; i++) {
+ rows.add(testTable.sqlLiteralGenerator.apply(i));
+ }
+ insertInto(testTable.tableId(), rows.toArray(new String[0]));
+ }
+
+ private void waitUtilExpectedResult(TestTable testTable, int endNum)
+ throws InterruptedException {
+ List<String> expectedResult = new ArrayList<>();
+ for (int i = 0; i < endNum; i++) {
+ expectedResult.add(
+ String.format(
+ "DataChangeEvent{tableId=%s, before=[], after=[%s], op=INSERT, meta=()}",
+ testTable.tableId(), testTable.valuesDataGenerator.apply(i)));
+ }
+ waitUtilExpectedResultInternal(testTable.tableId(), expectedResult);
+ }
+
+ @Override
+ protected List<String> getActualResult(TableId tableId) {
+ return Arrays.stream(outCaptor.toString().split("\n"))
+ .filter(
+ x ->
+ x.contains(
+ String.format(
+ "DataChangeEvent{tableId=%s",
+ tableId.getSchemaName())))
+ .map(x -> x.substring(x.indexOf("DataChangeEvent")))
+ .collect(Collectors.toList());
+ }
+
+ private static class TestTable {
+ private final String database;
+ private final String tableName;
+ private final Schema schema;
+ private final Function<Integer, String> sqlLiteralGenerator;
+ private final Function<Integer, String> valuesDataGenerator;
+
+ private TestTable(
+ String database,
+ String tableName,
+ Schema schema,
+ Function<Integer, String> sqlLiteralGenerator,
+ Function<Integer, String> valuesDataGenerator) {
+ this.database = database;
+ this.tableName = tableName;
+ this.schema = schema;
+ this.sqlLiteralGenerator = sqlLiteralGenerator;
+ this.valuesDataGenerator = valuesDataGenerator;
+ }
+
+ private Identifier identifier() {
+ return Identifier.create(database, tableName);
+ }
+
+ private TableId tableId() {
+ return TableId.tableId(database, tableName);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplitSerializerTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplitSerializerTest.java
new file mode 100644
index 0000000000..901c7d0b1a
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplitSerializerTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.source.DataSplit;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newFile;
+import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link TableAwareFileStoreSourceSplit} and {@link
+ * TableAwareFileStoreSourceSplit.Serializer}.
+ */
+public class TableAwareFileStoreSourceSplitSerializerTest {
+ @Test
+ public void test() throws Exception {
+ Identifier identifier = Identifier.create("test_database", "test_table");
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(row(1))
+ .withBucket(2)
+ .withDataFiles(Arrays.asList(newFile(0), newFile(1)))
+ .isStreaming(false)
+ .rawConvertible(false)
+ .withBucketPath("/temp/2") // not used
+ .build();
+ TableAwareFileStoreSourceSplit split =
+ new TableAwareFileStoreSourceSplit("split-1", dataSplit, 0L, identifier, null, 1L);
+
+ TableAwareFileStoreSourceSplit.Serializer serializer =
+ new TableAwareFileStoreSourceSplit.Serializer();
+ byte[] serialized = serializer.serialize(split);
+ TableAwareFileStoreSourceSplit deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+ assertThat(deserialized).isEqualTo(split);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpointSerializerTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpointSerializerTest.java
new file mode 100644
index 0000000000..84906edd73
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCCheckpointSerializerTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.enumerator;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.table.source.DataSplit;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newFile;
+import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link CDCCheckpoint.Serializer}. */
+public class CDCCheckpointSerializerTest {
+ @Test
+ public void test() throws Exception {
+ Identifier identifier = Identifier.create("test_database", "test_table");
+ List<TableAwareFileStoreSourceSplit> splits = new ArrayList<>();
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(row(1))
+ .withBucket(2)
+ .withDataFiles(Arrays.asList(newFile(0), newFile(1)))
+ .isStreaming(false)
+ .rawConvertible(false)
+ .withBucketPath("/temp/2") // not used
+ .build();
+ splits.add(new TableAwareFileStoreSourceSplit("1", dataSplit, 0, identifier, null, 1L));
+
+ Map<Identifier, Long> nextSnapshotIdMap = new HashMap<>();
+ nextSnapshotIdMap.put(identifier, 3L);
+
+ CDCCheckpoint checkpoint = new CDCCheckpoint(splits, nextSnapshotIdMap);
+
+ CDCCheckpoint.Serializer serializer = new CDCCheckpoint.Serializer();
+ CDCCheckpoint newCheckpoint =
+ serializer.deserialize(serializer.getVersion(), serializer.serialize(checkpoint));
+
+ assertThat(newCheckpoint).isEqualTo(checkpoint);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumeratorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumeratorTest.java
new file mode 100644
index 0000000000..70d2505554
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/enumerator/CDCSourceEnumeratorTest.java
@@ -0,0 +1,530 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.enumerator;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.pipeline.cdc.CDCOptions;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.flink.source.FileSplitEnumeratorTestBase;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataFilePlan;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext;
+import org.apache.flink.util.Preconditions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.UUID;
+
+import static org.apache.paimon.flink.pipeline.cdc.CDCOptions.toCDCOption;
+import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests for {@link CDCSourceEnumerator}. */
+public class CDCSourceEnumeratorTest
+ extends FileSplitEnumeratorTestBase<TableAwareFileStoreSourceSplit> {
+ @TempDir public java.nio.file.Path warehouseFolder;
+
+ private static final String DATABASE = "default";
+ private static final String TABLE = "test_table";
+ private static final int NUM_TABLES = 20;
+ private static final Schema SCHEMA =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .option("bucket", String.valueOf(NUM_TABLES))
+ .option("bucket-key", "a")
+ .build();
+
+ private Catalog catalog;
+
+ @BeforeEach
+ public void beforeEach() throws Exception {
+ Options options = new Options();
+ options.setString("warehouse", warehouseFolder.toAbsolutePath().toString());
+ CatalogContext catalogContext = CatalogContext.create(options);
+ catalog = CatalogFactory.createCatalog(catalogContext);
+ catalog.createDatabase(DATABASE, true);
+ for (int i = 0; i < NUM_TABLES; i++) {
+ catalog.createTable(Identifier.create(DATABASE, TABLE + i), SCHEMA, false);
+ }
+ }
+
+ @Test
+ public void testSplitAllocationIsOrdered() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(1);
+
+ List<TableAwareFileStoreSourceSplit> initialSplits = new ArrayList<>();
+ for (int i = 1; i <= 4; i++) {
+ initialSplits.add(createSnapshotSplit(i, 0, Collections.emptyList()));
+ }
+ List<TableAwareFileStoreSourceSplit> expectedSplits = new ArrayList<>(initialSplits);
+ final CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(initialSplits)
+ .setDiscoveryInterval(3)
+ .withSplitMaxPerTask(1)
+ .build();
+
+ // The first time split is allocated, split1 and split2 should be allocated
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(0, "test-host");
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ List<TableAwareFileStoreSourceSplit> assignedSplits =
+ assignments.get(0).getAssignedSplits();
+ assertThat(assignedSplits).hasSameElementsAs(expectedSplits.subList(0, 2));
+
+ // split1 and split2 is added back
+ enumerator.addSplitsBack(assignedSplits, 0);
+ context.getSplitAssignments().clear();
+ assertThat(context.getSplitAssignments()).isEmpty();
+
+ // The split is allocated for the second time, and split1 is allocated first
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(0, "test-host");
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ assignedSplits = assignments.get(0).getAssignedSplits();
+ assertThat(assignedSplits).hasSameElementsAs(expectedSplits.subList(0, 2));
+
+ // continuing to allocate split
+ context.getSplitAssignments().clear();
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(0, "test-host");
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ assignedSplits = assignments.get(0).getAssignedSplits();
+ assertThat(assignedSplits).hasSameElementsAs(expectedSplits.subList(2, 4));
+ }
+
+ @Test
+ public void testSplitWithBatch() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(1);
+
+ List<TableAwareFileStoreSourceSplit> initialSplits = new ArrayList<>();
+ for (int i = 1; i <= 18; i++) {
+ initialSplits.add(createSnapshotSplit(i, i, Collections.emptyList()));
+ }
+ final CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(initialSplits)
+ .setDiscoveryInterval(3)
+ .withSplitMaxPerTask(1)
+ .build();
+
+ // The first time split is allocated, split1 and split2 should be allocated
+ enumerator.handleSplitRequest(0, "test-host");
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(assignments.get(0).getAssignedSplits()).hasSize(1);
+
+ // test second batch assign
+ enumerator.handleSplitRequest(0, "test-host");
+
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(assignments.get(0).getAssignedSplits()).hasSize(2);
+
+ // test third batch assign
+ enumerator.handleSplitRequest(0, "test-host");
+
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(assignments.get(0).getAssignedSplits()).hasSize(3);
+ }
+
+ @Test
+ public void testSplitAllocationIsFair() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(1);
+
+ List<TableAwareFileStoreSourceSplit> initialSplits = new ArrayList<>();
+ for (int i = 1; i <= 2; i++) {
+ initialSplits.add(createSnapshotSplit(i, 0, Collections.emptyList()));
+ initialSplits.add(createSnapshotSplit(i, 1, Collections.emptyList()));
+ }
+
+ List<TableAwareFileStoreSourceSplit> expectedSplits = new ArrayList<>(initialSplits);
+
+ final CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(initialSplits)
+ .setDiscoveryInterval(3)
+ .withSplitMaxPerTask(1)
+ .build();
+
+ // each time a split is allocated from bucket-0 and bucket-1
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(0, "test-host");
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ List<TableAwareFileStoreSourceSplit> assignedSplits =
+ assignments.get(0).getAssignedSplits();
+ assertThat(assignedSplits).hasSameElementsAs(expectedSplits.subList(0, 2));
+
+ // clear assignments
+ context.getSplitAssignments().clear();
+ assertThat(context.getSplitAssignments()).isEmpty();
+
+ // continuing to allocate the rest splits
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(0, "test-host");
+ assignments = context.getSplitAssignments();
+ // Only subtask-0 is allocated.
+ assertThat(assignments).containsOnlyKeys(0);
+ assignedSplits = assignments.get(0).getAssignedSplits();
+ assertThat(assignedSplits).hasSameElementsAs(expectedSplits.subList(2, 4));
+ }
+
+ @Test
+ public void testRemoveReadersAwaitSuccessful() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(2);
+
+ TreeMap<Long, TableScan.Plan> results = new TreeMap<>();
+ StreamDataTableScan scan = new MockScan(results);
+ CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(Collections.emptyList())
+ .setDiscoveryInterval(1)
+ .setScan(scan)
+ .withSplitMaxPerTask(1)
+ .unawareBucket(true)
+ .build();
+ enumerator.start();
+ enumerator.handleSplitRequest(1, "test-host");
+
+ long snapshot = 0;
+ List<DataSplit> splits = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ splits.add(createDataSplit(snapshot, i, Collections.emptyList()));
+ }
+ results.put(1L, new DataFilePlan(splits));
+
+ context.registeredReaders().remove(1);
+ // assign to task 0
+ assertThatCode(() -> enumerator.handleSplitRequest(0, "test-host"))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testSnapshotEnumerator() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(2);
+
+ TreeMap<Long, TableScan.Plan> results = new TreeMap<>();
+ MockScan scan = new MockScan(results);
+ CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(Collections.emptyList())
+ .setDiscoveryInterval(1)
+ .setScan(scan)
+ .withSplitMaxPerTask(1)
+ .build();
+ enumerator.start();
+
+ long snapshot = 0;
+ List<DataSplit> splits = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ splits.add(createDataSplit(snapshot, i, Collections.emptyList()));
+ }
+ results.put(1L, new DataFilePlan(splits));
+ context.triggerAllActions();
+
+ // assign to task 0
+ enumerator.handleSplitRequest(0, "test-host");
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(toDataSplits(assignments.get(0).getAssignedSplits()))
+ .containsExactly(splits.get(0));
+
+ // assign to task 0
+ enumerator.handleSplitRequest(0, "test-host");
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(toDataSplits(assignments.get(0).getAssignedSplits()))
+ .containsExactly(splits.get(0), splits.get(1));
+
+ // always assign same table split to same task
+ enumerator.handleSplitRequest(1, "test-host");
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsKey(0);
+ assertThat(toDataSplits(assignments.get(0).getAssignedSplits()))
+ .containsExactly(splits.get(0), splits.get(1));
+ }
+
+ @Test
+ public void testAssignSameTableSplitToSameTask() {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(2);
+
+ TreeMap<Long, TableScan.Plan> results = new TreeMap<>();
+ MockScan scan = new MockScan(results);
+ CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(Collections.emptyList())
+ .setDiscoveryInterval(1)
+ .setScan(scan)
+ .withSplitMaxPerTask(1)
+ .build();
+ enumerator.start();
+
+ long snapshot = 0;
+ List<DataSplit> splits = new ArrayList<>();
+ splits.add(createDataSplit(snapshot, 0, Collections.emptyList()));
+ results.put(1L, new DataFilePlan(splits));
+ context.triggerAllActions();
+
+ enumerator.handleSplitRequest(0, "test-host");
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(toDataSplits(assignments.get(0).getAssignedSplits()))
+ .containsExactly(splits.get(0));
+
+ splits.add(createDataSplit(snapshot, 1, Collections.emptyList()));
+ results.put(1L, new DataFilePlan(splits.subList(1, 2)));
+ context.triggerAllActions();
+
+ enumerator.handleSplitRequest(1, "test-host");
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsOnlyKeys(0, 1);
+ assertThat(toDataSplits(assignments.get(0).getAssignedSplits()))
+ .containsExactly(splits.get(0));
+ assertThat(toDataSplits(assignments.get(1).getAssignedSplits()))
+ .containsExactly(splits.get(1));
+ }
+
+ @Test
+ public void testEnumeratorSnapshotState() throws Exception {
+ final TestingSplitEnumeratorContext<TableAwareFileStoreSourceSplit> context =
+ getSplitEnumeratorContext(1);
+
+ final CDCSourceEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(Collections.emptyList())
+ .setDiscoveryInterval(3)
+ .withSplitMaxPerTask(1)
+ .build();
+
+ List<TableAwareFileStoreSourceSplit> splits = new ArrayList<>();
+ for (int i = 1; i <= 2; i++) {
+ Identifier identifier = Identifier.create(DATABASE, TABLE + 0);
+ FileStoreTable fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
+ enumerator.setScan(identifier, fileStoreTable, new MockScan(new TreeMap<>()));
+ splits.add(createSnapshotSplit(i, 0, Collections.emptyList()));
+ }
+ enumerator.addSplits(splits);
+ enumerator.handleSplitRequest(0, "test-host");
+
+ Map<
+ Integer,
+ TestingSplitEnumeratorContext.SplitAssignmentState<
+ TableAwareFileStoreSourceSplit>>
+ assignments = context.getSplitAssignments();
+ assertThat(assignments).containsOnlyKeys(0);
+ assertThat(assignments.get(0).getAssignedSplits()).containsExactly(splits.get(0));
+ CDCCheckpoint checkpoint = enumerator.snapshotState(1L);
+ assertThat(checkpoint.getSplits()).containsExactly(splits.get(1));
+ }
+
+ private class Builder {
+ protected SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context;
+ protected Collection<TableAwareFileStoreSourceSplit> initialSplits =
+ Collections.emptyList();
+ protected long discoveryInterval = Long.MAX_VALUE;
+
+ protected StreamDataTableScan scan;
+ protected boolean unawareBucket = false;
+
+ protected int splitMaxPerTask = 10;
+
+ public Builder setSplitEnumeratorContext(
+ SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context) {
+ this.context = context;
+ return this;
+ }
+
+ public Builder setInitialSplits(Collection<TableAwareFileStoreSourceSplit> initialSplits) {
+ this.initialSplits = initialSplits;
+ return this;
+ }
+
+ public Builder setDiscoveryInterval(long discoveryInterval) {
+ this.discoveryInterval = discoveryInterval;
+ return this;
+ }
+
+ public Builder setScan(StreamDataTableScan scan) {
+ this.scan = scan;
+ return this;
+ }
+
+ public Builder unawareBucket(boolean unawareBucket) {
+ this.unawareBucket = unawareBucket;
+ return this;
+ }
+
+ public Builder withSplitMaxPerTask(int splitMaxPerTask) {
+ this.splitMaxPerTask = splitMaxPerTask;
+ return this;
+ }
+
+ public CDCSourceEnumerator build() {
+ Options options = new Options();
+ options.setString("warehouse", warehouseFolder.toAbsolutePath().toString());
+ options.set(CoreOptions.SCAN_MAX_SPLITS_PER_TASK, splitMaxPerTask);
+ Configuration cdcConfig = new Configuration();
+ cdcConfig.set(toCDCOption(CDCOptions.DATABASE), DATABASE);
+
+ Map<Identifier, Long> nextSnapshotIdMap = new HashMap<>();
+ for (TableAwareFileStoreSourceSplit split : initialSplits) {
+ nextSnapshotIdMap.put(split.getIdentifier(), 1L);
+ }
+
+ TestCDCSourceEnumerator enumerator;
+ try {
+ enumerator =
+ new TestCDCSourceEnumerator(
+ context,
+ new org.apache.flink.configuration.Configuration(),
+ discoveryInterval,
+ CatalogContext.create(options),
+ cdcConfig,
+ new CDCCheckpoint(initialSplits, nextSnapshotIdMap));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (scan != null) {
+ try {
+ for (String tableName : catalog.listTables(DATABASE)) {
+ FileStoreTable table =
+ (FileStoreTable)
+ catalog.getTable(Identifier.create(DATABASE, tableName));
+ enumerator.setScan(Identifier.create(DATABASE, tableName), table, scan);
+ }
+ } catch (Catalog.DatabaseNotExistException | Catalog.TableNotExistException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return enumerator;
+ }
+ }
+
+ @Override
+ protected TableAwareFileStoreSourceSplit createSnapshotSplit(
+ int snapshotId, int bucket, List<DataFileMeta> files, int... partitions) {
+ Identifier identifier = Identifier.create(DATABASE, TABLE + bucket);
+ return new TableAwareFileStoreSourceSplit(
+ UUID.randomUUID().toString(),
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(row(partitions))
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(true)
+ .withBucketPath("/temp/xxx") // not used
+ .build(),
+ 0,
+ identifier,
+ null,
+ 1L);
+ }
+
+ private static class TestCDCSourceEnumerator extends CDCSourceEnumerator {
+
+ public TestCDCSourceEnumerator(
+ SplitEnumeratorContext<TableAwareFileStoreSourceSplit> context,
+ org.apache.flink.configuration.Configuration flinkConfig,
+ long discoveryInterval,
+ CatalogContext catalogContext,
+ Configuration cdcConfig,
+ @Nullable CDCCheckpoint checkpoint) {
+ super(context, flinkConfig, discoveryInterval, catalogContext, cdcConfig, checkpoint);
+ }
+
+ @Override
+ protected TableAwareFileStoreSourceSplit toTableAwareSplit(
+ String splitId,
+ Split split,
+ FileStoreTable table,
+ Identifier identifier,
+ @Nullable Long lastSchemaId) {
+ Preconditions.checkState(split instanceof DataSplit);
+ return new TableAwareFileStoreSourceSplit(
+ splitId, split, 0, identifier, lastSchemaId, 1L);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
new file mode 100644
index 0000000000..e12ff8f19e
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
@@ -0,0 +1,575 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.reader;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.pipeline.cdc.source.CDCSource;
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+import org.apache.paimon.flink.source.FileStoreSourceReaderTest.DummyMetricGroup;
+import org.apache.paimon.flink.source.TestChangelogDataReadWrite;
+import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.utils.RecordWriter;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.OperationType;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.reader.BulkFormat.RecordIterator;
+import org.apache.flink.connector.file.src.util.RecordAndPosition;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.paimon.flink.LogicalTypeConversion.toDataType;
+import static org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newSourceSplit;
+import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CDCSourceSplitReader}. */
+public class CDCSourceSplitReaderTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private static final String DATABASE = "default";
+ private static final String TABLE = "test_table";
+ private static final Schema SCHEMA =
+ new Schema(
+ toDataType(
+ new RowType(
+ Collections.singletonList(
+ new RowType.RowField("v", new BigIntType()))))
+ .getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ null);
+
+ private String tablePath;
+
+ @BeforeEach
+ public void beforeEach() throws Exception {
+ Options options = new Options();
+ options.setString("warehouse", tempDir.toUri().toString());
+ CatalogContext catalogContext = CatalogContext.create(options);
+ Catalog catalog = CatalogFactory.createCatalog(catalogContext);
+ catalog.createDatabase(DATABASE, true);
+ catalog.createTable(Identifier.create(DATABASE, TABLE), SCHEMA, false);
+ tablePath = catalog.getTable(Identifier.create(DATABASE, TABLE)).options().get("path");
+ }
+
+ @Test
+ public void testPrimaryKey() throws Exception {
+ innerTestOnce(0);
+ }
+
+ @Test
+ public void testPrimaryKeySkip() throws Exception {
+ innerTestOnce(4);
+ }
+
+ @Test
+ public void testSplitReaderWakeupAble() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input);
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files, 0));
+ reader.fetch();
+
+ Thread thread =
+ new Thread(
+ () -> {
+ try {
+ // block on object pool
+ reader.fetch();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ thread.start();
+ thread.join(20000);
+ assertThat(thread.isAlive()).isTrue();
+ reader.wakeUp();
+ thread.join(15000);
+ assertThat(thread.isAlive()).isFalse();
+ }
+
+ private CDCSourceSplitReader createReader(TableRead tableRead) {
+ return new TestCDCSourceSplitReader(
+ new FileStoreSourceReaderMetrics(new DummyMetricGroup()), tableRead);
+ }
+
+ private void innerTestOnce(int skip) throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input);
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files, skip));
+
+ RecordsWithSplitIds<RecordIterator<Event>> records = reader.fetch();
+
+ List<Tuple2<RowKind, Long>> expected =
+ input.stream()
+ .map(t -> new Tuple2<>(RowKind.INSERT, t.f1))
+ .collect(Collectors.toList());
+
+ List<Tuple2<RowKind, Long>> result = readRecords(records, "id1", skip);
+ assertThat(result).isEqualTo(expected.subList(skip, expected.size()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testPrimaryKeyWithDelete() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input = kvs();
+ RecordWriter<KeyValue> writer = rw.createMergeTreeWriter(row(1), 0);
+ for (Tuple2<Long, Long> tuple2 : input) {
+ writer.write(
+ new KeyValue()
+ .replace(
+ GenericRow.of(tuple2.f0),
+ org.apache.paimon.types.RowKind.INSERT,
+ GenericRow.of(tuple2.f1)));
+ }
+ writer.write(
+ new KeyValue()
+ .replace(
+ GenericRow.of(222L),
+ org.apache.paimon.types.RowKind.DELETE,
+ GenericRow.of(333L)));
+ List<DataFileMeta> files = writer.prepareCommit(true).newFilesIncrement().newFiles();
+ writer.close();
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files, true));
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+
+ List<Tuple2<RowKind, Long>> expected =
+ input.stream()
+ .map(t -> new Tuple2<>(RowKind.INSERT, t.f1))
+ .collect(Collectors.toList());
+ expected.add(new Tuple2<>(RowKind.DELETE, 333L));
+
+ List<Tuple2<RowKind, Long>> result = readRecords(records, "id1", 0);
+ assertThat(result).isEqualTo(expected);
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testMultipleBatchInSplit() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input1 = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input1);
+
+ List<Tuple2<Long, Long>> input2 = kvs(6);
+ List<DataFileMeta> files2 = rw.writeFiles(row(1), 0, input2);
+ files.addAll(files2);
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files));
+
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 0,
+ input1.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 6,
+ input2.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testRestore() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input);
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files, 3));
+
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 3,
+ input.subList(3, input.size()).stream()
+ .map(t -> t.f1)
+ .collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testRestoreMultipleBatchInSplit() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input1 = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input1);
+
+ List<Tuple2<Long, Long>> input2 = kvs(6);
+ List<DataFileMeta> files2 = rw.writeFiles(row(1), 0, input2);
+ files.addAll(files2);
+
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files, 7));
+
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 7,
+ Stream.concat(input1.stream(), input2.stream())
+ .skip(7)
+ .map(t -> t.f1)
+ .collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testMultipleSplits() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input1 = kvs();
+ List<DataFileMeta> files1 = rw.writeFiles(row(1), 0, input1);
+ assignSplit(reader, newSourceSplit("id1", row(1), 0, files1));
+
+ List<Tuple2<Long, Long>> input2 = kvs();
+ List<DataFileMeta> files2 = rw.writeFiles(row(2), 1, input2);
+ assignSplit(reader, newSourceSplit("id2", row(2), 1, files2));
+
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 0,
+ input1.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id2",
+ 0,
+ input2.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id2", "id2", 0, null);
+
+ reader.close();
+ }
+
+ @Test
+ public void testPauseOrResumeSplits() throws Exception {
+ TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+
+ List<Tuple2<Long, Long>> input1 = kvs();
+ List<DataFileMeta> files = rw.writeFiles(row(1), 0, input1);
+
+ List<Tuple2<Long, Long>> input2 = kvs(6);
+ List<DataFileMeta> files2 = rw.writeFiles(row(1), 0, input2);
+ files.addAll(files2);
+
+ TableAwareFileStoreSourceSplit split1 = newSourceSplit("id1", row(1), 0, files);
+ assignSplit(reader, split1);
+
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 0,
+ input1.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ // pause split1
+ reader.pauseOrResumeSplits(Collections.singletonList(split1), Collections.emptyList());
+ records = reader.fetch();
+ assertRecords(records, null, null, 0, Collections.emptyList());
+
+ // assign next split
+ List<Tuple2<Long, Long>> input3 = kvs(12);
+ List<DataFileMeta> files3 = rw.writeFiles(row(1), 0, input3);
+ TableAwareFileStoreSourceSplit split2 = newSourceSplit("id2", row(1), 0, files3);
+ assignSplit(reader, split2);
+
+ records = reader.fetch();
+ assertRecords(records, null, null, 0, Collections.emptyList());
+
+ // resume split1
+ reader.pauseOrResumeSplits(Collections.emptyList(), Collections.singletonList(split1));
+ records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id1",
+ 6,
+ input2.stream().map(t -> t.f1).collect(Collectors.toList()));
+
+ records = reader.fetch();
+ assertRecords(records, "id1", "id1", 0, null);
+
+ // fetch split2
+ records = reader.fetch();
+ assertRecords(
+ records,
+ null,
+ "id2",
+ 0,
+ input3.stream().map(t -> t.f1).collect(Collectors.toList()));
+ records = reader.fetch();
+ assertRecords(records, "id2", "id2", 0, null);
+
+ reader.close();
+ }
+
+ private void assertRecords(
+ RecordsWithSplitIds<BulkFormat.RecordIterator<Event>> records,
+ String finishedSplit,
+ String nextSplit,
+ long startRecordSkipCount,
+ List<Long> expected) {
+ if (finishedSplit != null) {
+ assertThat(records.finishedSplits()).isEqualTo(Collections.singleton(finishedSplit));
+ return;
+ }
+
+ List<Tuple2<RowKind, Long>> result = readRecords(records, nextSplit, startRecordSkipCount);
+ assertThat(result.stream().map(t -> t.f1).collect(Collectors.toList())).isEqualTo(expected);
+ }
+
+ private List<Tuple2<RowKind, Long>> readRecords(
+ RecordsWithSplitIds<RecordIterator<Event>> records,
+ String nextSplit,
+ long startRecordSkipCount) {
+ assertThat(records.finishedSplits()).isEmpty();
+ assertThat(records.nextSplit()).isEqualTo(nextSplit);
+ List<Tuple2<RowKind, Long>> result = new ArrayList<>();
+ RecordIterator<Event> iterator;
+ while ((iterator = records.nextRecordFromSplit()) != null) {
+ RecordAndPosition<Event> record;
+ while ((record = iterator.next()) != null) {
+ assertThat(record.getRecord()).isInstanceOf(DataChangeEvent.class);
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) record.getRecord();
+ if (dataChangeEvent.op() == OperationType.INSERT) {
+ result.add(
+ new Tuple2<>(
+ RowKind.INSERT,
+ ((DataChangeEvent) record.getRecord()).after().getLong(0)));
+ } else {
+ result.add(
+ new Tuple2<>(
+ RowKind.DELETE,
+ ((DataChangeEvent) record.getRecord()).before().getLong(0)));
+ }
+ assertThat(record.getRecordSkipCount()).isEqualTo(++startRecordSkipCount);
+ }
+ }
+ records.recycle();
+ return result;
+ }
+
+ private List<Tuple2<Long, Long>> kvs() {
+ return kvs(0);
+ }
+
+ private List<Tuple2<Long, Long>> kvs(long keyBase) {
+ List<Tuple2<Long, Long>> kvs = new ArrayList<>();
+ kvs.add(new Tuple2<>(keyBase + 1L, 1L));
+ kvs.add(new Tuple2<>(keyBase + 2L, 2L));
+ kvs.add(new Tuple2<>(keyBase + 3L, 2L));
+ kvs.add(new Tuple2<>(keyBase + 4L, -1L));
+ kvs.add(new Tuple2<>(keyBase + 5L, 1L));
+ kvs.add(new Tuple2<>(keyBase + 6L, -2L));
+ return kvs;
+ }
+
+ private void assignSplit(CDCSourceSplitReader reader, TableAwareFileStoreSourceSplit split) {
+ SplitsChange<TableAwareFileStoreSourceSplit> splitsChange =
+ new SplitsAddition<>(Collections.singletonList(split));
+ reader.handleSplitsChanges(splitsChange);
+ }
+
+ public static TableAwareFileStoreSourceSplit newSourceSplit(
+ String id, BinaryRow partition, int bucket, List<DataFileMeta> files) {
+ return newSourceSplit(id, partition, bucket, files, false, 0);
+ }
+
+ public static TableAwareFileStoreSourceSplit newSourceSplit(
+ String id,
+ BinaryRow partition,
+ int bucket,
+ List<DataFileMeta> files,
+ boolean isIncremental) {
+ return newSourceSplit(id, partition, bucket, files, isIncremental, 0);
+ }
+
+ public static TableAwareFileStoreSourceSplit newSourceSplit(
+ String id,
+ BinaryRow partition,
+ int bucket,
+ List<DataFileMeta> files,
+ long recordsToSkip) {
+ return newSourceSplit(id, partition, bucket, files, false, recordsToSkip);
+ }
+
+ public static TableAwareFileStoreSourceSplit newSourceSplit(
+ String id,
+ BinaryRow partition,
+ int bucket,
+ List<DataFileMeta> files,
+ boolean isIncremental,
+ long recordsToSkip) {
+ DataSplit split =
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(partition)
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(isIncremental)
+ .rawConvertible(false)
+ .withBucketPath("/temp/" + bucket) // no used
+ .build();
+ return new TableAwareFileStoreSourceSplit(
+ id, split, recordsToSkip, Identifier.create(DATABASE, TABLE), 1L, 1L);
+ }
+
+ private static class TestCDCSourceSplitReader extends CDCSourceSplitReader {
+ private final TableRead tableRead;
+
+ public TestCDCSourceSplitReader(FileStoreSourceReaderMetrics metrics, TableRead tableRead) {
+ super(metrics, new TestTableManager(tableRead));
+ this.tableRead = tableRead;
+ }
+
+ @Override
+ protected LazyRecordReader createLazyRecordReader(Split split) {
+ return new TestLazyRecordReader(split, tableRead);
+ }
+ }
+
+ private static class TestLazyRecordReader extends CDCSourceSplitReader.LazyRecordReader {
+ private final TableRead tableRead;
+ private RecordReader<InternalRow> lazyRecordReader;
+
+ protected TestLazyRecordReader(Split split, TableRead tableRead) {
+ super(split, null, new TestTableManager(tableRead));
+ this.tableRead = tableRead;
+ }
+
+ @Override
+ public RecordReader<InternalRow> recordReader() throws IOException {
+ if (lazyRecordReader == null) {
+ lazyRecordReader = tableRead.createReader(split);
+ }
+ return lazyRecordReader;
+ }
+ }
+
+ private static class TestTableManager extends CDCSource.TableManager {
+ private final TableRead tableRead;
+
+ public TestTableManager(TableRead tableRead) {
+ super(null, null, null);
+ this.tableRead = tableRead;
+ }
+
+ @Override
+ public @Nullable TableSchema getTableSchema(
+ Identifier identifier, @Nullable Long schemaId) {
+ return TableSchema.create(1L, SCHEMA);
+ }
+
+ @Override
+ public TableRead getTableRead(Identifier identifier, TableSchema schema) {
+ return tableRead;
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
index c3d7710097..1955e6c439 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
@@ -345,7 +345,7 @@ public class FileStoreSourceSplitReader
* An empty implementation of {@link RecordsWithSplitIds}. It is used to indicate that the
* {@link FileStoreSourceSplitReader} is paused or wakeup.
*/
- private static class EmptyRecordsWithSplitIds<T> implements RecordsWithSplitIds<T> {
+ public static class EmptyRecordsWithSplitIds<T> implements RecordsWithSplitIds<T> {
@Nullable
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitState.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitState.java
index d8c9596f08..85aacbd708 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitState.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitState.java
@@ -20,7 +20,6 @@ package org.apache.paimon.flink.source;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
import org.apache.flink.connector.file.src.util.RecordAndPosition;
-import org.apache.flink.table.data.RowData;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -40,7 +39,7 @@ public final class FileStoreSourceSplitState {
this.recordsToSkip = split.recordsToSkip();
}
- public void setPosition(RecordAndPosition<RowData> position) {
+ public void setPosition(RecordAndPosition<?> position) {
checkArgument(position.getOffset() == CheckpointedPosition.NO_OFFSET);
this.recordsToSkip = position.getRecordSkipCount();
}
@@ -52,4 +51,14 @@ public final class FileStoreSourceSplitState {
public FileStoreSourceSplit toSourceSplit() {
return split.updateWithRecordsToSkip(recordsToSkip);
}
+
+ @Override
+ public String toString() {
+ return "FileStoreSourceSplitState{"
+ + "split="
+ + split
+ + ", recordsToSkip="
+ + recordsToSkip
+ + '}';
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
index 90c35350eb..051a09e874 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
@@ -19,25 +19,16 @@
package org.apache.paimon.flink.source;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.manifest.PartitionEntry;
-import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.table.source.DataFilePlan;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.EndOfScanException;
-import org.apache.paimon.table.source.SnapshotNotExistPlan;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.table.source.TableScan;
-import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext;
-import org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService;
-import org.apache.flink.runtime.source.coordinator.ExecutorNotifier;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
-import javax.annotation.Nullable;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -46,10 +37,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
-import java.util.concurrent.Callable;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
import static org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext.SplitAssignmentState;
import static org.apache.paimon.io.DataFileTestUtils.row;
@@ -57,7 +46,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for the {@link ContinuousFileSplitEnumerator}. */
-public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBase {
+public class ContinuousFileSplitEnumeratorTest
+ extends FileSplitEnumeratorTestBase<FileStoreSourceSplit> {
@Test
public void testSplitAllocationIsOrdered() {
@@ -871,12 +861,6 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
enumerator.notifyCheckpointComplete(checkpointId);
}
- private void scanNextSnapshot(
- TestingAsyncSplitEnumeratorContext<FileStoreSourceSplit> context) {
- context.workerExecutor.triggerPeriodicScheduledTasks();
- context.triggerAlCoordinatorAction();
- }
-
private static PendingSplitsCheckpoint checkpointWithoutException(
ContinuousFileSplitEnumerator enumerator, long checkpointId) {
try {
@@ -886,26 +870,6 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
}
}
- private static List<DataSplit> toDataSplits(Collection<FileStoreSourceSplit> splits) {
- return splits.stream()
- .map(FileStoreSourceSplit::split)
- .map(split -> (DataSplit) split)
- .collect(Collectors.toList());
- }
-
- private static DataSplit createDataSplit(
- long snapshotId, int bucket, List<DataFileMeta> files) {
- return DataSplit.builder()
- .withSnapshot(snapshotId)
- .withPartition(row(1))
- .withBucket(bucket)
- .withDataFiles(files)
- .isStreaming(true)
- .rawConvertible(false)
- .withBucketPath("") // not used
- .build();
- }
-
private static class Builder {
private SplitEnumeratorContext<FileStoreSourceSplit> context;
private Collection<FileStoreSourceSplit> initialSplits = Collections.emptyList();
@@ -967,108 +931,19 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
}
}
- private static class MockScan implements StreamTableScan {
-
- private final TreeMap<Long, Plan> results;
- private @Nullable Long nextSnapshotId;
- private boolean allowEnd = true;
- private Long nextSnapshotIdForConsumer;
-
- public MockScan(TreeMap<Long, Plan> results) {
- this.results = results;
- this.nextSnapshotId = null;
- this.nextSnapshotIdForConsumer = null;
- }
-
- @Override
- public TableScan withMetricRegistry(MetricRegistry registry) {
- return this;
- }
-
- @Override
- public Plan plan() {
- Map.Entry<Long, Plan> planEntry = results.pollFirstEntry();
- if (planEntry == null) {
- if (allowEnd) {
- throw new EndOfScanException();
- } else {
- return SnapshotNotExistPlan.INSTANCE;
- }
- }
- nextSnapshotId = planEntry.getKey() + 1;
- return planEntry.getValue();
- }
-
- @Override
- public List<PartitionEntry> listPartitionEntries() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Long checkpoint() {
- return nextSnapshotId;
- }
-
- @Override
- public void notifyCheckpointComplete(@Nullable Long nextSnapshot) {
- nextSnapshotIdForConsumer = nextSnapshot;
- }
-
- @Nullable
- @Override
- public Long watermark() {
- return null;
- }
-
- @Override
- public void restore(Long state) {}
-
- public void allowEnd(boolean allowEnd) {
- this.allowEnd = allowEnd;
- }
-
- public Long getNextSnapshotIdForConsumer() {
- return nextSnapshotIdForConsumer;
- }
- }
-
- private static class TestingAsyncSplitEnumeratorContext<SplitT extends SourceSplit>
- extends TestingSplitEnumeratorContext<SplitT> {
-
- private final ManuallyTriggeredScheduledExecutorService workerExecutor;
- private final ExecutorNotifier notifier;
-
- public TestingAsyncSplitEnumeratorContext(int parallelism) {
- super(parallelism);
- this.workerExecutor = new ManuallyTriggeredScheduledExecutorService();
- this.notifier = new ExecutorNotifier(workerExecutor, super.getExecutorService());
- }
-
- @Override
- public <T> void callAsync(Callable<T> callable, BiConsumer<T, Throwable> handler) {
- notifier.notifyReadyAsync(callable, handler);
- }
-
- @Override
- public <T> void callAsync(
- Callable<T> callable,
- BiConsumer<T, Throwable> handler,
- long initialDelay,
- long period) {
- notifier.notifyReadyAsync(callable, handler, initialDelay, period);
- }
-
- public void triggerAllWorkerAction() {
- this.workerExecutor.triggerPeriodicScheduledTasks();
- this.workerExecutor.triggerAll();
- }
-
- public void triggerAlCoordinatorAction() {
- super.triggerAllActions();
- }
-
- public void triggerNextCoordinatorAction() {
- super.getExecutorService().trigger();
- }
+ @Override
+ protected FileStoreSourceSplit createSnapshotSplit(
+ int snapshotId, int bucket, List<DataFileMeta> files, int... partitions) {
+ return new FileStoreSourceSplit(
+ UUID.randomUUID().toString(),
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(row(partitions))
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(true)
+ .withBucketPath("/temp/xxx") // not used
+ .build(),
+ 0);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileSplitEnumeratorTestBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileSplitEnumeratorTestBase.java
index 446cbec0e9..a33def9cf6 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileSplitEnumeratorTestBase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileSplitEnumeratorTestBase.java
@@ -19,26 +19,45 @@
package org.apache.paimon.flink.source;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DataTableScan;
+import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.SnapshotNotExistPlan;
+import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.snapshot.StartingContext;
+import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext;
+import org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService;
+import org.apache.flink.runtime.source.coordinator.ExecutorNotifier;
+import javax.annotation.Nullable;
+
+import java.util.Collection;
import java.util.List;
-import java.util.UUID;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
import static org.apache.paimon.io.DataFileTestUtils.row;
/** Base class for split enumerators. */
-public abstract class FileSplitEnumeratorTestBase {
+public abstract class FileSplitEnumeratorTestBase<SplitT extends FileStoreSourceSplit> {
- protected TestingSplitEnumeratorContext<FileStoreSourceSplit> getSplitEnumeratorContext(
- int parallelism) {
+ protected TestingSplitEnumeratorContext<SplitT> getSplitEnumeratorContext(int parallelism) {
return getSplitEnumeratorContext(parallelism, parallelism);
}
- protected TestingSplitEnumeratorContext<FileStoreSourceSplit> getSplitEnumeratorContext(
+ protected TestingSplitEnumeratorContext<SplitT> getSplitEnumeratorContext(
int parallelism, int registeredReaders) {
- final TestingSplitEnumeratorContext<FileStoreSourceSplit> context =
+ final TestingSplitEnumeratorContext<SplitT> context =
new TestingSplitEnumeratorContext<>(parallelism);
for (int i = 0; i < registeredReaders; i++) {
context.registerReader(i, "test-host");
@@ -46,23 +65,166 @@ public abstract class FileSplitEnumeratorTestBase {
return context;
}
- protected FileStoreSourceSplit createSnapshotSplit(
- int snapshotId, int bucket, List<DataFileMeta> files) {
+ protected SplitT createSnapshotSplit(int snapshotId, int bucket, List<DataFileMeta> files) {
return createSnapshotSplit(snapshotId, bucket, files, 1);
}
- protected FileStoreSourceSplit createSnapshotSplit(
- int snapshotId, int bucket, List<DataFileMeta> files, int... partitions) {
- return new FileStoreSourceSplit(
- UUID.randomUUID().toString(),
- DataSplit.builder()
- .withSnapshot(snapshotId)
- .withPartition(row(partitions))
- .withBucket(bucket)
- .withDataFiles(files)
- .isStreaming(true)
- .withBucketPath("/temp/xxx") // not used
- .build(),
- 0);
+ protected abstract SplitT createSnapshotSplit(
+ int snapshotId, int bucket, List<DataFileMeta> files, int... partitions);
+
+ protected void scanNextSnapshot(TestingAsyncSplitEnumeratorContext<SplitT> context) {
+ context.workerExecutor.triggerPeriodicScheduledTasks();
+ context.triggerAlCoordinatorAction();
+ }
+
+ protected List<DataSplit> toDataSplits(Collection<SplitT> splits) {
+ return splits.stream()
+ .map(SplitT::split)
+ .map(split -> (DataSplit) split)
+ .collect(Collectors.toList());
+ }
+
+ protected static DataSplit createDataSplit(
+ long snapshotId, int bucket, List<DataFileMeta> files) {
+ return DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(row(1))
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(true)
+ .rawConvertible(false)
+ .withBucketPath("") // not used
+ .build();
+ }
+
+ /** A mock {@link StreamTableScan} that can manually specify generated plans. */
+ protected static class MockScan implements StreamDataTableScan {
+
+ private final TreeMap<Long, Plan> results;
+ private @Nullable Long nextSnapshotId;
+ private boolean allowEnd = true;
+ private Long nextSnapshotIdForConsumer;
+
+ public MockScan(TreeMap<Long, Plan> results) {
+ this.results = results;
+ this.nextSnapshotId = null;
+ this.nextSnapshotIdForConsumer = null;
+ }
+
+ @Override
+ public InnerTableScan withFilter(Predicate predicate) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InnerTableScan withMetricRegistry(MetricRegistry registry) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Plan plan() {
+ Map.Entry<Long, Plan> planEntry = results.pollFirstEntry();
+ if (planEntry == null) {
+ if (allowEnd) {
+ throw new EndOfScanException();
+ } else {
+ return SnapshotNotExistPlan.INSTANCE;
+ }
+ }
+ nextSnapshotId = planEntry.getKey() + 1;
+ return planEntry.getValue();
+ }
+
+ @Override
+ public List<PartitionEntry> listPartitionEntries() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Long checkpoint() {
+ return nextSnapshotId;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(@Nullable Long nextSnapshot) {
+ nextSnapshotIdForConsumer = nextSnapshot;
+ }
+
+ @Nullable
+ @Override
+ public Long watermark() {
+ return null;
+ }
+
+ @Override
+ public void restore(Long state) {}
+
+ public void allowEnd(boolean allowEnd) {
+ this.allowEnd = allowEnd;
+ }
+
+ public Long getNextSnapshotIdForConsumer() {
+ return nextSnapshotIdForConsumer;
+ }
+
+ @Override
+ public StartingContext startingContext() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void restore(@Nullable Long nextSnapshotId, boolean scanAllSnapshot) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ /**
+ * A {@link TestingSplitEnumeratorContext} that supports manually controlling asynchronous
+ * calls.
+ */
+ protected static class TestingAsyncSplitEnumeratorContext<SplitT extends SourceSplit>
+ extends TestingSplitEnumeratorContext<SplitT> {
+
+ private final ManuallyTriggeredScheduledExecutorService workerExecutor;
+ private final ExecutorNotifier notifier;
+
+ public TestingAsyncSplitEnumeratorContext(int parallelism) {
+ super(parallelism);
+ this.workerExecutor = new ManuallyTriggeredScheduledExecutorService();
+ this.notifier = new ExecutorNotifier(workerExecutor, super.getExecutorService());
+ }
+
+ @Override
+ public <SplitT> void callAsync(
+ Callable<SplitT> callable, BiConsumer<SplitT, Throwable> handler) {
+ notifier.notifyReadyAsync(callable, handler);
+ }
+
+ @Override
+ public <SplitT> void callAsync(
+ Callable<SplitT> callable,
+ BiConsumer<SplitT, Throwable> handler,
+ long initialDelay,
+ long period) {
+ notifier.notifyReadyAsync(callable, handler, initialDelay, period);
+ }
+
+ public void triggerAllWorkerAction() {
+ this.workerExecutor.triggerPeriodicScheduledTasks();
+ this.workerExecutor.triggerAll();
+ }
+
+ public void triggerAlCoordinatorAction() {
+ super.triggerAllActions();
+ }
+
+ public void triggerNextCoordinatorAction() {
+ super.getExecutorService().trigger();
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/StaticFileStoreSplitEnumeratorTestBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/StaticFileStoreSplitEnumeratorTestBase.java
index f10de11c5d..5c045d0f5e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/StaticFileStoreSplitEnumeratorTestBase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/StaticFileStoreSplitEnumeratorTestBase.java
@@ -19,6 +19,8 @@
package org.apache.paimon.flink.source;
import org.apache.paimon.flink.FlinkConnectorOptions;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -37,13 +39,16 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import static org.apache.paimon.flink.source.StaticFileStoreSource.createSplitAssigner;
+import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.assertj.core.api.Assertions.assertThat;
/** Base test class of {@link StaticFileStoreSplitEnumerator}. */
-public abstract class StaticFileStoreSplitEnumeratorTestBase extends FileSplitEnumeratorTestBase {
+public abstract class StaticFileStoreSplitEnumeratorTestBase
+ extends FileSplitEnumeratorTestBase<FileStoreSourceSplit> {
@Test
public void testDynamicPartitionFilteringAfterStarted() {
@@ -183,4 +188,20 @@ public abstract class StaticFileStoreSplitEnumeratorTestBase extends FileSplitEn
return true;
}
}
+
+ @Override
+ protected FileStoreSourceSplit createSnapshotSplit(
+ int snapshotId, int bucket, List<DataFileMeta> files, int... partitions) {
+ return new FileStoreSourceSplit(
+ UUID.randomUUID().toString(),
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(row(partitions))
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(true)
+ .withBucketPath("/temp/xxx") // not used
+ .build(),
+ 0);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
index 1aca14fa31..cb7a854286 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
@@ -152,10 +152,10 @@ public class TestChangelogDataReadWrite {
return new KeyValueTableRead(() -> read, () -> rawFileRead, null);
}
- public List<DataFileMeta> writeFiles(
- BinaryRow partition, int bucket, List<Tuple2<Long, Long>> kvs) throws Exception {
+ public <T> List<DataFileMeta> writeFiles(
+ BinaryRow partition, int bucket, List<Tuple2<Long, T>> kvs) throws Exception {
RecordWriter<KeyValue> writer = createMergeTreeWriter(partition, bucket);
- for (Tuple2<Long, Long> tuple2 : kvs) {
+ for (Tuple2<Long, T> tuple2 : kvs) {
writer.write(
new KeyValue()
.replace(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumeratorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumeratorTest.java
index 43dd09682b..a42d697160 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumeratorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumeratorTest.java
@@ -27,11 +27,13 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -51,13 +53,16 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
+import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for the {@link AlignedContinuousFileSplitEnumerator}. */
-public class AlignedContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBase {
+public class AlignedContinuousFileSplitEnumeratorTest
+ extends FileSplitEnumeratorTestBase<FileStoreSourceSplit> {
private static final RowType ROW_TYPE =
RowType.of(
@@ -250,4 +255,20 @@ public class AlignedContinuousFileSplitEnumeratorTest extends FileSplitEnumerato
-1);
}
}
+
+ @Override
+ protected FileStoreSourceSplit createSnapshotSplit(
+ int snapshotId, int bucket, List<DataFileMeta> files, int... partitions) {
+ return new FileStoreSourceSplit(
+ UUID.randomUUID().toString(),
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(row(partitions))
+ .withBucket(bucket)
+ .withDataFiles(files)
+ .isStreaming(true)
+ .withBucketPath("/temp/xxx") // not used
+ .build(),
+ 0);
+ }
}