You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "aiwenmo (via GitHub)" <gi...@apache.org> on 2024/12/23 15:57:30 UTC
[PR] [FLINK-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
aiwenmo opened a new pull request, #3812:
URL: https://github.com/apache/flink-cdc/pull/3812
(no comment)
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2561920778
# Code implementation
Topology graph: Source -> PreTransform -> PostTransform -> SchemaBatchOperator-> PartitionBy(Batch) -> BatchSink
1. add SchemaBatchOperator which removed the processing of schema change event and removed Coordinator.
2. add RegularPrePartitionBatchOperator which removed SchemaEvolutionClient.
3. add DataBatchSinkFunctionOperator and DataBatchSinkWriterOperator which removed SchemaEvolutionClient.
4. remove SchemaRegistry in batch mode.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2009421299
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,23 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ public List<Set<TableId>> groupSourceTablesByRouteRule(Set<TableId> tableIdSet) {
Review Comment:
Ditto.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -76,7 +77,7 @@ public List<TableId> route(TableId sourceTableId) {
return routingCache.getUnchecked(sourceTableId);
}
- private List<TableId> calculateRoute(TableId sourceTableId) {
+ public List<TableId> calculateRoute(TableId sourceTableId) {
Review Comment:
@aiwenmo could you add some feedback about this comment？
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034623697
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,35 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ /*
+ * groupSourceTablesByRouteRule
+ *
+ * @param: tableIdSet
+ * @return: java.util.List<java.util.Set<org.apache.flink.cdc.common.event.TableId>>
+ *
+ * Group the source tables that conform to the same routing rule together.
+ * The total number of groups is less than or equal to the number of routing rules.
+ * For the source tables within each group, their table structures will be merged to obtain
+ * the widest table structure in that group. The structures of all tables within the group
+ * will be expanded to this widest table structure.
+ */
Review Comment:
ditto.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2003444291
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java:
##########
@@ -158,8 +176,7 @@ private <CommT> void addCommittingTopology(
((WithPreCommitTopology<Event, CommT>) sink).addPreCommitTopology(written);
}
- // TODO: Hard coding stream mode and checkpoint
- boolean isBatchMode = false;
+ // TODO: Hard coding checkpoint
boolean isCheckpointingEnabled = true;
Review Comment:
Some sinks need to rely on checkpointing to complete the writing.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2017811453
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +94,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ if (!alreadySendAllCreateTableTables) {
Review Comment:
In snapshot mode, we will:
* obtain and check startup mode, and check the flag
In any other modes, we will:
* obtain and check startup mode
I would suggest naming `alreadySendAllCreateTableTables` => `shouldEmitAllCtesInSnapshotMode`, and set it to `true` in snapshot mode, `false` (in other modes), and the checking could be simplified to:
```java
if (shouldEmitAllCtesInSnapshotMode) {
createTableEventCache.forEach(
(tableId, createTableEvent) -> output.collect(createTableEvent)
);
shouldEmitAllCtesInSnapshotMode = false;
}
```
so we don't have to check the startup mode every time when we receive a SourceRecord.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaBatchOperator.java:
##########
Review Comment:
minor: use consistent names for new operator classes: either `BatchXXXOperator` (like `BatchPreTransformOp`) or `XXXBatchOperator` (`SchemaBatchOperator`).
Personally I prefer the former one since it is easy to distinguish it from normal Streaming operators.
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java:
##########
Review Comment:
I think TransformE2e and UdfE2e has nothing to do with batch mode. Shall we refactor original cases with `@ParameterizedTest`, or just remove these cases to avoid code inflation and longer CI execution time?
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/PartitioningTranslator.java:
##########
@@ -46,15 +47,44 @@ public DataStream<Event> translateRegular(
int downstreamParallelism,
OperatorID schemaOperatorID,
HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
- return input.transform(
- "PrePartition",
- new PartitioningEventTypeInfo(),
- new RegularPrePartitionOperator(
- schemaOperatorID, downstreamParallelism, hashFunctionProvider))
- .setParallelism(upstreamParallelism)
- .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
- .map(new PostPartitionProcessor(), new EventTypeInfo())
- .name("PostPartition");
+ return translateRegular(
+ input,
+ upstreamParallelism,
+ downstreamParallelism,
+ false,
+ schemaOperatorID,
+ hashFunctionProvider);
+ }
+
+ public DataStream<Event> translateRegular(
+ DataStream<Event> input,
+ int upstreamParallelism,
+ int downstreamParallelism,
+ boolean isBatchMode,
+ OperatorID schemaOperatorID,
+ HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
+ if (isBatchMode) {
+ return input.transform(
+ "BatchPrePartition",
+ new PartitioningEventTypeInfo(),
+ new RegularPrePartitionBatchOperator(
+ downstreamParallelism, hashFunctionProvider))
+ .setParallelism(upstreamParallelism)
+ .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
+ .map(new PostPartitionProcessor(), new EventTypeInfo())
+ .name("BatchPostPartition")
+ .setParallelism(downstreamParallelism);
+ } else {
+ return input.transform(
+ "PrePartition",
+ new PartitioningEventTypeInfo(),
+ new RegularPrePartitionOperator(
+ schemaOperatorID, downstreamParallelism, hashFunctionProvider))
+ .setParallelism(upstreamParallelism)
+ .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
+ .map(new PostPartitionProcessor(), new EventTypeInfo())
+ .name("PostPartition");
Review Comment:
nit: operations after PrePartition seems identical and could be extracted from the if-else block
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +355,57 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
+ public static List<CreateTableEvent> deduceMergedCreateTableEventInBatchMode(
Review Comment:
Yes, but `SchemaDerivator` itself shouldn't be aware of streaming / batch execution mode. Similar initial deducing logic could be ported to streaming mode later.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +94,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ if (!alreadySendAllCreateTableTables) {
+ createTableEventCache.forEach(
+ (tableId, createTableEvent) -> {
+ output.collect(createTableEvent);
+ alreadySendAllCreateTableTables = true;
Review Comment:
nit: move `alreadySendAllCreateTableTables = true` out of the loop
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivatorTest.java:
##########
@@ -599,4 +600,113 @@ void testNormalizeSchemaChangeEventsInIgnoreMode() {
new DropTableEvent(NORMALIZE_TEST_TABLE_ID)))
.isEmpty();
}
+
+ @Test
+ void testDeduceMergedCreateTableEventInBatchMode() {
Review Comment:
ditto, remove `inBatchMode`
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -129,6 +130,13 @@ public DataSource createDataSource(Context context) {
ZoneId serverTimeZone = getServerTimeZone(config);
StartupOptions startupOptions = getStartupOptions(config);
+ // Batch mode only supports StartupMode.SNAPSHOT.
+ Configuration pipelineConfiguration = context.getPipelineConfiguration();
+ if (pipelineConfiguration != null
+ && pipelineConfiguration.contains(PipelineOptions.PIPELINE_BATCH_MODE_ENABLED)) {
+ startupOptions = StartupOptions.snapshot();
+ }
Review Comment:
Just a suggestion, what about throwing an exception explicitly if one enables batch mode with a non-snapshotting source? That would prevent some silent behavior change of `startupOptions` config.
Alternatively, we may add a interface in `DataSourceFactory` like this to make things clearer:
```java
@PublicEvolving
public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
/** Checking if this {@link DataSource} could be created in batch mode. */
boolean supportsBatchPipeline(Context context);
}
```
and verifies it during translating pipeline job graph. WDYT?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2769153836
Hi, @aiwenmo. Overall, it looks good to me and we can verify its reliability through the rich testing you have added.
However, perhaps we still need some log information to assist in analysis in case of abnormal situations, so you can refer to https://github.com/apache/flink-cdc/pull/3960/files and add some logs in BatchPreTransformOperator//DataBatchSinkWriterOperator/DataBatchSinkFunctionOperator to help better troubleshooting.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034611868
##########
docs/content.zh/docs/core-concept/data-pipeline.md:
##########
@@ -115,4 +115,5 @@ under the License.
|-----------------|---------------------------------------|-------------------|
| name | 这个 pipeline 的名称，会用在 Flink 集群中作为作业的名称。 | optional |
| parallelism | pipeline的全局并发度，默认值是1。 | optional |
-| local-time-zone | 作业级别的本地时区。 | optional |
\ No newline at end of file
+| local-time-zone | 作业级别的本地时区。 | optional |
+| batch-mode.enabled | 仅使用批处理模式来同步当前快照数据。 | optional |
Review Comment:
Hi. @ruanhang1993 .The previous writing was consistent with what you described, but @lvyanquan suggested changing it to this form. I think it's necessary to communicate and confirm.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2048647527
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/DataSourceFactory.java:
##########
@@ -28,4 +28,7 @@ public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
+
+ /** Checking if this {@link DataSource} could be created in batch mode. */
+ default void verifyRuntimeMode(Context context) {}
Review Comment:
This is not intuitive to me, why do we need to introduce a method to verify runtime mode(especially the java doc said batch mode) in Factory?
##########
flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java:
##########
@@ -366,6 +366,7 @@ private void testSchemaEvolutionTypesParsing(
.put("parallelism", "4")
.put("schema.change.behavior", "evolve")
.put("schema-operator.rpc-timeout", "1 h")
+ .put("runtime-mode", "STREAMING")
Review Comment:
The change is unnecessary if the configuration has a default value IIUC?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -274,6 +276,22 @@ public DataSource createDataSource(Context context) {
return new MySqlDataSource(configFactory, readableMetadataList);
}
+ @Override
+ public void verifyRuntimeMode(Context context) {
+ final Configuration config = context.getFactoryConfiguration();
+ StartupOptions startupOptions = getStartupOptions(config);
+ // Batch mode only supports StartupMode.SNAPSHOT.
+ Configuration pipelineConfiguration = context.getPipelineConfiguration();
+ if (pipelineConfiguration != null
+ && pipelineConfiguration.contains(PipelineOptions.PIPELINE_RUNTIME_MODE)
+ && RuntimeMode.BATCH.equals(
+ pipelineConfiguration.get(PipelineOptions.PIPELINE_RUNTIME_MODE))
+ && !StartupOptions.snapshot().equals(startupOptions)) {
+ throw new IllegalArgumentException(
+ "Batch mode is only supported for MySQL source in snapshot mode.");
Review Comment:
```suggestion
"Only "snapshot" of MySQLDataSource StartupOption is supported in BATCH pipeline, but actual MySQLDataSource StartupOption is {}.");
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperatorFactory.java:
##########
@@ -36,17 +36,31 @@ public class DataSinkWriterOperatorFactory<CommT>
YieldingOperatorFactory<CommittableMessage<CommT>> {
private final Sink<Event> sink;
+ private final boolean isBatchMode;
private final OperatorID schemaOperatorID;
- public DataSinkWriterOperatorFactory(Sink<Event> sink, OperatorID schemaOperatorID) {
+ public DataSinkWriterOperatorFactory(
+ Sink<Event> sink, boolean isBatchMode, OperatorID schemaOperatorID) {
this.sink = sink;
+ this.isBatchMode = isBatchMode;
this.schemaOperatorID = schemaOperatorID;
}
@SuppressWarnings("unchecked")
@Override
public <T extends StreamOperator<CommittableMessage<CommT>>> T createStreamOperator(
StreamOperatorParameters<CommittableMessage<CommT>> parameters) {
+
+ if (isBatchMode) {
Review Comment:
just a personal flavor,I think use `isBounded` in code is more aligned with flink's stream concept(unbounded stream and bounded stream )
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<RuntimeMode> PIPELINE_RUNTIME_MODE =
Review Comment:
Flink has `execution.runtime-mode` configuration, and I assume the two configuration has same semantics, could we follow Flink's design[1] ?
[1] https://github.com/apache/flink/blob/7c03bc834b642ff40828739961a9a4100b9fb073/flink-core-api/src/main/java/org/apache/flink/api/common/RuntimeExecutionMode.java#L39C5-L39C14
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +94,15 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (shouldEmitAllCtesInSnapshotMode
+ && StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
Review Comment:
We can calculate `Boolean isBounded = StartupOptions.snapshot().equals(sourceConfig.getStartupOptions()` in constructor instead of calculate it per element.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -71,6 +72,9 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
// Used when startup mode is initial
private Set<TableId> alreadySendCreateTableTables;
+ // Used when startup mode is snapshot
+ private boolean shouldEmitAllCtesInSnapshotMode = true;
Review Comment:
`Ctes` means?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1992850275
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java:
##########
@@ -336,6 +336,92 @@ public void testSyncWholeDatabase() throws Exception {
}
}
+ @Test
+ public void testSyncWholeDatabaseInBatchMode() throws Exception {
+ String databaseName = mysqlInventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: mysql\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
Review Comment:
It will be great if we reduce this config, and we can extract pipeline config from `Context` when using `createDataSource(Context context)` method.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2000926072
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +91,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ if (alreadySendCreateTableTables.isEmpty()) {
+ createTableEventCache.forEach(
+ (tableId, createTableEvent) -> {
+ output.collect(createTableEvent);
+ alreadySendCreateTableTables.add(tableId);
Review Comment:
alreadySendCreateTableTables is useless in snapshot-only mode.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1992559705
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -89,7 +90,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ createTableEventCache.forEach(
+ createTableEvent -> {
+ output.collect(createTableEvent);
+ TableId tableId = TableId.parse(createTableEvent.tableId().identifier());
+ alreadySendCreateTableTables.add(tableId);
+ });
+ alreadySendCreateTableForBinlogSplit = true;
+ } else if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
Review Comment:
Hi. I'm rebasing and testing.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2717696091
Hi. I'm in the process of coding and testing.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2754614941
Hi. I have rebased 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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1988229313
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java:
##########
@@ -267,7 +268,13 @@ public DataSource createDataSource(Context context) {
}
String metadataList = config.get(METADATA_LIST);
List<MySqlReadableMetadata> readableMetadataList = listReadableMetadata(metadataList);
- return new MySqlDataSource(configFactory, readableMetadataList);
+ Configuration pipelineConfiguration = context.getPipelineConfiguration();
+ boolean isBatchMode = false;
+ if (pipelineConfiguration != null
+ && pipelineConfiguration.contains(PipelineOptions.PIPELINE_BATCH_MODE_ENABLED)) {
+ isBatchMode = pipelineConfiguration.get(PipelineOptions.PIPELINE_BATCH_MODE_ENABLED);
+ }
+ return new MySqlDataSource(configFactory, isBatchMode, readableMetadataList);
Review Comment:
No such constructor method for MySqlDataSource.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +354,50 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
+ public static List<CreateTableEvent> deduceMergedCreateTableEventInBatchMode(
Review Comment:
Please add an unit test for this method.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java:
##########
@@ -81,7 +81,6 @@ public EventSourceProvider getEventSourceProvider() {
(sourceReaderMetrics, sourceConfig) ->
new MySqlPipelineRecordEmitter(
deserializer, sourceReaderMetrics, sourceConfig));
-
Review Comment:
Unnecessary change.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2003485777
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +355,57 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
+ public static List<CreateTableEvent> deduceMergedCreateTableEventInBatchMode(
Review Comment:
At present, it only merges CreateTableEvent in batch mode.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2017811453
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +94,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ if (!alreadySendAllCreateTableTables) {
Review Comment:
In snapshot mode, we will:
* obtain and check startup mode
* then check the flag
In any other modes, we will:
* obtain and check startup mode
I would suggest naming `alreadySendAllCreateTableTables` => `shouldEmitAllCtesInSnapshotMode`, and set it to `true` in snapshot mode, `false` (in other modes), and the checking could be simplified to:
```java
if (shouldEmitAllCtesInSnapshotMode) {
createTableEventCache.forEach(
(tableId, createTableEvent) -> output.collect(createTableEvent)
);
shouldEmitAllCtesInSnapshotMode = false;
}
```
so we don't have to check the startup mode every time when we receive a SourceRecord.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2009513615
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,23 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ public List<Set<TableId>> groupSourceTablesByRouteRule(Set<TableId> tableIdSet) {
Review Comment:
It makes use of "routes". In the latest code, I've added documentation for it.
Now that I think about it, it can also be placed in SchemaDerivator. I'll give it a try tonight.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2003452471
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -134,6 +135,19 @@ public static Set<Schema> reverseLookupDependingUpstreamSchemas(
.collect(Collectors.toSet());
}
+ /** For an evolved table ID, reverse lookup all upstream schemas that needs to be fit in. */
+ public static Set<Schema> reverseLookupDependingUpstreamSchemas(
+ final TableIdRouter tableIdRouter,
+ final TableId evolvedTableId,
+ final Set<TableId> allOriginalTables,
+ final Schema originalSchema) {
+ return reverseLookupDependingUpstreamTables(
+ tableIdRouter, evolvedTableId, allOriginalTables)
+ .stream()
+ .map(utid -> originalSchema)
+ .collect(Collectors.toSet());
+ }
+
Review Comment:
This code is useless and should be deleted.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1976998390
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<RunTimeMode> PIPELINE_RUNTIME_MODE =
+ ConfigOptions.key("runtime-mode")
Review Comment:
What about making this Boolean type like `batch-mode.enabled` or `snapshot-only.enabled` to simplify user config?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.ChangeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
+
+ /** A set of {@link TableId} that already processed {@link CreateTableEvent}. */
+ private final Set<TableId> processedTableIds;
+
+ private transient volatile Map<TableId, Schema> originalSchemaMap;
+
+ public DataBatchSinkFunctionOperator(
+ SinkFunction<Event> userFunction, OperatorID schemaOperatorID) {
+ super(userFunction);
+ processedTableIds = new HashSet<>();
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<Object>> output) {
+ super.setup(containingTask, config, output);
+ originalSchemaMap = new HashMap<>();
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Event> element) throws Exception {
+ Event event = element.getValue();
+
+ // FlushEvent triggers flush
+ if (event instanceof FlushEvent) {
+ return;
+ }
+
+ // CreateTableEvent marks the table as processed directly
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ processedTableIds.add(createTableEvent.tableId());
+ originalSchemaMap.put(createTableEvent.tableId(), createTableEvent.getSchema());
+ super.processElement(element);
+ return;
+ }
+
+ // Check if the table is processed before emitting all other events, because we have to make
+ // sure that sink have a view of the full schema before processing any change events,
+ // including schema changes.
+ ChangeEvent changeEvent = (ChangeEvent) event;
+ if (!processedTableIds.contains(changeEvent.tableId())) {
Review Comment:
We don‘t need to consider the case of task failover in batch mode as it should re-emit from Source.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionBatchOperator.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.runtime.partitioning;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+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.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
+import org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Operator for processing events from {@link SchemaOperator} before {@link EventPartitioner} with
+ * regular topology in batch mode.
+ */
+@Internal
+public class RegularPrePartitionBatchOperator extends AbstractStreamOperator<PartitioningEvent>
+ implements OneInputStreamOperator<Event, PartitioningEvent>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Duration CACHE_EXPIRE_DURATION = Duration.ofDays(1);
+
+ private final int downstreamParallelism;
+ private final HashFunctionProvider<DataChangeEvent> hashFunctionProvider;
+
+ private transient LoadingCache<TableId, HashFunction<DataChangeEvent>> cachedHashFunctions;
Review Comment:
We can simply use Map here as the job will finish.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -69,27 +71,44 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
private Set<TableId> alreadySendCreateTableTables;
// Used when startup mode is not initial
+ private boolean isBatchMode = false;
private boolean alreadySendCreateTableForBinlogSplit = false;
private List<CreateTableEvent> createTableEventCache;
+ private boolean alreadySendAllCreateTable = false;
public MySqlPipelineRecordEmitter(
DebeziumDeserializationSchema<Event> debeziumDeserializationSchema,
MySqlSourceReaderMetrics sourceReaderMetrics,
- MySqlSourceConfig sourceConfig) {
+ MySqlSourceConfig sourceConfig,
+ boolean isBatchMode) {
super(
debeziumDeserializationSchema,
sourceReaderMetrics,
sourceConfig.isIncludeSchemaChanges());
this.sourceConfig = sourceConfig;
this.alreadySendCreateTableTables = new HashSet<>();
this.createTableEventCache = generateCreateTableEvent(sourceConfig);
+ this.isBatchMode = isBatchMode;
}
@Override
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (isBatchMode
+ && StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())
Review Comment:
I intend not to pass `isBatchMode` to source again, StartupOptions.snapshot is enough to make us confirm that all CreateTableEvents can be emitted.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaBatchOperator.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.runtime.operators.schema.regular;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.CreateTableCompletedEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.route.RouteRule;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.runtime.operators.schema.common.SchemaDerivator;
+import org.apache.flink.cdc.runtime.operators.schema.common.SchemaManager;
+import org.apache.flink.cdc.runtime.operators.schema.common.TableIdRouter;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** The operator will apply create table event and router mapper in batch mode. */
+@Internal
+public class SchemaBatchOperator extends AbstractStreamOperator<Event>
+ implements OneInputStreamOperator<Event, Event>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(SchemaBatchOperator.class);
+
+ // Final fields that are set in constructor
+ private final String timezone;
+ private final List<RouteRule> routingRules;
+
+ // Transient fields that are set during open()
+ private transient volatile Map<TableId, Schema> originalSchemaMap;
+ private transient volatile Map<TableId, Schema> evolvedSchemaMap;
+ private transient TableIdRouter router;
+ private transient SchemaDerivator derivator;
+ protected transient SchemaManager schemaManager;
+ protected MetadataApplier metadataApplier;
+ private boolean alreadyMergedCreateTableTables = false;
+
+ public SchemaBatchOperator(
+ List<RouteRule> routingRules, MetadataApplier metadataApplier, String timezone) {
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ this.timezone = timezone;
+ this.routingRules = routingRules;
+ this.metadataApplier = metadataApplier;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<Event>> output) {
+ super.setup(containingTask, config, output);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ this.originalSchemaMap = new HashMap<>();
+ this.evolvedSchemaMap = new HashMap<>();
+ this.router = new TableIdRouter(routingRules);
+ this.derivator = new SchemaDerivator();
+ this.schemaManager = new SchemaManager(SchemaChangeBehavior.IGNORE);
+ }
+
+ /**
+ * This method is guaranteed to not be called concurrently with other methods of the operator.
+ */
+ @Override
+ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
+ Event event = streamRecord.getValue();
+ // System.out.println("Print Event:" + event.toString());
+ // Only catch create table event and data change event in batch mode
+ if (event instanceof CreateTableEvent) {
+ handleCreateTableEvent((CreateTableEvent) event);
+ } else if (event instanceof CreateTableCompletedEvent) {
Review Comment:
I don't think we need to add an additional event.
Instead, when we receive a DataChangeEvent for the first time, we can assume that all CreateTableEvents have been sent.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkWriterOperator.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.ChangeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * An operator that processes records to be written into a {@link Sink} in batch mode.
+ *
+ * <p>The operator is a proxy of SinkWriterOperator in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ *
+ * @param <CommT> the type of the committable (to send to downstream operators)
+ */
+@Internal
+public class DataBatchSinkWriterOperator<CommT>
+ extends AbstractStreamOperator<CommittableMessage<CommT>>
+ implements OneInputStreamOperator<Event, CommittableMessage<CommT>>, BoundedOneInput {
+
+ private final Sink<Event> sink;
+
+ private final ProcessingTimeService processingTimeService;
+
+ private final MailboxExecutor mailboxExecutor;
+
+ /** Operator that actually execute sink logic. */
+ private Object flinkWriterOperator;
+
+ /**
+ * The internal {@link SinkWriter} of flinkWriterOperator, obtained it through reflection to
+ * deal with {@link FlushEvent}.
+ */
+ private SinkWriter<Event> copySinkWriter;
+
+ /** A set of {@link TableId} that already processed {@link CreateTableEvent}. */
+ private final Set<TableId> processedTableIds;
+
+ private transient volatile Map<TableId, Schema> originalSchemaMap;
+
+ public DataBatchSinkWriterOperator(
+ Sink<Event> sink,
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor) {
+ this.sink = sink;
+ this.processingTimeService = processingTimeService;
+ this.mailboxExecutor = mailboxExecutor;
+ this.processedTableIds = new HashSet<>();
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<CommittableMessage<CommT>>> output) {
+ super.setup(containingTask, config, output);
+ flinkWriterOperator = createFlinkWriterOperator();
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .setup(containingTask, config, output);
+ originalSchemaMap = new HashMap<>();
+ }
+
+ @Override
+ public void open() throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator().open();
+ copySinkWriter = getFieldValue("sinkWriter");
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .initializeState(context);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .snapshotState(context);
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .processWatermark(mark);
+ }
+
+ @Override
+ public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {
+ super.processWatermarkStatus(watermarkStatus);
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .processWatermarkStatus(watermarkStatus);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Event> element) throws Exception {
+ Event event = element.getValue();
+
+ // FlushEvent triggers flush
+ if (event instanceof FlushEvent) {
+ handleFlushEvent(((FlushEvent) event));
+ return;
+ }
+
+ // CreateTableEvent marks the table as processed directly
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ processedTableIds.add(createTableEvent.tableId());
+ originalSchemaMap.put(createTableEvent.tableId(), createTableEvent.getSchema());
+ this.<OneInputStreamOperator<Event, CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .processElement(element);
+ return;
+ }
+
+ // Check if the table is processed before emitting all other events, because we have to make
+ // sure that sink have a view of the full schema before processing any change events,
+ // including schema changes.
+ ChangeEvent changeEvent = (ChangeEvent) event;
+ if (!processedTableIds.contains(changeEvent.tableId())) {
Review Comment:
Ditto.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaBatchOperator.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.runtime.operators.schema.regular;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.CreateTableCompletedEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.route.RouteRule;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.runtime.operators.schema.common.SchemaDerivator;
+import org.apache.flink.cdc.runtime.operators.schema.common.SchemaManager;
+import org.apache.flink.cdc.runtime.operators.schema.common.TableIdRouter;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** The operator will apply create table event and router mapper in batch mode. */
+@Internal
+public class SchemaBatchOperator extends AbstractStreamOperator<Event>
+ implements OneInputStreamOperator<Event, Event>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(SchemaBatchOperator.class);
+
+ // Final fields that are set in constructor
+ private final String timezone;
+ private final List<RouteRule> routingRules;
+
+ // Transient fields that are set during open()
+ private transient volatile Map<TableId, Schema> originalSchemaMap;
+ private transient volatile Map<TableId, Schema> evolvedSchemaMap;
+ private transient TableIdRouter router;
+ private transient SchemaDerivator derivator;
+ protected transient SchemaManager schemaManager;
+ protected MetadataApplier metadataApplier;
+ private boolean alreadyMergedCreateTableTables = false;
+
+ public SchemaBatchOperator(
+ List<RouteRule> routingRules, MetadataApplier metadataApplier, String timezone) {
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ this.timezone = timezone;
+ this.routingRules = routingRules;
+ this.metadataApplier = metadataApplier;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<Event>> output) {
+ super.setup(containingTask, config, output);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ this.originalSchemaMap = new HashMap<>();
+ this.evolvedSchemaMap = new HashMap<>();
+ this.router = new TableIdRouter(routingRules);
+ this.derivator = new SchemaDerivator();
+ this.schemaManager = new SchemaManager(SchemaChangeBehavior.IGNORE);
+ }
+
+ /**
+ * This method is guaranteed to not be called concurrently with other methods of the operator.
+ */
+ @Override
+ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
+ Event event = streamRecord.getValue();
+ // System.out.println("Print Event:" + event.toString());
Review Comment:
Can be removed.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1992559031
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java:
##########
@@ -336,6 +336,92 @@ public void testSyncWholeDatabase() throws Exception {
}
}
+ @Test
+ public void testSyncWholeDatabaseInBatchMode() throws Exception {
+ String databaseName = mysqlInventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: mysql\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
Review Comment:
It is currently needed. Since batch mode is only applicable to snapshots, should it be optimized so that snapshot is the default option if not specified?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034656382
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
Review Comment:
Are you referring to this comment I left earlier?
> minor: use consistent names for new operator classes: either BatchXXXOperator (like BatchPreTransformOp) or XXXBatchOperator (SchemaBatchOperator).
> Personally I prefer the former one since it is easy to distinguish it from normal Streaming operators.
I think it's essentially the same as what Hang just suggested: favor `BatchDataSinkFunctionOperator` instead of `DataBatchSinkFunctionOperator`.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034645271
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/DataSourceFactory.java:
##########
@@ -28,4 +28,10 @@ public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
+
+ /** Checking if this {@link DataSource} could be created in batch mode. */
+ default void verifyBatchMode(Context context) {
Review Comment:
+1 for @ruanhang1993's suggestion.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034621913
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java:
##########
@@ -73,6 +73,12 @@ public SchemaManager() {
behavior = SchemaChangeBehavior.EVOLVE;
}
+ public SchemaManager(SchemaChangeBehavior schemaChangeBehavior) {
+ evolvedSchemas = new ConcurrentHashMap<>();
+ originalSchemas = new ConcurrentHashMap<>();
+ behavior = schemaChangeBehavior;
Review Comment:
I will update the code tonight.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2771406207
Hi @leonardBang, PTAL.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2051474586
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/DataSourceFactory.java:
##########
@@ -28,4 +28,7 @@ public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
+
+ /** Checking if this {@link DataSource} could be created in batch mode. */
+ default void verifyRuntimeMode(Context context) {}
Review Comment:
Hi. @leonardBang The purpose of adding this method is to remind developers to handle streaming and batch processing when developing new data sources. It's also feasible to remove the definition of this method and let developers implement it themselves in the createDataSource method. So, should we remove it now? I'm glad to hear your opinion.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2566053278
During the test, a new bug was discovered and has been fixed. This PR relies on this fix. #3826
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034656382
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
Review Comment:
Are you referring to this comment I left earlier?
> minor: use consistent names for new operator classes: either BatchXXXOperator (like BatchPreTransformOp) or XXXBatchOperator (SchemaBatchOperator).
> Personally I prefer the former one since it is easy to distinguish it from normal Streaming operators.
I think it's essentially the same as what Hang just suggested: favor `BatchDataSinkFunctionOperator` instead of `DataBatchSinkFunctionOperator`. The same below.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034617315
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/DataSourceFactory.java:
##########
@@ -28,4 +28,10 @@ public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
+
+ /** Checking if this {@link DataSource} could be created in batch mode. */
+ default void verifyBatchMode(Context context) {
Review Comment:
ditto. @yuxiqian
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2051486078
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -71,6 +72,9 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
// Used when startup mode is initial
private Set<TableId> alreadySendCreateTableTables;
+ // Used when startup mode is snapshot
+ private boolean shouldEmitAllCtesInSnapshotMode = true;
Review Comment:
`Ctes` means CreateTableEvents. How do you think it should be modified to make it clearer?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2051868198
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -71,6 +72,9 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
// Used when startup mode is initial
private Set<TableId> alreadySendCreateTableTables;
+ // Used when startup mode is snapshot
+ private boolean shouldEmitAllCtesInSnapshotMode = true;
Review Comment:
Clear is better than long in my opinion
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2753650761
Hi @aiwenmo, could you please rebase this PR with latest `master` when available?
Code style verifier has been updated to enforce JUnit 5 + AssertJ framework and these classes might need to be migrated:
* JUnit 4 style test annotations should be changed to JUnit 5 equivalents
* `org.junit.Test` => `org.junit.jupiter.api.Test`
* `@Before`, `@BeforeClass` => `@BeforeEach`, `@BeforeAll`
* `@After`, `@AfterClass` => `@AfterEach`, `@AfterAll`
* JUnit Assertions / Hamcrest Assertions are not allowed, including:
* `org.junit.Assert`
* `org.junit.jupiter.api.Assertions`
* `org.hamcrest.*`
Use `org.assertj.core.api.Assertions` instead.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1991296352
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
+
+ public DataBatchSinkFunctionOperator(SinkFunction<Event> userFunction) {
+ super(userFunction);
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ }
+
+ @Override
+ public void processElement(StreamRecord<Event> element) throws Exception {
+ Event event = element.getValue();
+
+ // FlushEvent triggers flush
+ if (event instanceof FlushEvent) {
Review Comment:
Actually, there is no FlushEvent here because SchemaBatchOperator will not emit FlushEvent to downstream.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkWriterOperator.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+
+/**
+ * An operator that processes records to be written into a {@link Sink} in batch mode.
+ *
+ * <p>The operator is a proxy of SinkWriterOperator in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ *
+ * @param <CommT> the type of the committable (to send to downstream operators)
+ */
+@Internal
+public class DataBatchSinkWriterOperator<CommT>
+ extends AbstractStreamOperator<CommittableMessage<CommT>>
+ implements OneInputStreamOperator<Event, CommittableMessage<CommT>>, BoundedOneInput {
+
+ private final Sink<Event> sink;
+
+ private final ProcessingTimeService processingTimeService;
+
+ private final MailboxExecutor mailboxExecutor;
+
+ /** Operator that actually execute sink logic. */
+ private Object flinkWriterOperator;
+
+ /**
+ * The internal {@link SinkWriter} of flinkWriterOperator, obtained it through reflection to
+ * deal with {@link FlushEvent}.
+ */
+ private SinkWriter<Event> copySinkWriter;
+
+ public DataBatchSinkWriterOperator(
+ Sink<Event> sink,
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor) {
+ this.sink = sink;
+ this.processingTimeService = processingTimeService;
+ this.mailboxExecutor = mailboxExecutor;
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<CommittableMessage<CommT>>> output) {
+ super.setup(containingTask, config, output);
+ flinkWriterOperator = createFlinkWriterOperator();
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .setup(containingTask, config, output);
+ }
+
+ @Override
+ public void open() throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator().open();
+ copySinkWriter = getFieldValue("sinkWriter");
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .initializeState(context);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .snapshotState(context);
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .processWatermark(mark);
+ }
+
+ @Override
+ public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {
+ super.processWatermarkStatus(watermarkStatus);
+ this.<AbstractStreamOperator<CommittableMessage<CommT>>>getFlinkWriterOperator()
+ .processWatermarkStatus(watermarkStatus);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Event> element) throws Exception {
+ Event event = element.getValue();
+
+ // FlushEvent triggers flush
+ if (event instanceof FlushEvent) {
Review Comment:
Ditto.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreBatchTransformOperator.java:
##########
@@ -0,0 +1,389 @@
+/*
+ * 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.runtime.operators.transform;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.DropTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.event.TruncateTableEvent;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.schema.Selectors;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+import org.apache.flink.cdc.runtime.parser.TransformParser;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * A data process function that filters out columns which aren't (directly & indirectly) referenced
+ * in batch mode. There is no need to consider the task failover scenario in batch mode, as it
+ * should be re - emitted from the source.
+ */
+public class PreBatchTransformOperator extends AbstractStreamOperator<Event>
+ implements OneInputStreamOperator<Event, Event>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<TransformRule> transformRules;
+ private transient List<PreTransformer> transforms;
+ private final Map<TableId, PreTransformChangeInfo> preTransformChangeInfoMap;
+ private final List<Tuple2<Selectors, SchemaMetadataTransform>> schemaMetadataTransformers;
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
+ private List<UserDefinedFunctionDescriptor> udfDescriptors;
+ private Map<TableId, PreTransformProcessor> preTransformProcessorMap;
+ private Map<TableId, Boolean> hasAsteriskMap;
+
+ public static PreBatchTransformOperator.Builder newBuilder() {
+ return new PreBatchTransformOperator.Builder();
+ }
+
+ /** Builder of {@link PreBatchTransformOperator}. */
+ public static class Builder {
+ private final List<TransformRule> transformRules = new ArrayList<>();
+
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
+ new ArrayList<>();
+
+ public PreBatchTransformOperator.Builder addTransform(
+ String tableInclusions, @Nullable String projection, @Nullable String filter) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ "",
+ "",
+ "",
+ null,
+ new SupportedMetadataColumn[0]));
+ return this;
+ }
+
+ public PreBatchTransformOperator.Builder addTransform(
+ String tableInclusions,
+ @Nullable String projection,
+ @Nullable String filter,
+ String primaryKey,
+ String partitionKey,
+ String tableOption,
+ @Nullable String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ primaryKey,
+ partitionKey,
+ tableOption,
+ postTransformConverter,
+ supportedMetadataColumns));
+ return this;
+ }
+
+ public PreBatchTransformOperator.Builder addUdfFunctions(
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ this.udfFunctions.addAll(udfFunctions);
+ return this;
+ }
+
+ public PreBatchTransformOperator build() {
+ return new PreBatchTransformOperator(transformRules, udfFunctions);
+ }
+ }
+
+ private PreBatchTransformOperator(
+ List<TransformRule> transformRules,
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ this.preTransformChangeInfoMap = new ConcurrentHashMap<>();
+ this.preTransformProcessorMap = new ConcurrentHashMap<>();
+ this.schemaMetadataTransformers = new ArrayList<>();
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
+
+ this.transformRules = transformRules;
+ this.udfFunctions = udfFunctions;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<Event>> output) {
+ super.setup(containingTask, config, output);
+ this.udfDescriptors =
+ this.udfFunctions.stream()
+ .map(udf -> new UserDefinedFunctionDescriptor(udf.f0, udf.f1, udf.f2))
+ .collect(Collectors.toList());
+
+ // Initialize data fields in advance because they might be accessed in
+ // `::initializeState` function when restoring from a previous state.
+ this.transforms = new ArrayList<>();
+ for (TransformRule transformRule : transformRules) {
+ String tableInclusions = transformRule.getTableInclusions();
+ String projection = transformRule.getProjection();
+ String filter = transformRule.getFilter();
+ String primaryKeys = transformRule.getPrimaryKey();
+ String partitionKeys = transformRule.getPartitionKey();
+ String tableOptions = transformRule.getTableOption();
+ Selectors selectors =
+ new Selectors.SelectorsBuilder().includeTables(tableInclusions).build();
+ transforms.add(
+ new PreTransformer(
+ selectors,
+ TransformProjection.of(projection).orElse(null),
+ TransformFilter.of(filter, udfDescriptors).orElse(null)));
+ schemaMetadataTransformers.add(
+ new Tuple2<>(
+ selectors,
+ new SchemaMetadataTransform(primaryKeys, partitionKeys, tableOptions)));
+ }
+ this.preTransformProcessorMap = new ConcurrentHashMap<>();
+ this.hasAsteriskMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void finish() throws Exception {
+ super.finish();
+ clearOperator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ clearOperator();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Event> element) throws Exception {
+ Event event = element.getValue();
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ preTransformProcessorMap.remove(createTableEvent.tableId());
+ output.collect(new StreamRecord<>(cacheCreateTable(createTableEvent)));
+ } else if (event instanceof DropTableEvent) {
+ preTransformProcessorMap.remove(((DropTableEvent) event).tableId());
+ output.collect(new StreamRecord<>(event));
+ } else if (event instanceof TruncateTableEvent) {
+ output.collect(new StreamRecord<>(event));
+ } else if (event instanceof SchemaChangeEvent) {
+ SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
+ preTransformProcessorMap.remove(schemaChangeEvent.tableId());
+ cacheChangeSchema(schemaChangeEvent)
+ .ifPresent(e -> output.collect(new StreamRecord<>(e)));
Review Comment:
No DropTableEvent, TruncateTableEvent and other SchemaChangeEvents will be sent, so we can only deal with CreateTableEvent and DataChangeEvent.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1992850275
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java:
##########
@@ -336,6 +336,92 @@ public void testSyncWholeDatabase() throws Exception {
}
}
+ @Test
+ public void testSyncWholeDatabaseInBatchMode() throws Exception {
+ String databaseName = mysqlInventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: mysql\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
Review Comment:
It will be great if we reduce this config, and we can extract pipeline config from Context when using `createDataSource(Context context)` method.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2002273866
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java:
##########
@@ -158,8 +176,7 @@ private <CommT> void addCommittingTopology(
((WithPreCommitTopology<Event, CommT>) sink).addPreCommitTopology(written);
}
- // TODO: Hard coding stream mode and checkpoint
- boolean isBatchMode = false;
+ // TODO: Hard coding checkpoint
boolean isCheckpointingEnabled = true;
Review Comment:
Just curious, is it possible to enable checkpointing in batch mode?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -134,6 +135,19 @@ public static Set<Schema> reverseLookupDependingUpstreamSchemas(
.collect(Collectors.toSet());
}
+ /** For an evolved table ID, reverse lookup all upstream schemas that needs to be fit in. */
+ public static Set<Schema> reverseLookupDependingUpstreamSchemas(
+ final TableIdRouter tableIdRouter,
+ final TableId evolvedTableId,
+ final Set<TableId> allOriginalTables,
+ final Schema originalSchema) {
+ return reverseLookupDependingUpstreamTables(
+ tableIdRouter, evolvedTableId, allOriginalTables)
+ .stream()
+ .map(utid -> originalSchema)
+ .collect(Collectors.toSet());
+ }
+
Review Comment:
Seems this method will always return `Set{originalSchema}` or `Set{}`. What is it used for?
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/PartitioningTranslator.java:
##########
@@ -46,6 +47,34 @@ public DataStream<Event> translateRegular(
int downstreamParallelism,
OperatorID schemaOperatorID,
HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
+ return translateRegular(
+ input,
+ upstreamParallelism,
+ downstreamParallelism,
+ false,
+ schemaOperatorID,
+ hashFunctionProvider);
+ }
+
+ public DataStream<Event> translateRegular(
+ DataStream<Event> input,
+ int upstreamParallelism,
+ int downstreamParallelism,
+ boolean isBatchMode,
+ OperatorID schemaOperatorID,
+ HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
+ if (isBatchMode) {
+ return input.transform(
+ "BatchPrePartition",
+ new PartitioningEventTypeInfo(),
+ new RegularPrePartitionBatchOperator(
+ downstreamParallelism, hashFunctionProvider))
+ .setParallelism(upstreamParallelism)
+ .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
+ .map(new PostPartitionProcessor(), new EventTypeInfo())
+ .name("BatchPostPartition")
+ .setParallelism(downstreamParallelism);
+ }
Review Comment:
ditto: prefer
```java
if (isBatchMode) {
// batch code
} else {
// streaming code
}
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperatorFactory.java:
##########
@@ -59,6 +79,9 @@ public <T extends StreamOperator<CommittableMessage<CommT>>> T createStreamOpera
@Override
public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ if (isBatchMode) {
+ return DataBatchSinkWriterOperator.class;
+ }
return DataSinkWriterOperator.class;
Review Comment:
nit: wrap streaming case in `else { ... }`
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java:
##########
@@ -111,6 +112,12 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
SchemaChangeBehavior schemaChangeBehavior =
pipelineDefConfig.get(PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR);
+ boolean isBatchMode = false;
+ if (pipelineDefConfig.get(PipelineOptions.PIPELINE_BATCH_MODE_ENABLED)) {
+ isBatchMode = true;
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ }
Review Comment:
```suggestion
boolean isBatchMode =
pipelineDefConfig.get(PipelineOptions.PIPELINE_BATCH_MODE_ENABLED);
if (isBatchMode) {
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
} else {
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
}
```
nit: though `STREAMING` is the default option for now, manually specifying it would be clearer.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +355,57 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
Review Comment:
```suggestion
/** Deduce initial downstream schemas based on upstream tables and routing rules. */
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperatorFactory.java:
##########
@@ -36,17 +36,37 @@ public class DataSinkWriterOperatorFactory<CommT>
YieldingOperatorFactory<CommittableMessage<CommT>> {
private final Sink<Event> sink;
+ private final boolean isBatchMode;
private final OperatorID schemaOperatorID;
public DataSinkWriterOperatorFactory(Sink<Event> sink, OperatorID schemaOperatorID) {
this.sink = sink;
+ this.isBatchMode = false;
+ this.schemaOperatorID = schemaOperatorID;
+ }
Review Comment:
Better remove this constructor and specify streaming / batch mode explicitly
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +355,57 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
+ public static List<CreateTableEvent> deduceMergedCreateTableEventInBatchMode(
Review Comment:
```suggestion
public static List<CreateTableEvent> deduceInitialCreateTableEvents(
```
Removed `inBatchMode`, since it might be helpful in streaming mode, too.
##########
flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml:
##########
@@ -58,6 +58,7 @@ pipeline:
parallelism: 4
schema.change.behavior: evolve
schema-operator.rpc-timeout: 1 h
+ batch-mode.enabled: false
Review Comment:
minor: since batchMode is disabled by default, maybe we can turn it on here to verify if it could be enabled correctly?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -76,7 +77,7 @@ public List<TableId> route(TableId sourceTableId) {
return routingCache.getUnchecked(sourceTableId);
}
- private List<TableId> calculateRoute(TableId sourceTableId) {
+ public List<TableId> calculateRoute(TableId sourceTableId) {
Review Comment:
Don't modify this, it's an internal method. Call `TableIdRouter#route` externally so the results could be cached.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<Boolean> PIPELINE_BATCH_MODE_ENABLED =
+ ConfigOptions.key("batch-mode.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enabled batch mode of the pipeline");
Review Comment:
```suggestion
.withDescription("Run pipeline job in batch mode instead of streaming mode");
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,23 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ public List<Set<TableId>> groupSourceTablesByRouteRule(Set<TableId> tableIdSet) {
Review Comment:
I doubt if it's really a generic and reusable method, without corresponding JavaDocs and test cases. Maybe just write it as a `for` loop in `SchemaDerivator`?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java:
##########
@@ -341,4 +355,57 @@ public Optional<DataChangeEvent> coerceDataRecord(
return Optional.of(dataChangeEvent);
}
+
+ /** Deduce merged CreateTableEvent in batch mode. */
+ public static List<CreateTableEvent> deduceMergedCreateTableEventInBatchMode(
+ TableIdRouter router, List<CreateTableEvent> createTableEvents) {
+ Set<TableId> originalTables =
+ createTableEvents.stream()
+ .map(CreateTableEvent::tableId)
+ .collect(Collectors.toSet());
+
+ List<Set<TableId>> sourceTablesByRouteRule =
+ router.groupSourceTablesByRouteRule(originalTables);
+ Map<TableId, Schema> sourceTableIdToSchemaMap =
+ createTableEvents.stream()
+ .collect(
+ Collectors.toMap(
+ CreateTableEvent::tableId, CreateTableEvent::getSchema));
+ Map<TableId, Schema> sinkTableIdToSchemaMap = new HashMap<>();
+ Set<TableId> routedTables = new HashSet<>();
+ for (Set<TableId> sourceTables : sourceTablesByRouteRule) {
+ List<Schema> toBeMergedSchemas = new ArrayList<>();
+ for (TableId tableId : sourceTables) {
+ toBeMergedSchemas.add(sourceTableIdToSchemaMap.get(tableId));
+ routedTables.add(tableId);
+ }
+ if (toBeMergedSchemas.isEmpty()) {
+ continue;
+ }
+ Schema mergedSchema = null;
+ for (Schema toBeMergedSchema : toBeMergedSchemas) {
+ if (mergedSchema == null) {
+ mergedSchema = toBeMergedSchema;
+ continue;
+ }
+ mergedSchema =
+ SchemaMergingUtils.getLeastCommonSchema(mergedSchema, toBeMergedSchema);
+ }
Review Comment:
call `SchemaUtils#getCommonSchema`
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreBatchTransformOperator.java:
##########
@@ -0,0 +1,337 @@
+/*
+ * 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.runtime.operators.transform;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.schema.Selectors;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.runtime.parser.TransformParser;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * A data process function that filters out columns which aren't (directly & indirectly) referenced
+ * in batch mode. There is no need to consider the task failover scenario in batch mode, as it
+ * should be re - emitted from the source.
+ */
+public class PreBatchTransformOperator extends AbstractStreamOperator<Event>
Review Comment:
"PreTransform" is a phrase. Call it `BatchPreTransformOperator`?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2742002897
PTAL @leonardBang @lvyanquan @yuxiqian
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2009509031
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -76,7 +77,7 @@ public List<TableId> route(TableId sourceTableId) {
return routingCache.getUnchecked(sourceTableId);
}
- private List<TableId> calculateRoute(TableId sourceTableId) {
+ public List<TableId> calculateRoute(TableId sourceTableId) {
Review Comment:
I've already replaced `TableIdRouter#calculateRoute` with `TableIdRouter#route`.I forgot to remove the change here.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2011141794
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,23 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ public List<Set<TableId>> groupSourceTablesByRouteRule(Set<TableId> tableIdSet) {
Review Comment:
Sorry. The attempt couldn't be achieved. The current way of writing is more optimal.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2563877827
5. DataSource will send CreateTableCompletedEvent after sending all CreateTableEvent.
6. add CreateTableCompletedEvent to notify SchemaBatchOperator to merge all CreateTableEvent.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034618276
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<Boolean> PIPELINE_BATCH_MODE_ENABLED =
+ ConfigOptions.key("batch-mode.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Run pipeline job in batch mode instead of streaming mode");
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r1992509136
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java:
##########
@@ -336,6 +336,92 @@ public void testSyncWholeDatabase() throws Exception {
}
}
+ @Test
+ public void testSyncWholeDatabaseInBatchMode() throws Exception {
+ String databaseName = mysqlInventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: mysql\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
Review Comment:
Do we need to add `scan.startup.mode: snapshot` in MySQL source config?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -89,7 +90,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
+ if (StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ createTableEventCache.forEach(
+ createTableEvent -> {
+ output.collect(createTableEvent);
+ TableId tableId = TableId.parse(createTableEvent.tableId().identifier());
+ alreadySendCreateTableTables.add(tableId);
+ });
+ alreadySendCreateTableForBinlogSplit = true;
+ } else if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
Review Comment:
You may need to rebase to solve the conflict as there is another change in MySqlPipelineRecordEmitter.java that introduced in https://github.com/apache/flink-cdc/commit/eee0cc06f7bf842034d0e22bc45f8ffb811f7f78#diff-7da556d0b9e83dcd71a4a2319c7588c813a5c14ac0dfc180fd096ae7b2a4a5b0.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2017971522
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/PartitioningTranslator.java:
##########
@@ -46,15 +47,44 @@ public DataStream<Event> translateRegular(
int downstreamParallelism,
OperatorID schemaOperatorID,
HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
- return input.transform(
- "PrePartition",
- new PartitioningEventTypeInfo(),
- new RegularPrePartitionOperator(
- schemaOperatorID, downstreamParallelism, hashFunctionProvider))
- .setParallelism(upstreamParallelism)
- .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
- .map(new PostPartitionProcessor(), new EventTypeInfo())
- .name("PostPartition");
+ return translateRegular(
+ input,
+ upstreamParallelism,
+ downstreamParallelism,
+ false,
+ schemaOperatorID,
+ hashFunctionProvider);
+ }
+
+ public DataStream<Event> translateRegular(
+ DataStream<Event> input,
+ int upstreamParallelism,
+ int downstreamParallelism,
+ boolean isBatchMode,
+ OperatorID schemaOperatorID,
+ HashFunctionProvider<DataChangeEvent> hashFunctionProvider) {
+ if (isBatchMode) {
+ return input.transform(
+ "BatchPrePartition",
+ new PartitioningEventTypeInfo(),
+ new RegularPrePartitionBatchOperator(
+ downstreamParallelism, hashFunctionProvider))
+ .setParallelism(upstreamParallelism)
+ .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
+ .map(new PostPartitionProcessor(), new EventTypeInfo())
+ .name("BatchPostPartition")
+ .setParallelism(downstreamParallelism);
+ } else {
+ return input.transform(
+ "PrePartition",
+ new PartitioningEventTypeInfo(),
+ new RegularPrePartitionOperator(
+ schemaOperatorID, downstreamParallelism, hashFunctionProvider))
+ .setParallelism(upstreamParallelism)
+ .partitionCustom(new EventPartitioner(), new PartitioningEventKeySelector())
+ .map(new PostPartitionProcessor(), new EventTypeInfo())
+ .name("PostPartition");
Review Comment:
No modification is needed after discussion.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2051478326
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<RuntimeMode> PIPELINE_RUNTIME_MODE =
Review Comment:
Regarding this issue, there have been numerous exchanges of views and modifications with other reviewers before. Now, I want to know the following information more clearly:
1.Should we change to the same parameter definition as that of Flink, with our parameter having a lower priority than Flink's own parameters?
2.Is it feasible to use "snapshot" along with Flink's own parameters for control? In this way, when users enable the batch mode, they need to explicitly specify Flink's `execution.runtime-mode` parameter.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2032935652
##########
docs/content.zh/docs/core-concept/data-pipeline.md:
##########
@@ -115,4 +115,5 @@ under the License.
|-----------------|---------------------------------------|-------------------|
| name | 这个 pipeline 的名称，会用在 Flink 集群中作为作业的名称。 | optional |
| parallelism | pipeline的全局并发度，默认值是1。 | optional |
-| local-time-zone | 作业级别的本地时区。 | optional |
\ No newline at end of file
+| local-time-zone | 作业级别的本地时区。 | optional |
+| batch-mode.enabled | 仅使用批处理模式来同步当前快照数据。 | optional |
Review Comment:
I think it is better to use a option `runtime-mode` in enum type with a default value `STREAMING`, like `RuntimeExecutionMode` in flink.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -95,4 +96,35 @@ private TableId resolveReplacement(
}
return TableId.parse(route.f1);
}
+
+ /*
+ * groupSourceTablesByRouteRule
+ *
+ * @param: tableIdSet
+ * @return: java.util.List<java.util.Set<org.apache.flink.cdc.common.event.TableId>>
+ *
+ * Group the source tables that conform to the same routing rule together.
+ * The total number of groups is less than or equal to the number of routing rules.
+ * For the source tables within each group, their table structures will be merged to obtain
+ * the widest table structure in that group. The structures of all tables within the group
+ * will be expanded to this widest table structure.
+ */
Review Comment:
```suggestion
/**
* Group the source tables that conform to the same routing rule together.
* The total number of groups is less than or equal to the number of routing rules.
* For the source tables within each group, their table structures will be merged to obtain
* the widest table structure in that group. The structures of all tables within the group
* will be expanded to this widest table structure.
*
* @param tableIdSet The tables need to be grouped by the router
* @return The tables grouped by the router
*/
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java:
##########
@@ -73,6 +73,12 @@ public SchemaManager() {
behavior = SchemaChangeBehavior.EVOLVE;
Review Comment:
```suggestion
this(new ConcurrentHashMap<>(), new ConcurrentHashMap<>(), SchemaChangeBehavior.EVOLVE);
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkWriterOperator.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+
+/**
+ * An operator that processes records to be written into a {@link Sink} in batch mode.
+ *
+ * <p>The operator is a proxy of SinkWriterOperator in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ *
+ * @param <CommT> the type of the committable (to send to downstream operators)
+ */
+@Internal
+public class DataBatchSinkWriterOperator<CommT>
Review Comment:
```suggestion
public class BatchDataSinkWriterOperator<CommT>
```
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/DataSourceFactory.java:
##########
@@ -28,4 +28,10 @@ public interface DataSourceFactory extends Factory {
/** Creates a {@link DataSource} instance. */
DataSource createDataSource(Context context);
+
+ /** Checking if this {@link DataSource} could be created in batch mode. */
+ default void verifyBatchMode(Context context) {
Review Comment:
```suggestion
default void verifyRuntimeMode(Context context, RuntimeMode runtimeMode) {
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java:
##########
@@ -73,6 +73,12 @@ public SchemaManager() {
behavior = SchemaChangeBehavior.EVOLVE;
}
+ public SchemaManager(SchemaChangeBehavior schemaChangeBehavior) {
+ evolvedSchemas = new ConcurrentHashMap<>();
+ originalSchemas = new ConcurrentHashMap<>();
+ behavior = schemaChangeBehavior;
Review Comment:
```suggestion
this(new ConcurrentHashMap<>(), new ConcurrentHashMap<>(), schemaChangeBehavior)
```
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java:
##########
@@ -45,6 +45,12 @@ public class PipelineOptions {
.defaultValue(1)
.withDescription("Parallelism of the pipeline");
+ public static final ConfigOption<Boolean> PIPELINE_BATCH_MODE_ENABLED =
+ ConfigOptions.key("batch-mode.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Run pipeline job in batch mode instead of streaming mode");
Review Comment:
```java
public static final ConfigOption<Boolean> PIPELINE_RUNTIME_MODE =
ConfigOptions.key("runtime-mode")
.enumType(RuntimeMode.class)
.defaultValue(STREAMING)
.withDescription("xxx");
```
Maybe we could reuse the flink `RuntimeExecutionMode` instead of introducing a new enum.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionBatchOperator.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.runtime.partitioning;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+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.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Operator for processing events from {@link SchemaOperator} before {@link EventPartitioner} with
+ * regular topology in batch mode.
+ */
+@Internal
+public class RegularPrePartitionBatchOperator extends AbstractStreamOperator<PartitioningEvent>
Review Comment:
```suggestion
public class BatchRegularPrePartitionOperator extends AbstractStreamOperator<PartitioningEvent>
```
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkFunctionOperator.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * An operator that processes records to be written into a {@link SinkFunction} in batch mode.
+ *
+ * <p>The operator is a proxy of {@link StreamSink} in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ */
+@Internal
+public class DataBatchSinkFunctionOperator extends StreamSink<Event> {
Review Comment:
```suggestion
public class BatchDataSinkFunctionOperator extends StreamSink<Event> {
```
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2792290534
Hi. @ruanhang1993 All checks have passed in my repository. Please run the failed work again. thx
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034624663
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataBatchSinkWriterOperator.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.runtime.operators.sink;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.FlushEvent;
+import org.apache.flink.cdc.runtime.operators.sink.exception.SinkWrapperException;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+
+/**
+ * An operator that processes records to be written into a {@link Sink} in batch mode.
+ *
+ * <p>The operator is a proxy of SinkWriterOperator in Flink.
+ *
+ * <p>The operator is always part of a sink pipeline and is the first operator.
+ *
+ * @param <CommT> the type of the committable (to send to downstream operators)
+ */
+@Internal
+public class DataBatchSinkWriterOperator<CommT>
Review Comment:
ditto.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionBatchOperator.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.runtime.partitioning;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+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.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Operator for processing events from {@link SchemaOperator} before {@link EventPartitioner} with
+ * regular topology in batch mode.
+ */
+@Internal
+public class RegularPrePartitionBatchOperator extends AbstractStreamOperator<PartitioningEvent>
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2034620832
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaManager.java:
##########
@@ -73,6 +73,12 @@ public SchemaManager() {
behavior = SchemaChangeBehavior.EVOLVE;
Review Comment:
I will update the code tonight.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2003397123
##########
flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml:
##########
@@ -58,6 +58,7 @@ pipeline:
parallelism: 4
schema.change.behavior: evolve
schema-operator.rpc-timeout: 1 h
+ batch-mode.enabled: false
Review Comment:
Many parameters are not effective in batch mode, so "false" is written here.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2000903783
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java:
##########
@@ -200,7 +201,22 @@ public SplitEnumerator<MySqlSplit, PendingSplitsState> createEnumerator(
validator.validate();
final MySqlSplitAssigner splitAssigner;
- if (!sourceConfig.getStartupOptions().isStreamOnly()) {
+ // In batch mode, only split snapshots.
Review Comment:
This change will also modify the behavior of Stream mode, should we update the comment to `In snapshot-only startup option`?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java:
##########
@@ -200,7 +201,22 @@ public SplitEnumerator<MySqlSplit, PendingSplitsState> createEnumerator(
validator.validate();
final MySqlSplitAssigner splitAssigner;
- if (!sourceConfig.getStartupOptions().isStreamOnly()) {
+ // In batch mode, only split snapshots.
Review Comment:
This change will also modify the behavior of Stream mode, should we update the comment to `In snapshot-moly mode`?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java:
##########
@@ -200,7 +201,22 @@ public SplitEnumerator<MySqlSplit, PendingSplitsState> createEnumerator(
validator.validate();
final MySqlSplitAssigner splitAssigner;
- if (!sourceConfig.getStartupOptions().isStreamOnly()) {
+ // In batch mode, only split snapshots.
Review Comment:
This change will also modify the behavior of Stream mode, should we update the comment to `In snapshot-moly mode`?
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2009509031
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/TableIdRouter.java:
##########
@@ -76,7 +77,7 @@ public List<TableId> route(TableId sourceTableId) {
return routingCache.getUnchecked(sourceTableId);
}
- private List<TableId> calculateRoute(TableId sourceTableId) {
+ public List<TableId> calculateRoute(TableId sourceTableId) {
Review Comment:
I've already replaced `TableIdRouter#calculateRoute` with `TableIdRouter#route`.
I forgot to remove the change here.
I'll submit the code tonight.
--
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-36931][cdc] FlinkCDC YAML supports batch mode [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#issuecomment-2717688411
I think an e2e test that run in batch mode with transform module is necessary to verify the whole pipeline is runnable.
--
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