You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "yuxiqian (via GitHub)" <gi...@apache.org> on 2024/10/28 10:09:54 UTC
[PR] [FLINK-36610] MySQL CDC supports parsing gh-ost generated schema changes [flink-cdc]
yuxiqian opened a new pull request, #3668:
URL: https://github.com/apache/flink-cdc/pull/3668
This closes FLINK-36610.
GitHub's [gh-ost](https://github.com/github/gh-ost/) project allows us to alter MySQL table schema without triggers. It is implemented based on a "replicate and swap" technique, which could not be correctly recognized by MySQL CDC for now, since `RENAME TABLE` DDL isn't supported yet.
This PR adds basic support for table altering schema change events emitted by gh-ost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904115337
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java:
##########
@@ -281,4 +281,12 @@ public class MySqlDataSourceOptions {
.withDescription(
"List of readable metadata from SourceRecord to be passed to downstream, split by `,`. "
+ "Available readable metadata are: op_ts.");
+
+ @Experimental
+ public static final ConfigOption<Boolean> PARSE_ONLINE_SCHEMA_CHANGES =
+ ConfigOptions.key("scan.parse.online.schema.changes")
Review Comment:
We can add `enabled` to key name to indicate that the configOption is boolean type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904119776
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
We can move the logic of isOnLineSchemaChangeEvent and setTableId to [MySqlStreamingChangeEventSource#handleQueryEvent](https://github.com/apache/flink-cdc/blob/ddb5f00df5bab21e8c91e4a31574d446d8b7a3f7/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java#L660) to simplify the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904871634
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Thanks for the suggestion, but I noticed that it's not very convenient to peel it out there since `SchemaChangeEvent` is still in raw state (DDL expression only) and no tableId information available yet.
More specifically, there's no TableId information available here and must be parsed manually from DDL string.
```
// MySqlStreamingChangeEventSource.java
eventDispatcher.dispatchSchemaChangeEvent(
partition,
tableId,
(receiver) -> {
try {
receiver.schemaChangeEvent(schemaChangeEvent);
} catch (Exception e) {
throw new DebeziumException(e);
}
});
```
Any suggestions on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904871634
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Thanks for the suggestion, but I noticed that it's not very convenient to peel it out there since `SchemaChangeEvent` is still in raw state (DDL expression only) and no table information available yet.
More specifically, there's no TableId information available here and must be parsed manually from DDL string.
```java
// MySqlStreamingChangeEventSource.java
eventDispatcher.dispatchSchemaChangeEvent(
partition,
tableId,
(receiver) -> {
try {
receiver.schemaChangeEvent(schemaChangeEvent);
} catch (Exception e) {
throw new DebeziumException(e);
}
});
```
Any suggestions on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2449697859
@MOBIN-F Sure! Feel free to work on this PR if helpful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "GOODBOY008 (via GitHub)" <gi...@apache.org>.
GOODBOY008 commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905613729
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
+ Struct value = (Struct) dataRecord.value();
+ Document historyRecordDocument;
+ try {
+ historyRecordDocument = getHistoryRecord(dataRecord).document();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ HistoryRecord newHistoryRecord =
+ new HistoryRecord(
+ historyRecordDocument.set(
+ "ddl",
+ historyRecordDocument
+ .get("ddl")
Review Comment:
```suggestion
.get(HistoryRecord.Fields.DDL_STATEMENTS)
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
+ Struct value = (Struct) dataRecord.value();
+ Document historyRecordDocument;
+ try {
+ historyRecordDocument = getHistoryRecord(dataRecord).document();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ HistoryRecord newHistoryRecord =
+ new HistoryRecord(
+ historyRecordDocument.set(
+ "ddl",
Review Comment:
ditto
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlOnLineSchemaMigrationTableITCase.java:
##########
@@ -0,0 +1,598 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.table;
+
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.utils.TestCaseUtils;
+import org.apache.flink.cdc.connectors.mysql.source.MySqlSourceTestBase;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Random;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.api.common.JobStatus.RUNNING;
+import static org.apache.flink.cdc.common.utils.TestCaseUtils.DEFAULT_INTERVAL;
+import static org.apache.flink.cdc.common.utils.TestCaseUtils.DEFAULT_TIMEOUT;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationTableITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String TEST_USER = "mysqluser";
+ private static final String TEST_PASSWORD = "mysqlpw";
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ private final StreamTableEnvironment tEnv =
+ StreamTableEnvironment.create(
+ env, EnvironmentSettings.newInstance().inStreamingMode().build());
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Containers are started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.close();
+ LOG.info("Containers are stopped.");
+ }
+
+ @Before
+ public void before() {
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(DEFAULT_PARALLELISM);
+ env.enableCheckpointing(200);
+ customerDatabase.createAndInitialize();
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE debezium_source ("
+ + " `id` INT NOT NULL,"
+ + " name STRING,"
+ + " address STRING,"
+ + " phone_number STRING,"
+ + " primary key (`id`) not enforced"
+ + ") WITH ("
+ + " 'connector' = 'mysql-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '%s',"
+ + " 'server-time-zone' = 'UTC',"
+ + " 'server-id' = '%s'"
+ + ")",
+ MYSQL8_CONTAINER.getHost(),
+ MYSQL8_CONTAINER.getDatabasePort(),
+ TEST_USER,
+ TEST_PASSWORD,
+ customerDatabase.getDatabaseName(),
+ "customers",
+ true,
+ getServerId());
+
+ tEnv.executeSql(sourceDDL);
+
+ // async submit job
+ TableResult result = tEnv.executeSql("SELECT * FROM debezium_source");
+
+ // wait for the source startup, we don't have a better way to wait it, use sleep for now
+ TestCaseUtils.repeatedCheck(
+ () -> result.getJobClient().get().getJobStatus().get().equals(RUNNING),
+ DEFAULT_TIMEOUT,
+ DEFAULT_INTERVAL,
+ Arrays.asList(InterruptedException.class, NoSuchElementException.class));
+
+ CloseableIterator<Row> iterator = result.collect();
+
+ {
+ String[] expected =
+ new String[] {
+ "+I[101, user_1, Shanghai, 123567891234]",
+ "+I[102, user_2, Shanghai, 123567891234]",
+ "+I[103, user_3, Shanghai, 123567891234]",
+ "+I[109, user_4, Shanghai, 123567891234]",
+ "+I[110, user_5, Shanghai, 123567891234]",
+ "+I[111, user_6, Shanghai, 123567891234]",
+ "+I[118, user_7, Shanghai, 123567891234]",
+ "+I[121, user_8, Shanghai, 123567891234]",
+ "+I[123, user_9, Shanghai, 123567891234]",
+ "+I[1009, user_10, Shanghai, 123567891234]",
+ "+I[1010, user_11, Shanghai, 123567891234]",
+ "+I[1011, user_12, Shanghai, 123567891234]",
+ "+I[1012, user_13, Shanghai, 123567891234]",
+ "+I[1013, user_14, Shanghai, 123567891234]",
+ "+I[1014, user_15, Shanghai, 123567891234]",
+ "+I[1015, user_16, Shanghai, 123567891234]",
+ "+I[1016, user_17, Shanghai, 123567891234]",
+ "+I[1017, user_18, Shanghai, 123567891234]",
+ "+I[1018, user_19, Shanghai, 123567891234]",
+ "+I[1019, user_20, Shanghai, 123567891234]",
+ "+I[2000, user_21, Shanghai, 123567891234]"
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ // Wait for a little while until we're in Binlog streaming mode.
+ Thread.sleep(5_000);
+
+ {
+ LOG.info("Step 3: Evolve schema with gh-ost - ADD COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=add column ext int first",
+ "--allow-on-master", // because we don't have a replica
+ "--initially-drop-old-table", // drop previously generated temporary tables
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ // The new column `ext` has been inserted now
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (17, 10000, 'Alice', 'Beijing', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ }
+
+ String[] expected =
+ new String[] {
+ "+I[10000, Alice, Beijing, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ {
+ LOG.info("Step 4: Evolve schema with gh-ost - MODIFY COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=modify column ext double",
+ "--allow-on-master",
+ "--initially-drop-old-table",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (2.718281828, 10001, 'Bob', 'Chongqing', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ }
+
+ String[] expected =
+ new String[] {
+ "+I[10001, Bob, Chongqing, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ {
+ LOG.info("Step 5: Evolve schema with gh-ost - DROP COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=drop column ext",
+ "--allow-on-master",
+ "--initially-drop-old-table",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10002, 'Cicada', 'Urumqi', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ }
+
+ String[] expected =
+ new String[] {
+ "+I[10002, Cicada, Urumqi, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+ }
+
+ @Test
+ public void testPtOscSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Start pipeline job");
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE debezium_source ("
+ + " `id` INT NOT NULL,"
+ + " name STRING,"
+ + " address STRING,"
+ + " phone_number STRING,"
+ + " primary key (`id`) not enforced"
+ + ") WITH ("
+ + " 'connector' = 'mysql-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '%s',"
+ + " 'server-time-zone' = 'UTC',"
+ + " 'server-id' = '%s'"
+ + ")",
+ MYSQL8_CONTAINER.getHost(),
+ MYSQL8_CONTAINER.getDatabasePort(),
+ TEST_USER,
+ TEST_PASSWORD,
+ customerDatabase.getDatabaseName(),
+ "customers",
+ true,
+ getServerId());
+
+ tEnv.executeSql(sourceDDL);
+
+ // async submit job
+ TableResult result = tEnv.executeSql("SELECT * FROM debezium_source");
+
+ // wait for the source startup, we don't have a better way to wait it, use sleep for now
+ TestCaseUtils.repeatedCheck(
+ () -> result.getJobClient().get().getJobStatus().get().equals(RUNNING),
+ DEFAULT_TIMEOUT,
+ DEFAULT_INTERVAL,
+ Arrays.asList(InterruptedException.class, NoSuchElementException.class));
+
+ CloseableIterator<Row> iterator = result.collect();
+ {
+ String[] expected =
+ new String[] {
+ "+I[101, user_1, Shanghai, 123567891234]",
+ "+I[102, user_2, Shanghai, 123567891234]",
+ "+I[103, user_3, Shanghai, 123567891234]",
+ "+I[109, user_4, Shanghai, 123567891234]",
+ "+I[110, user_5, Shanghai, 123567891234]",
+ "+I[111, user_6, Shanghai, 123567891234]",
+ "+I[118, user_7, Shanghai, 123567891234]",
+ "+I[121, user_8, Shanghai, 123567891234]",
+ "+I[123, user_9, Shanghai, 123567891234]",
+ "+I[1009, user_10, Shanghai, 123567891234]",
+ "+I[1010, user_11, Shanghai, 123567891234]",
+ "+I[1011, user_12, Shanghai, 123567891234]",
+ "+I[1012, user_13, Shanghai, 123567891234]",
+ "+I[1013, user_14, Shanghai, 123567891234]",
+ "+I[1014, user_15, Shanghai, 123567891234]",
+ "+I[1015, user_16, Shanghai, 123567891234]",
+ "+I[1016, user_17, Shanghai, 123567891234]",
+ "+I[1017, user_18, Shanghai, 123567891234]",
+ "+I[1018, user_19, Shanghai, 123567891234]",
+ "+I[1019, user_20, Shanghai, 123567891234]",
+ "+I[2000, user_21, Shanghai, 123567891234]"
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ // Wait for a little while until we're in Binlog streaming mode.
+ Thread.sleep(5_000);
+
+ LOG.info("Step 2: Evolve schema with pt-osc - ADD COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "add column ext int FIRST",
+ "--charset=utf8",
+ "--recursion-method=NONE", // Do not look for slave nodes
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ // The new column `ext` has been inserted now
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (17, 10000, 'Alice', 'Beijing', '123567891234');",
+ customerDatabase.getDatabaseName()));
+
+ String[] expected =
+ new String[] {
+ "+I[10000, Alice, Beijing, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ LOG.info("Step 3: Evolve schema with pt-osc - MODIFY COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "modify column ext double",
+ "--charset=utf8",
+ "--recursion-method=NONE",
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (2.718281828, 10001, 'Bob', 'Chongqing', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ String[] expected =
+ new String[] {
+ "+I[10001, Bob, Chongqing, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+
+ LOG.info("Step 4: Evolve schema with pt-osc - DROP COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "drop column ext",
+ "--charset=utf8",
+ "--recursion-method=NONE",
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10002, 'Cicada', 'Urumqi', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ String[] expected =
+ new String[] {
+ "+I[10002, Cicada, Urumqi, 123567891234]",
+ };
+ assertEqualsInAnyOrder(Arrays.asList(expected), fetchRows(iterator, expected.length));
+ }
+ }
+
+ private static void execInContainer(Container<?> container, String prompt, String... commands)
+ throws IOException, InterruptedException {
+ {
+ LOG.info(
+ "Starting to {} with the following command: `{}`",
+ prompt,
+ String.join(" ", commands));
+ Container.ExecResult execResult = container.execInContainer(commands);
+ if (execResult.getExitCode() == 0) {
+ LOG.info("Successfully {}. Stdout: {}", prompt, execResult.getStdout());
+ } else {
+ LOG.error(
+ "Failed to {}. Exit code: {}, Stdout: {}, Stderr: {}",
+ prompt,
+ execResult.getExitCode(),
+ execResult.getStdout(),
+ execResult.getStderr());
+ throw new IOException("Failed to execute commands: " + Arrays.toString(commands));
Review Comment:
```suggestion
throw new IOException("Failed to execute commands: " + String.join(" ", commands));
```
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
+ .startupOptions(StartupOptions.initial())
+ .serverId(getServerId(env.getParallelism()))
+ .serverTimeZone("UTC")
+ .includeSchemaChanges(SCHEMA_CHANGE_ENABLED.defaultValue())
+ .parseOnLineSchemaChanges(true);
+
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider) new MySqlDataSource(configFactory).getEventSourceProvider();
+ CloseableIterator<Event> events =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ MySqlDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo())
+ .executeAndCollect();
+ Thread.sleep(5_000);
+
+ List<Event> expected = new ArrayList<>();
+ Schema schemaV1 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ expected.add(new CreateTableEvent(tableId, schemaV1));
+ expected.addAll(getSnapshotExpected(tableId, schemaV1));
+ List<Event> actual = fetchResults(events, expected.size());
+ assertEqualsInAnyOrder(
+ expected.stream().map(Object::toString).collect(Collectors.toList()),
+ actual.stream().map(Object::toString).collect(Collectors.toList()));
+
+ // Wait for a little while until we're in Binlog streaming mode.
+ Thread.sleep(5_000);
+
+ LOG.info("Step 3: Evolve schema with gh-ost - ADD COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=add column ext int",
+ "--allow-on-master", // because we don't have a replica
+ "--initially-drop-old-table", // drop previously generated temporary tables
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ // The new column `ext` has been inserted now
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10000, 'Alice', 'Beijing', '123567891234', 17);",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV2 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.INT())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ new PhysicalColumn("ext", DataTypes.INT(), null)))),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(schemaV2, 10000, "Alice", "Beijing", "123567891234", 17))),
+ fetchResults(events, 2));
+
+ LOG.info("Step 4: Evolve schema with gh-ost - MODIFY COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=modify column ext double",
+ "--allow-on-master",
+ "--initially-drop-old-table",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10001, 'Bob', 'Chongqing', '123567891234', 2.718281828);",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV3 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.DOUBLE())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new AlterColumnTypeEvent(
+ tableId, Collections.singletonMap("ext", DataTypes.DOUBLE())),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(
+ schemaV3,
+ 10001,
+ "Bob",
+ "Chongqing",
+ "123567891234",
+ 2.718281828))),
+ fetchResults(events, 2));
+
+ LOG.info("Step 5: Evolve schema with gh-ost - DROP COLUMN");
+ execInContainer(
+ MYSQL8_CONTAINER,
+ "evolve schema",
+ "gh-ost",
+ "--user=" + TEST_USER,
+ "--password=" + TEST_PASSWORD,
+ "--database=" + customerDatabase.getDatabaseName(),
+ "--table=customers",
+ "--alter=drop column ext",
+ "--allow-on-master",
+ "--initially-drop-old-table",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10002, 'Cicada', 'Urumqi', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV4 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new DropColumnEvent(tableId, Collections.singletonList("ext")),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(schemaV4, 10002, "Cicada", "Urumqi", "123567891234"))),
+ fetchResults(events, 2));
+ }
+
+ @Test
+ public void testPtOscSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Start pipeline job");
+
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
+ .startupOptions(StartupOptions.initial())
+ .serverId(getServerId(env.getParallelism()))
+ .serverTimeZone("UTC")
+ .includeSchemaChanges(SCHEMA_CHANGE_ENABLED.defaultValue())
+ .parseOnLineSchemaChanges(true);
+
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider) new MySqlDataSource(configFactory).getEventSourceProvider();
+ CloseableIterator<Event> events =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ MySqlDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo())
+ .executeAndCollect();
+ Thread.sleep(5_000);
+
+ List<Event> expected = new ArrayList<>();
+ Schema schemaV1 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ expected.add(new CreateTableEvent(tableId, schemaV1));
+ expected.addAll(getSnapshotExpected(tableId, schemaV1));
+ List<Event> actual = fetchResults(events, expected.size());
+ assertEqualsInAnyOrder(
+ expected.stream().map(Object::toString).collect(Collectors.toList()),
+ actual.stream().map(Object::toString).collect(Collectors.toList()));
+
+ // Wait for a little while until we're in Binlog streaming mode.
+ Thread.sleep(5_000);
+
+ LOG.info("Step 2: Evolve schema with pt-osc - ADD COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "add column ext int",
+ "--charset=utf8",
+ "--recursion-method=NONE", // Do not look for slave nodes
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ // The new column `ext` has been inserted now
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10000, 'Alice', 'Beijing', '123567891234', 17);",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV2 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.INT())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ new PhysicalColumn("ext", DataTypes.INT(), null)))),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(schemaV2, 10000, "Alice", "Beijing", "123567891234", 17))),
+ fetchResults(events, 2));
+
+ LOG.info("Step 3: Evolve schema with pt-osc - MODIFY COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "modify column ext double",
+ "--charset=utf8",
+ "--recursion-method=NONE",
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10001, 'Bob', 'Chongqing', '123567891234', 2.718281828);",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV3 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.DOUBLE())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new AlterColumnTypeEvent(
+ tableId, Collections.singletonMap("ext", DataTypes.DOUBLE())),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(
+ schemaV3,
+ 10001,
+ "Bob",
+ "Chongqing",
+ "123567891234",
+ 2.718281828))),
+ fetchResults(events, 2));
+
+ LOG.info("Step 4: Evolve schema with pt-osc - DROP COLUMN");
+ 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,D=" + customerDatabase.getDatabaseName(),
+ "--alter",
+ "drop column ext",
+ "--charset=utf8",
+ "--recursion-method=NONE",
+ "--print",
+ "--execute");
+
+ try (Connection connection = customerDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`customers` VALUES (10002, 'Cicada', 'Urumqi', '123567891234');",
+ customerDatabase.getDatabaseName()));
+ }
+
+ Schema schemaV4 =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+
+ assertEquals(
+ Arrays.asList(
+ new DropColumnEvent(tableId, Collections.singletonList("ext")),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(schemaV4, 10002, "Cicada", "Urumqi", "123567891234"))),
+ fetchResults(events, 2));
+ }
+
+ private static void execInContainer(Container<?> container, String prompt, String... commands)
+ throws IOException, InterruptedException {
+ {
+ LOG.info(
+ "Starting to {} with the following command: `{}`",
+ prompt,
+ String.join(" ", commands));
+ Container.ExecResult execResult = container.execInContainer(commands);
+ if (execResult.getExitCode() == 0) {
+ LOG.info("Successfully {}. Stdout: {}", prompt, execResult.getStdout());
+ } else {
+ LOG.error(
+ "Failed to {}. Exit code: {}, Stdout: {}, Stderr: {}",
+ prompt,
+ execResult.getExitCode(),
+ execResult.getStdout(),
+ execResult.getStderr());
+ throw new IOException("Failed to execute commands: " + Arrays.toString(commands));
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904882691
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Can we reuse and modify the tableId here?
https://github.com/apache/flink-cdc/blob/ddb5f00df5bab21e8c91e4a31574d446d8b7a3f7/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java#L664-L667.
And we can use
```
schemaChangeEvents.get(i).getType().equals(SchemaChangeEvent.SchemaChangeEventType.ALTER)
```
to judge the type of raw DDL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905378005
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -489,4 +529,75 @@ private static Optional<WatermarkKind> getWatermarkKind(SourceRecord record) {
}
return Optional.empty();
}
+
+ /**
+ * This utility method checks if given source record is a gh-ost/pt-osc initiated schema change
+ * event by checking the "alter" ddl.
+ */
+ public static boolean isOnLineSchemaChangeEvent(SourceRecord record) {
+ if (!isSchemaChangeEvent(record)) {
+ return false;
+ }
+ Struct value = (Struct) record.value();
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ // 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)
+ // 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, we only need the "ALTER" one that happens on the `_gho`/`_new`
+ // table.
+ String ddl =
+ mapper.readTree(value.getString(HISTORY_RECORD_FIELD))
+ .get("ddl")
+ .asText()
+ .toLowerCase();
+ if (ddl.toLowerCase().startsWith("alter")) {
Review Comment:
This toLowerCase() is unnecessary?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905394070
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
Review Comment:
I think it would be more reliable to validate capturing the changes of entire database and making twice online table change changes for the same table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2575213081
Hi @leonardBang, @ruanhang1993 could you please help to trigger CI check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1836009961
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -489,4 +529,75 @@ private static Optional<WatermarkKind> getWatermarkKind(SourceRecord record) {
}
return Optional.empty();
}
+
+ /**
+ * This utility method checks if given source record is a gh-ost/pt-osc initiated schema change
+ * event by checking the "alter" ddl.
+ */
+ public static boolean isOnLineSchemaChangeEvent(SourceRecord record) {
+ if (!isSchemaChangeEvent(record)) {
+ return false;
+ }
+ Struct value = (Struct) record.value();
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ // 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)
+ // 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, we only need the "ALTER" one that happens on the `_gho`/`_new`
+ // table.
+ String ddl =
+ mapper.readTree(value.getString(HISTORY_RECORD_FIELD))
+ .get("ddl")
+ .asText()
+ .toLowerCase();
+ if (ddl.toLowerCase().startsWith("alter")) {
+ String tableName =
+ value.getStruct(Envelope.FieldName.SOURCE).getString(TABLE_NAME_KEY);
+ return OSC_TABLE_ID_PATTERN.matcher(tableName).matches();
+ }
+
+ return false;
+ } catch (JsonProcessingException e) {
+ return false;
+ }
+ }
+
+ private static final Pattern OSC_TABLE_ID_PATTERN = Pattern.compile("^_(.*)_(gho|new)$");
Review Comment:
Seems it's not necessary since actual schema evolution DDLs are always performed on `_XXX_gho` tables with gh-ost and `_XXX_new` with os-ptc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost generated schema changes [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2449692445
hi, @yuxiqian [pt-osc](https://docs.percona.com/percona-toolkit/pt-online-schema-change.html) is a product with similar functions to gh-ost. I think we should also support it. I am willing to do this part of the work. Can I submit code on this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "GOODBOY008 (via GitHub)" <gi...@apache.org>.
GOODBOY008 merged PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2576800570
Seems there's some [compatibility issue](https://forums.percona.com/t/percona-toolkit-execution-failure/34610) between Percona Toolkit 3.7 and MySQL. Locked to a lower version should solve this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905359170
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Got it, that's because the temporary table is not included in the table filter condition.
We can still get the table name from source struct of schemaChangeEvent, but this loses the meaning of doing so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905394070
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
Review Comment:
I think it would be more reliable to validate capturing the changes of entire database.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904902270
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Seems we can't retrieve TableIds here because `schemaChangeEvent.getTables()` is always empty for schema changes on shadow tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2576521439
Thanks for @lvyanquan and @GOODBOY008's kindly review, addressed your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "czy006 (via GitHub)" <gi...@apache.org>.
czy006 commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1836006217
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -489,4 +529,75 @@ private static Optional<WatermarkKind> getWatermarkKind(SourceRecord record) {
}
return Optional.empty();
}
+
+ /**
+ * This utility method checks if given source record is a gh-ost/pt-osc initiated schema change
+ * event by checking the "alter" ddl.
+ */
+ public static boolean isOnLineSchemaChangeEvent(SourceRecord record) {
+ if (!isSchemaChangeEvent(record)) {
+ return false;
+ }
+ Struct value = (Struct) record.value();
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ // 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)
+ // 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, we only need the "ALTER" one that happens on the `_gho`/`_new`
+ // table.
+ String ddl =
+ mapper.readTree(value.getString(HISTORY_RECORD_FIELD))
+ .get("ddl")
+ .asText()
+ .toLowerCase();
+ if (ddl.toLowerCase().startsWith("alter")) {
+ String tableName =
+ value.getStruct(Envelope.FieldName.SOURCE).getString(TABLE_NAME_KEY);
+ return OSC_TABLE_ID_PATTERN.matcher(tableName).matches();
+ }
+
+ return false;
+ } catch (JsonProcessingException e) {
+ return false;
+ }
+ }
+
+ private static final Pattern OSC_TABLE_ID_PATTERN = Pattern.compile("^_(.*)_(gho|new)$");
Review Comment:
As I know, gh-ost also has **_del** and **_ghc** tables. Do we need to join them **^_(.*)_(gho|ghc|del|new)$** ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2568928957
Hello @czy006 @MOBIN-F
Any idea when this PR could be merged?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905385783
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java:
##########
@@ -281,4 +281,12 @@ public class MySqlDataSourceOptions {
.withDescription(
"List of readable metadata from SourceRecord to be passed to downstream, split by `,`. "
+ "Available readable metadata are: op_ts.");
+
+ @Experimental
+ public static final ConfigOption<Boolean> PARSE_ONLINE_SCHEMA_CHANGES =
+ ConfigOptions.key("scan.parse.online.schema.changes.enabled")
Review Comment:
A doc update for this option is welcomed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905394070
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
Review Comment:
I think it would be more reliable to validate capturing the changes of entire database and make some online table change changes for the same table.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
Review Comment:
I think it would be more reliable to validate capturing the changes of entire database.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2467286354
Added testcases for DataStream / Table API, too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "GOODBOY008 (via GitHub)" <gi...@apache.org>.
GOODBOY008 commented on PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#issuecomment-2576782890
@yuxiqian Mysql UT failed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1904871634
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -384,6 +390,40 @@ public static TableId getTableId(SourceRecord dataRecord) {
return new TableId(dbName, null, tableName);
}
+ public static SourceRecord setTableId(
+ SourceRecord dataRecord, TableId originalTableId, TableId tableId) {
Review Comment:
Thanks for the suggestion, but I noticed that it's not very convenient to peel it out there since `SchemaChangeEvent` is still in raw state (DDL expression only) and no tableId information available yet.
More specifically, there's no TableId information available here and must be parsed manually from DDL string.
```java
// MySqlStreamingChangeEventSource.java
eventDispatcher.dispatchSchemaChangeEvent(
partition,
tableId,
(receiver) -> {
try {
receiver.schemaChangeEvent(schemaChangeEvent);
} catch (Exception e) {
throw new DebeziumException(e);
}
});
```
Any suggestions on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36610] MySQL CDC supports parsing gh-ost / pt-osc generated schema changes [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3668:
URL: https://github.com/apache/flink-cdc/pull/3668#discussion_r1905394070
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java:
##########
@@ -0,0 +1,591 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+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.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
+import org.apache.flink.cdc.connectors.mysql.testutils.MySqlVersion;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_PASSWORD;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.TEST_USER;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.fetchResults;
+import static org.apache.flink.cdc.connectors.mysql.testutils.MySqSourceTestUtils.getServerId;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * IT case for Evolving MySQL schema with gh-ost/pt-osc utility. See <a
+ * href="https://github.com/github/gh-ost">github/gh-ost</a>/<a
+ * href="https://docs.percona.com/percona-toolkit/pt-online-schema-change.html">doc/pt-osc</a> for
+ * more details.
+ */
+public class MySqlOnLineSchemaMigrationITCase extends MySqlSourceTestBase {
+ private static final MySqlContainer MYSQL8_CONTAINER =
+ createMySqlContainer(MySqlVersion.V8_0, "docker/server-gtids/expire-seconds/my.cnf");
+
+ private static final String PERCONA_TOOLKIT = "perconalab/percona-toolkit:3.5.7";
+
+ protected static final GenericContainer<?> PERCONA_TOOLKIT_CONTAINER =
+ createPerconaToolkitContainer();
+
+ private final UniqueDatabase customerDatabase =
+ new UniqueDatabase(MYSQL8_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
+
+ private final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ private static final String GH_OST_DOWNLOAD_LINK =
+ DockerClientFactory.instance().client().versionCmd().exec().getArch().equals("amd64")
+ ? "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-amd64-20231207144046.tar.gz"
+ : "https://github.com/github/gh-ost/releases/download/v1.1.6/gh-ost-binary-linux-arm64-20231207144046.tar.gz";
+
+ @BeforeClass
+ public static void beforeClass() {
+ LOG.info("Starting MySql8 containers...");
+ Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
+ Startables.deepStart(Stream.of(PERCONA_TOOLKIT_CONTAINER)).join();
+ LOG.info("Container MySql8 is started.");
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ LOG.info("Stopping MySql8 containers...");
+ MYSQL8_CONTAINER.stop();
+ PERCONA_TOOLKIT_CONTAINER.stop();
+ LOG.info("Container MySql8 is stopped.");
+ }
+
+ @Before
+ public void before() {
+ customerDatabase.createAndInitialize();
+ TestValuesTableFactory.clearAllData();
+ env.setParallelism(4);
+ env.enableCheckpointing(200);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ }
+
+ @After
+ public void after() {
+ customerDatabase.dropDatabase();
+ }
+
+ private static void installGhOstCli(Container<?> container) {
+ try {
+ execInContainer(
+ container,
+ "download gh-ost tarball",
+ "curl",
+ "-L",
+ "-o",
+ "/tmp/gh-ost.tar.gz",
+ GH_OST_DOWNLOAD_LINK);
+ execInContainer(
+ container, "unzip binary", "tar", "-xzvf", "/tmp/gh-ost.tar.gz", "-C", "/bin");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static GenericContainer<?> createPerconaToolkitContainer() {
+ GenericContainer<?> perconaToolkit =
+ new GenericContainer<>(PERCONA_TOOLKIT)
+ // keep container alive
+ .withCommand("tail", "-f", "/dev/null")
+ .withNetwork(NETWORK)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return perconaToolkit;
+ }
+
+ @Test
+ public void testGhOstSchemaMigrationFromScratch() throws Exception {
+ LOG.info("Step 1: Install gh-ost command line utility");
+ installGhOstCli(MYSQL8_CONTAINER);
+
+ LOG.info("Step 2: Start pipeline job");
+ env.setParallelism(1);
+ TableId tableId = TableId.tableId(customerDatabase.getDatabaseName(), "customers");
+ MySqlSourceConfigFactory configFactory =
+ new MySqlSourceConfigFactory()
+ .hostname(MYSQL8_CONTAINER.getHost())
+ .port(MYSQL8_CONTAINER.getDatabasePort())
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(customerDatabase.getDatabaseName())
+ .tableList(customerDatabase.getDatabaseName() + "\\.customers")
Review Comment:
I think it would be more reliable to validate capturing the changes of entire database and making twice online table change changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org