diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java index fa4b85f1b29..4b25bf1fb13 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java @@ -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; @@ -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 optionalSchema = getLatestOriginalSchema(schemaChangeEvent.tableId()); checkArgument( @@ -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 optionalSchema = getLatestEvolvedSchema(schemaChangeEvent.tableId()); checkArgument( diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java index f9aad19b856..392cdb3627c 100755 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java @@ -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; @@ -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 { @@ -348,9 +355,8 @@ private void startSchemaChange() throws TimeoutException { Set affectedTableIds = deduceSummary.f0; Map 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>> futures = @@ -417,32 +423,52 @@ private Tuple2, List> deduceEvolvedSchemaChanges Set 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 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 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 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 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 @@ -463,6 +489,16 @@ private Tuple2, List> deduceEvolvedSchemaChanges return Tuple2.of(affectedSinkTableIds, evolvedSchemaChanges); } + @VisibleForTesting + boolean containsDropTableEventForSinkTable( + List 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); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java index 114629e2dba..cfb958c8f43 100755 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java @@ -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; @@ -135,9 +136,15 @@ public void processElement(StreamRecord 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)) { @@ -168,6 +175,12 @@ public void processElement(StreamRecord 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 @@ -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() diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java index 3a241aac489..de340f149a4 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java @@ -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; @@ -284,20 +285,76 @@ private List deduceEvolvedSchemaChanges(SchemaChangeEvent eve Set 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 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 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 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 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 toBeMergedSchemas = SchemaDerivator.reverseLookupDependingUpstreamSchemas( router, evolvedTableId, schemaManager); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperator.java index b659b7234f7..c0a401cdc4a 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperator.java @@ -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.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; @@ -161,9 +162,13 @@ public void processElement(StreamRecord streamRecord) throws Exception { private void handleSchemaChangeEvent(SchemaChangeEvent originalEvent) throws Exception { // First, update original schema map unconditionally and it will never fail TableId tableId = originalEvent.tableId(); - originalSchemaMap.compute( - tableId, - (tId, schema) -> SchemaUtils.applySchemaChangeEvent(schema, originalEvent)); + if (originalEvent instanceof DropTableEvent) { + originalSchemaMap.remove(tableId); + } else { + originalSchemaMap.compute( + tableId, + (tId, schema) -> SchemaUtils.applySchemaChangeEvent(schema, originalEvent)); + } schemaOperatorMetrics.increaseSchemaChangeEvents(1); // First, send FlushEvent or it might be blocked later @@ -188,7 +193,15 @@ private void handleSchemaChangeEvent(SchemaChangeEvent originalEvent) throws Exc response.getAppliedSchemaChangeEvents(); // Update local evolved schema map's cache - evolvedSchemaMap.putAll(response.getEvolvedSchemas()); + response.getEvolvedSchemas() + .forEach( + (tId, schema) -> { + if (schema == null) { + evolvedSchemaMap.remove(tId); + } else { + evolvedSchemaMap.put(tId, schema); + } + }); // and emit the finished event to downstream for (SchemaChangeEvent finishedEvent : finishedSchemaChangeEvents) { @@ -207,6 +220,12 @@ private void handleDataChangeEvent(DataChangeEvent dataChangeEvent) { // Then, for each routing terminus, coerce data records to the expected schema for (TableId sinkTableId : router.route(tableId)) { Schema evolvedSchema = evolvedSchemaMap.get(sinkTableId); + if (originalSchema == null || evolvedSchema == null) { + throw new IllegalStateException( + String.format( + "Unable to coerce data record from %s (schema: %s) to %s (schema: %s)", + tableId, originalSchema, sinkTableId, evolvedSchema)); + } DataChangeEvent coercedDataRecord = derivator .coerceDataRecord( diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java index 4286b3548e6..9751c7612a9 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java @@ -20,6 +20,7 @@ import org.apache.flink.cdc.common.annotation.Internal; import org.apache.flink.cdc.common.event.ChangeEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; +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.SchemaChangeEventType; @@ -100,6 +101,13 @@ public void processElement(StreamRecord element) throws Exception { return; } + if (event instanceof DropTableEvent) { + // Allow a later table with the same identifier to be initialized again. + processedTableIds.remove(((DropTableEvent) event).tableId()); + super.processElement(element); + return; + } + // Check if the table is processed before emitting all other events, because we have to // make // sure that sink have a view of the full schema before processing any change events, @@ -119,7 +127,8 @@ public void processElement(StreamRecord element) throws Exception { // ----------------------------- Helper functions ------------------------------- private void handleFlushEvent(FlushEvent event) throws Exception { userFunction.finish(); - if (event.getSchemaChangeEventType() != SchemaChangeEventType.CREATE_TABLE) { + if (event.getSchemaChangeEventType() != SchemaChangeEventType.CREATE_TABLE + && event.getSchemaChangeEventType() != SchemaChangeEventType.DROP_TABLE) { event.getTableIds().stream() .filter(tableId -> !processedTableIds.contains(tableId)) .forEach( diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperator.java index bac2929cec8..56df234b3c5 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperator.java @@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.annotation.Internal; import org.apache.flink.cdc.common.event.ChangeEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; +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.SchemaChangeEventType; @@ -173,6 +174,16 @@ public void processElement(StreamRecord element) throws Exception { return; } + if (event instanceof DropTableEvent) { + // Allow a later table with the same identifier to be initialized again. + processedTableIds.remove(((DropTableEvent) event).tableId()); + this + .>> + getFlinkWriterOperator() + .processElement(element); + return; + } + // Check if the table is processed before emitting all other events, because we have to // make // sure that sink have a view of the full schema before processing any change events, diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java index 7c6db17354e..c38a0fc33a6 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java @@ -200,7 +200,11 @@ private void processEvent(Event event) { output.collect(new StreamRecord<>(cacheCreateTable(createTableEvent))); } } else if (event instanceof DropTableEvent) { - preTransformProcessorMap.remove(((DropTableEvent) event).tableId()); + TableId tableId = ((DropTableEvent) event).tableId(); + // A table with the same identifier may be recreated with a different schema. + preTransformProcessorMap.remove(tableId); + preTransformChangeInfoMap.remove(tableId); + hasAsteriskMap.remove(tableId); output.collect(new StreamRecord<>(event)); } else if (event instanceof TruncateTableEvent) { output.collect(new StreamRecord<>(event)); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java index 4750c685a95..e9d5c7f8b26 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java @@ -19,6 +19,7 @@ import org.apache.flink.cdc.common.annotation.Internal; 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.SchemaChangeEvent; import org.apache.flink.cdc.common.event.TableId; @@ -72,7 +73,13 @@ public void open() throws Exception { @Override public void processElement(StreamRecord element) throws Exception { Event event = element.getValue(); - if (event instanceof SchemaChangeEvent) { + if (event instanceof DropTableEvent) { + TableId tableId = ((DropTableEvent) event).tableId(); + schemaMap.remove(tableId); + hashFunctionMap.remove(tableId); + // DropTableEvent ends the table lifecycle and no longer has a latest schema. + broadcastEvent(event); + } else if (event instanceof SchemaChangeEvent) { SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event; TableId tableId = schemaChangeEvent.tableId(); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java index a5d81bfb19b..82dfc4a2c58 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java @@ -19,6 +19,7 @@ import org.apache.flink.cdc.common.annotation.Internal; 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; @@ -85,7 +86,12 @@ public void open() throws Exception { @Override public void processElement(StreamRecord element) throws Exception { Event event = element.getValue(); - if (event instanceof SchemaChangeEvent) { + if (event instanceof DropTableEvent) { + TableId tableId = ((DropTableEvent) event).tableId(); + cachedHashFunctions.invalidate(tableId); + // DropTableEvent ends the table lifecycle and no longer has a latest schema. + broadcastEvent(event); + } else if (event instanceof SchemaChangeEvent) { // Update hash function TableId tableId = ((SchemaChangeEvent) event).tableId(); cachedHashFunctions.put(tableId, recreateHashFunction(tableId)); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManagerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManagerTest.java index f977089b47a..478059ef0af 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManagerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManagerTest.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.DropColumnEvent; +import org.apache.flink.cdc.common.event.DropTableEvent; import org.apache.flink.cdc.common.event.RenameColumnEvent; import org.apache.flink.cdc.common.event.TableId; import org.apache.flink.cdc.common.schema.Column; @@ -72,6 +73,27 @@ void testHandlingCreateTableEvent() { .doesNotThrowAnyException(); } + @Test + void testHandlingDropTableEvent() { + SchemaManager schemaManager = new SchemaManager(); + schemaManager.applyOriginalSchemaChange(new CreateTableEvent(CUSTOMERS, CUSTOMERS_SCHEMA)); + schemaManager.applyEvolvedSchemaChange(new CreateTableEvent(CUSTOMERS, CUSTOMERS_SCHEMA)); + + schemaManager.applyOriginalSchemaChange(new DropTableEvent(CUSTOMERS)); + schemaManager.applyEvolvedSchemaChange(new DropTableEvent(CUSTOMERS)); + + assertThat(schemaManager.getLatestOriginalSchema(CUSTOMERS)).isEmpty(); + assertThat(schemaManager.getLatestEvolvedSchema(CUSTOMERS)).isEmpty(); + + schemaManager.applyOriginalSchemaChange(new CreateTableEvent(CUSTOMERS, PRODUCTS_SCHEMA)); + schemaManager.applyEvolvedSchemaChange(new CreateTableEvent(CUSTOMERS, PRODUCTS_SCHEMA)); + + assertThat(schemaManager.getLatestOriginalSchema(CUSTOMERS)).contains(PRODUCTS_SCHEMA); + assertThat(schemaManager.getLatestEvolvedSchema(CUSTOMERS)).contains(PRODUCTS_SCHEMA); + assertThat(schemaManager.getOriginalSchema(CUSTOMERS, 0)).isEqualTo(PRODUCTS_SCHEMA); + assertThat(schemaManager.getEvolvedSchema(CUSTOMERS, 0)).isEqualTo(PRODUCTS_SCHEMA); + } + @Test void testHandlingAddColumnEvent() { SchemaManager schemaManager = new SchemaManager(); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaEvolveTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaEvolveTest.java index b7424cb5d61..e38be065391 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaEvolveTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaEvolveTest.java @@ -35,7 +35,10 @@ import org.apache.flink.cdc.common.types.DataTypes; import org.apache.flink.cdc.runtime.operators.AbstractStreamOperatorAdapter; import org.apache.flink.cdc.runtime.operators.schema.common.SchemaTestBase; +import org.apache.flink.cdc.runtime.operators.schema.distributed.event.SchemaChangeRequest; import org.apache.flink.cdc.runtime.testutils.operators.DistributedEventOperatorTestHarness; +import org.apache.flink.cdc.runtime.testutils.operators.MockedOperatorCoordinatorContext; +import org.apache.flink.cdc.runtime.testutils.schema.CollectingMetadataApplier; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.util.function.BiConsumerWithException; @@ -45,6 +48,7 @@ import java.time.Duration; import java.util.Collections; import java.util.LinkedList; +import java.util.concurrent.Executors; import java.util.function.Function; import java.util.function.Supplier; @@ -220,7 +224,170 @@ void testLenientSchemaEvolution() throws Exception { truncateTableEvent.getType()), genInsert(TABLE_ID, "ISDSBS", 6, "Ferris", 0.001, null, false, null), new FlushEvent( - 0, Collections.singletonList(TABLE_ID), dropTableEvent.getType())); + 0, Collections.singletonList(TABLE_ID), dropTableEvent.getType()), + dropTableEvent); + } + + @Test + void testDropAndRecreateTable() throws Exception { + Schema schemaV1 = + Schema.newBuilder() + .physicalColumn("id", DataTypes.INT()) + .physicalColumn("name", DataTypes.STRING()) + .build(); + Schema schemaV2 = + Schema.newBuilder() + .physicalColumn("id", DataTypes.BIGINT()) + .physicalColumn("score", DataTypes.INT()) + .build(); + CreateTableEvent createTableV1 = new CreateTableEvent(TABLE_ID, schemaV1); + DropTableEvent dropTableEvent = new DropTableEvent(TABLE_ID); + CreateTableEvent createTableV2 = new CreateTableEvent(TABLE_ID, schemaV2); + + Assertions.assertThat( + runInHarness( + () -> + new SchemaOperator( + ROUTING_RULES, + RouteMode.ALL_MATCH, + Duration.ofMinutes(3), + SchemaChangeBehavior.EVOLVE, + "UTC"), + (SchemaOperator op) -> + new DistributedEventOperatorTestHarness<>( + op, + 20, + Duration.ofSeconds(3), + Duration.ofMinutes(3)), + (operator, harness) -> { + operator.processElement(wrap(createTableV1)); + operator.processElement(wrap(dropTableEvent)); + operator.processElement(wrap(createTableV2)); + operator.processElement( + wrap(genInsert(TABLE_ID, "LI", 1L, 100))); + })) + .map(StreamRecord::getValue) + .containsExactly( + new FlushEvent( + 0, Collections.singletonList(TABLE_ID), createTableV1.getType()), + createTableV1, + new FlushEvent( + 0, Collections.singletonList(TABLE_ID), dropTableEvent.getType()), + dropTableEvent, + new FlushEvent( + 0, Collections.singletonList(TABLE_ID), createTableV2.getType()), + createTableV2, + genInsert(TABLE_ID, "LI", 1L, 100)); + } + + @Test + void testDropTableClearsAllSourcePartitionSchemas() throws Exception { + Schema schema = + Schema.newBuilder() + .physicalColumn("id", DataTypes.INT()) + .physicalColumn("name", DataTypes.STRING()) + .build(); + CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_ID, schema); + DropTableEvent dropTableEvent = new DropTableEvent(TABLE_ID); + + Assertions.assertThat( + runInHarness( + () -> + new SchemaOperator( + ROUTING_RULES, + RouteMode.ALL_MATCH, + Duration.ofMinutes(3), + SchemaChangeBehavior.EVOLVE, + "UTC"), + (SchemaOperator op) -> + new DistributedEventOperatorTestHarness<>( + op, + 20, + Duration.ofSeconds(3), + Duration.ofMinutes(3)), + (operator, harness) -> { + operator.processElement(wrap(createTableEvent, 0, 0)); + harness.clearOutputRecords(); + + // Simulate a stale schema copy for the same table id in + // another source partition. + operator.processElement(wrap(createTableEvent, 1, 0)); + harness.clearOutputRecords(); + + operator.processElement(wrap(dropTableEvent, 0, 0)); + })) + .map(StreamRecord::getValue) + .containsExactly( + new FlushEvent( + 0, Collections.singletonList(TABLE_ID), dropTableEvent.getType()), + dropTableEvent); + } + + @Test + void testDropTableEventMatchesOnlyAffectedSinkTable() throws Exception { + SchemaCoordinator coordinator = + new SchemaCoordinator( + "schema-coordinator", + new MockedOperatorCoordinatorContext( + DistributedEventOperatorTestHarness.SCHEMA_OPERATOR_ID, + Thread.currentThread().getContextClassLoader()), + Executors.newSingleThreadExecutor(), + new CollectingMetadataApplier(Duration.ZERO), + ROUTING_RULES, + RouteMode.ALL_MATCH, + SchemaChangeBehavior.EVOLVE, + Duration.ofMinutes(3)); + coordinator.start(); + try { + LinkedList schemaChangeRequests = new LinkedList<>(); + + // A one-to-one dropped source table only affects its identical sink table. + schemaChangeRequests.add( + new SchemaChangeRequest( + 0, 0, new DropTableEvent(TableId.parse("db_1.table_1")))); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_1.table_1"))) + .isTrue(); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_1.table_2"))) + .isFalse(); + + // A re-routed dropped source table affects the routed sink table, not its source name. + schemaChangeRequests.clear(); + schemaChangeRequests.add( + new SchemaChangeRequest( + 0, 0, new DropTableEvent(TableId.parse("db_2.table_1")))); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_2.table_2"))) + .isTrue(); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_2.table_1"))) + .isFalse(); + + // A broadcast dropped source table can affect multiple routed sink tables. + schemaChangeRequests.clear(); + schemaChangeRequests.add( + new SchemaChangeRequest( + 0, 0, new DropTableEvent(TableId.parse("db_4.table_1")))); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_4.table_a"))) + .isTrue(); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_4.table_b"))) + .isTrue(); + Assertions.assertThat( + coordinator.containsDropTableEventForSinkTable( + schemaChangeRequests, TableId.parse("db_4.table_d"))) + .isFalse(); + } finally { + coordinator.close(); + } } @Test @@ -335,6 +502,48 @@ void testIgnoreSchemaEvolution() throws Exception { genInsert(TABLE_ID, "ISFS", 6, "Ferris", null, null)); } + @Test + void testIgnoreDropTableRejectsLaterDataWithoutRecreate() { + CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_ID, INITIAL_SCHEMA); + DropTableEvent dropTableEvent = new DropTableEvent(TABLE_ID); + + Assertions.assertThatThrownBy( + () -> + runInHarness( + () -> + new SchemaOperator( + ROUTING_RULES, + RouteMode.ALL_MATCH, + Duration.ofMinutes(3), + SchemaChangeBehavior.IGNORE, + "UTC"), + (op) -> + new DistributedEventOperatorTestHarness<>( + op, + 20, + Duration.ofSeconds(3), + Duration.ofMinutes(3)), + (operator, harness) -> { + operator.processElement(wrap(createTableEvent)); + harness.clearOutputRecords(); + + operator.processElement(wrap(dropTableEvent)); + harness.clearOutputRecords(); + + operator.processElement( + wrap( + genInsert( + TABLE_ID, + "ISFS", + 2, + "Bob", + 31.415926f, + "late"))); + })) + .isExactlyInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unable to coerce data record"); + } + @Test void testExceptionSchemaEvolution() throws Exception { CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_ID, INITIAL_SCHEMA); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaEvolveTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaEvolveTest.java index 7bdf37ed49a..03108efbf7c 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaEvolveTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaEvolveTest.java @@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.DropColumnEvent; +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.RenameColumnEvent; @@ -352,6 +353,224 @@ tableId, buildRecord(INT, 12, STRING, "Jane", FLOAT, 11f)), harness.close(); } + @Test + void testDropAndRecreateTable() throws Exception { + TableId tableId = CUSTOMERS_TABLE_ID; + Schema schemaV1 = + Schema.newBuilder() + .physicalColumn("id", INT) + .physicalColumn("name", STRING) + .primaryKey("id") + .build(); + Schema schemaV2 = + Schema.newBuilder() + .physicalColumn("id", BIGINT) + .physicalColumn("name", STRING) + .physicalColumn("score", INT) + .primaryKey("id") + .build(); + + SchemaChangeBehavior behavior = SchemaChangeBehavior.EVOLVE; + + SchemaOperator schemaOperator = + new SchemaOperator( + new ArrayList<>(), RouteMode.ALL_MATCH, Duration.ofSeconds(30), behavior); + RegularEventOperatorTestHarness harness = + RegularEventOperatorTestHarness.withDurationAndBehavior( + schemaOperator, 17, Duration.ofSeconds(3), behavior); + harness.open(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV1))); + harness.clearOutputRecords(); + + processEvent(schemaOperator, Collections.singletonList(new DropTableEvent(tableId))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isNull(); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isNull(); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Arrays.asList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.DROP_TABLE), + new DropTableEvent(tableId))); + harness.clearOutputRecords(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV2))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isEqualTo(schemaV2); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV2); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Arrays.asList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.CREATE_TABLE), + new CreateTableEvent(tableId, schemaV2))); + + harness.close(); + } + + @Test + void testDropAndRecreateTableWithDropTableExcluded() throws Exception { + TableId tableId = CUSTOMERS_TABLE_ID; + Schema schemaV1 = + Schema.newBuilder() + .physicalColumn("id", INT) + .physicalColumn("name", STRING) + .primaryKey("id") + .build(); + Schema schemaV2 = + Schema.newBuilder() + .physicalColumn("id", INT) + .physicalColumn("name", STRING) + .physicalColumn("score", INT) + .primaryKey("id") + .build(); + + SchemaChangeBehavior behavior = SchemaChangeBehavior.LENIENT; + SchemaOperator schemaOperator = + new SchemaOperator( + new ArrayList<>(), RouteMode.ALL_MATCH, Duration.ofSeconds(30), behavior); + RegularEventOperatorTestHarness harness = + RegularEventOperatorTestHarness.withDurationAndFineGrainedBehavior( + schemaOperator, + 17, + Duration.ofSeconds(3), + behavior, + Sets.difference( + Arrays.stream(SchemaChangeEventTypeFamily.ALL) + .collect(Collectors.toSet()), + Collections.singleton(SchemaChangeEventType.DROP_TABLE))); + harness.open(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV1))); + harness.clearOutputRecords(); + + processEvent(schemaOperator, Collections.singletonList(new DropTableEvent(tableId))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isNull(); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV1); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Collections.singletonList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.DROP_TABLE))); + harness.clearOutputRecords(); + + Assertions.assertThatThrownBy( + () -> + processEvent( + schemaOperator, + Collections.singletonList( + DataChangeEvent.insertEvent( + tableId, + buildRecord(INT, 2, STRING, "Bob"))))) + .isExactlyInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unable to coerce data record"); + harness.clearOutputRecords(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV2))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isEqualTo(schemaV2); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV2); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Arrays.asList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.CREATE_TABLE), + new AddColumnEvent( + tableId, + Collections.singletonList( + new AddColumnEvent.ColumnWithPosition( + Column.physicalColumn("score", INT)))))); + + harness.close(); + } + + @Test + void testIgnoreDropAndRecreateTableKeepsEvolvedSchema() throws Exception { + TableId tableId = CUSTOMERS_TABLE_ID; + Schema schemaV1 = + Schema.newBuilder() + .physicalColumn("id", INT) + .physicalColumn("name", STRING) + .primaryKey("id") + .build(); + Schema schemaV2 = + Schema.newBuilder() + .physicalColumn("id", INT) + .physicalColumn("name", STRING) + .physicalColumn("score", INT) + .primaryKey("id") + .build(); + + SchemaChangeBehavior behavior = SchemaChangeBehavior.IGNORE; + SchemaOperator schemaOperator = + new SchemaOperator( + new ArrayList<>(), RouteMode.ALL_MATCH, Duration.ofSeconds(30), behavior); + RegularEventOperatorTestHarness harness = + RegularEventOperatorTestHarness.withDurationAndBehavior( + schemaOperator, 17, Duration.ofSeconds(3), behavior); + harness.open(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV1))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isEqualTo(schemaV1); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV1); + harness.clearOutputRecords(); + + processEvent(schemaOperator, Collections.singletonList(new DropTableEvent(tableId))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isNull(); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV1); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Collections.singletonList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.DROP_TABLE))); + harness.clearOutputRecords(); + + processEvent( + schemaOperator, Collections.singletonList(new CreateTableEvent(tableId, schemaV2))); + Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isEqualTo(schemaV2); + Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV1); + Assertions.assertThat( + harness.getOutputRecords().stream() + .map(StreamRecord::getValue) + .collect(Collectors.toList())) + .isEqualTo( + Collections.singletonList( + new FlushEvent( + 0, + Collections.singletonList(tableId), + SchemaChangeEventType.CREATE_TABLE))); + + harness.close(); + } + /** Tests try-evolve behavior without exceptions. */ @Test void testTryEvolveSchema() throws Exception { diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorAdapter.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorAdapter.java index 61c56134835..7bda090a192 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorAdapter.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorAdapter.java @@ -19,6 +19,7 @@ import org.apache.flink.cdc.common.event.ChangeEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; +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.SchemaChangeEventType; @@ -110,6 +111,12 @@ public void processElement(StreamRecord element) throws Exception { return; } + if (event instanceof DropTableEvent) { + processedTableIds.remove(((DropTableEvent) event).tableId()); + output.collect(element); + return; + } + // Check if the table is processed before emitting all other events, because we have to make // sure that sink have a view of the full schema before processing any change events, // including schema changes. @@ -135,7 +142,8 @@ public void endInput() {} private void handleFlushEvent(FlushEvent event) throws Exception { // omit copySinkWriter/userFunction flush from testing - if (event.getSchemaChangeEventType() != SchemaChangeEventType.CREATE_TABLE) { + if (event.getSchemaChangeEventType() != SchemaChangeEventType.CREATE_TABLE + && event.getSchemaChangeEventType() != SchemaChangeEventType.DROP_TABLE) { event.getTableIds().stream() .filter(tableId -> !processedTableIds.contains(tableId)) .forEach( diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorWithSchemaEvolveTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorWithSchemaEvolveTest.java index bc2985351dc..af2a9de539d 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorWithSchemaEvolveTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorWithSchemaEvolveTest.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.event.AddColumnEvent; 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; @@ -323,4 +324,56 @@ void testDataChangeEventAfterFailover() throws Exception { dataSinkWriterOperatorHarness.clearOutputRecords(); } } + + @Test + void testDataChangeEventAfterDropTable() throws Exception { + DataSinkOperatorAdapter dataSinkWriterOperator = new DataSinkOperatorAdapter(); + try (RegularEventOperatorTestHarness + dataSinkWriterOperatorHarness = setupHarness(dataSinkWriterOperator)) { + CreateTableEvent createTableEvent = + new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_SCHEMA); + processSchemaChangeEvent( + dataSinkWriterOperator, + dataSinkWriterOperatorHarness, + CUSTOMERS_TABLEID, + createTableEvent); + dataSinkWriterOperatorHarness.clearOutputRecords(); + + DropTableEvent dropTableEvent = new DropTableEvent(CUSTOMERS_TABLEID); + processSchemaChangeEvent( + dataSinkWriterOperator, + dataSinkWriterOperatorHarness, + CUSTOMERS_TABLEID, + dropTableEvent); + assertOutputEvents( + dataSinkWriterOperatorHarness, Collections.singletonList(dropTableEvent)); + dataSinkWriterOperatorHarness.clearOutputRecords(); + + dataSinkWriterOperatorHarness.registerOriginalSchema( + CUSTOMERS_TABLEID, CUSTOMERS_LATEST_SCHEMA); + dataSinkWriterOperatorHarness.registerEvolvedSchema( + CUSTOMERS_TABLEID, CUSTOMERS_LATEST_SCHEMA); + + BinaryRecordDataGenerator recordDataGenerator = + new BinaryRecordDataGenerator( + ((RowType) CUSTOMERS_LATEST_SCHEMA.toRowDataType())); + DataChangeEvent insertEvent = + DataChangeEvent.insertEvent( + CUSTOMERS_TABLEID, + recordDataGenerator.generate( + new Object[] { + new BinaryStringData("1"), + new BinaryStringData("2"), + new BinaryStringData("3"), + })); + + processDataChangeEvent(dataSinkWriterOperator, insertEvent); + assertOutputEvents( + dataSinkWriterOperatorHarness, + Arrays.asList( + new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_LATEST_SCHEMA), + insertEvent)); + dataSinkWriterOperatorHarness.clearOutputRecords(); + } + } } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java index 0425751d79a..16a061103c2 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.event.AddColumnEvent; 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.TableId; import org.apache.flink.cdc.common.schema.Column; @@ -314,6 +315,46 @@ void testEventTransform() throws Exception { transformFunctionEventEventOperatorTestHarness.close(); } + @Test + void testDropAndRecreateTable() throws Exception { + PreTransformOperator transform = + PreTransformOperator.newBuilder() + .addTransform( + CUSTOMERS_TABLEID.identifier(), + "*, concat(col1,col2) col12", + null, + "col2", + "col12", + "key1=value1,key2=value2", + null, + new SupportedMetadataColumn[0]) + .build(); + RegularEventOperatorTestHarness harness = + RegularEventOperatorTestHarness.with(transform, 1); + harness.open(); + + transform.processElement( + new StreamRecord<>(new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_SCHEMA))); + Assertions.assertThat(harness.getOutputRecords().poll()) + .isEqualTo( + new StreamRecord<>(new CreateTableEvent(CUSTOMERS_TABLEID, EXPECT_SCHEMA))); + + DropTableEvent dropTableEvent = new DropTableEvent(CUSTOMERS_TABLEID); + transform.processElement(new StreamRecord<>(dropTableEvent)); + Assertions.assertThat(harness.getOutputRecords().poll()) + .isEqualTo(new StreamRecord<>(dropTableEvent)); + + transform.processElement( + new StreamRecord<>( + new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_LATEST_SCHEMA))); + Assertions.assertThat(harness.getOutputRecords().poll()) + .isEqualTo( + new StreamRecord<>( + new CreateTableEvent(CUSTOMERS_TABLEID, EXPECT_LATEST_SCHEMA))); + + harness.close(); + } + @Test void testNullabilityColumn() throws Exception { PreTransformOperator transform = diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java index 2f0e55abe58..9d50d8b3920 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java @@ -20,6 +20,7 @@ import org.apache.flink.cdc.common.data.binary.BinaryStringData; 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.FlushEvent; import org.apache.flink.cdc.common.event.SchemaChangeEventType; import org.apache.flink.cdc.common.event.TableId; @@ -35,10 +36,11 @@ import org.junit.jupiter.api.Test; import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; -/** Unit test for {@link RegularPrePartitionOperator}. */ +/** Unit test for pre-partition operators. */ class PrePartitionOperatorTest { private static final TableId CUSTOMERS = TableId.tableId("my_company", "my_branch", "customers"); @@ -73,6 +75,60 @@ void testBroadcastingSchemaChangeEvent() throws Exception { } } + @Test + void testBroadcastingDropTableEvent() throws Exception { + try (RegularEventOperatorTestHarness + testHarness = createTestHarness()) { + // Initialization + testHarness.open(); + + RegularPrePartitionOperator operator = testHarness.getOperator(); + DropTableEvent dropTableEvent = new DropTableEvent(CUSTOMERS); + operator.processElement(new StreamRecord<>(dropTableEvent)); + + assertThat(testHarness.getOutputRecords()).hasSize(DOWNSTREAM_PARALLELISM); + for (int i = 0; i < DOWNSTREAM_PARALLELISM; i++) { + assertThat(testHarness.getOutputRecords().poll()) + .isEqualTo( + new StreamRecord<>(PartitioningEvent.ofRegular(dropTableEvent, i))); + } + } + } + + @Test + void testDistributedDropTableEventDoesNotRecreateHashFunction() throws Exception { + AtomicInteger hashFunctionCreations = new AtomicInteger(); + DistributedPrePartitionOperator operator = + new DistributedPrePartitionOperator( + DOWNSTREAM_PARALLELISM, + (tableId, schema) -> { + hashFunctionCreations.incrementAndGet(); + return event -> 0; + }); + try (RegularEventOperatorTestHarness + testHarness = + RegularEventOperatorTestHarness.with(operator, DOWNSTREAM_PARALLELISM)) { + testHarness.open(); + + CreateTableEvent createTableEvent = new CreateTableEvent(CUSTOMERS, CUSTOMERS_SCHEMA); + operator.processElement(new StreamRecord<>(createTableEvent)); + assertThat(hashFunctionCreations).hasValue(1); + testHarness.clearOutputRecords(); + + DropTableEvent dropTableEvent = new DropTableEvent(CUSTOMERS); + operator.processElement(new StreamRecord<>(dropTableEvent)); + + assertThat(hashFunctionCreations).hasValue(1); + assertThat(testHarness.getOutputRecords()).hasSize(DOWNSTREAM_PARALLELISM); + for (int i = 0; i < DOWNSTREAM_PARALLELISM; i++) { + assertThat(testHarness.getOutputRecords().poll()) + .isEqualTo( + new StreamRecord<>( + PartitioningEvent.ofDistributed(dropTableEvent, 0, i))); + } + } + } + @Test void testBroadcastingFlushEvent() throws Exception { try (RegularEventOperatorTestHarness