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 1030263e841..6df94c28b0d 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 @@ -21,16 +21,12 @@ import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.SchemaChangeEvent; import org.apache.flink.cdc.common.schema.Schema; -import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory; -import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit; -import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase; import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState; import org.apache.flink.cdc.connectors.base.source.metrics.SourceReaderMetrics; import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect; import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; import org.apache.flink.cdc.connectors.postgres.source.schema.PostgresSchemaRecord; -import org.apache.flink.cdc.connectors.postgres.source.utils.TableDiscoveryUtils; import org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils; import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema; import org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema; @@ -41,14 +37,13 @@ import io.debezium.relational.Table; import io.debezium.relational.TableId; import io.debezium.relational.history.TableChanges; +import io.debezium.relational.history.TableChanges.TableChange; import org.apache.kafka.connect.data.Field; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.source.SourceRecord; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.sql.SQLException; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -60,7 +55,6 @@ import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.isSchemaChangeEvent; import static org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils.toCdcTableId; import static org.apache.flink.cdc.connectors.postgres.utils.SchemaChangeUtil.inferSchemaChangeEvent; -import static org.apache.flink.cdc.connectors.postgres.utils.SchemaChangeUtil.toCreateTableEvent; /** The {@link RecordEmitter} implementation for PostgreSQL pipeline connector. */ public class PostgresPipelineRecordEmitter extends PostgresSourceRecordEmitter { @@ -69,13 +63,9 @@ public class PostgresPipelineRecordEmitter extends PostgresSourceRecordEmitte // Used when startup mode is initial private final Set alreadySendCreateTableTables; - private final boolean isBounded; private final boolean includeDatabaseInTableId; private final Map createTableEventCache; - // Used when startup mode is not initial - private boolean shouldEmitAllCreateTableEventsInSnapshotMode = true; - public PostgresPipelineRecordEmitter( DebeziumDeserializationSchema debeziumDeserializationSchema, SourceReaderMetrics sourceReaderMetrics, @@ -94,61 +84,27 @@ public PostgresPipelineRecordEmitter( this.createTableEventCache = ((DebeziumEventDeserializationSchema) debeziumDeserializationSchema) .getCreateTableEventCache(); - generateCreateTableEvent(sourceConfig); - this.isBounded = StartupOptions.snapshot().equals(sourceConfig.getStartupOptions()); - } - - @Override - public void applySplit(SourceSplitBase split) { - if ((isBounded) && createTableEventCache.isEmpty() && split instanceof SnapshotSplit) { - // TableSchemas in SnapshotSplit only contains one table. - createTableEventCache.putAll(generateCreateTableEvent(sourceConfig)); - } else { - for (Map.Entry entry : - split.getTableSchemas().entrySet()) { - TableChanges.TableChange tableChange = entry.getValue(); - - Table table = tableChange.getTable(); - CreateTableEvent createTableEvent = - toCreateTableEvent(table, sourceConfig, postgresDialect); - ((DebeziumEventDeserializationSchema) debeziumDeserializationSchema) - .applyChangeEvent(createTableEvent); - } - } } @Override protected void processElement( SourceRecord element, SourceOutput output, SourceSplitState splitState) throws Exception { - if (shouldEmitAllCreateTableEventsInSnapshotMode && isBounded) { - // In snapshot mode, we simply emit all schemas at once. - createTableEventCache.forEach( - (tableId, createTableEvent) -> { - output.collect((T) createTableEvent); - }); - shouldEmitAllCreateTableEventsInSnapshotMode = false; - } else if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) { + if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) { TableId tableId = splitState.asSnapshotSplitState().toSourceSplit().getTableId(); - maybeSendCreateTableEventFromCache(tableId, output); + sendCreateTableEventIfNeeded(tableId, output, splitState); } else if (isDataChangeRecord(element)) { - handleDataChangeRecord(element, output); + handleDataChangeRecord(element, output, splitState); } else if (isSchemaChangeEvent(element) && sourceConfig.isIncludeSchemaChanges()) { handleSchemaChangeRecord(element, output, splitState); } super.processElement(element, output, splitState); } - private void handleDataChangeRecord(SourceRecord element, SourceOutput output) { + private void handleDataChangeRecord( + SourceRecord element, SourceOutput output, SourceSplitState splitState) { TableId tableId = getTableId(element); - maybeSendCreateTableEventFromCache(tableId, output); - // In rare case, we may miss some CreateTableEvents before DataChangeEvents. - // Don't send CreateTableEvent for SchemaChangeEvents as it's the latest schema. - if (!createTableEventCache.containsKey(tableId)) { - CreateTableEvent createTableEvent = getCreateTableEvent(sourceConfig, tableId); - sendCreateTableEvent(createTableEvent, output); - createTableEventCache.put(tableId, createTableEvent); - } + sendCreateTableEventIfNeeded(tableId, output, splitState); } private void handleSchemaChangeRecord( @@ -163,32 +119,79 @@ private void handleSchemaChangeRecord( splitState.toSourceSplit().getTableSchemas(); PostgresSchemaRecord schemaRecord = (PostgresSchemaRecord) element; Table schemaAfter = schemaRecord.getTable(); - maybeSendCreateTableEventFromCache(schemaAfter.id(), output); Table schemaBefore = null; if (existedTableSchemas.containsKey(schemaAfter.id())) { schemaBefore = existedTableSchemas.get(schemaAfter.id()).getTable(); } + if (schemaBefore != null) { + sendCreateTableEventIfNeeded(schemaAfter.id(), output, splitState); + } List schemaChangeEvents = inferSchemaChangeEvent( schemaAfter.id(), schemaBefore, schemaAfter, sourceConfig, postgresDialect); LOG.info("Inferred Schema change events: {}", schemaChangeEvents); - schemaChangeEvents.forEach(schemaChangeEvent -> output.collect((T) schemaChangeEvent)); + schemaChangeEvents.forEach( + schemaChangeEvent -> { + output.collect((T) schemaChangeEvent); + if (schemaChangeEvent instanceof CreateTableEvent) { + cacheCreateTableEvent( + schemaAfter.id(), (CreateTableEvent) schemaChangeEvent); + alreadySendCreateTableTables.add(schemaAfter.id()); + } + }); } - private void maybeSendCreateTableEventFromCache(TableId tableId, SourceOutput output) { - if (!alreadySendCreateTableTables.contains(tableId)) { - CreateTableEvent createTableEvent = createTableEventCache.get(tableId); - if (createTableEvent != null) { - sendCreateTableEvent(createTableEvent, output); - } - alreadySendCreateTableTables.add(tableId); + private void sendCreateTableEventIfNeeded( + TableId tableId, SourceOutput output, SourceSplitState splitState) { + if (alreadySendCreateTableTables.contains(tableId)) { + return; } + + CreateTableEvent createTableEvent = getOrCreateCreateTableEvent(tableId, splitState); + sendCreateTableEvent(createTableEvent, output); + alreadySendCreateTableTables.add(tableId); } private void sendCreateTableEvent(CreateTableEvent createTableEvent, SourceOutput output) { output.collect((T) createTableEvent); } + private CreateTableEvent getOrCreateCreateTableEvent( + TableId tableId, SourceSplitState splitState) { + CreateTableEvent createTableEvent = createTableEventCache.get(tableId); + if (createTableEvent == null) { + createTableEvent = getCreateTableEventFromSplit(tableId, splitState); + } + if (createTableEvent == null) { + createTableEvent = getCreateTableEvent(sourceConfig, tableId); + } + cacheCreateTableEvent(tableId, createTableEvent); + return createTableEvent; + } + + private CreateTableEvent getCreateTableEventFromSplit( + TableId tableId, SourceSplitState splitState) { + TableChange tableChange = splitState.toSourceSplit().getTableSchemas().get(tableId); + if (tableChange == null || tableChange.getTable() == null) { + return null; + } + try (PostgresConnection jdbc = postgresDialect.openJdbcConnection()) { + return new CreateTableEvent( + toCdcTableId( + tableId, + sourceConfig.getDatabaseList().get(0), + includeDatabaseInTableId), + PostgresSchemaUtils.toSchema( + tableChange.getTable(), + sourceConfig.getDbzConnectorConfig(), + jdbc.getTypeRegistry())); + } + } + + private void cacheCreateTableEvent(TableId tableId, CreateTableEvent createTableEvent) { + createTableEventCache.put(tableId, createTableEvent); + } + private CreateTableEvent getCreateTableEvent( PostgresSourceConfig sourceConfig, TableId tableId) { try (PostgresConnection jdbc = postgresDialect.openJdbcConnection()) { @@ -213,31 +216,4 @@ private TableId getTableId(SourceRecord dataRecord) { String tableName = source.getString(TABLE_NAME_KEY); return new TableId(null, schemaName, tableName); } - - private Map generateCreateTableEvent( - PostgresSourceConfig sourceConfig) { - try (PostgresConnection jdbc = postgresDialect.openJdbcConnection()) { - Map createTableEventCache = new HashMap<>(); - List capturedTableIds = - TableDiscoveryUtils.listTables( - sourceConfig.getDatabaseList().get(0), - jdbc, - sourceConfig.getTableFilters(), - sourceConfig.includePartitionedTables()); - for (TableId tableId : capturedTableIds) { - Schema schema = PostgresSchemaUtils.getTableSchema(tableId, sourceConfig, jdbc); - createTableEventCache.put( - tableId, - new CreateTableEvent( - toCdcTableId( - tableId, - this.sourceConfig.getDatabaseList().get(0), - includeDatabaseInTableId), - schema)); - } - return createTableEventCache; - } catch (SQLException e) { - throw new RuntimeException("Cannot start emitter to fetch table schema.", e); - } - } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitterTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitterTest.java new file mode 100644 index 00000000000..8d54cba23ba --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitterTest.java @@ -0,0 +1,103 @@ +/* + * 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. + */ + +package org.apache.flink.cdc.connectors.postgres.source.reader; + +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit; +import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect; +import org.apache.flink.cdc.connectors.postgres.source.PostgresEventDeserializer; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory; +import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; + +import io.debezium.connector.postgresql.connection.PostgresConnection; +import io.debezium.relational.TableId; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThatCode; + +/** Tests for {@link PostgresPipelineRecordEmitter}. */ +class PostgresPipelineRecordEmitterTest { + + private static final TableId DBZ_TABLE_ID = new TableId(null, "public", "users"); + + @Test + void testInitializationDoesNotLoadTableSchemas() { + PostgresSourceConfig sourceConfig = createSourceConfig(StartupOptions.initial()); + + assertThatCode(() -> createEmitter(sourceConfig)).doesNotThrowAnyException(); + } + + @Test + void testSnapshotSplitAssignmentDoesNotLoadAllTableSchemas() { + PostgresSourceConfig sourceConfig = createSourceConfig(StartupOptions.snapshot()); + PostgresPipelineRecordEmitter emitter = createEmitter(sourceConfig); + + assertThatCode(() -> emitter.applySplit(createSnapshotSplit())).doesNotThrowAnyException(); + } + + private static PostgresPipelineRecordEmitter createEmitter( + PostgresSourceConfig sourceConfig) { + return new PostgresPipelineRecordEmitter<>( + new PostgresEventDeserializer(DebeziumChangelogMode.ALL), + null, + sourceConfig, + new PostgresOffsetFactory(), + new FailingPostgresDialect(sourceConfig)); + } + + private static PostgresSourceConfig createSourceConfig(StartupOptions startupOptions) { + PostgresSourceConfigFactory configFactory = new PostgresSourceConfigFactory(); + configFactory.hostname("localhost"); + configFactory.port(5432); + configFactory.username("user"); + configFactory.password("password"); + configFactory.database("db0"); + configFactory.tableList("public.users"); + configFactory.startupOptions(startupOptions); + return configFactory.create(0); + } + + private static SnapshotSplit createSnapshotSplit() { + return new SnapshotSplit( + DBZ_TABLE_ID, + "public.users:0", + RowType.of(new IntType()), + null, + null, + null, + Collections.emptyMap()); + } + + private static class FailingPostgresDialect extends PostgresDialect { + private FailingPostgresDialect(PostgresSourceConfig sourceConfig) { + super(sourceConfig); + } + + @Override + public PostgresConnection openJdbcConnection() { + throw new AssertionError("JDBC connection should be opened lazily."); + } + } +}