This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new a9322335b0 [FLINK-40512] Cache typeRegistry to reduce the busy work on
inferring schema change event (#4519)
a9322335b0 is described below
commit a9322335b0d701d24b38090f4b873c612d7de22c
Author: Hongshun Wang <[email protected]>
AuthorDate: Wed Sep 2 19:09:31 2026 +0800
[FLINK-40512] Cache typeRegistry to reduce the busy work on inferring
schema change event (#4519)
---
.../reader/PostgresPipelineRecordEmitter.java | 18 ++++++++++++++-
.../postgres/utils/SchemaChangeUtil.java | 26 +++++-----------------
2 files changed, 22 insertions(+), 22 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java
index 6df94c28b0..05ba856f0b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java
@@ -32,6 +32,7 @@ import
org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema;
import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import io.debezium.connector.postgresql.TypeRegistry;
import io.debezium.connector.postgresql.connection.PostgresConnection;
import io.debezium.data.Envelope;
import io.debezium.relational.Table;
@@ -66,6 +67,8 @@ public class PostgresPipelineRecordEmitter<T> extends
PostgresSourceRecordEmitte
private final boolean includeDatabaseInTableId;
private final Map<TableId, CreateTableEvent> createTableEventCache;
+ private TypeRegistry typeRegistry;
+
public PostgresPipelineRecordEmitter(
DebeziumDeserializationSchema<T> debeziumDeserializationSchema,
SourceReaderMetrics sourceReaderMetrics,
@@ -128,7 +131,11 @@ public class PostgresPipelineRecordEmitter<T> extends
PostgresSourceRecordEmitte
}
List<SchemaChangeEvent> schemaChangeEvents =
inferSchemaChangeEvent(
- schemaAfter.id(), schemaBefore, schemaAfter,
sourceConfig, postgresDialect);
+ schemaAfter.id(),
+ schemaBefore,
+ schemaAfter,
+ sourceConfig,
+ getTypeRegistry());
LOG.info("Inferred Schema change events: {}", schemaChangeEvents);
schemaChangeEvents.forEach(
schemaChangeEvent -> {
@@ -205,6 +212,15 @@ public class PostgresPipelineRecordEmitter<T> extends
PostgresSourceRecordEmitte
}
}
+ private TypeRegistry getTypeRegistry() {
+ if (typeRegistry == null) {
+ try (PostgresConnection jdbc =
postgresDialect.openJdbcConnection()) {
+ typeRegistry = jdbc.getTypeRegistry();
+ }
+ }
+ return typeRegistry;
+ }
+
private TableId getTableId(SourceRecord dataRecord) {
Struct value = (Struct) dataRecord.value();
Struct source = value.getStruct(Envelope.FieldName.SOURCE);
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/SchemaChangeUtil.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/SchemaChangeUtil.java
index c92c83f0b5..53e77e3bfd 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/SchemaChangeUtil.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/SchemaChangeUtil.java
@@ -25,12 +25,10 @@ import org.apache.flink.cdc.common.event.RenameColumnEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.types.DataType;
-import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
import
org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
import io.debezium.connector.postgresql.PostgresConnectorConfig;
import io.debezium.connector.postgresql.TypeRegistry;
-import io.debezium.connector.postgresql.connection.PostgresConnection;
import io.debezium.relational.Column;
import io.debezium.relational.Table;
@@ -83,10 +81,11 @@ public class SchemaChangeUtil {
@Nullable Table tableBefore,
Table tableAfter,
PostgresSourceConfig sourceConfig,
- PostgresDialect dialect) {
+ TypeRegistry typeRegistry) {
if (tableBefore == null) {
- return Collections.singletonList(toCreateTableEvent(tableAfter,
sourceConfig, dialect));
+ return Collections.singletonList(
+ toCreateTableEvent(tableAfter, sourceConfig,
typeRegistry));
}
TableId cdcTableId =
@@ -95,23 +94,8 @@ public class SchemaChangeUtil {
sourceConfig.getDatabaseList().get(0),
sourceConfig.isIncludeDatabaseInTableId());
PostgresConnectorConfig dbzConfig =
sourceConfig.getDbzConnectorConfig();
-
- try (PostgresConnection connection = dialect.openJdbcConnection()) {
- TypeRegistry typeRegistry = connection.getTypeRegistry();
- return inferMinimalSchemaChanges(
- cdcTableId,
- tableBefore.columns(),
- tableAfter.columns(),
- dbzConfig,
- typeRegistry);
- }
- }
-
- public static CreateTableEvent toCreateTableEvent(
- Table table, PostgresSourceConfig sourceConfig, PostgresDialect
dialect) {
- try (PostgresConnection connection = dialect.openJdbcConnection()) {
- return toCreateTableEvent(table, sourceConfig,
connection.getTypeRegistry());
- }
+ return inferMinimalSchemaChanges(
+ cdcTableId, tableBefore.columns(), tableAfter.columns(),
dbzConfig, typeRegistry);
}
private static CreateTableEvent toCreateTableEvent(
