Package org.apache.flink.cdc.connectors.mysql.source.utils

Class OnlineSchemaChangeUtils

  • public class OnlineSchemaChangeUtils
extends Object
    Utility class for handling gh-ost/pt-osc online schema change events.

    • Method Summary

      All Methods Static Methods Concrete Methods 
      Modifier and Type Method Description
      static boolean isOnLineSchemaChangeEvent​(org.apache.kafka.connect.source.SourceRecord record)
      Checks whether the given source record is a gh-ost/pt-osc initiated schema change event by inspecting the ALTER DDL statement targeting a shadow/new table.
      static Optional<String> parseOnLineSchemaRenameEvent​(org.apache.kafka.connect.source.SourceRecord record)
      Parses a gh-ost/pt-osc RENAME TABLE event and returns the original (user-facing) table name if the event represents the completion of an online schema change.
      static io.debezium.relational.TableId peelTableId​(io.debezium.relational.TableId tableId)
      Peels out a gh-ost/pt-osc mangled TableId back to the original user-facing one.

    • Method Detail

      • isOnLineSchemaChangeEvent

        public static boolean isOnLineSchemaChangeEvent​(org.apache.kafka.connect.source.SourceRecord record)
        Checks whether the given source record is a gh-ost/pt-osc initiated schema change event by inspecting the ALTER DDL statement targeting a shadow/new table.

        There will be these schema change events generated in total during one transaction.

        gh-ost: 

         DROP TABLE IF EXISTS `db`.`_tb1_gho`
 DROP TABLE IF EXISTS `db`.`_tb1_del`
 DROP TABLE IF EXISTS `db`.`_tb1_ghc`
 create /* gh-ost */ table `db`.`_tb1_ghc` ...
 create /* gh-ost */ table `db`.`_tb1_gho` like `db`.`tb1`
 alter /* gh-ost */ table `db`.`_tb1_gho` add column c varchar(255)
 alter /* gh-ost */ table `db`.`_tb1_gho` AUTO_INCREMENT=N  (only present when the table has an AUTO_INCREMENT column)
 create /* gh-ost */ table `db`.`_tb1_del` ...
 DROP TABLE IF EXISTS `db`.`_tb1_del`
 rename /* gh-ost */ table `db`.`tb1` to `db`.`_tb1_del`
 rename /* gh-ost */ table `db`.`_tb1_gho` to `db`.`tb1`
 DROP TABLE IF EXISTS `db`.`_tb1_ghc`
 DROP TABLE IF EXISTS `db`.`_tb1_del`

        pt-osc: 

         CREATE TABLE `db`.`_test_tb1_new`
 ALTER TABLE `db`.`_test_tb1_new` add column c varchar(50)
 CREATE TRIGGER `pt_osc_db_test_tb1_del`...
 CREATE TRIGGER `pt_osc_db_test_tb1_upd`...
 CREATE TRIGGER `pt_osc_db_test_tb1_ins`...
 ANALYZE TABLE `db`.`_test_tb1_new` /* pt-online-schema-change */
 RENAME TABLE `db`.`test_tb1` TO `db`.`_test_tb1_old`, `db`.`_test_tb1_new` TO `db`.`test_tb1`
 DROP TABLE IF EXISTS `_test_tb1_old` /* generated by server */
 DROP TRIGGER IF EXISTS `db`.`pt_osc_db_test_tb1_del`
 DROP TRIGGER IF EXISTS `db`.`pt_osc_db_test_tb1_upd`
 DROP TRIGGER IF EXISTS `db`.`pt_osc_db_test_tb1_ins`

        Among all these, only the ALTER statement targeting the _gho/_new table is stored temporarily, and emitted when the subsequent RENAME TABLE event arrives.

      • parseOnLineSchemaRenameEvent

        public static Optional<String> parseOnLineSchemaRenameEvent​(org.apache.kafka.connect.source.SourceRecord record)
        Parses a gh-ost/pt-osc RENAME TABLE event and returns the original (user-facing) table name if the event represents the completion of an online schema change.
        Returns:
        the original table name if the record is an OSC completion rename, or Optional.empty() otherwise.

      • peelTableId

        public static io.debezium.relational.TableId peelTableId​(io.debezium.relational.TableId tableId)
        Peels out a gh-ost/pt-osc mangled TableId back to the original user-facing one.

        For example, _customers_gho → customers, _orders_new → orders.