Skip to content

[FLINK-40453][postgres] Not load table schema when initializing record emitter #4513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
naivedogger wants to merge 1 commit into
base: master
Choose a base branch
from
+166 −87
Open

[FLINK-40453][postgres] Not load table schema when initializing record emitter #4513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 63 additions & 87 deletions ...org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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<T> extends PostgresSourceRecordEmitter<T> {
Expand All @@ -69,13 +63,9 @@ public class PostgresPipelineRecordEmitter<T> extends PostgresSourceRecordEmitte

// Used when startup mode is initial
private final Set<TableId> alreadySendCreateTableTables;
private final boolean isBounded;
private final boolean includeDatabaseInTableId;
private final Map<TableId, CreateTableEvent> createTableEventCache;

// Used when startup mode is not initial
private boolean shouldEmitAllCreateTableEventsInSnapshotMode = true;

public PostgresPipelineRecordEmitter(
DebeziumDeserializationSchema<T> debeziumDeserializationSchema,
SourceReaderMetrics sourceReaderMetrics,
Expand All @@ -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<TableId, TableChanges.TableChange> 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<T> 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<T> output) {
private void handleDataChangeRecord(
SourceRecord element, SourceOutput<T> 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(
Expand All @@ -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<SchemaChangeEvent> 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<T> 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<T> 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<T> 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()) {
Expand All @@ -213,31 +216,4 @@ private TableId getTableId(SourceRecord dataRecord) {
String tableName = source.getString(TABLE_NAME_KEY);
return new TableId(null, schemaName, tableName);
}

private Map<TableId, CreateTableEvent> generateCreateTableEvent(
PostgresSourceConfig sourceConfig) {
try (PostgresConnection jdbc = postgresDialect.openJdbcConnection()) {
Map<TableId, CreateTableEvent> createTableEventCache = new HashMap<>();
List<TableId> 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);
}
}
}
103 changes: 103 additions & 0 deletions ...apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<Event> emitter = createEmitter(sourceConfig);

assertThatCode(() -> emitter.applySplit(createSnapshotSplit())).doesNotThrowAnyException();
}

private static PostgresPipelineRecordEmitter<Event> 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.");
}
}
}
Loading