[FLINK-37386] Emit CreateTableEvent only when met the related SourceRecord. #3932
Conversation
lvyanquan
changed the title
|
Hi @yuxiqian. Could you help to review this? |
| } else { | ||
| if (isDataChangeRecord(element) || isSchemaChangeEvent(element)) { | ||
| TableId tableId = getTableId(element); | ||
| if (!alreadySendCreateTableTables.contains(tableId)) { |
There was a problem hiding this comment.
Seems alreadySendCreateTableTables and createTableEventCache are not being stored into MySQL source state persistently. Will that block users from recovering from an earlier binlog state where schemas are different from current state?
| TableId tableId = getTableId(element); | ||
| if (!alreadySendCreateTableTables.contains(tableId)) { | ||
| CreateTableEvent createTableEvent = createTableEventCache.get(tableId); | ||
| // New created table in binlog reading phase. |
There was a problem hiding this comment.
--- Create Table A --> Table A Binlog --> Drop Table A --> Startup Offset --->
^
|
What if we start from here?
Seems testDanglingDropTableEventInBinlog will fail.
There was a problem hiding this comment.
This is indeed a problem, but it is different from the one I hope to solve here.
For DataChange, perhaps we can derive the schema from SourceRecord instead of using SQL to query the latest schema. But to make this problem more difficult, if we start reading from a position where SchemaChange happened, we cannot derive the original schema.
--- Create Table A --> Alter Table A Add a column--> Table A Binlog --> Startup Offset --->
^
|
What if we start from here?
So I think starting from a position where historical schema is different with the current schema is still an unresolved issue for us.
There was a problem hiding this comment.
Thanks for clarifying this. Should we throw an exception for such dangling schema change events here?
|
The issue is related to #3912 (comment) @gongzexin as you may be interested about it. |
Emit CreateTableEvent only when met the related SourceRecord to avoid downstream backpressure.