Skip to content

[FLINK-39965][runtime] Correct schema state after DropTableEvent #4448

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
taoran92 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-39965][runtime] Correct schema state after DropTableEvent #4448

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
7 changes: 7 additions & 0 deletions ...ime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.flink.cdc.common.annotation.Internal;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
Expand Down Expand Up @@ -151,6 +152,9 @@ public Schema getOriginalSchema(TableId tableId, int version) {
public void applyOriginalSchemaChange(SchemaChangeEvent schemaChangeEvent) {
if (schemaChangeEvent instanceof CreateTableEvent) {
handleCreateTableEvent(originalSchemas, ((CreateTableEvent) schemaChangeEvent));
} else if (schemaChangeEvent instanceof DropTableEvent) {
// Drop ends the table schema lifecycle.
originalSchemas.remove(schemaChangeEvent.tableId());
} else {
Optional<Schema> optionalSchema = getLatestOriginalSchema(schemaChangeEvent.tableId());
checkArgument(
Expand All @@ -168,6 +172,9 @@ public void applyOriginalSchemaChange(SchemaChangeEvent schemaChangeEvent) {
public void applyEvolvedSchemaChange(SchemaChangeEvent schemaChangeEvent) {
if (schemaChangeEvent instanceof CreateTableEvent) {
handleCreateTableEvent(evolvedSchemas, ((CreateTableEvent) schemaChangeEvent));
} else if (schemaChangeEvent instanceof DropTableEvent) {
// Drop ends the table schema lifecycle.
evolvedSchemas.remove(schemaChangeEvent.tableId());
} else {
Optional<Schema> optionalSchema = getLatestEvolvedSchema(schemaChangeEvent.tableId());
checkArgument(
Expand Down
96 changes: 66 additions & 30 deletions ...ain/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.RouteMode;
Expand Down Expand Up @@ -310,10 +311,16 @@ private void handleSchemaEvolveRequest(
private void updateUpstreamSchemaTable(
TableId tableId, int sourcePartition, SchemaChangeEvent schemaChangeEvent) {
Schema oldSchema = upstreamSchemaTable.get(tableId, sourcePartition);
upstreamSchemaTable.put(
tableId,
sourcePartition,
SchemaUtils.applySchemaChangeEvent(oldSchema, schemaChangeEvent));
if (schemaChangeEvent instanceof DropTableEvent) {
// DropTableEvent is a table-level DDL and ends the table lifecycle for all source
// partitions.
upstreamSchemaTable.row(tableId).clear();
} else {
upstreamSchemaTable.put(
tableId,
sourcePartition,
SchemaUtils.applySchemaChangeEvent(oldSchema, schemaChangeEvent));
}
}

private void startSchemaChange() throws TimeoutException {
Expand Down Expand Up @@ -348,9 +355,8 @@ private void startSchemaChange() throws TimeoutException {
Set<TableId> affectedTableIds = deduceSummary.f0;
Map<TableId, Schema> evolvedSchemaView = new HashMap<>();
for (TableId tableId : affectedTableIds) {
schemaManager
.getLatestEvolvedSchema(tableId)
.ifPresent(schema -> evolvedSchemaView.put(tableId, schema));
evolvedSchemaView.put(
tableId, schemaManager.getLatestEvolvedSchema(tableId).orElse(null));
}

List<Tuple2<SchemaChangeRequest, CompletableFuture<CoordinationResponse>>> futures =
Expand Down Expand Up @@ -417,32 +423,52 @@ private Tuple2<Set<TableId>, List<SchemaChangeEvent>> deduceEvolvedSchemaChanges
Set<TableId> upstreamDependencies =
SchemaDerivator.reverseLookupDependingUpstreamTables(
router, affectedSinkTableId, upstreamSchemaTable);
Preconditions.checkState(
!upstreamDependencies.isEmpty(),
"An affected sink table's upstream dependency cannot be empty.");
LOG.info("Step 3.2 - upstream dependency tables are: {}", upstreamDependencies);

// Then, grab all upstream schemas from all known partitions and merge them.
Set<Schema> toBeMergedSchemas =
SchemaDerivator.reverseLookupDependingUpstreamSchemas(
router, affectedSinkTableId, upstreamSchemaTable);
LOG.info("Step 3.3 - Upstream dependency schemas are: {}.", toBeMergedSchemas);

// In distributed topology, schema will never be narrowed because current schema is
// in the merging base. Notice that current schema might be NULL if it's the first
// time we met a CreateTableEvent.
Schema mergedSchema = currentSinkSchema;
for (Schema toBeMergedSchema : toBeMergedSchemas) {
mergedSchema =
SchemaMergingUtils.getLeastCommonSchema(mergedSchema, toBeMergedSchema);
List<SchemaChangeEvent> localEvolvedSchemaChanges;
if (upstreamDependencies.isEmpty()) {
if (containsDropTableEventForSinkTable(
validSchemaChangeRequests, affectedSinkTableId)) {
if (currentSinkSchema != null) {
localEvolvedSchemaChanges = new ArrayList<>();
localEvolvedSchemaChanges.add(new DropTableEvent(affectedSinkTableId));
LOG.info(
"Step 3.3 - No upstream dependency remains and could be forwarded as {}.",
localEvolvedSchemaChanges);
} else {
localEvolvedSchemaChanges = new ArrayList<>();
LOG.info(
"Step 3.3 - No upstream dependency or current evolved schema remains.");
}
} else {
throw new IllegalStateException(
"An affected sink table's upstream dependency cannot be empty.");
}
} else {
// Then, grab all upstream schemas from all known partitions and merge them.
Set<Schema> toBeMergedSchemas =
SchemaDerivator.reverseLookupDependingUpstreamSchemas(
router, affectedSinkTableId, upstreamSchemaTable);
LOG.info("Step 3.3 - Upstream dependency schemas are: {}.", toBeMergedSchemas);

// In distributed topology, schema will never be narrowed because current schema is
// in the merging base. Notice that current schema might be NULL if it's the first
// time we met a CreateTableEvent.
Schema mergedSchema = currentSinkSchema;
for (Schema toBeMergedSchema : toBeMergedSchemas) {
mergedSchema =
SchemaMergingUtils.getLeastCommonSchema(mergedSchema, toBeMergedSchema);
}
LOG.info("Step 3.4 - Deduced widest schema is: {}.", mergedSchema);

// Detect what schema changes we need to apply to get expected sink table.
localEvolvedSchemaChanges =
SchemaMergingUtils.getSchemaDifference(
affectedSinkTableId, currentSinkSchema, mergedSchema);
LOG.info(
"Step 3.5 - Corresponding schema changes are: {}.",
localEvolvedSchemaChanges);
}
LOG.info("Step 3.4 - Deduced widest schema is: {}.", mergedSchema);

// Detect what schema changes we need to apply to get expected sink table.
List<SchemaChangeEvent> localEvolvedSchemaChanges =
SchemaMergingUtils.getSchemaDifference(
affectedSinkTableId, currentSinkSchema, mergedSchema);
LOG.info("Step 3.5 - Corresponding schema changes are: {}.", localEvolvedSchemaChanges);

// Finally, we normalize schema change events, including rewriting events by current
// schema change behavior configuration, dropping explicitly excluded schema change
Expand All @@ -463,6 +489,16 @@ private Tuple2<Set<TableId>, List<SchemaChangeEvent>> deduceEvolvedSchemaChanges
return Tuple2.of(affectedSinkTableIds, evolvedSchemaChanges);
}

@VisibleForTesting
boolean containsDropTableEventForSinkTable(
List<SchemaChangeRequest> schemaChangeRequests, TableId affectedSinkTableId) {
return schemaChangeRequests.stream()
.map(SchemaChangeRequest::getSchemaChangeEvent)
.filter(event -> event instanceof DropTableEvent)
.map(event -> ((DropTableEvent) event).tableId())
.anyMatch(droppedTable -> router.route(droppedTable).contains(affectedSinkTableId));
}

private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent schemaChangeEvent) {
try {
metadataApplier.applySchemaChange(schemaChangeEvent);
Expand Down
29 changes: 25 additions & 4 deletions ...c/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.FlushEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
Expand Down Expand Up @@ -135,9 +136,15 @@ public void processElement(StreamRecord<PartitioningEvent> streamRecord) throws

// First, update upstream schema map unconditionally and it will never fail
Schema beforeSchema = upstreamSchemaTable.get(tableId, sourcePartition);
Schema afterSchema =
SchemaUtils.applySchemaChangeEvent(beforeSchema, schemaChangeEvent);
upstreamSchemaTable.put(tableId, sourcePartition, afterSchema);
if (schemaChangeEvent instanceof DropTableEvent) {
// DropTableEvent is a table-level DDL and ends the table lifecycle for all source
// partitions.
upstreamSchemaTable.row(tableId).clear();
} else {
Schema afterSchema =
SchemaUtils.applySchemaChangeEvent(beforeSchema, schemaChangeEvent);
upstreamSchemaTable.put(tableId, sourcePartition, afterSchema);
}

// Check if we need to send a schema change request to the coordinator
if (!(schemaChangeEvent instanceof CreateTableEvent)) {
Expand Down Expand Up @@ -168,6 +175,12 @@ public void processElement(StreamRecord<PartitioningEvent> streamRecord) throws
// Then, for each routing terminus, coerce data records to the expected schema
for (TableId sinkTableId : tableIdRouter.route(tableId)) {
Schema evolvedSchema = evolvedSchemaMap.get(sinkTableId);
if (upstreamSchema == null || evolvedSchema == null) {
throw new IllegalStateException(
String.format(
"Unable to coerce data record from %s (schema: %s) to %s (schema: %s)",
tableId, upstreamSchema, sinkTableId, evolvedSchema));
}

DataChangeEvent coercedDataRecord =
derivator
Expand Down Expand Up @@ -214,7 +227,15 @@ private void requestSchemaChange(
LOG.info("{}> Evolve request response: {}", subTaskId, response);

// Update local evolved schema cache
evolvedSchemaMap.putAll(response.getEvolvedSchemas());
response.getEvolvedSchemas()
.forEach(
(tableId, schema) -> {
if (schema == null) {
evolvedSchemaMap.remove(tableId);
} else {
evolvedSchemaMap.put(tableId, schema);
}
});

// And emit schema change events to downstream
response.getEvolvedSchemaChangeEvents()
Expand Down
75 changes: 66 additions & 9 deletions ...rc/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
Expand Down Expand Up @@ -284,20 +285,76 @@ private List<SchemaChangeEvent> deduceEvolvedSchemaChanges(SchemaChangeEvent eve
Set<TableId> upstreamDependencies =
SchemaDerivator.reverseLookupDependingUpstreamTables(
router, evolvedTableId, originalTables);
Preconditions.checkArgument(
!upstreamDependencies.isEmpty(),
"An affected sink table's upstream dependency cannot be empty.");
LOG.info("Step 3.2 - upstream dependency tables are: {}", upstreamDependencies);

List<SchemaChangeEvent> rawSchemaChangeEvents = new ArrayList<>();
if (upstreamDependencies.size() == 1) {
if (event instanceof DropTableEvent) {
// A dropped upstream table has been removed from original schemas before
// derivation. If no upstream table still routes to this downstream table, the
// downstream table should be dropped as well. Otherwise, keep the existing widened
// downstream schema because the remaining upstream tables still depend on it.
if (upstreamDependencies.isEmpty()) {
if (currentEvolvedSchema != null) {
SchemaChangeEvent rawEvent = event.copy(evolvedTableId);
rawSchemaChangeEvents.add(rawEvent);
LOG.info(
"Step 3.3 - No upstream dependency remains and could be forwarded as {}.",
rawEvent);
} else {
LOG.info(
"Step 3.3 - No upstream dependency or current evolved schema remains.");
}
} else {
Set<Schema> toBeMergedSchemas =
SchemaDerivator.reverseLookupDependingUpstreamSchemas(
router, evolvedTableId, schemaManager);
LOG.info("Step 3.3 - Upstream dependency schemas are: {}.", toBeMergedSchemas);

// Keep current schema as the merge base to avoid narrowing downstream schema
// while other upstream tables are still routed to this sink table.
Schema mergedSchema = currentEvolvedSchema;
for (Schema toBeMergedSchema : toBeMergedSchemas) {
mergedSchema =
SchemaMergingUtils.getLeastCommonSchema(
mergedSchema, toBeMergedSchema);
}
LOG.info("Step 3.4 - Deduced widest schema is: {}.", mergedSchema);

List<SchemaChangeEvent> rawEvents =
SchemaMergingUtils.getSchemaDifference(
evolvedTableId, currentEvolvedSchema, mergedSchema);
LOG.info(
"Step 3.5 - It's an many-to-one routing and causes schema changes: {}.",
rawEvents);

rawSchemaChangeEvents.addAll(rawEvents);
}
} else if (upstreamDependencies.size() == 1) {
// If it's a one-by-one routing rule, we can simply forward it to downstream sink.
SchemaChangeEvent rawEvent = event.copy(evolvedTableId);
rawSchemaChangeEvents.add(rawEvent);
LOG.info(
"Step 3.3 - It's an one-by-one routing and could be forwarded as {}.",
rawEvent);
if (event instanceof CreateTableEvent && currentEvolvedSchema != null) {
// The downstream table may still exist if a previous DropTableEvent was
// excluded. In that case, evolve it with schema differences instead of sending
// another CreateTableEvent.
List<SchemaChangeEvent> rawEvents =
SchemaMergingUtils.getSchemaDifference(
evolvedTableId,
currentEvolvedSchema,
((CreateTableEvent) event).getSchema());
rawSchemaChangeEvents.addAll(rawEvents);
LOG.info(
"Step 3.3 - The downstream table exists and causes schema changes: {}.",
rawEvents);
} else {
SchemaChangeEvent rawEvent = event.copy(evolvedTableId);
rawSchemaChangeEvents.add(rawEvent);
LOG.info(
"Step 3.3 - It's an one-by-one routing and could be forwarded as {}.",
rawEvent);
}
} else {
Preconditions.checkArgument(
!upstreamDependencies.isEmpty(),
"An affected sink table's upstream dependency cannot be empty.");
Set<Schema> toBeMergedSchemas =
SchemaDerivator.reverseLookupDependingUpstreamSchemas(
router, evolvedTableId, schemaManager);
Expand Down
Loading
Loading