Skip to content

[FLINK-34865][pipeline-connector/mysql] Support sync table comment in binlog phase #3869

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leonardBang merged 2 commits into from
Jan 17, 2025
Merged

[FLINK-34865][pipeline-connector/mysql] Support sync table comment in binlog phase #3869

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d444048
support sync table coments in binlog phase
qg-lin Jan 16, 2025
3c58a20
Merge branch 'apache:master' into FLINK-34865
qg-lin Jan 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions ...a/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ public void exitColumnCreateTable(MySqlParser.ColumnCreateTableContext ctx) {
if (tableEditor.hasPrimaryKey()) {
builder.primaryKey(tableEditor.primaryKeyColumnNames());
}
builder.comment(tableEditor.create().comment());
changes.add(
new CreateTableEvent(
toCdcTableId(tableEditor.tableId()), builder.build()));
Expand Down Expand Up @@ -413,6 +414,21 @@ public void exitDropTable(MySqlParser.DropTableContext ctx) {
super.exitDropTable(ctx);
}

@Override
public void enterTableOptionComment(MySqlParser.TableOptionCommentContext ctx) {
if (!parser.skipComments()) {
parser.runIfNotNull(
() -> {
if (ctx.COMMENT() != null) {
tableEditor.setComment(
parser.withoutQuotes(ctx.STRING_LITERAL().getText()));
}
},
tableEditor);
}
super.enterTableOptionComment(ctx);
}

private org.apache.flink.cdc.common.schema.Column toCdcColumn(Column dbzColumn) {
return org.apache.flink.cdc.common.schema.Column.physicalColumn(
dbzColumn.name(),
Expand Down
70 changes: 70 additions & 0 deletions ...mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.INCLUDE_COMMENTS_ENABLED;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PASSWORD;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCHEMA_CHANGE_ENABLED;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SERVER_TIME_ZONE;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES;
Expand Down Expand Up @@ -1076,6 +1077,75 @@ public void testIncludeComments() throws Exception {
actual.stream().map(Object::toString).collect(Collectors.toList()));
}

@Test
public void testIncludeCommentsForScanBinlogNewlyAddedTableEnabled() throws Exception {
env.setParallelism(1);
inventoryDatabase.createAndInitialize();
TableId tableId = TableId.tableId(inventoryDatabase.getDatabaseName(), "products");
TableId newTableId =
TableId.tableId(inventoryDatabase.getDatabaseName(), "products_with_comments2");

Map<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), MYSQL8_CONTAINER.getHost());
options.put(PORT.key(), String.valueOf(MYSQL8_CONTAINER.getDatabasePort()));
options.put(USERNAME.key(), TEST_USER);
options.put(PASSWORD.key(), TEST_PASSWORD);
options.put(SERVER_TIME_ZONE.key(), "UTC");
options.put(INCLUDE_COMMENTS_ENABLED.key(), "true");
options.put(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED.key(), "true");
options.put(TABLES.key(), inventoryDatabase.getDatabaseName() + ".products\\.*");
Factory.Context context =
new FactoryHelper.DefaultContext(
Configuration.fromMap(options), null, this.getClass().getClassLoader());

MySqlDataSourceFactory factory = new MySqlDataSourceFactory();
MySqlDataSource dataSource = (MySqlDataSource) factory.createDataSource(context);
FlinkSourceProvider sourceProvider =
(FlinkSourceProvider) dataSource.getEventSourceProvider();

CloseableIterator<Event> events =
env.fromSource(
sourceProvider.getSource(),
WatermarkStrategy.noWatermarks(),
MySqlDataSourceFactory.IDENTIFIER,
new EventTypeInfo())
.executeAndCollect();
Thread.sleep(5_000);
Copy link
Member

@yuxiqian yuxiqian Jan 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we can fetchResults here to ensure snapshot stage to be over before generating binlog events, instead of sleeping.


String createTableSql =
String.format(
"CREATE TABLE IF NOT EXISTS `%s`.`%s` (\n"
+ " id INTEGER NOT NULL AUTO_INCREMENT COMMENT 'column comment of id' PRIMARY KEY,\n"
+ " name VARCHAR(255) NOT NULL DEFAULT 'flink' COMMENT 'column comment of name',\n"
+ " weight FLOAT COMMENT 'column comment of weight'\n"
+ ")\n"
+ "COMMENT 'table comment of products';",
inventoryDatabase.getDatabaseName(), "products_with_comments2");
executeSql(inventoryDatabase, createTableSql);

// add some column
String addColumnSql =
String.format(
"ALTER TABLE `%s`.`products_with_comments2` ADD COLUMN `description` VARCHAR(512) comment 'column comment of description';",
inventoryDatabase.getDatabaseName());
executeSql(inventoryDatabase, addColumnSql);

List<Event> expectedEvents = new ArrayList<>();
CreateTableEvent productCreateTableEvent = getProductsCreateTableEvent(tableId);
expectedEvents.add(productCreateTableEvent);
// generate snapshot data
List<Event> productExpectedSnapshot = getSnapshotExpected(tableId);
expectedEvents.addAll(productExpectedSnapshot);

List<Event> newTableExpectedEvents = getEventsWithComments(newTableId);
expectedEvents.addAll(newTableExpectedEvents);

List<Event> actual = fetchResults(events, expectedEvents.size());
assertEqualsInAnyOrder(
expectedEvents.stream().map(Object::toString).collect(Collectors.toList()),
actual.stream().map(Object::toString).collect(Collectors.toList()));
}

private void executeSql(UniqueDatabase database, String sql) throws SQLException {
try (Connection connection = database.getJdbcConnection();
Statement statement = connection.createStatement()) {
Expand Down
Loading