[FLINK-39633][postgres] Fix NPE in incremental snapshot backfill when WAL carries records for other captured tables

[JNSimba]

This closes https://issues.apache.org/jira/browse/FLINK-39633.

Problem

During the snapshot backfill phase of PostgreSQL CDC (incremental snapshot), the WAL stream from the logical replication slot may carry change records for captured tables other than the one whose chunk is currently being snapshotted. IncrementalSourceScanFetcher#isChangeRecordInChunkRange does not filter records by tableId before delegating to JdbcSourceFetchTaskContext#isRecordBetween, which calls getDatabaseSchema().tableFor(record.tableId). If the foreign table's schema is not yet present in RelationAwarePostgresSchema's lazy cache, the lookup returns null and ChunkUtils.getSplitColumn throws NPE on null.primaryKeyColumns():

java.lang.NullPointerException: Cannot invoke "io.debezium.relational.Table.primaryKeyColumns()" because "table" is null
    at org.apache.flink.cdc.connectors.postgres.source.utils.ChunkUtils.getSplitColumn(ChunkUtils.java:45)
    at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext.getSplitType(PostgresSourceFetchTaskContext.java:291)
    at org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext.isRecordBetween(JdbcSourceFetchTaskContext.java:76)
    at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.isChangeRecordInChunkRange(IncrementalSourceScanFetcher.java:265)

The streaming reader (IncrementalSourceStreamFetcher#shouldEmit) already guards against this case via finishedSplitsInfo.containsKey(tableId) before invoking isRecordBetween. The scan fetcher was missing the symmetric guard.

Fix

Filter change records by tableId in IncrementalSourceScanFetcher#isChangeRecordInChunkRange before delegating to isRecordBetween. Records whose tableId does not equal currentSnapshotSplit.getTableId() are skipped.

Test plan

• Added IncrementalSourceScanFetcherTest with three cases:
  • foreign-table change record is filtered out (no NPE, isRecordBetween is never invoked)
  • same-table change record is still delegated to isRecordBetween
  • non-data change records (watermark / signal) short-circuit without calling getTableId
• All existing flink-cdc-base unit tests still pass (16/16)

[Copilot]

Pull request overview

Fixes an NPE during PostgreSQL incremental snapshot backfill when the WAL stream includes change events for other captured tables by filtering scan-fetcher change records by tableId before delegating to range checks.

Changes:

• Add tableId equality filtering in IncrementalSourceScanFetcher#isChangeRecordInChunkRange to ignore foreign-table change records.
• Expose isChangeRecordInChunkRange and a snapshot-split setter for testing via @VisibleForTesting.
• Add unit tests covering foreign-table filtering, same-table delegation, and non-data-change short-circuiting.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java	Filters WAL change records by TableId during snapshot backfill to prevent NPEs from schema cache misses.
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcherTest.java	Adds regression tests ensuring foreign-table records are skipped and non-data events don’t trigger tableId/range checks.

---

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