You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ru...@apache.org on 2024/12/20 03:09:34 UTC
(flink-cdc) branch master updated: [FLINK-36784][common] Support to add metadata columns for data in the meta fields of DataChangeEvent at transform (#3758)
This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new a0ae3320f [FLINK-36784][common] Support to add metadata columns for data in the meta fields of DataChangeEvent at transform (#3758)
a0ae3320f is described below
commit a0ae3320fae042874c7a5ad190906d282410f3b6
Author: Hang Ruan <ru...@hotmail.com>
AuthorDate: Fri Dec 20 11:09:30 2024 +0800
[FLINK-36784][common] Support to add metadata columns for data in the meta fields of DataChangeEvent at transform (#3758)
Co-authored-by: Kunni <lv...@alibaba-inc.com>
---
.../apache/flink/cdc/common/source/DataSource.java | 5 +
.../cdc/common/source/SupportedMetadataColumn.java | 46 ++++++
.../cdc/composer/flink/FlinkPipelineComposer.java | 12 +-
.../flink/translator/DataSourceTranslator.java | 43 +++---
.../flink/translator/TransformTranslator.java | 13 +-
.../flink/FlinkPipelineComposerITCase.java | 28 ++--
.../flink/FlinkPipelineComposerLenientITCase.java | 24 ++--
.../flink/FlinkPipelineTransformITCase.java | 1 +
.../cdc/composer/flink/FlinkPipelineUdfITCase.java | 92 ++++++------
.../mysql/factory/MySqlDataSourceFactory.java | 33 ++++-
.../connectors/mysql/source/MySqlDataSource.java | 23 ++-
.../mysql/source/MySqlDataSourceOptions.java | 9 ++
.../mysql/source/MySqlEventDeserializer.java | 27 +++-
.../mysql/source/OpTsMetadataColumn.java | 51 +++++++
.../mysql/source/MySqlPipelineITCase.java | 158 +++++++++++++++++++++
.../values/source/OpTsMetadataColumn.java | 51 +++++++
.../connectors/values/source/ValuesDataSource.java | 6 +
.../values/source/ValuesDataSourceHelper.java | 35 ++++-
.../cdc/pipeline/tests/TransformE2eITCase.java | 68 ++++++---
.../operators/transform/PostTransformOperator.java | 51 +++++--
.../operators/transform/PostTransformer.java | 10 +-
.../operators/transform/PreTransformOperator.java | 17 ++-
.../transform/ProjectionColumnProcessor.java | 46 +++++-
.../transform/TransformFilterProcessor.java | 47 +++++-
.../transform/TransformProjectionProcessor.java | 59 ++++++--
.../runtime/operators/transform/TransformRule.java | 11 +-
.../flink/cdc/runtime/parser/TransformParser.java | 52 +++++--
.../transform/PostTransformOperatorTest.java | 13 +-
.../transform/PreTransformOperatorTest.java | 31 ++--
.../cdc/runtime/parser/TransformParserTest.java | 15 +-
30 files changed, 876 insertions(+), 201 deletions(-)
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/DataSource.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/DataSource.java
index 1027207a4..096648f8b 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/DataSource.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/DataSource.java
@@ -31,4 +31,9 @@ public interface DataSource {
/** Get the {@link MetadataAccessor} for accessing metadata from external systems. */
MetadataAccessor getMetadataAccessor();
+
+ /** Get the {@link SupportedMetadataColumn}s of the source. */
+ default SupportedMetadataColumn[] supportedMetadataColumns() {
+ return new SupportedMetadataColumn[0];
+ }
}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/SupportedMetadataColumn.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/SupportedMetadataColumn.java
new file mode 100644
index 000000000..6bd2f51d6
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/SupportedMetadataColumn.java
@@ -0,0 +1,46 @@
+/*
+ * 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.common.source;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.types.DataType;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/** A metadata column that the source supports to read from the meta field. */
+@Experimental
+public interface SupportedMetadataColumn extends Serializable {
+ /** Column name. */
+ String getName();
+
+ /** The data type of this column in Flink CDC. */
+ DataType getType();
+
+ /** The returned java class of the reader. */
+ Class<?> getJavaClass();
+
+ /**
+ * Read the metadata from the {@link DataChangeEvent#meta()}.
+ *
+ * @param metadata the metadata returned from {@link DataChangeEvent#meta()}
+ * @return the value of this metadata found by metadata name
+ */
+ Object read(Map<String, String> metadata);
+}
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index 579eb9607..4578a49fd 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.source.DataSource;
import org.apache.flink.cdc.composer.PipelineComposer;
import org.apache.flink.cdc.composer.PipelineExecution;
import org.apache.flink.cdc.composer.definition.PipelineDef;
@@ -99,9 +100,10 @@ public class FlinkPipelineComposer implements PipelineComposer {
// Build Source Operator
DataSourceTranslator sourceTranslator = new DataSourceTranslator();
+ DataSource dataSource =
+ sourceTranslator.createDataSource(pipelineDef.getSource(), env, pipelineDefConfig);
DataStream<Event> stream =
- sourceTranslator.translate(
- pipelineDef.getSource(), env, pipelineDefConfig, parallelism);
+ sourceTranslator.translate(pipelineDef.getSource(), env, parallelism, dataSource);
// Build PreTransformOperator for processing Schema Event
TransformTranslator transformTranslator = new TransformTranslator();
@@ -110,7 +112,8 @@ public class FlinkPipelineComposer implements PipelineComposer {
stream,
pipelineDef.getTransforms(),
pipelineDef.getUdfs(),
- pipelineDef.getModels());
+ pipelineDef.getModels(),
+ dataSource.supportedMetadataColumns());
// Schema operator
SchemaOperatorTranslator schemaOperatorTranslator =
@@ -129,7 +132,8 @@ public class FlinkPipelineComposer implements PipelineComposer {
pipelineDef.getTransforms(),
pipelineDef.getConfig().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
pipelineDef.getUdfs(),
- pipelineDef.getModels());
+ pipelineDef.getModels(),
+ dataSource.supportedMetadataColumns());
// Build DataSink in advance as schema operator requires MetadataApplier
DataSinkTranslator sinkTranslator = new DataSinkTranslator();
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java
index 90bbea73e..4323871ff 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java
@@ -39,14 +39,29 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@Internal
public class DataSourceTranslator {
+ public DataSource createDataSource(
+ SourceDef sourceDef, StreamExecutionEnvironment env, Configuration pipelineConfig) {
+ // Search the data source factory
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier(
+ sourceDef.getType(), DataSourceFactory.class);
+ // Add source JAR to environment
+ FactoryDiscoveryUtils.getJarPathByIdentifier(sourceFactory)
+ .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
+ DataSource dataSource =
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ sourceDef.getConfig(),
+ pipelineConfig,
+ Thread.currentThread().getContextClassLoader()));
+ return dataSource;
+ }
+
public DataStreamSource<Event> translate(
SourceDef sourceDef,
StreamExecutionEnvironment env,
- Configuration pipelineConfig,
- int sourceParallelism) {
- // Create data source
- DataSource dataSource = createDataSource(sourceDef, env, pipelineConfig);
-
+ int sourceParallelism,
+ DataSource dataSource) {
// Get source provider
EventSourceProvider eventSourceProvider = dataSource.getEventSourceProvider();
if (eventSourceProvider instanceof FlinkSourceProvider) {
@@ -78,24 +93,6 @@ public class DataSourceTranslator {
}
}
- private DataSource createDataSource(
- SourceDef sourceDef, StreamExecutionEnvironment env, Configuration pipelineConfig) {
- // Search the data source factory
- DataSourceFactory sourceFactory =
- FactoryDiscoveryUtils.getFactoryByIdentifier(
- sourceDef.getType(), DataSourceFactory.class);
- // Add source JAR to environment
- FactoryDiscoveryUtils.getJarPathByIdentifier(sourceFactory)
- .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
- DataSource dataSource =
- sourceFactory.createDataSource(
- new FactoryHelper.DefaultContext(
- sourceDef.getConfig(),
- pipelineConfig,
- Thread.currentThread().getContextClassLoader()));
- return dataSource;
- }
-
private String generateDefaultSourceName(SourceDef sourceDef) {
return String.format("Flink CDC Event Source: %s", sourceDef.getType());
}
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
index c84911fc7..bd2fadbc2 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.composer.flink.translator;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.composer.definition.ModelDef;
import org.apache.flink.cdc.composer.definition.TransformDef;
import org.apache.flink.cdc.composer.definition.UdfDef;
@@ -46,7 +47,8 @@ public class TransformTranslator {
DataStream<Event> input,
List<TransformDef> transforms,
List<UdfDef> udfFunctions,
- List<ModelDef> models) {
+ List<ModelDef> models,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
if (transforms.isEmpty()) {
return input;
}
@@ -61,7 +63,8 @@ public class TransformTranslator {
transform.getPrimaryKeys(),
transform.getPartitionKeys(),
transform.getTableOptions(),
- transform.getPostTransformConverter());
+ transform.getPostTransformConverter(),
+ supportedMetadataColumns);
}
preTransformFunctionBuilder.addUdfFunctions(
@@ -77,7 +80,8 @@ public class TransformTranslator {
List<TransformDef> transforms,
String timezone,
List<UdfDef> udfFunctions,
- List<ModelDef> models) {
+ List<ModelDef> models,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
if (transforms.isEmpty()) {
return input;
}
@@ -93,7 +97,8 @@ public class TransformTranslator {
transform.getPrimaryKeys(),
transform.getPartitionKeys(),
transform.getTableOptions(),
- transform.getPostTransformConverter());
+ transform.getPostTransformConverter(),
+ supportedMetadataColumns);
}
}
postTransformFunctionBuilder.addTimezone(timezone);
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
index 7820981ef..c8877ef6e 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
@@ -351,13 +351,13 @@ class FlinkPipelineComposerITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20], after=[2, x, 20], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20], after=[2, x, 20], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -383,7 +383,7 @@ class FlinkPipelineComposerITCase {
TransformDef transformDef =
new TransformDef(
"default_namespace.default_schema.table1",
- "*,concat(col1,'0') as col12,__data_event_type__ as rk",
+ "*,concat(col1,'0') as col12,__data_event_type__ as rk,op_ts as opts",
"col1 <> '3'",
"col1",
"col12",
@@ -413,14 +413,14 @@ class FlinkPipelineComposerITCase {
String[] outputEvents = outCaptor.toString().trim().split("\n");
assertThat(outputEvents)
.containsExactly(
- "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING,`rk` STRING NOT NULL}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20, +I], op=INSERT, meta=()}",
+ "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING,`rk` STRING NOT NULL,`opts` BIGINT NOT NULL}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10, +I, 1], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20, +I, 2], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10, -D], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20, -U], after=[2, x, 20, +U], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10, -D, 4], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20, -U, 5], after=[2, x, 20, +U, 5], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -486,13 +486,13 @@ class FlinkPipelineComposerITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 11], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 22], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 11], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 22], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 11], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 22], after=[2, x, 22], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 11], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 22], after=[2, x, 22], op=UPDATE, meta=({op_ts=5})}");
}
@Test
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java
index 2dd6b8998..da33d83d5 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java
@@ -395,12 +395,12 @@ class FlinkPipelineComposerLenientITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`newCol2` STRING, position=LAST, existedColumnName=null}, ColumnWithPosition{column=`newCol3` STRING, position=LAST, existedColumnName=null}]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 10, null, null, 1], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 20, null, null, ], after=[2, null, 20, null, null, x], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 10, null, null, 1], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 20, null, null, ], after=[2, null, 20, null, null, x], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -456,12 +456,12 @@ class FlinkPipelineComposerLenientITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING,`rk` STRING NOT NULL}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20, +I], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 10, +I], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 20, +I], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`newCol2` STRING, position=LAST, existedColumnName=null}, ColumnWithPosition{column=`newCol3` STRING, position=LAST, existedColumnName=null}]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 10, -D, null, null, 1], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 20, -U, null, null, ], after=[2, null, 20, +U, null, null, x], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 10, -D, null, null, 1], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 20, -U, null, null, ], after=[2, null, 20, +U, null, null, x], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -526,12 +526,12 @@ class FlinkPipelineComposerLenientITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 11], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 22], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 11], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 22], op=INSERT, meta=({op_ts=2})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`newCol2` STRING, position=LAST, existedColumnName=null}, ColumnWithPosition{column=`newCol3` STRING, position=LAST, existedColumnName=null}]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 11, null, null, 1], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 22, null, null, ], after=[2, null, 22, null, null, x], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, null, 11, null, null, 1], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, 22, null, null, ], after=[2, null, 22, null, null, x], op=UPDATE, meta=({op_ts=5})}");
}
@Test
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index f0d1f5679..6a775e74a 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -1889,6 +1889,7 @@ class FlinkPipelineTransformITCase {
null,
null,
null,
+ null,
null)),
Collections.emptyList(),
pipelineConfig);
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
index 55d5b9600..b97b4b6de 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
@@ -166,14 +166,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`fmt` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, from 1 to z is lie], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, from 2 to z is lie], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, from 3 to z is lie], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, from 1 to z is lie], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, from 2 to z is lie], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, from 3 to z is lie], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, from 1 to z is lie], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , from 2 to z is lie], after=[2, x, from 2 to z is lie], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, from 1 to z is lie], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , from 2 to z is lie], after=[2, x, from 2 to z is lie], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -237,12 +237,12 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`collen` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 3], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 4], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 3], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 4], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 3], after=[2, x, 3], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 3], after=[2, x, 3], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -306,14 +306,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`tob` STRING,`toi` STRING,`tof` STRING,`tos` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Boolean: true, Integer: 1, Double: 3.14, String: str], after=[2, x, Boolean: true, Integer: 1, Double: 3.14, String: str], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Boolean: true, Integer: 1, Double: 3.14, String: str], after=[2, x, Boolean: true, Integer: 1, Double: 3.14, String: str], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -378,14 +378,14 @@ public class FlinkPipelineUdfITCase {
.contains("[ LifecycleFunction ] opened.")
.contains(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`stt` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, #0], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, #1], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, #2], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, #0], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, #1], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, #2], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, #3], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , #4], after=[2, x, #5], op=UPDATE, meta=()}")
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, #3], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , #4], after=[2, x, #5], op=UPDATE, meta=({op_ts=5})}")
.contains("[ LifecycleFunction ] closed. Called 6 times.");
}
@@ -450,14 +450,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`ans` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Forty-two], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Forty-two], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Forty-two], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Forty-two], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Forty-two], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Forty-two], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Forty-two], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Forty-two], after=[2, x, Forty-two], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Forty-two], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Forty-two], after=[2, x, Forty-two], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -535,14 +535,14 @@ public class FlinkPipelineUdfITCase {
.contains("[ LifecycleFunction ] opened.")
.contains(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`inccol` STRING,`typ` STRING,`fmt` STRING,`stt` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 3, Integer: 42, 1-42, #0], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 4, Integer: 42, 2-42, #1], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 5, Integer: 42, 3-42, #2], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 3, Integer: 42, 1-42, #0], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 4, Integer: 42, 2-42, #1], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 5, Integer: 42, 3-42, #2], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 3, Integer: 42, 1-42, #3], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 4, Integer: 42, 2-42, #4], after=[2, x, 4, Integer: 42, 2-42, #5], op=UPDATE, meta=()}")
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 3, Integer: 42, 1-42, #3], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 4, Integer: 42, 2-42, #4], after=[2, x, 4, Integer: 42, 2-42, #5], op=UPDATE, meta=({op_ts=5})}")
.contains("[ LifecycleFunction ] closed. Called 6 times.");
}
@@ -610,14 +610,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`fmt` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, from 1 to z is lie], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, from 2 to z is lie], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, from 3 to z is lie], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, from 1 to z is lie], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, from 2 to z is lie], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, from 3 to z is lie], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, from 1 to z is lie], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , from 2 to z is lie], after=[2, x, from 2 to z is lie], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, from 1 to z is lie], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , from 2 to z is lie], after=[2, x, from 2 to z is lie], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -680,12 +680,12 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`collen` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 3], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 4], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 3], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 4], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 3], after=[2, x, 3], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 3], after=[2, x, 3], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -748,14 +748,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.containsExactly(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`tob` STRING,`toi` STRING,`tof` STRING,`tos` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, Boolean: true, Integer: 1, Double: 3.14, String: str], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Boolean: true, Integer: 1, Double: 3.14, String: str], after=[2, x, Boolean: true, Integer: 1, Double: 3.14, String: str], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, Boolean: true, Integer: 1, Double: 3.14, String: str], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , Boolean: true, Integer: 1, Double: 3.14, String: str], after=[2, x, Boolean: true, Integer: 1, Double: 3.14, String: str], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
@@ -827,14 +827,14 @@ public class FlinkPipelineUdfITCase {
assertThat(outputEvents)
.contains(
"CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`inccol` STRING,`typ` STRING,`fmt` STRING}, primaryKeys=col1, options=({key1=value1})}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 3, Integer: 42, 1-42], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 4, Integer: 42, 2-42], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 5, Integer: 42, 3-42], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, 3, Integer: 42, 1-42], op=INSERT, meta=({op_ts=1})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, 4, Integer: 42, 2-42], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3, 3, 5, Integer: 42, 3-42], op=INSERT, meta=({op_ts=3})}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 3, Integer: 42, 1-42], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 4, Integer: 42, 2-42], after=[2, x, 4, Integer: 42, 2-42], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 3, Integer: 42, 1-42], after=[], op=DELETE, meta=({op_ts=4})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 4, Integer: 42, 2-42], after=[2, x, 4, Integer: 42, 2-42], op=UPDATE, meta=({op_ts=5})}");
}
@ParameterizedTest
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
index 118e8cdb1..3d89f18c0 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
@@ -26,12 +26,14 @@ import org.apache.flink.cdc.common.factories.Factory;
import org.apache.flink.cdc.common.factories.FactoryHelper;
import org.apache.flink.cdc.common.schema.Selectors;
import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.common.utils.StringUtils;
import org.apache.flink.cdc.connectors.mysql.source.MySqlDataSource;
import org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
import org.apache.flink.cdc.connectors.mysql.source.config.ServerIdRange;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetBuilder;
+import org.apache.flink.cdc.connectors.mysql.table.MySqlReadableMetadata;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.connectors.mysql.utils.MySqlSchemaUtils;
import org.apache.flink.cdc.connectors.mysql.utils.OptionUtils;
@@ -46,6 +48,8 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -62,6 +66,7 @@ import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOption
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.CONNECT_TIMEOUT;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.HEARTBEAT_INTERVAL;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.HOSTNAME;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.METADATA_LIST;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PASSWORD;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
@@ -234,8 +239,33 @@ public class MySqlDataSourceFactory implements DataSourceFactory {
LOG.info("Add chunkKeyColumn {}.", chunkKeyColumnMap);
configFactory.chunkKeyColumn(chunkKeyColumnMap);
}
+ String metadataList = config.get(METADATA_LIST);
+ List<MySqlReadableMetadata> readableMetadataList = listReadableMetadata(metadataList);
+ return new MySqlDataSource(configFactory, readableMetadataList);
+ }
- return new MySqlDataSource(configFactory);
+ private List<MySqlReadableMetadata> listReadableMetadata(String metadataList) {
+ if (StringUtils.isNullOrWhitespaceOnly(metadataList)) {
+ return new ArrayList<>();
+ }
+ Set<String> readableMetadataList =
+ Arrays.stream(metadataList.split(","))
+ .map(String::trim)
+ .collect(Collectors.toSet());
+ List<MySqlReadableMetadata> foundMetadata = new ArrayList<>();
+ for (MySqlReadableMetadata metadata : MySqlReadableMetadata.values()) {
+ if (readableMetadataList.contains(metadata.getKey())) {
+ foundMetadata.add(metadata);
+ readableMetadataList.remove(metadata.getKey());
+ }
+ }
+ if (readableMetadataList.isEmpty()) {
+ return foundMetadata;
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "[%s] cannot be found in mysql metadata.",
+ String.join(", ", readableMetadataList)));
}
@Override
@@ -276,6 +306,7 @@ public class MySqlDataSourceFactory implements DataSourceFactory {
options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
options.add(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
+ options.add(METADATA_LIST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
index d1dc487c0..17cb5cd5a 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
@@ -24,11 +24,16 @@ import org.apache.flink.cdc.common.source.DataSource;
import org.apache.flink.cdc.common.source.EventSourceProvider;
import org.apache.flink.cdc.common.source.FlinkSourceProvider;
import org.apache.flink.cdc.common.source.MetadataAccessor;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
import org.apache.flink.cdc.connectors.mysql.source.reader.MySqlPipelineRecordEmitter;
+import org.apache.flink.cdc.connectors.mysql.table.MySqlReadableMetadata;
import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode;
+import java.util.ArrayList;
+import java.util.List;
+
/** A {@link DataSource} for mysql cdc connector. */
@Internal
public class MySqlDataSource implements DataSource {
@@ -36,16 +41,27 @@ public class MySqlDataSource implements DataSource {
private final MySqlSourceConfigFactory configFactory;
private final MySqlSourceConfig sourceConfig;
+ private List<MySqlReadableMetadata> readableMetadataList;
+
public MySqlDataSource(MySqlSourceConfigFactory configFactory) {
+ this(configFactory, new ArrayList<>());
+ }
+
+ public MySqlDataSource(
+ MySqlSourceConfigFactory configFactory,
+ List<MySqlReadableMetadata> readableMetadataList) {
this.configFactory = configFactory;
this.sourceConfig = configFactory.createConfig(0);
+ this.readableMetadataList = readableMetadataList;
}
@Override
public EventSourceProvider getEventSourceProvider() {
MySqlEventDeserializer deserializer =
new MySqlEventDeserializer(
- DebeziumChangelogMode.ALL, sourceConfig.isIncludeSchemaChanges());
+ DebeziumChangelogMode.ALL,
+ sourceConfig.isIncludeSchemaChanges(),
+ readableMetadataList);
MySqlSource<Event> source =
new MySqlSource<>(
@@ -67,4 +83,9 @@ public class MySqlDataSource implements DataSource {
public MySqlSourceConfig getSourceConfig() {
return sourceConfig;
}
+
+ @Override
+ public SupportedMetadataColumn[] supportedMetadataColumns() {
+ return new SupportedMetadataColumn[] {new OpTsMetadataColumn()};
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
index 580d370b5..f37703f69 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
@@ -272,4 +272,13 @@ public class MySqlDataSourceOptions {
+ "The difference between scan.newly-added-table.enabled and scan.binlog.newly-added-table.enabled options is: \n"
+ "scan.newly-added-table.enabled: do re-snapshot & binlog-reading for newly added table when restored; \n"
+ "scan.binlog.newly-added-table.enabled: only do binlog-reading for newly added table during binlog reading phase.");
+
+ @Experimental
+ public static final ConfigOption<String> METADATA_LIST =
+ ConfigOptions.key("metadata.list")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "List of readable metadata from SourceRecord to be passed to downstream, split by `,`. "
+ + "Available readable metadata are: op_ts.");
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
index 548e603fa..a50ac8729 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
@@ -22,8 +22,10 @@ import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.connectors.mysql.source.parser.CustomMySqlAntlrDdlParser;
+import org.apache.flink.cdc.connectors.mysql.table.MySqlReadableMetadata;
import org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema;
import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode;
+import org.apache.flink.table.data.TimestampData;
import com.esri.core.geometry.ogc.OGCGeometry;
import com.fasterxml.jackson.databind.JsonNode;
@@ -39,6 +41,7 @@ import org.apache.kafka.connect.source.SourceRecord;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -63,10 +66,20 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema {
private transient Tables tables;
private transient CustomMySqlAntlrDdlParser customParser;
+ private List<MySqlReadableMetadata> readableMetadataList;
+
public MySqlEventDeserializer(
DebeziumChangelogMode changelogMode, boolean includeSchemaChanges) {
+ this(changelogMode, includeSchemaChanges, new ArrayList<>());
+ }
+
+ public MySqlEventDeserializer(
+ DebeziumChangelogMode changelogMode,
+ boolean includeSchemaChanges,
+ List<MySqlReadableMetadata> readableMetadataList) {
super(new MySqlSchemaDataTypeInference(), changelogMode);
this.includeSchemaChanges = includeSchemaChanges;
+ this.readableMetadataList = readableMetadataList;
}
@Override
@@ -118,7 +131,19 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema {
@Override
protected Map<String, String> getMetadata(SourceRecord record) {
- return Collections.emptyMap();
+ Map<String, String> metadataMap = new HashMap<>();
+ readableMetadataList.forEach(
+ (mySqlReadableMetadata -> {
+ Object metadata = mySqlReadableMetadata.getConverter().read(record);
+ if (mySqlReadableMetadata.equals(MySqlReadableMetadata.OP_TS)) {
+ metadataMap.put(
+ mySqlReadableMetadata.getKey(),
+ String.valueOf(((TimestampData) metadata).getMillisecond()));
+ } else {
+ metadataMap.put(mySqlReadableMetadata.getKey(), String.valueOf(metadata));
+ }
+ }));
+ return metadataMap;
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/OpTsMetadataColumn.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/OpTsMetadataColumn.java
new file mode 100644
index 000000000..5a2393f2d
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/OpTsMetadataColumn.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source;
+
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import java.util.Map;
+
+/** A {@link SupportedMetadataColumn} for op_ts. */
+public class OpTsMetadataColumn implements SupportedMetadataColumn {
+
+ @Override
+ public String getName() {
+ return "op_ts";
+ }
+
+ @Override
+ public DataType getType() {
+ return DataTypes.BIGINT().notNull();
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
+
+ @Override
+ public Object read(Map<String, String> metadata) {
+ if (metadata.containsKey(getName())) {
+ return Long.parseLong(metadata.get(getName()));
+ }
+ throw new IllegalArgumentException("op_ts doesn't exist in the metadata: " + metadata);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
index f2b4ccb82..276a2f12d 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.connectors.mysql.source;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
@@ -28,8 +29,11 @@ import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
+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.factories.Factory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.source.FlinkSourceProvider;
@@ -64,8 +68,10 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -258,6 +264,158 @@ public class MySqlPipelineITCase extends MySqlSourceTestBase {
return result;
}
+ @Test
+ public void testInitialStartupModeWithOpTs() throws Exception {
+ inventoryDatabase.createAndInitialize();
+ Configuration sourceConfiguration = new Configuration();
+ sourceConfiguration.set(MySqlDataSourceOptions.HOSTNAME, MYSQL8_CONTAINER.getHost());
+ sourceConfiguration.set(MySqlDataSourceOptions.PORT, MYSQL8_CONTAINER.getDatabasePort());
+ sourceConfiguration.set(MySqlDataSourceOptions.USERNAME, TEST_USER);
+ sourceConfiguration.set(MySqlDataSourceOptions.PASSWORD, TEST_PASSWORD);
+ sourceConfiguration.set(
+ MySqlDataSourceOptions.TABLES, inventoryDatabase.getDatabaseName() + ".products");
+ sourceConfiguration.set(
+ MySqlDataSourceOptions.SERVER_ID, getServerId(env.getParallelism()));
+ sourceConfiguration.set(MySqlDataSourceOptions.SERVER_TIME_ZONE, "UTC");
+ sourceConfiguration.set(MySqlDataSourceOptions.METADATA_LIST, "op_ts");
+ Factory.Context context =
+ new FactoryHelper.DefaultContext(
+ sourceConfiguration, new Configuration(), this.getClass().getClassLoader());
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider)
+ new MySqlDataSourceFactory()
+ .createDataSource(context)
+ .getEventSourceProvider();
+ CloseableIterator<Event> events =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ MySqlDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo())
+ .executeAndCollect();
+ Thread.sleep(10_000);
+ TableId tableId = TableId.tableId(inventoryDatabase.getDatabaseName(), "products");
+ CreateTableEvent createTableEvent = getProductsCreateTableEvent(tableId);
+ // generate snapshot data
+ Map<String, String> meta = new HashMap<>();
+ meta.put("op_ts", "0");
+ List<Event> expectedSnapshot =
+ getSnapshotExpected(tableId).stream()
+ .map(
+ event -> {
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+ return DataChangeEvent.insertEvent(
+ dataChangeEvent.tableId(),
+ dataChangeEvent.after(),
+ meta);
+ })
+ .collect(Collectors.toList());
+ String startTime = String.valueOf(System.currentTimeMillis());
+ Thread.sleep(1000);
+ List<Event> expectedBinlog = new ArrayList<>();
+ try (Connection connection = inventoryDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ expectedBinlog.addAll(executeAlterAndProvideExpected(tableId, statement));
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(255).notNull(),
+ DataTypes.FLOAT(),
+ DataTypes.VARCHAR(45),
+ DataTypes.VARCHAR(55)
+ },
+ new String[] {"id", "name", "weight", "col1", "col2"});
+ BinaryRecordDataGenerator generator = new BinaryRecordDataGenerator(rowType);
+ // insert more data
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`products` VALUES (default,'scooter',5.5,'c-10','c-20');",
+ inventoryDatabase.getDatabaseName())); // 110
+ expectedBinlog.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 110,
+ BinaryStringData.fromString("scooter"),
+ 5.5f,
+ BinaryStringData.fromString("c-10"),
+ BinaryStringData.fromString("c-20")
+ })));
+ statement.execute(
+ String.format(
+ "INSERT INTO `%s`.`products` VALUES (default,'football',6.6,'c-11','c-21');",
+ inventoryDatabase.getDatabaseName())); // 111
+ expectedBinlog.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 111,
+ BinaryStringData.fromString("football"),
+ 6.6f,
+ BinaryStringData.fromString("c-11"),
+ BinaryStringData.fromString("c-21")
+ })));
+ statement.execute(
+ String.format(
+ "UPDATE `%s`.`products` SET `col1`='c-12', `col2`='c-22' WHERE id=110;",
+ inventoryDatabase.getDatabaseName()));
+ expectedBinlog.add(
+ DataChangeEvent.updateEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 110,
+ BinaryStringData.fromString("scooter"),
+ 5.5f,
+ BinaryStringData.fromString("c-10"),
+ BinaryStringData.fromString("c-20")
+ }),
+ generator.generate(
+ new Object[] {
+ 110,
+ BinaryStringData.fromString("scooter"),
+ 5.5f,
+ BinaryStringData.fromString("c-12"),
+ BinaryStringData.fromString("c-22")
+ })));
+ statement.execute(
+ String.format(
+ "DELETE FROM `%s`.`products` WHERE `id` = 111;",
+ inventoryDatabase.getDatabaseName()));
+ expectedBinlog.add(
+ DataChangeEvent.deleteEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 111,
+ BinaryStringData.fromString("football"),
+ 6.6f,
+ BinaryStringData.fromString("c-11"),
+ BinaryStringData.fromString("c-21")
+ })));
+ }
+ List<Event> actual =
+ fetchResults(events, 1 + expectedSnapshot.size() + expectedBinlog.size());
+ assertThat(actual.get(0)).isEqualTo(createTableEvent);
+ assertThat(actual.subList(1, 10))
+ .containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0]));
+ for (int i = 0; i < expectedBinlog.size(); i++) {
+ if (expectedBinlog.get(i) instanceof SchemaChangeEvent) {
+ assertThat(expectedBinlog.get(i)).isEqualTo(actual.get(10 + i));
+ } else {
+ DataChangeEvent expectedEvent = (DataChangeEvent) expectedBinlog.get(i);
+ DataChangeEvent actualEvent = (DataChangeEvent) actual.get(10 + i);
+ assertThat(actualEvent.op()).isEqualTo(expectedEvent.op());
+ assertThat(actualEvent.before()).isEqualTo(expectedEvent.before());
+ assertThat(actualEvent.after()).isEqualTo(expectedEvent.after());
+ assertThat(actualEvent.meta().get("op_ts")).isGreaterThanOrEqualTo(startTime);
+ }
+ }
+ }
+
@Test
public void testParseAlterStatement() throws Exception {
env.setParallelism(1);
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/OpTsMetadataColumn.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/OpTsMetadataColumn.java
new file mode 100644
index 000000000..f6f1f437a
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/OpTsMetadataColumn.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.values.source;
+
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import java.util.Map;
+
+/** A {@link SupportedMetadataColumn} for op_ts. */
+public class OpTsMetadataColumn implements SupportedMetadataColumn {
+
+ @Override
+ public String getName() {
+ return "op_ts";
+ }
+
+ @Override
+ public DataType getType() {
+ return DataTypes.BIGINT();
+ }
+
+ @Override
+ public Class<?> getJavaClass() {
+ return Long.class;
+ }
+
+ @Override
+ public Object read(Map<String, String> metadata) {
+ if (metadata.containsKey(getName())) {
+ return Long.parseLong(metadata.get(getName()));
+ }
+ return null;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSource.java
index 61699a5d0..60ed244a5 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSource.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSource.java
@@ -34,6 +34,7 @@ import org.apache.flink.cdc.common.source.DataSource;
import org.apache.flink.cdc.common.source.EventSourceProvider;
import org.apache.flink.cdc.common.source.FlinkSourceProvider;
import org.apache.flink.cdc.common.source.MetadataAccessor;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.connectors.values.ValuesDatabase;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.core.io.SimpleVersionedSerializer;
@@ -80,6 +81,11 @@ public class ValuesDataSource implements DataSource {
return new ValuesDatabase.ValuesMetadataAccessor();
}
+ @Override
+ public SupportedMetadataColumn[] supportedMetadataColumns() {
+ return new SupportedMetadataColumn[] {new OpTsMetadataColumn()};
+ }
+
/**
* A flink {@link Source} implementation for end to end tests, splits are created by {@link
* ValuesDataSourceHelper}.
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
index 25999c1bf..af9c1890b 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java
@@ -550,7 +550,12 @@ public class ValuesDataSourceHelper {
new Object[] {
BinaryStringData.fromString("1"),
BinaryStringData.fromString("1")
- }));
+ }),
+ new HashMap<String, String>() {
+ {
+ put("op_ts", "1");
+ }
+ });
split1.add(insertEvent1);
DataChangeEvent insertEvent2 =
DataChangeEvent.insertEvent(
@@ -559,7 +564,12 @@ public class ValuesDataSourceHelper {
new Object[] {
BinaryStringData.fromString("2"),
BinaryStringData.fromString("2")
- }));
+ }),
+ new HashMap<String, String>() {
+ {
+ put("op_ts", "2");
+ }
+ });
split1.add(insertEvent2);
DataChangeEvent insertEvent3 =
DataChangeEvent.insertEvent(
@@ -568,7 +578,12 @@ public class ValuesDataSourceHelper {
new Object[] {
BinaryStringData.fromString("3"),
BinaryStringData.fromString("3")
- }));
+ }),
+ new HashMap<String, String>() {
+ {
+ put("op_ts", "3");
+ }
+ });
split1.add(insertEvent3);
// add column
@@ -599,7 +614,12 @@ public class ValuesDataSourceHelper {
new Object[] {
BinaryStringData.fromString("1"),
BinaryStringData.fromString("1")
- })));
+ }),
+ new HashMap<String, String>() {
+ {
+ put("op_ts", "4");
+ }
+ }));
// update
split1.add(
@@ -614,7 +634,12 @@ public class ValuesDataSourceHelper {
new Object[] {
BinaryStringData.fromString("2"),
BinaryStringData.fromString("x")
- })));
+ }),
+ new HashMap<String, String>() {
+ {
+ put("op_ts", "5");
+ }
+ }));
eventOfSplits.add(split1);
return eventOfSplits;
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
index 57cbf238f..c125de59b 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
@@ -49,6 +49,7 @@ import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeoutException;
+import java.util.regex.Pattern;
import java.util.stream.Stream;
/** E2e tests for the {@link PreTransformOperator} and {@link PostTransformOperator}. */
@@ -442,13 +443,14 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
+ " tables: %s.\\.*\n"
+ " server-id: 5400-5404\n"
+ " server-time-zone: UTC\n"
+ + " metadata.list: op_ts\n"
+ "sink:\n"
+ " type: values\n"
+ "transform:\n"
+ " - source-table: %s.TABLEALPHA\n"
- + " projection: \\*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type\n"
+ + " projection: \\*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type, op_ts AS opts\n"
+ " - source-table: %s.TABLEBETA\n"
- + " projection: \\*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type\n"
+ + " projection: \\*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type, op_ts AS opts\n"
+ "pipeline:\n"
+ " parallelism: %d",
INTER_CONTAINER_MYSQL_ALIAS,
@@ -467,25 +469,25 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
waitUntilSpecificEvent(
String.format(
- "CreateTableEvent{tableId=%s.TABLEALPHA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`PRICEALPHA` INT,`AGEALPHA` INT,`NAMEALPHA` VARCHAR(128),`identifier_name` STRING,`type` STRING NOT NULL}, primaryKeys=ID, options=()}",
+ "CreateTableEvent{tableId=%s.TABLEALPHA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`PRICEALPHA` INT,`AGEALPHA` INT,`NAMEALPHA` VARCHAR(128),`identifier_name` STRING,`type` STRING NOT NULL,`opts` BIGINT NOT NULL}, primaryKeys=ID, options=()}",
transformTestDatabase.getDatabaseName()),
60000L);
waitUntilSpecificEvent(
String.format(
- "CreateTableEvent{tableId=%s.TABLEBETA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`CODENAMESBETA` VARCHAR(17),`AGEBETA` INT,`NAMEBETA` VARCHAR(128),`identifier_name` STRING,`type` STRING NOT NULL}, primaryKeys=ID, options=()}",
+ "CreateTableEvent{tableId=%s.TABLEBETA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`CODENAMESBETA` VARCHAR(17),`AGEBETA` INT,`NAMEBETA` VARCHAR(128),`identifier_name` STRING,`type` STRING NOT NULL,`opts` BIGINT NOT NULL}, primaryKeys=ID, options=()}",
transformTestDatabase.getDatabaseName()),
60000L);
validateEvents(
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1008, 8, 199, 17, Alice, null.%s.TABLEALPHA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1010, 10, 99, 19, Carol, null.%s.TABLEALPHA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1009, 8.1, 0, 18, Bob, null.%s.TABLEALPHA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011, 11, 59, 20, Dave, null.%s.TABLEALPHA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2012, 12, Monterey, 22, Fred, null.%s.TABLEBETA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2011, 11, Big Sur, 21, Eva, null.%s.TABLEBETA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2014, 14, Sonoma, 24, Henry, null.%s.TABLEBETA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2013, 13, Ventura, 23, Gus, null.%s.TABLEBETA, +I], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1008, 8, 199, 17, Alice, null.%s.TABLEALPHA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1010, 10, 99, 19, Carol, null.%s.TABLEALPHA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1009, 8.1, 0, 18, Bob, null.%s.TABLEALPHA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011, 11, 59, 20, Dave, null.%s.TABLEALPHA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2012, 12, Monterey, 22, Fred, null.%s.TABLEBETA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2011, 11, Big Sur, 21, Eva, null.%s.TABLEBETA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2014, 14, Sonoma, 24, Henry, null.%s.TABLEBETA, +I, 0], op=INSERT, meta=({op_ts=0})}",
+ "DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2013, 13, Ventura, 23, Gus, null.%s.TABLEBETA, +I, 0], op=INSERT, meta=({op_ts=0})}");
// generate binlogs
String mysqlJdbcUrl =
@@ -496,10 +498,10 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName());
insertBinlogEvents(mysqlJdbcUrl);
- validateEvents(
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[1009, 8.1, 0, 18, Bob, null.%s.TABLEALPHA, -U], after=[1009, 100, 0, 18, Bob, null.%s.TABLEALPHA, +U], op=UPDATE, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[3007, 7, 79, 16, IINA, null.%s.TABLEALPHA, +I], op=INSERT, meta=()}",
- "DataChangeEvent{tableId=%s.TABLEBETA, before=[2011, 11, Big Sur, 21, Eva, null.%s.TABLEBETA, -D], after=[], op=DELETE, meta=()}");
+ validateEventsWithPattern(
+ "DataChangeEvent\\{tableId=%s.TABLEALPHA, before=\\[1009, 8.1, 0, 18, Bob, null.%s.TABLEALPHA, -U, \\d+\\], after=\\[1009, 100, 0, 18, Bob, null.%s.TABLEALPHA, \\+U, \\d+\\], op=UPDATE, meta=\\(\\{op_ts=\\d+\\}\\)\\}",
+ "DataChangeEvent\\{tableId=%s.TABLEALPHA, before=\\[\\], after=\\[3007, 7, 79, 16, IINA, null.%s.TABLEALPHA, \\+I, \\d+\\], op=INSERT, meta=\\(\\{op_ts=\\d+\\}\\)\\}",
+ "DataChangeEvent\\{tableId=%s.TABLEBETA, before=\\[2011, 11, Big Sur, 21, Eva, null.%s.TABLEBETA, -D, \\d+\\], after=\\[\\], op=DELETE, meta=\\(\\{op_ts=\\d+\\}\\)\\}");
}
private static void insertBinlogEvents(String mysqlJdbcUrl) throws SQLException {
@@ -1121,12 +1123,46 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
}
}
+ private void validateEventsWithPattern(String... patterns) throws Exception {
+ for (String pattern : patterns) {
+ waitUntilSpecificEventWithPattern(
+ String.format(
+ pattern,
+ transformTestDatabase.getDatabaseName(),
+ transformTestDatabase.getDatabaseName(),
+ transformTestDatabase.getDatabaseName()),
+ 20000L);
+ }
+ }
+
private void validateResult(List<String> expectedEvents) throws Exception {
for (String event : expectedEvents) {
waitUntilSpecificEvent(event, 6000L);
}
}
+ private void waitUntilSpecificEventWithPattern(String patternStr, long timeout)
+ throws Exception {
+ boolean result = false;
+ long endTimeout = System.currentTimeMillis() + timeout;
+ while (System.currentTimeMillis() < endTimeout) {
+ String stdout = taskManagerConsumer.toUtf8String();
+ Pattern pattern = Pattern.compile(patternStr);
+ if (pattern.matcher(stdout).find()) {
+ result = true;
+ break;
+ }
+ Thread.sleep(1000);
+ }
+ if (!result) {
+ throw new TimeoutException(
+ "failed to get events with pattern: "
+ + patternStr
+ + " from stdout: "
+ + taskManagerConsumer.toUtf8String());
+ }
+ }
+
private void waitUntilSpecificEvent(String event, long timeout) throws Exception {
boolean result = false;
long endTimeout = System.currentTimeMillis() + timeout;
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
index 68f4526f0..56e243d31 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
@@ -31,6 +31,7 @@ import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
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.udf.UserDefinedFunctionContext;
import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverter;
@@ -104,7 +105,8 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
String primaryKey,
String partitionKey,
String tableOptions,
- String postTransformConverter) {
+ String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
transformRules.add(
new TransformRule(
tableInclusions,
@@ -113,14 +115,23 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
primaryKey,
partitionKey,
tableOptions,
- postTransformConverter));
+ postTransformConverter,
+ supportedMetadataColumns));
return this;
}
public PostTransformOperator.Builder addTransform(
String tableInclusions, @Nullable String projection, @Nullable String filter) {
transformRules.add(
- new TransformRule(tableInclusions, projection, filter, "", "", "", null));
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ "",
+ "",
+ "",
+ null,
+ new SupportedMetadataColumn[0]));
return this;
}
@@ -192,7 +203,8 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
TransformFilter.of(filterExpression, udfDescriptors)
.orElse(null),
PostTransformConverters.of(
- transformRule.getPostTransformConverter()));
+ transformRule.getPostTransformConverter()),
+ transformRule.getSupportedMetadataColumns());
})
.collect(Collectors.toList());
this.transformProjectionProcessorMap = new ConcurrentHashMap<>();
@@ -265,7 +277,8 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
.map(TransformProjection::getProjection)
.orElse(null),
createTableEvent.getSchema().getColumns(),
- udfDescriptors)
+ udfDescriptors,
+ rule.getSupportedMetadataColumns())
.stream())
.map(ProjectionColumn::getColumnName)
.collect(Collectors.toSet());
@@ -351,13 +364,16 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
transformProjection,
timezone,
udfDescriptors,
- getUdfFunctionInstances()));
+ getUdfFunctionInstances(),
+ transform.getSupportedMetadataColumns()));
}
TransformProjectionProcessor postTransformProcessor =
transformProjectionProcessorMap.get(
Tuple2.of(tableId, transformProjection));
// update the columns of projection and add the column of projection into Schema
- newSchemas.add(postTransformProcessor.processSchemaChangeEvent(schema));
+ newSchemas.add(
+ postTransformProcessor.processSchemaChangeEvent(
+ schema, transform.getSupportedMetadataColumns()));
}
}
}
@@ -401,7 +417,8 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
transformFilter,
timezone,
udfDescriptors,
- getUdfFunctionInstances()));
+ getUdfFunctionInstances(),
+ transform.getSupportedMetadataColumns()));
}
TransformFilterProcessor transformFilterProcessor =
transformFilterProcessorMap.get(Tuple2.of(tableId, transformFilter));
@@ -427,7 +444,8 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
transformProjection,
timezone,
udfDescriptors,
- getUdfFunctionInstances()));
+ getUdfFunctionInstances(),
+ transform.getSupportedMetadataColumns()));
}
TransformProjectionProcessor postTransformProcessor =
transformProjectionProcessorMap.get(
@@ -474,17 +492,18 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
throws Exception {
BinaryRecordData before = (BinaryRecordData) dataChangeEvent.before();
BinaryRecordData after = (BinaryRecordData) dataChangeEvent.after();
+ Map<String, String> meta = dataChangeEvent.meta();
// insert and update event only process afterData, delete only process beforeData
if (after != null) {
if (transformFilterProcessor.process(
- after, epochTime, opTypeToRowKind(dataChangeEvent.op(), '+'))) {
+ after, epochTime, opTypeToRowKind(dataChangeEvent.op(), '+'), meta)) {
return Optional.of(dataChangeEvent);
} else {
return Optional.empty();
}
} else if (before != null) {
if (transformFilterProcessor.process(
- before, epochTime, opTypeToRowKind(dataChangeEvent.op(), '-'))) {
+ before, epochTime, opTypeToRowKind(dataChangeEvent.op(), '-'), meta)) {
return Optional.of(dataChangeEvent);
} else {
return Optional.empty();
@@ -502,13 +521,19 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
if (before != null) {
BinaryRecordData projectedBefore =
postTransformProcessor.processData(
- before, epochTime, opTypeToRowKind(dataChangeEvent.op(), '-'));
+ before,
+ epochTime,
+ opTypeToRowKind(dataChangeEvent.op(), '-'),
+ dataChangeEvent.meta());
dataChangeEvent = DataChangeEvent.projectBefore(dataChangeEvent, projectedBefore);
}
if (after != null) {
BinaryRecordData projectedAfter =
postTransformProcessor.processData(
- after, epochTime, opTypeToRowKind(dataChangeEvent.op(), '+'));
+ after,
+ epochTime,
+ opTypeToRowKind(dataChangeEvent.op(), '+'),
+ dataChangeEvent.meta());
dataChangeEvent = DataChangeEvent.projectAfter(dataChangeEvent, projectedAfter);
}
return Optional.of(dataChangeEvent);
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java
index 06045bf42..c448e9903 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.schema.Selectors;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverter;
import javax.annotation.Nullable;
@@ -27,6 +28,7 @@ import java.util.Optional;
/** Post-Transformation rule used by {@link PostTransformOperator}. */
public class PostTransformer {
private final Selectors selectors;
+ private final SupportedMetadataColumn[] supportedMetadataColumns;
private final Optional<TransformProjection> projection;
private final Optional<TransformFilter> filter;
@@ -36,11 +38,13 @@ public class PostTransformer {
Selectors selectors,
@Nullable TransformProjection projection,
@Nullable TransformFilter filter,
- Optional<PostTransformConverter> postTransformConverter) {
+ Optional<PostTransformConverter> postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
this.selectors = selectors;
this.projection = projection != null ? Optional.of(projection) : Optional.empty();
this.filter = filter != null ? Optional.of(filter) : Optional.empty();
this.postTransformConverter = postTransformConverter;
+ this.supportedMetadataColumns = supportedMetadataColumns;
}
public Selectors getSelectors() {
@@ -58,4 +62,8 @@ public class PostTransformer {
public Optional<PostTransformConverter> getPostTransformConverter() {
return postTransformConverter;
}
+
+ public SupportedMetadataColumn[] getSupportedMetadataColumns() {
+ return supportedMetadataColumns;
+ }
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
index 6ee0d72e9..4a3c51583 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
@@ -33,6 +33,7 @@ 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.SchemaUtils;
import org.apache.flink.cdc.runtime.parser.TransformParser;
import org.apache.flink.runtime.state.StateInitializationContext;
@@ -89,7 +90,15 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
public PreTransformOperator.Builder addTransform(
String tableInclusions, @Nullable String projection, @Nullable String filter) {
transformRules.add(
- new TransformRule(tableInclusions, projection, filter, "", "", "", null));
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ "",
+ "",
+ "",
+ null,
+ new SupportedMetadataColumn[0]));
return this;
}
@@ -100,7 +109,8 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
String primaryKey,
String partitionKey,
String tableOption,
- @Nullable String postTransformConverter) {
+ @Nullable String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
transformRules.add(
new TransformRule(
tableInclusions,
@@ -109,7 +119,8 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
primaryKey,
partitionKey,
tableOption,
- postTransformConverter));
+ postTransformConverter,
+ supportedMetadataColumns));
return this;
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
index ee7740d9e..292ba416a 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.parser.JaninoCompiler;
import org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns;
import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
@@ -32,6 +33,8 @@ import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METADATA_COLUMNS;
@@ -47,6 +50,7 @@ public class ProjectionColumnProcessor {
private String timezone;
private TransformExpressionKey transformExpressionKey;
private final List<UserDefinedFunctionDescriptor> udfDescriptors;
+ private final SupportedMetadataColumn[] supportedMetadataColumns;
private final transient List<Object> udfFunctionInstances;
private transient ExpressionEvaluator expressionEvaluator;
@@ -55,11 +59,13 @@ public class ProjectionColumnProcessor {
ProjectionColumn projectionColumn,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- final List<Object> udfFunctionInstances) {
+ final List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
this.tableInfo = tableInfo;
this.projectionColumn = projectionColumn;
this.timezone = timezone;
this.udfDescriptors = udfDescriptors;
+ this.supportedMetadataColumns = supportedMetadataColumns;
this.transformExpressionKey = generateTransformExpressionKey();
this.expressionEvaluator =
TransformExpressionCompiler.compileExpression(
@@ -72,18 +78,25 @@ public class ProjectionColumnProcessor {
ProjectionColumn projectionColumn,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
return new ProjectionColumnProcessor(
- tableInfo, projectionColumn, timezone, udfDescriptors, udfFunctionInstances);
+ tableInfo,
+ projectionColumn,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumns);
}
public ProjectionColumn getProjectionColumn() {
return projectionColumn;
}
- public Object evaluate(BinaryRecordData record, long epochTime, String opType) {
+ public Object evaluate(
+ BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
try {
- return expressionEvaluator.evaluate(generateParams(record, epochTime, opType));
+ return expressionEvaluator.evaluate(generateParams(record, epochTime, opType, meta));
} catch (InvocationTargetException e) {
LOG.error(
"Table:{} column:{} projection:{} execute failed. {}",
@@ -95,7 +108,8 @@ public class ProjectionColumnProcessor {
}
}
- private Object[] generateParams(BinaryRecordData record, long epochTime, String opType) {
+ private Object[] generateParams(
+ BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
List<Object> params = new ArrayList<>();
List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
@@ -119,6 +133,18 @@ public class ProjectionColumnProcessor {
continue;
}
+ boolean foundInMeta = false;
+ for (SupportedMetadataColumn supportedMetadataColumn : supportedMetadataColumns) {
+ if (supportedMetadataColumn.getName().equals(originalColumnName)) {
+ params.add(supportedMetadataColumn.read(meta));
+ foundInMeta = true;
+ break;
+ }
+ }
+ if (foundInMeta) {
+ continue;
+ }
+
boolean argumentFound = false;
for (int i = 0; i < columns.size(); i++) {
Column column = columns.get(i);
@@ -171,6 +197,14 @@ public class ProjectionColumnProcessor {
argumentNames.add(col.f0);
paramTypes.add(col.f2);
});
+ Stream.of(supportedMetadataColumns)
+ .filter(col -> col.getName().equals(originalColumnName))
+ .findFirst()
+ .ifPresent(
+ col -> {
+ argumentNames.add(col.getName());
+ paramTypes.add(col.getJavaClass());
+ });
}
argumentNames.add(JaninoCompiler.DEFAULT_TIME_ZONE);
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
index 430813d7f..220f61109 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
@@ -21,6 +21,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.parser.JaninoCompiler;
import org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns;
import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
@@ -31,8 +32,10 @@ import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METADATA_COLUMNS;
@@ -45,16 +48,19 @@ public class TransformFilterProcessor {
private TransformExpressionKey transformExpressionKey;
private final transient List<Object> udfFunctionInstances;
private transient ExpressionEvaluator expressionEvaluator;
+ private final Map<String, SupportedMetadataColumn> supportedMetadataColumns;
public TransformFilterProcessor(
PostTransformChangeInfo tableInfo,
TransformFilter transformFilter,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
this.tableInfo = tableInfo;
this.transformFilter = transformFilter;
this.timezone = timezone;
+ this.supportedMetadataColumns = supportedMetadataColumns;
this.transformExpressionKey = generateTransformExpressionKey();
this.udfFunctionInstances = udfFunctionInstances;
this.expressionEvaluator =
@@ -67,15 +73,27 @@ public class TransformFilterProcessor {
TransformFilter transformFilter,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ Map<String, SupportedMetadataColumn> supportedMetadataColumnsMap = new HashMap<>();
+ for (SupportedMetadataColumn supportedMetadataColumn : supportedMetadataColumns) {
+ supportedMetadataColumnsMap.put(
+ supportedMetadataColumn.getName(), supportedMetadataColumn);
+ }
return new TransformFilterProcessor(
- tableInfo, transformFilter, timezone, udfDescriptors, udfFunctionInstances);
+ tableInfo,
+ transformFilter,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumnsMap);
}
- public boolean process(BinaryRecordData record, long epochTime, String opType) {
+ public boolean process(
+ BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
try {
return (Boolean)
- expressionEvaluator.evaluate(generateParams(record, epochTime, opType));
+ expressionEvaluator.evaluate(generateParams(record, epochTime, opType, meta));
} catch (InvocationTargetException e) {
LOG.error(
"Table:{} filter:{} execute failed. {}",
@@ -110,10 +128,21 @@ public class TransformFilterProcessor {
argTypes.add(col.f2);
}
});
+
+ supportedMetadataColumns
+ .keySet()
+ .forEach(
+ colName -> {
+ if (scriptExpression.contains(colName) && !argNames.contains(colName)) {
+ argNames.add(colName);
+ argTypes.add(supportedMetadataColumns.get(colName).getJavaClass());
+ }
+ });
return Tuple2.of(argNames, argTypes);
}
- private Object[] generateParams(BinaryRecordData record, long epochTime, String opType) {
+ private Object[] generateParams(
+ BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
List<Object> params = new ArrayList<>();
List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
@@ -135,6 +164,12 @@ public class TransformFilterProcessor {
params.add(opType);
continue;
}
+
+ if (supportedMetadataColumns.containsKey(columnName)) {
+ params.add(supportedMetadataColumns.get(columnName).read(meta));
+ continue;
+ }
+
for (int i = 0; i < columns.size(); i++) {
Column column = columns.get(i);
if (column.getName().equals(columnName)) {
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
index 45ea35770..abedd9120 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
@@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.runtime.parser.TransformParser;
import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
@@ -30,6 +31,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -59,7 +61,8 @@ public class TransformProjectionProcessor {
TransformProjection transformProjection,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- final List<Object> udfFunctionInstances) {
+ final List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
this.postTransformChangeInfo = postTransformChangeInfo;
this.transformProjection = transformProjection;
this.timezone = timezone;
@@ -68,7 +71,8 @@ public class TransformProjectionProcessor {
// Create cached projection column processors after setting all other fields.
this.cachedProjectionColumnProcessors =
- cacheProjectionColumnProcessors(postTransformChangeInfo, transformProjection);
+ cacheProjectionColumnProcessors(
+ postTransformChangeInfo, transformProjection, supportedMetadataColumns);
}
public boolean hasTableInfo() {
@@ -80,32 +84,54 @@ public class TransformProjectionProcessor {
TransformProjection transformProjection,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
return new TransformProjectionProcessor(
- tableInfo, transformProjection, timezone, udfDescriptors, udfFunctionInstances);
+ tableInfo,
+ transformProjection,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumns);
}
public static TransformProjectionProcessor of(
TransformProjection transformProjection,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
return new TransformProjectionProcessor(
- null, transformProjection, timezone, udfDescriptors, udfFunctionInstances);
+ null,
+ transformProjection,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumns);
}
public static TransformProjectionProcessor of(
TransformProjection transformProjection,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances) {
+ List<Object> udfFunctionInstances,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
return new TransformProjectionProcessor(
- null, transformProjection, null, udfDescriptors, udfFunctionInstances);
+ null,
+ transformProjection,
+ null,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumns);
}
- public Schema processSchemaChangeEvent(Schema schema) {
+ public Schema processSchemaChangeEvent(
+ Schema schema, SupportedMetadataColumn[] supportedMetadataColumns) {
List<ProjectionColumn> projectionColumns =
TransformParser.generateProjectionColumns(
- transformProjection.getProjection(), schema.getColumns(), udfDescriptors);
+ transformProjection.getProjection(),
+ schema.getColumns(),
+ udfDescriptors,
+ supportedMetadataColumns);
transformProjection.setProjectionColumns(projectionColumns);
return schema.copy(
projectionColumns.stream()
@@ -113,7 +139,8 @@ public class TransformProjectionProcessor {
.collect(Collectors.toList()));
}
- public BinaryRecordData processData(BinaryRecordData payload, long epochTime, String opType) {
+ public BinaryRecordData processData(
+ BinaryRecordData payload, long epochTime, String opType, Map<String, String> meta) {
List<Object> valueList = new ArrayList<>();
List<Column> columns = postTransformChangeInfo.getPostTransformedSchema().getColumns();
@@ -124,7 +151,8 @@ public class TransformProjectionProcessor {
ProjectionColumn projectionColumn = projectionColumnProcessor.getProjectionColumn();
valueList.add(
DataTypeConverter.convert(
- projectionColumnProcessor.evaluate(payload, epochTime, opType),
+ projectionColumnProcessor.evaluate(
+ payload, epochTime, opType, meta),
projectionColumn.getDataType()));
} else {
Column column = columns.get(i);
@@ -159,7 +187,9 @@ public class TransformProjectionProcessor {
}
private List<ProjectionColumnProcessor> cacheProjectionColumnProcessors(
- PostTransformChangeInfo tableInfo, TransformProjection transformProjection) {
+ PostTransformChangeInfo tableInfo,
+ TransformProjection transformProjection,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
List<ProjectionColumnProcessor> cachedProjectionColumnProcessors = new ArrayList<>();
if (!hasTableInfo()) {
return cachedProjectionColumnProcessors;
@@ -184,7 +214,8 @@ public class TransformProjectionProcessor {
col,
timezone,
udfDescriptors,
- udfFunctionInstances))
+ udfFunctionInstances,
+ supportedMetadataColumns))
.orElse(null));
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java
index 7bdc5cffc..31a245257 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java
@@ -17,6 +17,8 @@
package org.apache.flink.cdc.runtime.operators.transform;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+
import javax.annotation.Nullable;
import java.io.Serializable;
@@ -34,6 +36,7 @@ public class TransformRule implements Serializable {
private final String partitionKey;
private final String tableOption;
private final @Nullable String postTransformConverter;
+ private final SupportedMetadataColumn[] supportedMetadataColumns;
public TransformRule(
String tableInclusions,
@@ -42,7 +45,8 @@ public class TransformRule implements Serializable {
String primaryKey,
String partitionKey,
String tableOption,
- @Nullable String postTransformConverter) {
+ @Nullable String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
this.tableInclusions = tableInclusions;
this.projection = projection;
this.filter = normalizeFilter(projection, filter);
@@ -50,6 +54,7 @@ public class TransformRule implements Serializable {
this.partitionKey = partitionKey;
this.tableOption = tableOption;
this.postTransformConverter = postTransformConverter;
+ this.supportedMetadataColumns = supportedMetadataColumns;
}
public String getTableInclusions() {
@@ -82,4 +87,8 @@ public class TransformRule implements Serializable {
public String getPostTransformConverter() {
return postTransformConverter;
}
+
+ public SupportedMetadataColumn[] getSupportedMetadataColumns() {
+ return supportedMetadataColumns;
+ }
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
index 16416daff..4dfcd09a5 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.runtime.parser;
import org.apache.flink.api.common.io.ParseException;
import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
@@ -83,6 +84,7 @@ import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import java.util.stream.Stream;
import static org.apache.flink.cdc.common.utils.StringUtils.isNullOrWhitespaceOnly;
import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METADATA_COLUMNS;
@@ -106,8 +108,10 @@ public class TransformParser {
private static RelNode sqlToRel(
List<Column> columns,
SqlNode sqlNode,
- List<UserDefinedFunctionDescriptor> udfDescriptors) {
- List<Column> columnsWithMetadata = copyFillMetadataColumn(columns);
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ List<Column> columnsWithMetadata =
+ copyFillMetadataColumn(columns, supportedMetadataColumns);
CalciteSchema rootSchema = CalciteSchema.createRootSchema(true);
SchemaPlus schema = rootSchema.plus();
Map<String, Object> operand = new HashMap<>();
@@ -265,7 +269,8 @@ public class TransformParser {
public static List<ProjectionColumn> generateProjectionColumns(
String projectionExpression,
List<Column> columns,
- List<UserDefinedFunctionDescriptor> udfDescriptors) {
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
if (isNullOrWhitespaceOnly(projectionExpression)) {
return new ArrayList<>();
}
@@ -273,8 +278,9 @@ public class TransformParser {
if (sqlSelect.getSelectList().isEmpty()) {
return new ArrayList<>();
}
+
expandWildcard(sqlSelect, columns);
- RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors);
+ RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors, supportedMetadataColumns);
Map<String, RelDataType> relDataTypeMap =
relNode.getRowType().getFieldList().stream()
.collect(
@@ -305,7 +311,7 @@ public class TransformParser {
String columnName = aliasNode.names.get(aliasNode.names.size() - 1);
Preconditions.checkArgument(
- !isMetadataColumn(columnName),
+ !isMetadataColumn(columnName, supportedMetadataColumns),
"Column name %s is reserved and shading it is not allowed.",
columnName);
@@ -320,7 +326,11 @@ public class TransformParser {
identifierExprNode.names.get(identifierExprNode.names.size() - 1);
projectionColumn =
resolveProjectionColumnFromIdentifier(
- relDataTypeMap, originalColumnMap, originalName, columnName);
+ relDataTypeMap,
+ originalColumnMap,
+ originalName,
+ columnName,
+ supportedMetadataColumns);
} else {
projectionColumn =
ProjectionColumn.ofCalculated(
@@ -339,7 +349,11 @@ public class TransformParser {
String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
projectionColumn =
resolveProjectionColumnFromIdentifier(
- relDataTypeMap, originalColumnMap, columnName, columnName);
+ relDataTypeMap,
+ originalColumnMap,
+ columnName,
+ columnName,
+ supportedMetadataColumns);
} else {
throw new ParseException("Unrecognized projection: " + sqlNode.toString());
}
@@ -366,8 +380,9 @@ public class TransformParser {
Map<String, RelDataType> relDataTypeMap,
Map<String, Column> originalColumnMap,
String identifier,
- String projectedColumnName) {
- if (isMetadataColumn(identifier)) {
+ String projectedColumnName,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ if (isMetadataColumn(identifier, supportedMetadataColumns)) {
// For a metadata column, we simply generate a projection column with the same
return ProjectionColumn.ofCalculated(
projectedColumnName,
@@ -406,7 +421,8 @@ public class TransformParser {
return JaninoCompiler.translateSqlNodeToJaninoExpression(where, udfDescriptors);
}
- public static List<String> parseComputedColumnNames(String projection) {
+ public static List<String> parseComputedColumnNames(
+ String projection, SupportedMetadataColumn[] supportedMetadataColumns) {
List<String> columnNames = new ArrayList<>();
if (isNullOrWhitespaceOnly(projection)) {
return columnNames;
@@ -436,7 +452,8 @@ public class TransformParser {
}
} else if (sqlNode instanceof SqlIdentifier) {
String columnName = sqlNode.toString();
- if (isMetadataColumn(columnName) && !columnNames.contains(columnName)) {
+ if (isMetadataColumn(columnName, supportedMetadataColumns)
+ && !columnNames.contains(columnName)) {
columnNames.add(columnName);
}
} else {
@@ -499,17 +516,24 @@ public class TransformParser {
return parseSelect(statement.toString());
}
- private static List<Column> copyFillMetadataColumn(List<Column> columns) {
+ private static List<Column> copyFillMetadataColumn(
+ List<Column> columns, SupportedMetadataColumn[] supportedMetadataColumns) {
// Add metaColumn for SQLValidator.validate
List<Column> columnsWithMetadata = new ArrayList<>(columns);
METADATA_COLUMNS.stream()
.map(col -> Column.physicalColumn(col.f0, col.f1))
.forEach(columnsWithMetadata::add);
+ Stream.of(supportedMetadataColumns)
+ .map(sCol -> Column.physicalColumn(sCol.getName(), sCol.getType()))
+ .forEach(columnsWithMetadata::add);
return columnsWithMetadata;
}
- private static boolean isMetadataColumn(String columnName) {
- return METADATA_COLUMNS.stream().anyMatch(col -> col.f0.equals(columnName));
+ private static boolean isMetadataColumn(
+ String columnName, SupportedMetadataColumn[] supportedMetadataColumns) {
+ return METADATA_COLUMNS.stream().anyMatch(col -> col.f0.equals(columnName))
+ || Stream.of(supportedMetadataColumns)
+ .anyMatch(col -> col.getName().equals(columnName));
}
public static SqlSelect parseFilterExpression(String filterExpression) {
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
index d266ed201..2c9baa84f 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.RowType;
import org.apache.flink.cdc.runtime.testutils.operators.EventOperatorTestHarness;
@@ -435,7 +436,14 @@ public class PostTransformOperatorTest {
PostTransformOperator transform =
PostTransformOperator.newBuilder()
.addTransform(
- DATATYPE_TABLEID.identifier(), "*", null, null, null, null, null)
+ DATATYPE_TABLEID.identifier(),
+ "*",
+ null,
+ null,
+ null,
+ null,
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PostTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -583,7 +591,8 @@ public class PostTransformOperatorTest {
"",
"",
"",
- "SOFT_DELETE")
+ "SOFT_DELETE",
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PostTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
index 6f7ac12e7..0ab43bc1c 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.cdc.common.event.Event;
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.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.RowType;
import org.apache.flink.cdc.runtime.testutils.operators.EventOperatorTestHarness;
@@ -192,7 +193,8 @@ public class PreTransformOperatorTest {
"col2",
"col12",
"key1=value1,key2=value2",
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -303,7 +305,8 @@ public class PreTransformOperatorTest {
"id",
"id",
"key1=value1,key2=value2",
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -334,7 +337,8 @@ public class PreTransformOperatorTest {
"id",
"id",
"key1=value1,key2=value2",
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -442,7 +446,8 @@ public class PreTransformOperatorTest {
"id",
"id",
"key1=value1,key2=value2",
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -559,7 +564,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.addTransform(
CUSTOMERS_TABLEID.identifier(),
"id, name as roleName",
@@ -567,7 +573,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -598,7 +605,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.addTransform(
CUSTOMERS_TABLEID.identifier(),
"id, age, name, sex, 'Juvenile' as roleName",
@@ -606,7 +614,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
@@ -636,7 +645,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.addTransform(
CUSTOMERS_TABLEID.identifier(),
"id, age, UPPER(name) as name, sex",
@@ -644,7 +654,8 @@ public class PreTransformOperatorTest {
"id",
null,
null,
- null)
+ null,
+ new SupportedMetadataColumn[0])
.build();
EventOperatorTestHarness<PreTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 0fd97c1b2..bc5d12005 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.runtime.parser;
import org.apache.flink.api.common.io.ParseException;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
import org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
@@ -183,7 +184,8 @@ public class TransformParserTest {
@Test
public void testParseComputedColumnNames() {
List<String> computedColumnNames =
- TransformParser.parseComputedColumnNames("CONCAT(id, order_id) as uniq_id, *");
+ TransformParser.parseComputedColumnNames(
+ "CONCAT(id, order_id) as uniq_id, *", new SupportedMetadataColumn[0]);
Assertions.assertThat(computedColumnNames.toArray()).isEqualTo(new String[] {"uniq_id"});
}
@@ -324,7 +326,8 @@ public class TransformParserTest {
TransformParser.generateProjectionColumns(
"id, upper(name) as name, age + 1 as newage, weight / (height * height) as bmi",
testColumns,
- Collections.emptyList());
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
List<String> expected =
Arrays.asList(
@@ -338,7 +341,8 @@ public class TransformParserTest {
TransformParser.generateProjectionColumns(
"*, __namespace_name__, __schema_name__, __table_name__",
testColumns,
- Collections.emptyList());
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
List<String> metadataExpected =
Arrays.asList(
@@ -359,7 +363,10 @@ public class TransformParserTest {
Assertions.assertThatThrownBy(
() ->
TransformParser.generateProjectionColumns(
- "id, 1 + 1", testColumns, Collections.emptyList()))
+ "id, 1 + 1",
+ testColumns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]))
.isExactlyInstanceOf(IllegalArgumentException.class)
.hasMessage(
"Unrecognized projection expression: 1 + 1. Should be <EXPR> AS <IDENTIFIER>");