[FLINK-38959][pipeline-connector][postgres]Added support for schema change in the postgres pipeline connector by linjianchang · Pull Request #4233 · apache/flink-cdc

linjianchang · 2026-01-22T01:21:09Z

Added support for schema change in the postgres pipeline connector

leonardBang · 2026-01-23T03:10:36Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+                        "SELECT a.attname attname,a.attnum AS age_oid \n" +
+                                "FROM pg_attribute a\n" +
+                                "JOIN pg_class c ON a.attrelid = c.oid\n" +


Is the official recommended way to obtain history schema of a Postgres table ?

Is the official recommended way to obtain history schema of a Postgres table ?

Here, we are only obtaining the OID of the schema field in the current state of the Postgres table, and determining whether the field has undergone a rename operation based on whether the OID remains consistent

zml1206 · 2026-01-26T09:17:58Z

Could a configuration option be added to control the on/off state? Also, I'm implementing a way to parse DDL SQL to support DDL, which requires adding an event table in Postgres. The two methods don't seem to conflict, is this approach acceptable to the community?

…hange in the postgres pipeline connector.

linjianchang · 2026-01-29T09:55:24Z

added to control the on/off state

Already added to control the on/off state，thanks for suggestion！

Copilot

Pull request overview

This pull request adds schema change event support to the Flink CDC Postgres pipeline connector. The implementation enables the connector to detect and emit schema change events (ADD/DROP/RENAME columns, ALTER column types) during change data capture operations.

Changes:

Added includeSchemaChanges configuration option to control schema change event emission
Implemented schema change detection logic by comparing before/after schemas and tracking column OIDs
Added comprehensive test coverage for various schema evolution scenarios (add, drop, rename, type changes)

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 14 comments.

Show a summary per file

File	Description
PostgresSourceBuilder.java	Added `includeSchemaChanges()` method to builder API
PostgresDataSourceOptions.java	Added `SCHEMA_CHANGE_ENABLED` configuration option with default value true
PostgresDataSourceFactory.java	Integrated schema change configuration into data source factory
PostgresSchemaUtils.java	Added methods to retrieve column OIDs for schema change tracking
PostgresDataSource.java	Initialized column OID maps and passed to deserializer
PostgresEventDeserializer.java	Core implementation of schema change detection and event generation logic
PostgresPipelineITCaseTest.java	Added parameterized tests for various schema evolution scenarios

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-01-30T09:13:20Z

...src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java

+                                        columns.stream()
+                                                .filter(e -> e.getName().equals(afterFieldName))
+                                                .findFirst()
+                                                .get();


NoSuchElementException risk: The code uses .get() on an Optional at line 349 without checking if the element is present. If no column matches the filter condition (e.getName().equals(afterFieldName)), this will throw a NoSuchElementException. Use .orElseThrow() with a descriptive error message or handle the absent case appropriately.

Suggested change

.get();

.orElseThrow(

() ->

new IllegalStateException(

"Failed to find column with name '"

+ afterFieldName

+ "' in table "

+ tableId));

Copilot · 2026-01-30T09:13:21Z

...src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java

