apache · EricJoy2048 · Jan 18, 2024 · Dec 10, 2023 · Dec 10, 2023 · Dec 12, 2023
diff --git a/docs/en/connector-v2/source/Postgre-CDC.md b/docs/en/connector-v2/source/Postgre-CDC.md
diff --git a/plugin-mapping.properties b/plugin-mapping.properties
@@ -115,5 +115,6 @@ seatunnel.source.AmazonSqs = connector-amazonsqs
 seatunnel.sink.AmazonSqs = connector-amazonsqs
 seatunnel.source.Paimon = connector-paimon
 seatunnel.sink.Paimon = connector-paimon
+seatunnel.source.Postgres-CDC = connector-cdc-postgres
 seatunnel.source.Oracle-CDC = connector-cdc-oracle
 seatunnel.sink.Pulsar = connector-pulsar
diff --git a/...ase/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/DataSourceDialect.java b/...ase/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/DataSourceDialect.java
@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.cdc.base.dialect;
 
+import org.apache.seatunnel.api.state.CheckpointListener;
 import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
 import org.apache.seatunnel.connectors.cdc.base.source.enumerator.splitter.ChunkSplitter;
 import org.apache.seatunnel.connectors.cdc.base.source.reader.external.FetchTask;
@@ -32,7 +33,8 @@
  *
  * @param <C> The source config of data source.
  */
-public interface DataSourceDialect<C extends SourceConfig> extends Serializable {
+public interface DataSourceDialect<C extends SourceConfig>
+        extends Serializable, CheckpointListener {
 
     /** Get the name of dialect. */
     String getName();
@@ -51,4 +53,13 @@ public interface DataSourceDialect<C extends SourceConfig> extends Serializable
 
     /** The task context used for fetch task to fetch data from external systems. */
     FetchTask.Context createFetchTaskContext(SourceSplitBase sourceSplitBase, C sourceConfig);
+
+    /**
+     * We have an empty default implementation here because most dialects do not have to implement
+     * the method.
+     *
+     * @see CheckpointListener#notifyCheckpointComplete(long)
+     */
+    @Override
+    default void notifyCheckpointComplete(long checkpointId) throws Exception {}
 }
diff --git a/...base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java b/...base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java
@@ -179,6 +179,7 @@ public SourceReader<T, SourceSplitBase> createReader(SourceReader.Context reader
                                 sourceConfig,
                                 schemaChangeResolver);
         return new IncrementalSourceReader<>(
+                dataSourceDialect,
                 elementsQueue,
                 splitReaderSupplier,
                 createRecordEmitter(sourceConfig, readerContext.getMetricsContext()),

diff --git a/.../java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java b/.../java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java
@@ -21,6 +21,7 @@
 import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
 import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsReportEvent;
 import org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
 import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit;
@@ -68,9 +69,12 @@ public class IncrementalSourceReader<T, C extends SourceConfig>
     private final C sourceConfig;
     private final DebeziumDeserializationSchema<T> debeziumDeserializationSchema;
 
+    private final DataSourceDialect<C> dataSourceDialect;
+
     private final AtomicBoolean needSendSplitRequest = new AtomicBoolean(false);
 
     public IncrementalSourceReader(
+            DataSourceDialect<C> dataSourceDialect,
             BlockingQueue<RecordsWithSplitIds<SourceRecords>> elementsQueue,
             Supplier<IncrementalSourceSplitReader<C>> splitReaderSupplier,
             RecordEmitter<SourceRecords, T, SourceSplitStateBase> recordEmitter,
@@ -84,6 +88,7 @@ public IncrementalSourceReader(
                 recordEmitter,
                 options,
                 context);
+        this.dataSourceDialect = dataSourceDialect;
         this.sourceConfig = sourceConfig;
         this.finishedUnackedSplits = new HashMap<>();
         this.subtaskId = context.getIndexOfSubtask();
@@ -106,7 +111,9 @@ public void pollNext(Collector<T> output) throws Exception {
     }
 
     @Override
-    public void notifyCheckpointComplete(long checkpointId) throws Exception {}
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {
+        dataSourceDialect.notifyCheckpointComplete(checkpointId);
+    }
 
     @Override
     public void addSplits(List<SourceSplitBase> splits) {

diff --git a/...ache/seatunnel/connectors/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java b/...ache/seatunnel/connectors/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java
@@ -60,7 +60,7 @@ public JdbcSourceFetchTaskContext(
 
     @Override
     public TableId getTableId(SourceRecord record) {
-        return null;
+        return SourceRecordUtils.getTableId(record);
     }
 
     @Override
@@ -71,7 +71,7 @@ public boolean isDataChangeRecord(SourceRecord record) {
     @Override
     public boolean isRecordBetween(SourceRecord record, Object[] splitStart, Object[] splitEnd) {
         SeaTunnelRowType splitKeyType =
-                getSplitType(getDatabaseSchema().tableFor(SourceRecordUtils.getTableId(record)));
+                getSplitType(getDatabaseSchema().tableFor(getTableId(record)));
         Object[] key = SourceRecordUtils.getSplitKey(splitKeyType, record, getSchemaNameAdjuster());
         return SourceRecordUtils.splitKeyRangeContains(key, splitStart, splitEnd);
     }

diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>connector-cdc</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-cdc-postgres</artifactId>
+
+    <properties />
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.apache.seatunnel</groupId>
+                <artifactId>connector-jdbc</artifactId>
+                <version>${project.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.seatunnel</groupId>
+                <artifactId>connector-cdc-base</artifactId>
+                <version>${project.version}</version>
+                <scope>compile</scope>
+            </dependency>
+
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-connector-postgres</artifactId>
+                <version>${debezium.version}</version>
+                <scope>compile</scope>
+            </dependency>
+
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-cdc-base</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-postgres</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.postgresql</groupId>
+                    <artifactId>postgresql</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-jdbc</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>2.4</version>
+                <configuration>
+                    <skip>false</skip>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>