loserwang1024 commented on code in PR #4403:
URL: https://github.com/apache/flink-cdc/pull/4403#discussion_r4535289619
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/utils/CustomPostgresSchema.java:
##########
@@ -119,26 +118,37 @@ private List<TableChange> readTableSchema(List<TableId>
tableIds) throws SQLExce
throw new FlinkRuntimeException("Failed to read schema", e);
}
- for (TableId tableId : tableIds) {
- Table table = Objects.requireNonNull(tables.forTable(tableId));
- // set the events to populate proper sourceInfo into offsetContext
- offsetContext.event(tableId, Instant.now());
-
- // TODO: check whether we always set isFromSnapshot = true
+ // Cache all tables discovered by readSchema to avoid redundant full
scans on future calls.
Review Comment:
Could we address the redundant metadata reads by grouping the requested,
uncached `TableId`s by schema and calling `readSchema` with the corresponding
schema pattern and an exact table filter, while caching only the requested
tables? This would reduce unnecessary column and primary-key metadata lookups
without retaining schemas that this instance may never need—for example, after
checkpoint recovery when only a few tables remain to be split. It also
preserves the existing lazy-loading behavior. The schema names should be
escaped as JDBC search patterns so that characters such as `_` and `%` are
treated literally.
```java
List<TableChange> readTableSchema(List<TableId> tableIds) throws
SQLException {
List<TableChange> tableChanges = new ArrayList<>();
final PostgresOffsetContext offsetContext =
PostgresOffsetContext.initialContext(dbzConfig, jdbcConnection,
Clock.SYSTEM);
PostgresPartition partition = new
PostgresPartition(dbzConfig.getLogicalName());
Tables tables = new Tables();
for (Map.Entry<String, Set<TableId>> entry :
groupTableIdsBySchema(tableIds).entrySet()) {
try {
jdbcConnection.readSchema(
tables,
dbzConfig.databaseName(),
// Restrict metadata discovery to the current schema.
entry.getKey(),
// Read only requested tables instead of all tables
matched by the global
// filter.
tableIdFilter(entry.getValue()),
null,
false);
} catch (SQLException e) {
throw new FlinkRuntimeException("Failed to read schema", e);
}
}
for (TableId tableId : tableIds) {
Table table = Objects.requireNonNull(tables.forTable(tableId));
// set the events to populate proper sourceInfo into offsetContext
offsetContext.event(tableId, Instant.now());
// TODO: check whether we always set isFromSnapshot = true

SchemaChangeEvent schemaChangeEvent =
SchemaChangeEvent.ofCreate(
partition,
offsetContext,
dbzConfig.databaseName(),
tableId.schema(),
null,
table,
true);
for (TableChanges.TableChange tableChange :
schemaChangeEvent.getTableChanges()) {
this.schemasByTableId.put(tableId, tableChange);
}
tableChanges.add(this.schemasByTableId.get(tableId));
}
return tableChanges;
}
static Map<String, Set<TableId>> groupTableIdsBySchema(List<TableId>
tableIds) {
Map<String, Set<TableId>> tableIdsBySchema = new HashMap<>();
for (TableId tableId : tableIds) {
tableIdsBySchema
.computeIfAbsent(tableId.schema(), ignored -> new
HashSet<>())
.add(tableId);
}
return tableIdsBySchema;
}
static Tables.TableFilter tableIdFilter(Set<TableId> tableIds) {
return tableIds::contains;
}
```
--
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]
