Rename __op_type__ to more generic __data_event_type__

yuxiqian · yuxiqian · commit cc14a3eef9d8 · 2024-07-16T19:08:36.000+08:00
diff --git a/docs/content.zh/docs/core-concept/transform.md b/docs/content.zh/docs/core-concept/transform.md
@@ -47,12 +47,12 @@ Multiple rules can be declared in one single pipeline YAML file.
 ## Fields definition
 There are some hidden columns used to access metadata information. They will only take effect when explicitly referenced in the transform rules.
 
-| Field              | Data Type | Description                                  |
-|--------------------|-----------|----------------------------------------------|
-| __namespace_name__ | String    | Name of the namespace that contains the row. |
-| __schema_name__    | String    | Name of the schema that contains the row.    |
-| __table_name__     | String    | Name of the table that contains the row.     |
-| __row_kind__       | String    | Operation type of data change event.         |
+| Field               | Data Type | Description                                  |
+|---------------------|-----------|----------------------------------------------|
+| __namespace_name__  | String    | Name of the namespace that contains the row. |
+| __schema_name__     | String    | Name of the schema that contains the row.    |
+| __table_name__      | String    | Name of the table that contains the row.     |
+| __data_event_type__ | String    | Operation type of data change event.         |
 
 ## Metadata relationship
 
diff --git a/docs/content/docs/core-concept/transform.md b/docs/content/docs/core-concept/transform.md
@@ -47,12 +47,12 @@ Multiple rules can be declared in one single pipeline YAML file.
 ## Fields definition
 There are some hidden columns used to access metadata information. They will only take effect when explicitly referenced in the transform rules.
 
