Copilot commented on code in PR #4367:
URL: https://github.com/apache/flink-cdc/pull/4367#discussion_r3055245469
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java:
##########
@@ -196,13 +200,64 @@ public Map<TableId, TableChange>
discoverDataCollectionSchemas(JdbcSourceConfig
try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
// fetch table schemas
Map<TableId, TableChange> tableSchemas = queryTableSchema(jdbc,
capturedTableIds);
+ // validate REPLICA IDENTITY for tables without primary key
+ validateReplicaIdentityForNoPkTables(jdbc, tableSchemas);
return tableSchemas;
Review Comment:
`discoverDataCollectionSchemas` always calls
`validateReplicaIdentityForNoPkTables(...)`. If REPLICA IDENTITY FULL is only
required for incremental snapshot mode, this should be gated by the
corresponding config; otherwise the validation/error message should be updated
to avoid implying this is incremental-snapshot-only.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java:
##########
@@ -196,13 +200,64 @@ public Map<TableId, TableChange>
discoverDataCollectionSchemas(JdbcSourceConfig
try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
// fetch table schemas
Map<TableId, TableChange> tableSchemas = queryTableSchema(jdbc,
capturedTableIds);
+ // validate REPLICA IDENTITY for tables without primary key
+ validateReplicaIdentityForNoPkTables(jdbc, tableSchemas);
return tableSchemas;
} catch (Exception e) {
throw new FlinkRuntimeException(
"Error to discover table schemas: " + e.getMessage(), e);
}
}
+ /**
+ * Validates that tables without primary key have REPLICA IDENTITY FULL
set. This is required
+ * for PostgreSQL to include the full before image in WAL for UPDATE and
DELETE events.
+ */
+ private void validateReplicaIdentityForNoPkTables(
+ JdbcConnection jdbc, Map<TableId, TableChange> tableSchemas)
throws SQLException {
+ for (Map.Entry<TableId, TableChange> entry : tableSchemas.entrySet()) {
+ TableId tableId = entry.getKey();
+ Table table = entry.getValue().getTable();
+ if (table.primaryKeyColumnNames().isEmpty()) {
+ String replicaIdentity = queryReplicaIdentity(jdbc, tableId);
+ if (!"f".equalsIgnoreCase(replicaIdentity)) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Table '%s.%s' has no primary key. "
+ + "To use incremental snapshot for
tables without primary key, "
+ + "REPLICA IDENTITY FULL must be
set. "
+ + "Please execute: ALTER TABLE
%s.%s REPLICA IDENTITY FULL",
+ tableId.schema(),
+ tableId.table(),
+ tableId.schema(),
+ tableId.table()));
+ }
+ LOG.info(
+ "Table '{}.{}' has no primary key but has REPLICA
IDENTITY FULL set.",
+ tableId.schema(),
+ tableId.table());
+ }
+ }
+ }
+
+ private String queryReplicaIdentity(JdbcConnection jdbc, TableId tableId)
throws SQLException {
+ String query =
+ String.format(
+ "SELECT relreplident FROM pg_class c "
+ + "JOIN pg_namespace n ON c.relnamespace =
n.oid "
+ + "WHERE n.nspname = '%s' AND c.relname =
'%s'",
+ tableId.schema(), tableId.table());
+ final String[] result = new String[1];
+ jdbc.query(
+ query,
+ rs -> {
+ if (rs.next()) {
+ result[0] = rs.getString(1);
+ }
+ });
Review Comment:
`queryReplicaIdentity(...)` returns `null` when the table isn't found; the
caller then throws the REPLICA IDENTITY FULL exception, which becomes
misleading. Consider explicitly handling the "no row returned" case (e.g.,
throw a dedicated exception indicating the table lookup failed).
```suggestion
});
if (result[0] == null) {
throw new FlinkRuntimeException(
String.format(
"Failed to query replica identity for table
'%s.%s': table was not found.",
tableId.schema(), tableId.table()));
}
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java:
##########
@@ -196,13 +200,64 @@ public Map<TableId, TableChange>
discoverDataCollectionSchemas(JdbcSourceConfig
try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
// fetch table schemas
Map<TableId, TableChange> tableSchemas = queryTableSchema(jdbc,
capturedTableIds);
+ // validate REPLICA IDENTITY for tables without primary key
+ validateReplicaIdentityForNoPkTables(jdbc, tableSchemas);
return tableSchemas;
} catch (Exception e) {
throw new FlinkRuntimeException(
"Error to discover table schemas: " + e.getMessage(), e);
}
}
+ /**
+ * Validates that tables without primary key have REPLICA IDENTITY FULL
set. This is required
+ * for PostgreSQL to include the full before image in WAL for UPDATE and
DELETE events.
+ */
+ private void validateReplicaIdentityForNoPkTables(
+ JdbcConnection jdbc, Map<TableId, TableChange> tableSchemas)
throws SQLException {
+ for (Map.Entry<TableId, TableChange> entry : tableSchemas.entrySet()) {
+ TableId tableId = entry.getKey();
+ Table table = entry.getValue().getTable();
+ if (table.primaryKeyColumnNames().isEmpty()) {
+ String replicaIdentity = queryReplicaIdentity(jdbc, tableId);
+ if (!"f".equalsIgnoreCase(replicaIdentity)) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Table '%s.%s' has no primary key. "
+ + "To use incremental snapshot for
tables without primary key, "
+ + "REPLICA IDENTITY FULL must be
set. "
+ + "Please execute: ALTER TABLE
%s.%s REPLICA IDENTITY FULL",
+ tableId.schema(),
+ tableId.table(),
+ tableId.schema(),
+ tableId.table()));
+ }
+ LOG.info(
+ "Table '{}.{}' has no primary key but has REPLICA
IDENTITY FULL set.",
+ tableId.schema(),
+ tableId.table());
+ }
+ }
+ }
+
+ private String queryReplicaIdentity(JdbcConnection jdbc, TableId tableId)
throws SQLException {
+ String query =
+ String.format(
+ "SELECT relreplident FROM pg_class c "
+ + "JOIN pg_namespace n ON c.relnamespace =
n.oid "
+ + "WHERE n.nspname = '%s' AND c.relname =
'%s'",
+ tableId.schema(), tableId.table());
Review Comment:
`queryReplicaIdentity(...)` builds SQL by string interpolation of
schema/table names. This can break for identifiers containing
quotes/case-sensitive names and is also vulnerable to SQL injection if table
identifiers come from user configuration; prefer a parameterized query and/or
proper identifier quoting.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/inventory_no_pk.sql:
##########
@@ -0,0 +1,36 @@
+-- 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.
+
+SET search_path TO inventory;
+
+-- Create table without primary key for no-PK table tests
+CREATE TABLE products_no_pk (
+ id INTEGER NOT NULL,
+ name VARCHAR(255) NOT NULL DEFAULT 'flink',
+ description VARCHAR(512),
+ weight FLOAT
+);
+ALTER TABLE products_no_pk REPLICA IDENTITY FULL;
+
+INSERT INTO products_no_pk
+VALUES (101,'scooter','Small 2-wheel scooter',3.14),
+ (102,'car battery','12V car battery',8.1),
+ (103,'12-pack drill bits','12-pack of drill bits with sizes ranging
from #40 to #3',0.8),
+ (104,'hammer','12oz carpenter''s hammer',0.75),
+ (105,'hammer','14oz carpenter''s hammer',0.875),
+ (106,'hammer','16oz carpenter''s hammer',1.0),
+ (107,'rocks','box of assorted rocks',5.3),
+ (108,'jacket','water resistent black wind breaker',0.1),
Review Comment:
Spelling: "water resistent" should be "water resistant" in test data to
avoid propagating typos into expected results/examples.
```suggestion
(108,'jacket','water resistant black wind breaker',0.1),
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java:
##########
@@ -49,6 +51,8 @@
@Internal
public abstract class JdbcSourceFetchTaskContext implements FetchTask.Context {
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcSourceFetchTaskContext.class);
+
Review Comment:
`LOG` is declared but never used in this class; please remove it (or use it)
to avoid unused-field/static-analysis violations.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java:
##########
@@ -1160,4 +1160,148 @@ void testUniqueIndexIncludingFunction(boolean
parallelismSnapshot) throws Except
tableResult.getJobClient().get().cancel().get();
RowUtils.USE_LEGACY_TO_STRING = true;
}
+
+ @Test
+ void testNoPKTableWithChunkKey() throws SQLException, ExecutionException,
InterruptedException {
+ setup(true);
+ initializePostgresTable(POSTGRES_CONTAINER, "inventory");
+ initializePostgresTable(POSTGRES_CONTAINER, "inventory_no_pk");
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE no_pk_source ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " description STRING,"
+ + " weight DECIMAL(10,3)"
+ + ") WITH ("
+ + " 'connector' = 'postgres-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'schema-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' =
'true',"
+ + "
'scan.incremental.snapshot.chunk.key-column' = 'id',"
+ + " 'scan.incremental.snapshot.chunk.size' =
'4',"
+ + " 'decoding.plugin.name' = 'pgoutput', "
+ + " 'slot.name' = '%s'"
+ + ")",
+ POSTGRES_CONTAINER.getHost(),
+ POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT),
+ POSTGRES_CONTAINER.getUsername(),
+ POSTGRES_CONTAINER.getPassword(),
+ POSTGRES_CONTAINER.getDatabaseName(),
+ "inventory",
+ "products_no_pk",
+ getSlotName());
+ String sinkDDL =
+ "CREATE TABLE sink ("
+ + " name STRING,"
+ + " weightSum DECIMAL(10,3),"
+ + " PRIMARY KEY (name) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")";
+ tEnv.executeSql(sourceDDL);
+ tEnv.executeSql(sinkDDL);
+
+ // async submit job
+ TableResult result =
+ tEnv.executeSql(
+ "INSERT INTO sink SELECT name, SUM(weight) FROM
no_pk_source GROUP BY name");
+
+ waitForSnapshotStarted("sink");
+
+ // wait a bit to make sure the replication slot is ready
+ Thread.sleep(5000);
+
+ // generate WAL
+ try (Connection connection = getJdbcConnection(POSTGRES_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "UPDATE inventory.products_no_pk SET description='18oz
carpenter hammer' WHERE id=106;");
+ statement.execute("UPDATE inventory.products_no_pk SET
weight='5.1' WHERE id=107;");
+ statement.execute(
+ "INSERT INTO inventory.products_no_pk VALUES
(110,'jacket','water resistent white wind breaker',0.2);");
+ statement.execute(
+ "INSERT INTO inventory.products_no_pk VALUES
(111,'scooter','Big 2-wheel scooter',5.18);");
+ statement.execute(
+ "UPDATE inventory.products_no_pk SET description='new
water resistent white wind breaker', weight='0.5' WHERE id=110;");
Review Comment:
Spelling: "water resistent" should be "water resistant" in the UPDATE
statement string to keep test data clean/consistent.
```suggestion
"INSERT INTO inventory.products_no_pk VALUES
(110,'jacket','water resistant white wind breaker',0.2);");
statement.execute(
"INSERT INTO inventory.products_no_pk VALUES
(111,'scooter','Big 2-wheel scooter',5.18);");
statement.execute(
"UPDATE inventory.products_no_pk SET description='new
water resistant white wind breaker', weight='0.5' WHERE id=110;");
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SplitKeyUtils.java:
##########
@@ -237,7 +238,29 @@ public static Object[] getSplitKey(
RowType splitBoundaryType, SourceRecord dataRecord,
SchemaNameAdjuster nameAdjuster) {
// the split key field contains single field now
String splitFieldName =
nameAdjuster.adjust(splitBoundaryType.getFieldNames().get(0));
- Struct key = (Struct) dataRecord.key();
- return new Object[] {key.get(splitFieldName)};
+ Struct target;
+ if (dataRecord.key() != null) {
+ target = (Struct) dataRecord.key();
+ } else {
+ // For tables without primary key, extract chunk key from value
struct
+ target = getStructContainingChunkKey(dataRecord);
+ }
+ return new Object[] {target.get(splitFieldName)};
+ }
+
+ /**
+ * For tables without primary key, the chunk key is not in the record key
(which is null).
+ * Instead, extract it from the value's after struct (for CREATE/READ) or
before struct (for
+ * UPDATE/DELETE).
+ */
+ public static Struct getStructContainingChunkKey(SourceRecord record) {
+ Struct value = (Struct) record.value();
+ Envelope.Operation op =
+
Envelope.Operation.forCode(value.getString(Envelope.FieldName.OPERATION));
+ if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
+ return value.getStruct(Envelope.FieldName.AFTER);
+ } else {
+ return value.getStruct(Envelope.FieldName.BEFORE);
+ }
Review Comment:
`getStructContainingChunkKey(...)` is only used inside `SplitKeyUtils`;
consider making it `private` to avoid expanding the public API surface
unnecessarily.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java:
##########
@@ -1160,4 +1160,148 @@ void testUniqueIndexIncludingFunction(boolean
parallelismSnapshot) throws Except
tableResult.getJobClient().get().cancel().get();
RowUtils.USE_LEGACY_TO_STRING = true;
}
+
+ @Test
+ void testNoPKTableWithChunkKey() throws SQLException, ExecutionException,
InterruptedException {
+ setup(true);
+ initializePostgresTable(POSTGRES_CONTAINER, "inventory");
+ initializePostgresTable(POSTGRES_CONTAINER, "inventory_no_pk");
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE no_pk_source ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " description STRING,"
+ + " weight DECIMAL(10,3)"
+ + ") WITH ("
+ + " 'connector' = 'postgres-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'schema-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' =
'true',"
+ + "
'scan.incremental.snapshot.chunk.key-column' = 'id',"
+ + " 'scan.incremental.snapshot.chunk.size' =
'4',"
+ + " 'decoding.plugin.name' = 'pgoutput', "
+ + " 'slot.name' = '%s'"
+ + ")",
+ POSTGRES_CONTAINER.getHost(),
+ POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT),
+ POSTGRES_CONTAINER.getUsername(),
+ POSTGRES_CONTAINER.getPassword(),
+ POSTGRES_CONTAINER.getDatabaseName(),
+ "inventory",
+ "products_no_pk",
+ getSlotName());
+ String sinkDDL =
+ "CREATE TABLE sink ("
+ + " name STRING,"
+ + " weightSum DECIMAL(10,3),"
+ + " PRIMARY KEY (name) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")";
+ tEnv.executeSql(sourceDDL);
+ tEnv.executeSql(sinkDDL);
+
+ // async submit job
+ TableResult result =
+ tEnv.executeSql(
+ "INSERT INTO sink SELECT name, SUM(weight) FROM
no_pk_source GROUP BY name");
+
+ waitForSnapshotStarted("sink");
+
+ // wait a bit to make sure the replication slot is ready
+ Thread.sleep(5000);
+
+ // generate WAL
+ try (Connection connection = getJdbcConnection(POSTGRES_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "UPDATE inventory.products_no_pk SET description='18oz
carpenter hammer' WHERE id=106;");
+ statement.execute("UPDATE inventory.products_no_pk SET
weight='5.1' WHERE id=107;");
+ statement.execute(
+ "INSERT INTO inventory.products_no_pk VALUES
(110,'jacket','water resistent white wind breaker',0.2);");
+ statement.execute(
+ "INSERT INTO inventory.products_no_pk VALUES
(111,'scooter','Big 2-wheel scooter',5.18);");
+ statement.execute(
+ "UPDATE inventory.products_no_pk SET description='new
water resistent white wind breaker', weight='0.5' WHERE id=110;");
Review Comment:
Spelling: "water resistent" should be "water resistant" in the inserted test
row (and adjust the subsequent UPDATE string accordingly).
```suggestion
"INSERT INTO inventory.products_no_pk VALUES
(110,'jacket','water resistant white wind breaker',0.2);");
statement.execute(
"INSERT INTO inventory.products_no_pk VALUES
(111,'scooter','Big 2-wheel scooter',5.18);");
statement.execute(
"UPDATE inventory.products_no_pk SET description='new
water resistant white wind breaker', weight='0.5' WHERE id=110;");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
