[FLINK-39582][postgres] Allow logical messages

eskabetxe · eskabetxe · commit 487521074195 · 2026-04-30T00:15:54.000+02:00
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/dialect/DataSourceDialect.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/dialect/DataSourceDialect.java
@@ -24,6 +24,7 @@
 import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
 import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
 
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
@@ -81,6 +82,15 @@ default Offset displayCommittedOffset(C sourceConfig) {
     /** The task context used for fetch task to fetch data from external systems. */
     FetchTask.Context createFetchTaskContext(C sourceConfig);
 
+    /**
+     * The stream fetcher used to fetch data of a stream split. Dialects may override this to
+     * provide connector-specific filtering or routing behavior.
+     */
+    default IncrementalSourceStreamFetcher createStreamFetcher(
+            FetchTask.Context taskContext, int subtaskId) {
+        return new IncrementalSourceStreamFetcher(taskContext, subtaskId);
+    }
+
     /**
      * We may need the offset corresponding to the checkpointId. For example, we should commit LSN
      * of checkpoint to postgres's slot.
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceRecordEmitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceRecordEmitter.java
@@ -132,7 +132,7 @@ protected TableChanges getTableChangeRecord(SourceRecord element) throws IOExcep
         return TABLE_CHANGE_SERIALIZER.deserialize(tableChanges, true);
     }
 
-    private void updateStreamSplitState(SourceSplitState splitState, SourceRecord element) {
+    protected void updateStreamSplitState(SourceSplitState splitState, SourceRecord element) {
         if (splitState.isStreamSplitState()) {
             Offset position = getOffsetPosition(element);
             splitState.asStreamSplitState().setStartingOffset(position);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceSplitReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceSplitReader.java
@@ -268,7 +268,7 @@ private IncrementalSourceScanFetcher getScanFetcher() {
     private IncrementalSourceStreamFetcher getStreamFetcher() {
         if (reusedStreamFetcher == null) {
             reusedStreamFetcher =
-                    new IncrementalSourceStreamFetcher(
+                    dataSourceDialect.createStreamFetcher(
                             dataSourceDialect.createFetchTaskContext(sourceConfig), subtaskId);
         }
         return reusedStreamFetcher;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java
@@ -185,7 +185,7 @@ public void close() {
      *  only the change event belong to [1024, 2048) and offset is after highWatermark1 should send.
      * </pre>
      */
-    private boolean shouldEmit(SourceRecord sourceRecord) {
+    protected boolean shouldEmit(SourceRecord sourceRecord) {
         if (taskContext.isDataChangeRecord(sourceRecord)) {
             TableId tableId = taskContext.getTableId(sourceRecord);
             Offset position = taskContext.getStreamOffset(sourceRecord);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
@@ -26,11 +26,13 @@
 import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
 import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
 import org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext;
 import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
 import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask;
 import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext;
 import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask;
+import org.apache.flink.cdc.connectors.postgres.source.reader.PostgresSourceStreamFetcher;
 import org.apache.flink.cdc.connectors.postgres.source.utils.CustomPostgresSchema;
 import org.apache.flink.cdc.connectors.postgres.source.utils.TableDiscoveryUtils;
 import org.apache.flink.util.FlinkRuntimeException;
@@ -239,6 +241,13 @@ public JdbcSourceFetchTaskContext createFetchTaskContext(JdbcSourceConfig taskSo
         return new PostgresSourceFetchTaskContext(taskSourceConfig, this);
     }
 
+    @Override
+    public IncrementalSourceStreamFetcher createStreamFetcher(
+            FetchTask.Context taskContext, int subtaskId) {
+        return new PostgresSourceStreamFetcher(
+                taskContext, subtaskId, sourceConfig.isLogicalMessageEnabled());
+    }
+
     @Override
     public void notifyCheckpointComplete(long checkpointId, Offset offset) throws Exception {
         if (streamFetchTask != null) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java
@@ -315,6 +315,12 @@ public PostgresSourceBuilder<T> includeDatabaseInTableId(boolean includeDatabase
         return this;
     }
 
+    /** Whether to emit Postgres logical decoding messages to the deserializer. */
+    public PostgresSourceBuilder<T> includeLogicalMessages(boolean includeLogicalMessages) {
+        this.configFactory.setIncludeLogicalMessages(includeLogicalMessages);
+        return this;
+    }
+
     /** Whether to infer schema change event on relation message. */
     public PostgresSourceBuilder<T> includeSchemaChanges(boolean includeSchemaChanges) {
         this.configFactory.includeSchemaChanges(includeSchemaChanges);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java
@@ -40,6 +40,7 @@ public class PostgresSourceConfig extends JdbcSourceConfig {
     private final int lsnCommitCheckpointsDelay;
     private final boolean includePartitionedTables;
     private final boolean includeDatabaseInTableId;
+    private final boolean includeLogicalMessages;
 
     public PostgresSourceConfig(
             int subtaskId,
@@ -71,7 +72,8 @@ public PostgresSourceConfig(
             int lsnCommitCheckpointsDelay,
             boolean assignUnboundedChunkFirst,
             boolean includePartitionedTables,
-            boolean includeDatabaseInTableId) {
+            boolean includeDatabaseInTableId,
+            boolean includeLogicalMessages) {
         super(
                 startupOptions,
                 databaseList,
@@ -103,6 +105,7 @@ public PostgresSourceConfig(
         this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay;
         this.includePartitionedTables = includePartitionedTables;
         this.includeDatabaseInTableId = includeDatabaseInTableId;
+        this.includeLogicalMessages = includeLogicalMessages;
     }
 
     /**
@@ -156,4 +159,9 @@ public PostgresConnectorConfig getDbzConnectorConfig() {
     public boolean isIncludeDatabaseInTableId() {
         return includeDatabaseInTableId;
     }
+
+    /** Returns whether to emit Postgres logical decoding messages to the deserializer. */
+    public boolean isLogicalMessageEnabled() {
+        return includeLogicalMessages;
+    }
 }
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java
@@ -57,6 +57,9 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory {
     private boolean includeDatabaseInTableId =
             PostgresSourceOptions.TABLE_ID_INCLUDE_DATABASE.defaultValue();
 
+    private boolean includeLogicalMessages =
+            PostgresSourceOptions.SCAN_LOGICAL_MESSAGE_ENABLED.defaultValue();
+
     /** Creates a new {@link PostgresSourceConfig} for the given subtask {@code subtaskId}. */
     @Override
     public PostgresSourceConfig create(int subtaskId) {
@@ -140,7 +143,8 @@ public PostgresSourceConfig create(int subtaskId) {
                 lsnCommitCheckpointsDelay,
                 assignUnboundedChunkFirst,
                 includePartitionedTables,
-                includeDatabaseInTableId);
+                includeDatabaseInTableId,
+                includeLogicalMessages);
     }
 
     /**
@@ -198,4 +202,9 @@ public void setIncludePartitionedTables(boolean includePartitionedTables) {
     public void setIncludeDatabaseInTableId(boolean includeDatabaseInTableId) {
         this.includeDatabaseInTableId = includeDatabaseInTableId;
     }
+
+    /** Set whether to emit Postgres logical decoding messages to the deserializer. */
+    public void setIncludeLogicalMessages(boolean includeLogicalMessages) {
+        this.includeLogicalMessages = includeLogicalMessages;
+    }
 }
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java
@@ -98,6 +98,16 @@ public class PostgresSourceOptions extends JdbcSourceOptions {
                                     + "(1) PUBLICATION must be created beforehand with parameter publish_via_partition_root=true\n"
                                     + "(2) Table list (regex or predefined list) should only match the parent table name, if table list matches both parent and child tables, snapshot data will be read twice.");
 
+    public static final ConfigOption<Boolean> SCAN_LOGICAL_MESSAGE_ENABLED =
+            ConfigOptions.key("scan.logical-message.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to emit Postgres logical messages produced by "
+                                    + "pg_logical_emit_message() to the deserializer. "
+                                    + "Disabled by default; logical messages are not bound to "
+                                    + "any table and are dropped by table-based watermark filtering.");
+
     public static final ConfigOption<Boolean> TABLE_ID_INCLUDE_DATABASE =
             ConfigOptions.key("table-id.include-database")
                     .booleanType()
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceRecordEmitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceRecordEmitter.java
@@ -17,7 +17,9 @@
 
 package org.apache.flink.cdc.connectors.postgres.source.reader;
 
+import org.apache.flink.api.connector.source.SourceOutput;
 import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState;
 import org.apache.flink.cdc.connectors.base.source.metrics.SourceReaderMetrics;
 import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceRecordEmitter;
 import org.apache.flink.cdc.connectors.postgres.source.schema.PostgresSchemaRecord;
@@ -43,6 +45,18 @@ public PostgresSourceRecordEmitter(
                 offsetFactory);
     }
 
+    @Override
+    protected void processElement(
+            SourceRecord element, SourceOutput<T> output, SourceSplitState splitState)
+            throws Exception {
+        if (PostgresSourceStreamFetcher.isLogicalMessage(element)) {
+            updateStreamSplitState(splitState, element);
+            emitElement(element, output);
+            return;
+        }
+        super.processElement(element, output, splitState);
+    }
+
     @Override
     protected TableChanges getTableChangeRecord(SourceRecord element) throws IOException {
         if (element instanceof PostgresSchemaRecord) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+    private final boolean includeLogicalMessages;
+
+    public PostgresSourceStreamFetcher(
+            FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+        super(taskContext, subtaskId);
+        this.includeLogicalMessages = includeLogicalMessages;
+    }
+
+    @Override
+    protected boolean shouldEmit(SourceRecord sourceRecord) {
+        if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+            return true;
+        }
+        return super.shouldEmit(sourceRecord);
+    }
+
+    static boolean isLogicalMessage(SourceRecord record) {
+        if (record.value() instanceof Struct) {
+            Struct struct = (Struct) record.value();
+            return struct.schema().field(Envelope.FieldName.OPERATION) != null
+                    && "m".equals(struct.getString(Envelope.FieldName.OPERATION));
+        }
+        return false;
+    }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherITCase.java
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherTest.java

Original file line number	Diff line number	Diff line change
`@@ -132,7 +132,7 @@ protected TableChanges getTableChangeRecord(SourceRecord element) throws IOExcep`
`132`	`132`	`return TABLE_CHANGE_SERIALIZER.deserialize(tableChanges, true);`
`133`	`133`	`}`
`134`	`134`
`135`		`- private void updateStreamSplitState(SourceSplitState splitState, SourceRecord element) {`
	`135`	`+ protected void updateStreamSplitState(SourceSplitState splitState, SourceRecord element) {`
`136`	`136`	`if (splitState.isStreamSplitState()) {`
`137`	`137`	`Offset position = getOffsetPosition(element);`
`138`	`138`	`splitState.asStreamSplitState().setStartingOffset(position);`
Original file line number	Diff line number	Diff line change
`@@ -268,7 +268,7 @@ private IncrementalSourceScanFetcher getScanFetcher() {`
`268`	`268`	`private IncrementalSourceStreamFetcher getStreamFetcher() {`
`269`	`269`	`if (reusedStreamFetcher == null) {`
`270`	`270`	`reusedStreamFetcher =`
`271`		`- new IncrementalSourceStreamFetcher(`
	`271`	`+ dataSourceDialect.createStreamFetcher(`
`272`	`272`	`dataSourceDialect.createFetchTaskContext(sourceConfig), subtaskId);`
`273`	`273`	`}`
`274`	`274`	`return reusedStreamFetcher;`