add auto increment ID support for pt-osc schema migration testing

Author: beryllw

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOscITCase.java

@@ -173,6 +173,30 @@ private void insertRecordsPhase2(UniqueDatabase database, int startIndex, int co
         }
     }
 
+    private void insertRecordsPhase1ForAutoId(UniqueDatabase database, int count) throws Exception {
+        try (Connection connection = database.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            for (int i = 0; i < count; i++) {
+                statement.execute(
+                        String.format(
+                                "insert into customers_auto_id (name, address, phone_number) values ('%s', '%s', '%s');",
+                                "flink_" + i, "Address Line #" + i, 1000000000L + i));
+            }
+        }
+    }
+
+    private void insertRecordsPhase2ForAutoId(UniqueDatabase database, int count) throws Exception {
+        try (Connection connection = database.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            for (int i = 0; i < count; i++) {
+                statement.execute(
+                        String.format(
+                                "insert into customers_auto_id (name, address, phone_number, ext) values ('%s', '%s', '%s', %s);",
+                                "flink_ext_" + i, "Address Line Ext #" + i, 1000000000L + i, i));
+            }
+        }
+    }
+
     @Test
     void testGhOstSchemaMigration() throws Exception {
         String databaseName = customerDatabase.getDatabaseName();
@@ -283,6 +307,63 @@ void testPtOscSchemaMigration() throws Exception {
         }
     }
 
+    @Test
+    void testPtOscSchemaMigrationWithAutoIncrementId() throws Exception {
+        String databaseName = customerDatabase.getDatabaseName();
+
+        LOG.info("Step 1: Start pipeline job for auto increment id table");
+
+        Thread yamlJob = runJob(databaseName, "customers_auto_id");
+        yamlJob.start();
+
+        LOG.info("Step 2: Insert initial records (Phase 1)");
+        insertRecordsPhase1ForAutoId(customerDatabase, 1000);
+
+        LOG.info("Step 3: Evolve schema with pt-osc - ADD COLUMN");
+
+        Thread thread =
+                new Thread(
+                        () -> {
+                            try {
+                                execInContainer(
+                                        PERCONA_TOOLKIT_CONTAINER,
+                                        "evolve schema",
+                                        "pt-online-schema-change",
+                                        "--user=" + TEST_USER,
+                                        "--host=" + INTER_CONTAINER_MYSQL_ALIAS,
+                                        "--password=" + TEST_PASSWORD,
+                                        "P=3306,t=customers_auto_id,D=" + databaseName,
+                                        "--alter",
+                                        "add column ext int first",
+                                        "--charset=utf8",
+                                        "--recursion-method=NONE", // Do not look for slave nodes
+                                        "--print",
+                                        "--execute");
+                            } catch (IOException | InterruptedException e) {
+                                throw new RuntimeException(e);
+                            }
+                        });
+
+        LOG.info("Insertion Phase 1 finishes");
+        thread.start();
+        insertRecordsPhase1ForAutoId(customerDatabase, 3000);
+        LOG.info("Insertion Phase 2 finishes");
+
+        thread.join();
+        insertRecordsPhase2ForAutoId(customerDatabase, 1000);
+        LOG.info("Insertion Phase 3 finishes");
+
+        // Initial 21 records + Phase1 1000 + Phase2 3000 + Phase3 1000 = 5021
+        try {
+            TestCaseUtils.repeatedCheck(
+                    () -> outCaptor.toString().split(System.lineSeparator()).length == 5021);
+        } catch (Exception e) {
+            LOG.error("Failed to verify results. Captured stdout: {}", outCaptor.toString(), e);
+        } finally {
+            yamlJob.interrupt();
+        }
+    }
+
     private static void execInContainer(Container<?> container, String prompt, String... commands)
             throws IOException, InterruptedException {
         {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/customer.sql

@@ -326,3 +326,34 @@ CREATE TABLE default_value_test (
 INSERT INTO default_value_test
 VALUES (1,'user1','Shanghai',123567),
        (2,'user2','Shanghai',123567);
+
+-- table has auto increment primary key for pt-osc testing
+CREATE TABLE customers_auto_id (
+  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+  name VARCHAR(255) NOT NULL DEFAULT 'flink',
+  address VARCHAR(1024),
+  phone_number VARCHAR(512)
+);
+
+INSERT INTO customers_auto_id
+VALUES (default, 'user_1', 'Shanghai', '123567891234'),
+       (default, 'user_2', 'Shanghai', '123567891234'),
+       (default, 'user_3', 'Shanghai', '123567891234'),
+       (default, 'user_4', 'Shanghai', '123567891234'),
+       (default, 'user_5', 'Shanghai', '123567891234'),
+       (default, 'user_6', 'Shanghai', '123567891234'),
+       (default, 'user_7', 'Shanghai', '123567891234'),
+       (default, 'user_8', 'Shanghai', '123567891234'),
+       (default, 'user_9', 'Shanghai', '123567891234'),
+       (default, 'user_10', 'Shanghai', '123567891234'),
+       (default, 'user_11', 'Shanghai', '123567891234'),
+       (default, 'user_12', 'Shanghai', '123567891234'),
+       (default, 'user_13', 'Shanghai', '123567891234'),
+       (default, 'user_14', 'Shanghai', '123567891234'),
+       (default, 'user_15', 'Shanghai', '123567891234'),
+       (default, 'user_16', 'Shanghai', '123567891234'),
+       (default, 'user_17', 'Shanghai', '123567891234'),
+       (default, 'user_18', 'Shanghai', '123567891234'),
+       (default, 'user_19', 'Shanghai', '123567891234'),
+       (default, 'user_20', 'Shanghai', '123567891234'),
+       (default, 'user_21', 'Shanghai', '123567891234');