-| Field              | Data Type | Description                                  |
-|--------------------|-----------|----------------------------------------------|
-| __namespace_name__ | String    | Name of the namespace that contains the row. |
-| __schema_name__    | String    | Name of the schema that contains the row.    |
-| __table_name__     | String    | Name of the table that contains the row.     |
-| __row_kind__       | String    | Operation type of data change event.         |
+| Field               | Data Type | Description                                  |
+|---------------------|-----------|----------------------------------------------|
+| __namespace_name__  | String    | Name of the namespace that contains the row. |
+| __schema_name__     | String    | Name of the schema that contains the row.    |
+| __table_name__      | String    | Name of the table that contains the row.     |
+| __data_event_type__ | String    | Operation type of data change event.         |
 
 ## Metadata relationship
 
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
@@ -358,7 +358,7 @@ void testOpTypeMetadataColumn(ValuesDataSink.SinkApi sinkApi) throws Exception {
         TransformDef transformDef =
                 new TransformDef(
                         "default_namespace.default_schema.table1",
-                        "*,concat(col1,'0') as col12,__row_kind__ as rk",
+                        "*,concat(col1,'0') as col12,__data_event_type__ as rk",
                         "col1 <> '3'",
                         "col1",
                         "col12",
@@ -384,7 +384,7 @@ void testOpTypeMetadataColumn(ValuesDataSink.SinkApi sinkApi) throws Exception {
         String[] outputEvents = outCaptor.toString().trim().split("\n");
         assertThat(outputEvents)
                 .containsExactly(
-                        "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`__row_kind__` STRING NOT NULL,`col12` STRING,`rk` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
+                        "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`__data_event_type__` STRING NOT NULL,`col12` STRING,`rk` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
                         "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[1, 1, +I, 10, +I], op=INSERT, meta=()}",
                         "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2, 2, +I, 20, +I], op=INSERT, meta=()}",
                         "AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
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
@@ -90,7 +90,7 @@ private Object[] generateParams(BinaryRecordData data, long epochTime, String op
                 params.add(tableInfo.getTableName());
                 continue;
             }
-            if (originalColumnName.equals(TransformParser.DEFAULT_ROW_KIND)) {
+            if (originalColumnName.equals(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {
                 params.add(opType);
                 continue;
             }
@@ -143,9 +143,9 @@ private TransformExpressionKey generateTransformExpressionKey() {
             paramTypes.add(String.class);
         }
 
-        if (scriptExpression.contains(TransformParser.DEFAULT_ROW_KIND)
-                && !argumentNames.contains(TransformParser.DEFAULT_ROW_KIND)) {
-            argumentNames.add(TransformParser.DEFAULT_ROW_KIND);
+        if (scriptExpression.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)
+                && !argumentNames.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {
+            argumentNames.add(TransformParser.DEFAULT_DATA_EVENT_TYPE);
             paramTypes.add(String.class);
         }
 
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
@@ -85,7 +85,7 @@ private Object[] generateParams(BinaryRecordData data, long epochTime, String op
                 params.add(tableInfo.getTableName());
                 continue;
             }
-            if (columnName.equals(TransformParser.DEFAULT_ROW_KIND)) {
+            if (columnName.equals(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {
                 params.add(opType);
                 continue;
             }
@@ -138,9 +138,9 @@ private TransformExpressionKey generateTransformExpressionKey() {
             paramTypes.add(String.class);
         }
 
-        if (scriptExpression.contains(TransformParser.DEFAULT_ROW_KIND)
-                && !argumentNames.contains(TransformParser.DEFAULT_ROW_KIND)) {
-            argumentNames.add(TransformParser.DEFAULT_ROW_KIND);
+        if (scriptExpression.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)
+                && !argumentNames.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {
+            argumentNames.add(TransformParser.DEFAULT_DATA_EVENT_TYPE);
             paramTypes.add(String.class);
         }
 
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
@@ -77,7 +77,7 @@ public class TransformParser {
     public static final String DEFAULT_NAMESPACE_NAME = "__namespace_name__";
     public static final String DEFAULT_SCHEMA_NAME = "__schema_name__";
     public static final String DEFAULT_TABLE_NAME = "__table_name__";
-    public static final String DEFAULT_ROW_KIND = "__row_kind__";
+    public static final String DEFAULT_DATA_EVENT_TYPE = "__data_event_type__";
 
     private static SqlParser getCalciteParser(String sql) {
         return SqlParser.create(
@@ -368,9 +368,10 @@ private static List<Column> copyFillMetadataColumn(
                 && !containsMetadataColumn(columnsWithMetadata, DEFAULT_TABLE_NAME)) {
             columnsWithMetadata.add(Column.physicalColumn(DEFAULT_TABLE_NAME, DataTypes.STRING()));
         }
-        if (transformStatement.contains(DEFAULT_ROW_KIND)
-                && !containsMetadataColumn(columnsWithMetadata, DEFAULT_ROW_KIND)) {
-            columnsWithMetadata.add(Column.physicalColumn(DEFAULT_ROW_KIND, DataTypes.STRING()));
+        if (transformStatement.contains(DEFAULT_DATA_EVENT_TYPE)
+                && !containsMetadataColumn(columnsWithMetadata, DEFAULT_DATA_EVENT_TYPE)) {
+            columnsWithMetadata.add(
+                    Column.physicalColumn(DEFAULT_DATA_EVENT_TYPE, DataTypes.STRING()));
         }
         return columnsWithMetadata;
     }
@@ -383,7 +384,7 @@ private static boolean isMetadataColumn(String columnName) {
         return DEFAULT_TABLE_NAME.equals(columnName)
                 || DEFAULT_SCHEMA_NAME.equals(columnName)
                 || DEFAULT_NAMESPACE_NAME.equals(columnName)
-                || DEFAULT_ROW_KIND.equals(columnName);
+                || DEFAULT_DATA_EVENT_TYPE.equals(columnName);
     }
 
     public static SqlSelect parseFilterExpression(String filterExpression) {

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ private Object[] generateParams(BinaryRecordData data, long epochTime, String op`
`90`	`90`	`params.add(tableInfo.getTableName());`
`91`	`91`	`continue;`
`92`	`92`	`}`
`93`		`- if (originalColumnName.equals(TransformParser.DEFAULT_ROW_KIND)) {`
	`93`	`+ if (originalColumnName.equals(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {`
`94`	`94`	`params.add(opType);`
`95`	`95`	`continue;`
`96`	`96`	`}`
`@@ -143,9 +143,9 @@ private TransformExpressionKey generateTransformExpressionKey() {`
`143`	`143`	`paramTypes.add(String.class);`
`144`	`144`	`}`
`145`	`145`
`146`		`- if (scriptExpression.contains(TransformParser.DEFAULT_ROW_KIND)`
`147`		`- && !argumentNames.contains(TransformParser.DEFAULT_ROW_KIND)) {`
`148`		`- argumentNames.add(TransformParser.DEFAULT_ROW_KIND);`
	`146`	`+ if (scriptExpression.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)`
	`147`	`+ && !argumentNames.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {`
	`148`	`+ argumentNames.add(TransformParser.DEFAULT_DATA_EVENT_TYPE);`
`149`	`149`	`paramTypes.add(String.class);`
`150`	`150`	`}`
`151`	`151`
Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,7 @@ private Object[] generateParams(BinaryRecordData data, long epochTime, String op`
`85`	`85`	`params.add(tableInfo.getTableName());`
`86`	`86`	`continue;`
`87`	`87`	`}`
`88`		`- if (columnName.equals(TransformParser.DEFAULT_ROW_KIND)) {`
	`88`	`+ if (columnName.equals(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {`
`89`	`89`	`params.add(opType);`
`90`	`90`	`continue;`
`91`	`91`	`}`
`@@ -138,9 +138,9 @@ private TransformExpressionKey generateTransformExpressionKey() {`
`138`	`138`	`paramTypes.add(String.class);`
`139`	`139`	`}`
`140`	`140`
`141`		`- if (scriptExpression.contains(TransformParser.DEFAULT_ROW_KIND)`
`142`		`- && !argumentNames.contains(TransformParser.DEFAULT_ROW_KIND)) {`
`143`		`- argumentNames.add(TransformParser.DEFAULT_ROW_KIND);`
	`141`	`+ if (scriptExpression.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)`
	`142`	`+ && !argumentNames.contains(TransformParser.DEFAULT_DATA_EVENT_TYPE)) {`
	`143`	`+ argumentNames.add(TransformParser.DEFAULT_DATA_EVENT_TYPE);`
`144`	`144`	`paramTypes.add(String.class);`
`145`	`145`	`}`
`146`	`146`