You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lvyanquan (via GitHub)" <gi...@apache.org> on 2024/08/21 05:50:17 UTC
[PR] [FLINK-FLINK-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
lvyanquan opened a new pull request, #3560:
URL: https://github.com/apache/flink-cdc/pull/3560
This allow to scan newly table DDL during incremental reading stage.
--
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-FLINK-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1725037516
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -128,6 +131,8 @@ public DataSource createDataSource(Context context) {
int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean scanIncrementalNewlyAddedTableEnabled =
Review Comment:
```suggestion
boolean scanInBinlogPhaseNewlyAddedTableEnabled =
```
--
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-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "zhongyuandong (via GitHub)" <gi...@apache.org>.
zhongyuandong commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1728568326
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -128,6 +131,8 @@ public DataSource createDataSource(Context context) {
int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean scanIncrementalNewlyAddedTableEnabled =
Review Comment:
I used flink-java1.8.1 flink-cdc3.0.1. Which version does it support? How to set it up?
--
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-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560
--
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-FLINK-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1724885602
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java:
##########
@@ -261,4 +261,12 @@ public class MySqlDataSourceOptions {
+ "If there is a need to use a dot (.) in a regular expression to match any character, "
+ "it is necessary to escape the dot with a backslash."
+ "eg. db0.\\.*, db1.user_table_[0-9]+, db[1-2].[app|web]_order_\\.*");
+
+ @Experimental
+ public static final ConfigOption<Boolean> SCAN_INCREMENTAL_NEWLY_ADDED_TABLE_ENABLED =
+ ConfigOptions.key("scan.incremental.newly-added-table.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to scan the ddl and dml statements of newly added tables or not in incremental reading stage, by default is false.");
Review Comment:
We should explain the option works phase and the behavior, and also the diff with SCAN_NEWLY_ADDED_TABLE_ENABLED : do re-snapshot & binlog-reading for newly added table when restore
SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED ： only do binlog-reading for newly added table during binlog reading phase
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java:
##########
@@ -261,4 +261,12 @@ public class MySqlDataSourceOptions {
+ "If there is a need to use a dot (.) in a regular expression to match any character, "
+ "it is necessary to escape the dot with a backslash."
+ "eg. db0.\\.*, db1.user_table_[0-9]+, db[1-2].[app|web]_order_\\.*");
+
+ @Experimental
+ public static final ConfigOption<Boolean> SCAN_INCREMENTAL_NEWLY_ADDED_TABLE_ENABLED =
+ ConfigOptions.key("scan.incremental.newly-added-table.enabled")
Review Comment:
scan.binlog.newly-added-table.enabled
--
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-FLINK-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1724886711
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MysqlPipelineNewlyAddedTableITCase.java:
##########
@@ -149,6 +153,40 @@ private MySqlConnection getConnection() {
return DebeziumUtils.createMySqlConnection(configuration, new Properties());
}
+ @Test
+ public void testAddNewlyTablesInIncrementalStage() throws Exception {
+ List<String> tables = Collections.singletonList("address_\\.*");
+ Map<String, String> options = new HashMap<>();
+ options.put(SCAN_INCREMENTAL_NEWLY_ADDED_TABLE_ENABLED.key(), "true");
+ options.put(SCAN_STARTUP_MODE.key(), "timestamp");
+ options.put(
+ SCAN_STARTUP_TIMESTAMP_MILLIS.key(), String.valueOf(System.currentTimeMillis()));
+
+ FlinkSourceProvider sourceProvider = getFlinkSourceProvider(tables, 4, options);
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+ env.enableCheckpointing(200);
+ DataStreamSource<Event> source =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ MySqlDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo());
+
+ TypeSerializer<Event> serializer =
+ source.getTransformation().getOutputType().createSerializer(env.getConfig());
+ CheckpointedCollectResultBuffer<Event> resultBuffer =
+ new CheckpointedCollectResultBuffer<>(serializer);
+ String accumulatorName = "dataStreamCollect_" + UUID.randomUUID();
+ CollectResultIterator<Event> iterator =
+ addCollector(env, source, resultBuffer, serializer, accumulatorName);
+ env.executeAsync("AddNewlyTablesInIncrementalStage");
Review Comment:
binlogReadingState
--
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-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#issuecomment-2302041115
@Jiabao-Sun @GOODBOY008 Would you like to take a look at this PR if you have time?
--
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-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "zhongyuandong (via GitHub)" <gi...@apache.org>.
zhongyuandong commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1728568326
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -128,6 +131,8 @@ public DataSource createDataSource(Context context) {
int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean scanIncrementalNewlyAddedTableEnabled =
Review Comment:
Which version does it support? How to set it up?
--
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-36115][pipeline-connector][mysql] add scan.incremental.newly-added-table.enabled option [flink-cdc]
Posted by "zhongyuandong (via GitHub)" <gi...@apache.org>.
zhongyuandong commented on code in PR #3560:
URL: https://github.com/apache/flink-cdc/pull/3560#discussion_r1728568326
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -128,6 +131,8 @@ public DataSource createDataSource(Context context) {
int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean scanIncrementalNewlyAddedTableEnabled =
Review Comment:
I used flink-java1.8.1 flink-cdc3.0.1. Which version does it support? How to set it up?
--
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