+                if ((newDelEntry.getValue() == null && newAddEntry.getValue() == null)
+                        || (newDelEntry.getValue() == null && newAddEntry.getValue() != null)) {
+                    int oldFieldIndex = oldSchema.field(newDelEntry.getKey()).index();
+                    int afterFieldIndex = afterSchema.field(newAddEntry.getKey()).index();
+                    if (oldFieldIndex == afterFieldIndex) {
+                        renameColumnMaps.put(newDelEntry.getKey(), newAddEntry.getKey());
+                    }


Logic issue in rename detection: The condition at line 385-386 compares null values but has problematic logic. When newDelEntry.getValue() is null and newAddEntry.getValue() is not null, it treats this as a potential rename. However, this is counterintuitive - if the deleted column has no OID (null) and the added column has an OID, they are likely different columns, not a rename. The logic should be reconsidered to ensure accurate rename detection.

Copilot · 2026-01-30T09:13:21Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+                jdbc.query(
+                        "SELECT b.relname,a.attname attname,a.attnum AS oid FROM pg_attribute a JOIN pg_class b ON a.attrelid = b.oid WHERE b.relname "
+                                + inClause
+                                + " and a.attnum > 0 and a.attisdropped = 'f' group by b.relname,a.attname,a.attnum",


Performance concern: The query at line 144-146 uses a GROUP BY clause unnecessarily when retrieving column OIDs. Since we're already filtering by attnum > 0 and attisdropped = 'f', each (relname, attname, attnum) combination should be unique without needing GROUP BY. Removing the GROUP BY would improve query performance.

Suggested change

+ " and a.attnum > 0 and a.attisdropped = 'f' group by b.relname,a.attname,a.attnum",

+ " and a.attnum > 0 and a.attisdropped = 'f'",

Copilot · 2026-01-30T09:13:21Z

...src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java

+    @Override
+    public List<? extends Event> deserialize(SourceRecord record) throws Exception {
+        List<Event> result = new ArrayList<>();
+        if (postgresSourceConfig.isIncludeSchemaChanges()) {


Potential NullPointerException: At line 118, postgresSourceConfig.isIncludeSchemaChanges() is called without checking if postgresSourceConfig is null. The constructor allows postgresSourceConfig to be null (see constructors at lines 78-93), which would cause a NullPointerException here. Add a null check before calling isIncludeSchemaChanges().

Suggested change

if (postgresSourceConfig.isIncludeSchemaChanges()) {

if (postgresSourceConfig != null && postgresSourceConfig.isIncludeSchemaChanges()) {

Copilot · 2026-01-30T09:13:22Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+            try {
+                jdbc.query(
+                        "SELECT a.attname attname,a.attnum as oid \n"
+                                + "FROM pg_attribute a\n"
+                                + "JOIN pg_class b ON a.attrelid = b.oid\n"
+                                + "WHERE b.relname = '"
+                                + tableId.getTableName()


Missing schema qualification in query: The query at lines 114-121 filters by table name (b.relname) only, without considering the schema/namespace. In PostgreSQL, multiple schemas can have tables with the same name. This could return incorrect column OIDs if tables with the same name exist in different schemas. The query should also filter by schema to ensure accurate results, especially since TableId includes schema information.

Suggested change

try {

jdbc.query(

"SELECT a.attname attname,a.attnum as oid \n"

+ "FROM pg_attribute a\n"

+ "JOIN pg_class b ON a.attrelid = b.oid\n"

+ "WHERE b.relname = '"

+ tableId.getTableName()

String tableName = tableId.getTableName().replace("'", "''");

String schemaName = tableId.getSchemaName().replace("'", "''");

try {

jdbc.query(

"SELECT a.attname attname,a.attnum as oid \n"

+ "FROM pg_attribute a\n"

+ "JOIN pg_class b ON a.attrelid = b.oid\n"

+ "JOIN pg_namespace n ON b.relnamespace = n.oid\n"

+ "WHERE b.relname = '"

+ tableName

+ "' AND n.nspname = '"

+ schemaName

Copilot · 2026-01-30T09:13:23Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java

+        beforeTableColumnsOidMaps =
+                PostgresSchemaUtils.getAllTablesColumnOids(
+                        postgresSourceConfig, postgresSourceConfig.getTableList());


Resource leak risk: The getAllTablesColumnOids method at line 76 opens a database connection internally. If this method throws an exception, the connection may not be properly closed. While the method uses try-with-resources internally, if an exception occurs between line 75 and 85 during deserializer construction, the partially initialized state could cause issues. Consider wrapping this in proper error handling.

Copilot · 2026-01-30T09:13:23Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+                            while (rs.next()) {
+                                TableId tableId = TableId.tableId(rs.getString(1));


Inconsistent TableId construction: At line 149, TableId.tableId() is called with only the table name (relname), which creates a single-component TableId. However, elsewhere in the code (e.g., line 145 in PostgresEventDeserializer), TableId.tableId(tableId.getTableName()) is also used with just the table name. This inconsistency can lead to lookup failures if the beforeTableColumnsOidMaps uses different TableId formats. The TableId should be constructed consistently with schema information to match how it's used throughout the code.

Copilot · 2026-01-30T09:13:23Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+                                + "WHERE b.relname = '"
+                                + tableId.getTableName()
+                                + "' AND a.attname "
+                                + inClause,
+                        rs -> {
+                            while (rs.next()) {
+                                oidMaps.put(rs.getString(1), rs.getInt(2));
+                            }
+                        });


SQL injection vulnerability: The tableId.getTableName() is directly concatenated into the SQL query without proper escaping or use of prepared statements. An attacker could exploit this by crafting a malicious table name. Consider using parameterized queries or properly escaping the table name.

Suggested change

+ "WHERE b.relname = '"

+ tableId.getTableName()

+ "' AND a.attname "

+ inClause,

rs -> {

while (rs.next()) {

oidMaps.put(rs.getString(1), rs.getInt(2));

}

});

+ "WHERE b.relname = ? AND a.attname "

+ inClause,

rs -> {

while (rs.next()) {

oidMaps.put(rs.getString(1), rs.getInt(2));

}

},

tableId.getTableName());

Copilot · 2026-01-30T09:13:23Z

...stgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

+            Map<TableId, Map<String, Integer>> tableOidMaps = new HashMap<>();
+            String inClause =
+                    tableList.stream()
+                            .map(table -> "'" + table.split("\\.")[1].replace("'", "''") + "'")


Potential ArrayIndexOutOfBoundsException: The code assumes that table.split("\.") will always have at least 2 elements (accessing index [1]). If a table name doesn't contain a dot, this will throw an exception. Add validation to ensure the split result has the expected number of elements before accessing index 1.

Suggested change

.map(table -> "'" + table.split("\\.")[1].replace("'", "''") + "'")

.map(

table -> {

String[] parts = table.split("\\.");

String tableName =

parts.length > 1 ? parts[1] : parts[0];

return "'" + tableName.replace("'", "''") + "'";

})

Copilot · 2026-01-30T09:13:24Z

...src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java

+                                afterField =
+                                        afterSchema.field(finalRenameColumnMaps.get(oldFieldName));
+                            }


Potential NullPointerException: At line 329, afterField could still be null if both afterSchema.field(oldFieldName) returns null and finalRenameColumnMaps.get(oldFieldName) also returns null or the field doesn't exist. Attempting to call afterField.name() will throw a NullPointerException. Add a null check before accessing afterField.

Suggested change

afterField =

afterSchema.field(finalRenameColumnMaps.get(oldFieldName));

}

String renamedFieldName = finalRenameColumnMaps.get(oldFieldName);

if (renamedFieldName != null) {

afterField = afterSchema.field(renamedFieldName);

}

}

if (afterField == null) {

// No corresponding field in the after-schema (even after rename); skip.

return;

}

lvyanquan · 2026-01-30T09:39:19Z

Hi @Mrart CC as you may be interested in this.

github-actions bot added the postgres-pipeline-connector label Jan 22, 2026

lvyanquan self-assigned this Jan 22, 2026

lvyanquan added this to the V3.6.0 milestone Jan 22, 2026

leonardBang reviewed Jan 23, 2026

View reviewed changes

linjianchang force-pushed the master-pgschemachang branch 2 times, most recently from c9048cc to 595f299 Compare January 23, 2026 08:43

[FLINK-38959][pipeline-connector][postgres]Added support for schema c…

01a67b4

…hange in the postgres pipeline connector.

linjianchang force-pushed the master-pgschemachang branch from 595f299 to 01a67b4 Compare January 29, 2026 09:54

github-actions bot added the postgres-cdc-connector label Jan 29, 2026

lvyanquan requested a review from Copilot January 30, 2026 08:53

Copilot started reviewing on behalf of lvyanquan January 30, 2026 08:55 View session

Copilot AI reviewed Jan 30, 2026

View reviewed changes

-                                                .get();
+                                                .orElseThrow(
+                                                        () ->
+                                                                new IllegalStateException(
+                                                                        "Failed to find column with name '"
+                                                                                + afterFieldName
+                                                                                + "' in table "
+                                                                                + tableId));

	+ " and a.attnum > 0 and a.attisdropped = 'f' group by b.relname,a.attname,a.attnum",
	+ " and a.attnum > 0 and a.attisdropped = 'f'",

	if (postgresSourceConfig.isIncludeSchemaChanges()) {
	if (postgresSourceConfig != null && postgresSourceConfig.isIncludeSchemaChanges()) {

		while (rs.next()) {
		TableId tableId = TableId.tableId(rs.getString(1));

-                            .map(table -> "'" + table.split("\\.")[1].replace("'", "''") + "'")
+                            .map(
+                                    table -> {
+                                        String[] parts = table.split("\\.");
+                                        String tableName =
+                                                parts.length > 1 ? parts[1] : parts[0];
+                                        return "'" + tableName.replace("'", "''") + "'";
+                                    })

-                                afterField =
-                                        afterSchema.field(finalRenameColumnMaps.get(oldFieldName));
-                            }
+                                String renamedFieldName = finalRenameColumnMaps.get(oldFieldName);
+                                if (renamedFieldName != null) {
+                                    afterField = afterSchema.field(renamedFieldName);
+                                }
+                            }
+                            if (afterField == null) {
+                                // No corresponding field in the after-schema (even after rename); skip.
+                                return;
+                            }

Conversation

linjianchang commented Jan 22, 2026

Uh oh!

leonardBang Jan 23, 2026

Choose a reason for hiding this comment

Uh oh!

linjianchang Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

zml1206 commented Jan 26, 2026

Uh oh!

linjianchang commented Jan 29, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

lvyanquan commented Jan 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants