[FLINK-39965][runtime] Correct schema state after DropTableEvent#4448
Open
taoran92 wants to merge 1 commit into
Open
[FLINK-39965][runtime] Correct schema state after DropTableEvent#4448taoran92 wants to merge 1 commit into
taoran92 wants to merge 1 commit into
Conversation
taoran92
force-pushed
the
fix-drop-table-schema-state
branch
from
f53db1d to
96d395e
Compare
DropTableEvent can remove a downstream table while the runtime still keeps original/evolved schema state and table-level caches for the same table id. When the upstream table is created again, the following CreateTableEvent may be treated as redundant, skipped by sink initialization, or forwarded as a duplicate create event if drop.table was filtered out. This change removes schema state for dropped tables and updates the regular schema derivation to distinguish between two cases: forward DropTableEvent only when no upstream dependency remains, and derive schema differences when a later CreateTableEvent targets an existing downstream schema. It also clears table-level caches in transform, pre-partition, and sink wrapper operators so recreated tables can be initialized with their new schema. This also avoids recreating pre-partition hash functions for dropped tables, covering the DROP TABLE failure reported in FLINK-39328.
taoran92
force-pushed
the
fix-drop-table-schema-state
branch
from
96d395e to
69821d1
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Jira: FLINK-39965 This PR fixes incorrect runtime state after
DropTableEventin Flink CDC pipeline jobs.When a source table is dropped, the downstream table may be removed while Flink CDC still keeps original/evolved schema state and table-level caches for the same
TableId. If the source table is later created again with the same identifier, the followingCreateTableEventmay be treated as redundant, skipped by sink initialization, or forwarded as a duplicate create event whendrop.tableis filtered out.This can cause sinks such as StarRocks to fail when writing data to the recreated table because the downstream table was dropped but not initialized again.
This PR makes
DropTableEventend the table lifecycle in runtime state and clears related table-level caches so recreated tables can be initialized with their latest schema. It also covers the pre-partition hash function issue reported in FLINK-39328.Brief change log
SchemaManagerwhen applyingDropTableEventDropTableEventonly when no upstream dependency remainsCreateTableEventwhendrop.tableis filtered out and the downstream table still existsnullevolved schema views for dropped sink tablesDropTableEventVerifying this change
This change is verified by unit tests:
SchemaManagerTest#testHandlingDropTableEventorg.apache.flink.cdc.runtime.operators.schema.regular.SchemaEvolveTest#testDropAndRecreateTableorg.apache.flink.cdc.runtime.operators.schema.regular.SchemaEvolveTest#testDropAndRecreateTableWithDropTableExcludedorg.apache.flink.cdc.runtime.operators.schema.regular.SchemaEvolveTest#testIgnoreDropAndRecreateTableKeepsEvolvedSchemaorg.apache.flink.cdc.runtime.operators.schema.distributed.SchemaEvolveTest#testDropAndRecreateTableorg.apache.flink.cdc.runtime.operators.schema.distributed.SchemaEvolveTest#testDropTableClearsAllSourcePartitionSchemasorg.apache.flink.cdc.runtime.operators.schema.distributed.SchemaEvolveTest#testDropTableEventMatchesOnlyAffectedSinkTableorg.apache.flink.cdc.runtime.operators.schema.distributed.SchemaEvolveTest#testIgnoreDropTableRejectsLaterDataWithoutRecreateDataSinkOperatorWithSchemaEvolveTest#testDataChangeEventAfterDropTablePreTransformOperatorTest#testDropAndRecreateTablePrePartitionOperatorTest#testBroadcastingDropTableEventPrePartitionOperatorTest#testDistributedDropTableEventDoesNotRecreateHashFunctionDoes this pull request potentially affect one of the following parts:
Documentation