[jira] [Created] (FLINK-39844) PostgreSQL CDC: filter table columns by schema name to fix LIKE-wildcard cross-schema column leakage

["Di Wu (Jira)" <ji...@apache.org>]

Di Wu created FLINK-39844:
-----------------------------

             Summary: PostgreSQL CDC: filter table columns by schema name to fix LIKE-wildcard cross-schema column   leakage
                 Key: FLINK-39844
                 URL: https://issues.apache.org/jira/browse/FLINK-39844
             Project: Flink
          Issue Type: Improvement
          Components: Flink CDC
    Affects Versions: cdc-3.6.0
            Reporter: Di Wu


*Background*

 

  During the snapshot phase, the PostgreSQL connector reads the column structure of captured tables

  via JDBC DatabaseMetaData#getColumns(catalog, schemaPattern, tableNamePattern,

  columnNamePattern).

 

  Per the JDBC spec, *both* *schemaPattern* *and* *tableNamePattern* *are* *LIKE* {*}patterns{*}:

  - _ matches any single character

  - % matches any sequence of characters

  

  So when a schema name or table name contains _ / % (both are legal identifier characters in

  PostgreSQL), getColumns may return columns from *other schemas / tables that were not meant to* 

  {*}match{*}.

 

  *Current State*

 

  An exact filter on the *table* *name* is already in place: for each column returned by getColumns,

  the result-set column 3 TABLE_NAME is compared for equality against the target TableId.table(),

  and the column is dropped on mismatch.

 

  However, the *schema* *dimension* *is* {*}missing{*}. Result-set column 2 TABLE_SCHEM is never validated.

 

  *Problem*

 

  Suppose a database has two schemas whose names differ by only a single character:

 

  - sch_test (the target)

  - schxtest (when _ acts as a wildcard, the pattern sch_test also matches it)

 

  Each schema contains a table with the same name, cross_schema_tbl. When the user only wants to

  capture sch_test.cross_schema_tbl:

 

  1. The connector calls getColumns(catalog, "sch_test", "cross_schema_tbl", null).

  2. PostgreSQL treats sch_test as a LIKE pattern, so schxtest matches too, and columns of the

  same-named table from *both* schemas are returned together.

  3. The existing table-name filter cannot stop this — both tables have TABLE_NAME equal to

  cross_schema_tbl.

  4. The same-named columns are loaded into a single table schema, and the snapshot phase throws

  IllegalStateException: Duplicate key Optional.empty, failing the job.

 

  The % case is analogous (when a schema name contains a literal %).

  

  *Impact*

 

  - {*}Trigger condition{*}: a single PostgreSQL instance contains multiple schemas whose names are

  LIKE-wildcard matches of one another, and they hold same-named tables.

  - {*}Consequence{*}: the job fails outright during the snapshot phase, or (when column names differ)

  columns from another schema are mistakenly merged in, producing an incorrect schema / dirty data.

 

  *Proposed* *Change*

 

  Add an *exact comparison on the schema name* next to the existing table-name comparison: read

  result-set column 2 TABLE_SCHEM and compare it for equality against TableId.schema().

 

  final String resultSchemaName = columnMetadata.getString(2);

  final String resultTableName = columnMetadata.getString(3);

  if (!tableId.table().equals(resultTableName)

          || (tableId.schema() != null && !tableId.schema().equals(resultSchemaName))) {

      return Optional.empty();

  }

  

  Edge case: when TableId.schema() is null (no schema specified), the schema check is skipped, to

  avoid treating the non-null schema name returned by the metadata as unequal and erroneously

  dropping all columns.

  

  *Compatibility*

 

  This is a pure after-the-fact filter. It only tightens the decision of "which columns belong to

  the target table"; it does not change the query, and it does not affect normal schemas/tables

  (those without wildcard characters). Existing test behavior is unchanged.

  

  *Testing*

 

  Add a cross-schema case to SimilarTableNamesITCase: create two schemas sch_test / schxtest that

  are wildcard matches of each other, each holding a same-named table cross_schema_tbl, and verify

  that only the target schema's snapshot and incremental data are captured, with no leakage from

  the look-alike schema.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)