[
https://issues.apache.org/jira/browse/FLINK-40152?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18096425#comment-18096425
]
Xin Chen commented on FLINK-40152:
----------------------------------
The {{isIncluded()}} check uses case-insensitive regex matching, which is
incorrect for PostgreSQL where quoted identifiers are case-sensitive.
== Reproduction ==
# Create two tables in PostgreSQL:
{code:java}
CREATE TABLE "mytable" (id INT PRIMARY KEY, name TEXT);
CREATE TABLE "MyTable" (id INT PRIMARY KEY, name TEXT);{code}
2. Configure Flink CDC to capture only mytable:
{code:java}
CREATE TABLE cdc_source (
id INT,
name STRING
) WITH (
'connector' = 'postgres-cdc',
'hostname' = 'localhost',
'port' = '5432',
'database-name' = 'mydb',
'schema-name' = 'public',
'table-name' = 'mytable',
'username' = 'postgres',
'password' = 'postgres',
'slot.name' = 'flink_test'
); {code}
3. Expected: Only mytable is captured Actual: Both mytable and MyTable are
captured
4. Actual: Both {{mytable}} and {{MyTable}} are captured
== Impact ==
* Users cannot selectively capture one table when case-variant tables exist
* May cause data duplication or unexpected behavior
* Violates PostgreSQL's case-sensitive identifier semantics
== Proposed Fix ==
Add case-sensitive post-filtering in
{{PostgresSourceBuilder.createEnumerator()}} or
{{TableDiscoveryUtils.listTables()}} to respect user-specified table names
exactly.
Example1 fix in {{{}createEnumerator(){}}}:
{code:java}
final List<TableId> remainingTables =
dataSourceDialect.discoverDataCollections(sourceConfig).stream()
.filter(t -> sourceConfig.getTableList().stream()
.anyMatch(userTable -> {
String tableIdStr = t.toString();
return tableIdStr.equals(userTable)
|| tableIdStr.endsWith("." + userTable);
}))
.collect(Collectors.toList()); {code}
Note: PostgresDialect.isDataCollectionIdCaseSensitive() already returns true,
indicating Flink CDC is aware of PostgreSQL's case sensitivity, but the actual
filtering logic does not honor this.
Example 2 fix in
com.ververica.cdc.connectors.postgres.source.utils.TableDiscoveryUtils#listTables（）
:
{code:java}
Set<TableId> capturedTables =
allTableIds.stream()
.filter(t -> tableFilters.dataCollectionFilter().isIncluded(t))
.filter(t -> tableList.contains(t.toString()))
.collect(Collectors.toSet()); {code}
> PostgreSQL CDC connector captures case-variant tables when only one is
> specified
> --------------------------------------------------------------------------------
>
> Key: FLINK-40152
> URL: https://issues.apache.org/jira/browse/FLINK-40152
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: 3.0.0, 2.4.0
> Reporter: Xin Chen
> Priority: Major
>
> == Problem ==
> When PostgreSQL has two tables with the same name but different cases (e.g.,
> `mytable` and `MyTable`), specifying only one table in `table-name` causes
> Flink CDC to capture BOTH tables.
> == Root Cause ==
> Debezium's `table.include.list` uses `Pattern.CASE_INSENSITIVE` for regex
> matching by default (see `io.debezium.util.Strings#includes()`). This causes
> `public.mytable` to also match `public.MyTable`.
> The issue is in `TableDiscoveryUtils.listTables()`:
> {code:java}
> Set<TableId> capturedTables =
> allTableIds.stream()
> .filter(t -> tableFilters.dataCollectionFilter().isIncluded(t))
> .collect(Collectors.toSet()); {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
