[INLONG-7908][Sort] PostgreSQL connector supports parallel read

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfig.java

@@ -38,6 +38,7 @@ public abstract class JdbcSourceConfig extends BaseSourceConfig {
     protected final String username;
     protected final String password;
     protected final List<String> databaseList;
+    protected final List<String> schemaList;
     protected final List<String> tableList;
     protected final int fetchSize;
     protected final String serverTimeZone;
@@ -49,6 +50,7 @@ public abstract class JdbcSourceConfig extends BaseSourceConfig {
     public JdbcSourceConfig(
             StartupOptions startupOptions,
             List<String> databaseList,
+            List<String> schemaList,
             List<String> tableList,
             int splitSize,
             int splitMetaGroupSize,
@@ -87,6 +89,7 @@ public JdbcSourceConfig(
         this.username = username;
         this.password = password;
         this.databaseList = databaseList;
+        this.schemaList = schemaList;
         this.tableList = tableList;
         this.fetchSize = fetchSize;
         this.serverTimeZone = serverTimeZone;

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/dialect/DataSourceDialect.java

@@ -26,6 +26,7 @@
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.state.CheckpointListener;
 
 import java.io.Serializable;
 import java.util.List;
@@ -38,7 +39,7 @@
  * Copy from com.ververica:flink-cdc-base:2.3.0.
  */
 @Experimental
-public interface DataSourceDialect<C extends SourceConfig> extends Serializable {
+public interface DataSourceDialect<C extends SourceConfig> extends Serializable, CheckpointListener {
 
     /** Get the name of dialect. */
     String getName();
@@ -70,4 +71,14 @@ 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 {
+    }
 }

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPoolFactory.java

@@ -46,7 +46,6 @@ public HikariDataSource createPooledDataSource(JdbcSourceConfig sourceConfig) {
         config.setConnectionTimeout(sourceConfig.getConnectTimeout().toMillis());
         config.addDataSourceProperty(SERVER_TIMEZONE_KEY, sourceConfig.getServerTimeZone());
         config.setDriverClassName(sourceConfig.getDriverClassName());
-
         // optional optimization configurations for pooled DataSource
         config.addDataSourceProperty("cachePrepStmts", "true");
         config.addDataSourceProperty("prepStmtCacheSize", "250");

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/HybridSplitAssigner.java

@@ -123,7 +123,8 @@ public Optional<SourceSplitBase> getNext() {
                 // assigning the stream split. Otherwise, records emitted from stream split
                 // might be out-of-order in terms of same primary key with snapshot splits.
                 isStreamSplitAssigned = true;
-                return Optional.of(createStreamSplit());
+                StreamSplit streamSplit = createStreamSplit();
+                return Optional.of(streamSplit);
             } else {
                 // stream split is not ready by now
                 return Optional.empty();

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsStateSerializer.java

@@ -38,6 +38,7 @@
 
 import static org.apache.inlong.sort.cdc.base.source.meta.split.SourceSplitSerializer.readTableSchemas;
 import static org.apache.inlong.sort.cdc.base.source.meta.split.SourceSplitSerializer.writeTableSchemas;
+import static org.apache.inlong.sort.cdc.base.util.RecordUtils.shouldUseCatalogBeforeSchema;
 
 /** The {@link SimpleVersionedSerializer Serializer} for the {@link PendingSplitsState}.
  * Copy from com.ververica:flink-cdc-base:2.3.0.
@@ -361,6 +362,8 @@ private void writeTableIds(Collection<TableId> tableIds, DataOutputSerializer ou
         final int size = tableIds.size();
         out.writeInt(size);
         for (TableId tableId : tableIds) {
+            boolean useCatalogBeforeSchema = shouldUseCatalogBeforeSchema(tableId);
+            out.writeBoolean(useCatalogBeforeSchema);
             out.writeUTF(tableId.toString());
         }
     }

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetDeserializerSerializer.java

@@ -84,12 +84,17 @@ default OffsetDeserializer createOffsetDeserializer() {
     default FinishedSnapshotSplitInfo deserialize(byte[] serialized) {
         try {
             final DataInputDeserializer in = new DataInputDeserializer(serialized);
-            TableId tableId = TableId.parse(in.readUTF());
+            String tableIdStr = in.readUTF();
             String splitId = in.readUTF();
             Object[] splitStart = serializedStringToRow(in.readUTF());
             Object[] splitEnd = serializedStringToRow(in.readUTF());
             OffsetFactory offsetFactory = (OffsetFactory) serializedStringToObject(in.readUTF());
             Offset highWatermark = readOffsetPosition(in);
+            boolean useCatalogBeforeSchema = true;
+            if (in.available() > 0) {
+                useCatalogBeforeSchema = in.readBoolean();
+            }
+            TableId tableId = TableId.parse(tableIdStr, useCatalogBeforeSchema);
             in.releaseArrays();
 
             return new FinishedSnapshotSplitInfo(

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/FinishedSnapshotSplitInfo.java

@@ -31,6 +31,7 @@
 import java.util.Objects;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.inlong.sort.cdc.base.util.RecordUtils.shouldUseCatalogBeforeSchema;
 
 /** The information used to describe a finished snapshot split.
  * Copy from com.ververica:flink-cdc-base:2.3.0.
@@ -140,6 +141,8 @@ public byte[] serialize() {
     }
 
     public byte[] serialize(final DataOutputSerializer out) throws IOException {
+        boolean useCatalogBeforeSchema = shouldUseCatalogBeforeSchema(this.getTableId());
+        out.writeBoolean(useCatalogBeforeSchema);
         out.writeUTF(this.getTableId().toString());
         out.writeUTF(this.getSplitId());
         out.writeUTF(SerializerUtils.rowToSerializedString(this.getSplitStart()));

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/split/SourceSplitSerializer.java

@@ -42,6 +42,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.inlong.sort.cdc.base.util.RecordUtils.shouldUseCatalogBeforeSchema;
+
 /** A serializer for the {@link SourceSplitBase}.
  * Copy from com.ververica:flink-cdc-base:2.3.0.
  * */
@@ -74,6 +76,8 @@ public byte[] serialize(SourceSplitBase split) throws IOException {
 
             final DataOutputSerializer out = SERIALIZER_CACHE.get();
             out.writeInt(SNAPSHOT_SPLIT_FLAG);
+            boolean useCatalogBeforeSchema = shouldUseCatalogBeforeSchema(snapshotSplit.getTableId());
+            out.writeBoolean(useCatalogBeforeSchema);
             out.writeUTF(snapshotSplit.getTableId().toString());
             out.writeUTF(snapshotSplit.splitId());
             out.writeUTF(snapshotSplit.getSplitKeyType().asSerializableString());
@@ -143,7 +147,8 @@ public SourceSplitBase deserializeSplit(int version, byte[] serialized) throws I
 
         int splitKind = in.readInt();
         if (splitKind == SNAPSHOT_SPLIT_FLAG) {
-            TableId tableId = TableId.parse(in.readUTF());
+            boolean useCatalogBeforeSchema = in.readBoolean();
+            TableId tableId = TableId.parse(in.readUTF(), useCatalogBeforeSchema);
             String splitId = in.readUTF();
             RowType splitKeyType = (RowType) LogicalTypeParser.parse(in.readUTF());
             Object[] splitBoundaryStart = SerializerUtils.serializedStringToRow(in.readUTF());
@@ -202,6 +207,8 @@ public static void writeTableSchemas(
         final int size = tableSchemas.size();
         out.writeInt(size);
         for (Map.Entry<TableId, TableChange> entry : tableSchemas.entrySet()) {
+            boolean useCatalogBeforeSchema = shouldUseCatalogBeforeSchema(entry.getKey());
+            out.writeBoolean(useCatalogBeforeSchema);
             out.writeUTF(entry.getKey().toString());
             final String tableChangeStr =
                     documentWriter.write(jsonSerializer.toDocument(entry.getValue()));
@@ -217,7 +224,8 @@ public static Map<TableId, TableChange> readTableSchemas(int version, DataInputD
         Map<TableId, TableChange> tableSchemas = new HashMap<>();
         final int size = in.readInt();
         for (int i = 0; i < size; i++) {
-            TableId tableId = TableId.parse(in.readUTF());
+            boolean useCatalogBeforeSchema = in.readBoolean();
+            TableId tableId = TableId.parse(in.readUTF(), useCatalogBeforeSchema);
             final String tableChangeStr;
             switch (version) {
                 case 1:
@@ -255,7 +263,8 @@ private List<FinishedSnapshotSplitInfo> readFinishedSplitsInfo(
         List<FinishedSnapshotSplitInfo> finishedSplitsInfo = new ArrayList<>();
         final int size = in.readInt();
         for (int i = 0; i < size; i++) {
-            TableId tableId = TableId.parse(in.readUTF());
+            boolean useCatalogBeforeSchema = in.readBoolean();
+            TableId tableId = TableId.parse(in.readUTF(), useCatalogBeforeSchema);
             String splitId = in.readUTF();
             Object[] splitStart = SerializerUtils.serializedStringToRow(in.readUTF());
             Object[] splitEnd = SerializerUtils.serializedStringToRow(in.readUTF());

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceReader.java

@@ -157,6 +157,11 @@ public List<SourceSplitBase> snapshotState(long checkpointId) {
         return stateSplits;
     }
 
+    @Override
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {
+        dialect.notifyCheckpointComplete(checkpointId);
+    }
+
     @Override
     protected void onSplitFinished(Map<String, SourceSplitState> finishedSplitIds) {
         for (SourceSplitState splitState : finishedSplitIds.values()) {
@@ -237,7 +242,7 @@ private StreamSplit discoverTableSchemasForStreamSplit(StreamSplit split) {
     public void handleSourceEvents(SourceEvent sourceEvent) {
         if (sourceEvent instanceof FinishedSnapshotSplitsAckEvent) {
             FinishedSnapshotSplitsAckEvent ackEvent = (FinishedSnapshotSplitsAckEvent) sourceEvent;
-            LOG.debug(
+            LOG.info(
                     "The subtask {} receives ack event for {} from enumerator.",
                     subtaskId,
                     ackEvent.getFinishedSplits());
@@ -246,12 +251,12 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
             }
         } else if (sourceEvent instanceof FinishedSnapshotSplitsRequestEvent) {
             // report finished snapshot splits
-            LOG.debug(
+            LOG.info(
                     "The subtask {} receives request to report finished snapshot splits.",
                     subtaskId);
             reportFinishedSnapshotSplitsIfNeed();
         } else if (sourceEvent instanceof StreamSplitMetaEvent) {
-            LOG.debug(
+            LOG.info(
                     "The subtask {} receives stream meta with group id {}.",
                     subtaskId,
                     ((StreamSplitMetaEvent) sourceEvent).getMetaGroupId());

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceRecordEmitter.java

@@ -45,6 +45,7 @@
 import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.getHistoryRecord;
 import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.getMessageTimestamp;
 import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.isDataChangeRecord;
+import static org.apache.inlong.sort.cdc.base.util.RecordUtils.isHeartbeatEvent;
 import static org.apache.inlong.sort.cdc.base.util.RecordUtils.isSchemaChangeEvent;
 
 /**
@@ -110,15 +111,24 @@ protected void processElement(
                 emitElement(element, output);
             }
         } else if (isDataChangeRecord(element)) {
-            if (splitState.isStreamSplitState()) {
-                Offset position = getOffsetPosition(element);
-                splitState.asStreamSplitState().setStartingOffset(position);
-            }
+            LOG.trace("Process DataChangeRecord: {}; splitState = {}", element, splitState);
+            updateStartingOffsetForSplit(splitState, element);
             reportMetrics(element);
             emitElement(element, output);
+        } else if (isHeartbeatEvent(element)) {
+            LOG.trace("Process Heartbeat: {}; splitState = {}", element, splitState);
+            updateStartingOffsetForSplit(splitState, element);
         } else {
             // unknown element
-            LOG.info("Meet unknown element {}, just skip.", element);
+            LOG.info(
+                    "Meet unknown element {} for splitState = {}, just skip.", element, splitState);
+        }
+    }
+
+    protected void updateStartingOffsetForSplit(SourceSplitState splitState, SourceRecord element) {
+        if (splitState.isStreamSplitState()) {
+            Offset position = getOffsetPosition(element);
+            splitState.asStreamSplitState().setStartingOffset(position);
         }
     }
 

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/FetchTask.java

@@ -47,6 +47,10 @@ public interface FetchTask<Split> {
     /** Returns the split that the task used. */
     Split getSplit();
 
+    /** Stops current task, most of the implementations don't need this. */
+    default void stop() {
+    }
+
     /** Base context used in the execution of fetch task. */
     interface Context {
 

inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/util/RecordUtils.java

@@ -60,6 +60,7 @@ public class RecordUtils {
     public static final String ORACLE_SCHEMA_CHANGE_EVENT_KEY_NAME = "io.debezium.connector.oracle.SchemaChangeKey";
     public static final String CONNECTOR = "connector";
     public static final String MYSQL_CONNECTOR = "mysql";
+    public static final String SCHEMA_HEARTBEAT_EVENT_KEY_NAME = "io.debezium.connector.common.Heartbeat";
 
     private RecordUtils() {
 
@@ -163,4 +164,16 @@ public static boolean isDdlRecord(Struct value) {
         return value.schema().field(HISTORY_RECORD_FIELD) != null;
     }
 
+    public static boolean isHeartbeatEvent(SourceRecord record) {
+        Schema valueSchema = record.valueSchema();
+        return valueSchema != null
+                && SCHEMA_HEARTBEAT_EVENT_KEY_NAME.equalsIgnoreCase(valueSchema.name());
+    }
+
+    public static boolean shouldUseCatalogBeforeSchema(TableId tableId) {
+        // if catalog is not defined but the schema is defined return this flag as false
+        // otherwise return true
+        return !(tableId.catalog() == null && tableId.schema() != null);
+    }
+
 }

inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfig.java

@@ -47,6 +47,7 @@ public class OracleSourceConfig extends JdbcSourceConfig {
     public OracleSourceConfig(
             StartupOptions startupOptions,
             List<String> databaseList,
+            List<String> schemaList,
             List<String> tableList,
             int splitSize,
             int splitMetaGroupSize,
@@ -72,6 +73,7 @@ public OracleSourceConfig(
         super(
                 startupOptions,
                 databaseList,
+                schemaList,
                 tableList,
                 splitSize,
                 splitMetaGroupSize,

inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java

@@ -125,6 +125,7 @@ public OracleSourceConfig create(int subtaskId) {
         return new OracleSourceConfig(
                 startupOptions,
                 databaseList,
+                schemaList,
                 tableList,
                 splitSize,
                 splitMetaGroupSize,