You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/12/17 06:46:30 UTC
(flink-cdc) branch master updated: [FLINK-36854][transform] Add missed comment and default value after transform
This is an automated email from the ASF dual-hosted git repository.
leonard 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 39b560e37 [FLINK-36854][transform] Add missed comment and default value after transform
39b560e37 is described below
commit 39b560e3771d953824cee006625903458f7d9be9
Author: yuxiqian <34...@users.noreply.github.com>
AuthorDate: Tue Dec 17 14:46:26 2024 +0800
[FLINK-36854][transform] Add missed comment and default value after transform
This closes #3780.
---
.../flink/FlinkPipelineComposerITCase.java | 2 +-
.../flink/FlinkPipelineComposerLenientITCase.java | 2 +-
.../flink/FlinkPipelineTransformITCase.java | 88 ++++++++++
.../cdc/pipeline/tests/TransformE2eITCase.java | 8 +-
.../operators/transform/PostTransformOperator.java | 2 +-
.../operators/transform/ProjectionColumn.java | 32 +++-
.../flink/cdc/runtime/parser/TransformParser.java | 194 +++++++++++----------
.../transform/PostTransformOperatorTest.java | 2 +-
.../transform/UnifiedTransformOperatorTest.java | 124 ++++++++++++-
.../cdc/runtime/parser/TransformParserTest.java | 25 +--
10 files changed, 360 insertions(+), 119 deletions(-)
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 679ff9cc6..6008848a5 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
@@ -411,7 +411,7 @@ 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}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
+ "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=()}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=AFTER, existedColumnName=col2}]}",
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 df47307d0..9d1fea7c7 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
@@ -453,7 +453,7 @@ class FlinkPipelineComposerLenientITCase {
String[] outputEvents = outCaptor.toString().trim().split(LINE_SEPARATOR);
assertThat(outputEvents)
.containsExactly(
- "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` STRING,`col2` STRING,`col12` STRING,`rk` STRING}, primaryKeys=col1, partitionKeys=col12, options=({key1=value1})}",
+ "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=()}",
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
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 739f78e43..3810f426a 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
@@ -51,6 +51,7 @@ import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
@@ -1515,6 +1516,93 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[12th, 15, Oops], after=[], op=DELETE, meta=()}");
}
+ @Test
+ void testTransformWithCommentsAndDefaultExpr() throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Setup value source
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.CUSTOM_SOURCE_EVENTS);
+
+ TableId myTable1 = TableId.tableId("default_namespace", "default_schema", "mytable1");
+ TableId myTable2 = TableId.tableId("default_namespace", "default_schema", "mytable2");
+ Schema table1Schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn("name", DataTypes.STRING(), "name column", "Jane Doe")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .partitionKey("id, age")
+ .primaryKey("id")
+ .build();
+ Schema table2Schema =
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.BIGINT(), "column for id", "AUTO_DECREMENT()")
+ .physicalColumn(
+ "name", DataTypes.VARCHAR(255), "column for name", "John Smith")
+ .physicalColumn("age", DataTypes.TINYINT(), "column for age", "91")
+ .physicalColumn(
+ "description",
+ DataTypes.STRING(),
+ "column for descriptions",
+ "not important")
+ .primaryKey("id")
+ .partitionKey("id, name")
+ .build();
+
+ List<Event> events = getTestEvents(table1Schema, table2Schema, myTable1, myTable2);
+
+ ValuesDataSourceHelper.setSourceEvents(Collections.singletonList(events));
+
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ // Setup value sink
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ // Setup pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.singletonList(
+ new TransformDef(
+ "default_namespace.default_schema.\\.*",
+ "*, name AS new_name, age + 1 AS new_age, 'extras' AS extras",
+ null,
+ null,
+ null,
+ null,
+ null)),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ // Check the order and content of all received events
+ String[] outputEvents = outCaptor.toString().trim().split("\n");
+
+ Assertions.assertThat(outputEvents)
+ .containsExactly(
+ "CreateTableEvent{tableId=default_namespace.default_schema.mytable1, schema=columns={`id` INT 'id column' 'AUTO_INCREMENT()',`name` STRING 'name column' 'Jane Doe',`age` INT 'age column' '17',`new_name` STRING 'name column' 'Jane Doe',`new_age` INT,`extras` STRING}, primaryKeys=id, partitionKeys=id, age, options=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, Alice, 18, Alice, 19, extras], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[2, Bob, 20, Bob, 21, extras], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[2, Bob, 20, Bob, 21, extras], after=[2, Bob, 30, Bob, 31, extras], op=UPDATE, meta=()}",
+ "CreateTableEvent{tableId=default_namespace.default_schema.mytable2, schema=columns={`id` BIGINT 'column for id' 'AUTO_DECREMENT()',`name` VARCHAR(255) 'column for name' 'John Smith',`age` TINYINT 'column for age' '91',`description` STRING 'column for descriptions' 'not important',`new_name` VARCHAR(255) 'column for name' 'John Smith',`new_age` INT,`extras` STRING}, primaryKeys=id, partitionKeys=id, name, options=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable2, before=[], after=[3, Carol, 15, student, Carol, 16, extras], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable2, before=[], after=[4, Derrida, 25, student, Derrida, 26, extras], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.mytable2, before=[4, Derrida, 25, student, Derrida, 26, extras], after=[], op=DELETE, meta=()}");
+ }
+
private List<Event> generateSchemaEvolutionEvents(TableId tableId) {
List<Event> events = new ArrayList<>();
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 ea3e2e017..57cbf238f 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
@@ -467,13 +467,13 @@ 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}, 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}, 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}, 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}, primaryKeys=ID, options=()}",
transformTestDatabase.getDatabaseName()),
60000L);
@@ -551,13 +551,13 @@ 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),`UID` INT,`NEWVERSION` STRING}, primaryKeys=ID, options=()}",
+ "CreateTableEvent{tableId=%s.TABLEALPHA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`PRICEALPHA` INT,`AGEALPHA` INT,`NAMEALPHA` VARCHAR(128),`UID` INT,`NEWVERSION` VARCHAR(17)}, 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),`UID` INT,`NEWVERSION` STRING}, primaryKeys=ID, options=()}",
+ "CreateTableEvent{tableId=%s.TABLEBETA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`CODENAMESBETA` VARCHAR(17),`AGEBETA` INT,`NAMEBETA` VARCHAR(128),`UID` INT,`NEWVERSION` VARCHAR(17)}, primaryKeys=ID, options=()}",
transformTestDatabase.getDatabaseName()),
60000L);
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 dc6f07751..23786c766 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
@@ -329,7 +329,7 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
return tableInfo;
}
- private Schema transformSchema(TableId tableId, Schema schema) throws Exception {
+ private Schema transformSchema(TableId tableId, Schema schema) {
List<Schema> newSchemas = new ArrayList<>();
for (PostTransformer transform : transforms) {
Selectors selectors = transform.getSelectors();
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
index 4bbe3d558..58f95392b 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.utils.StringUtils;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/**
@@ -96,11 +97,36 @@ public class ProjectionColumn implements Serializable {
return !StringUtils.isNullOrWhitespaceOnly(scriptExpression);
}
- public static ProjectionColumn of(String columnName, DataType dataType) {
- return new ProjectionColumn(Column.physicalColumn(columnName, dataType), null, null, null);
+ /**
+ * This projection is created with a plain column name. <br>
+ * Just like column {@code id} in {@code id, name AS new_name, age + 1 AS new_age}. <br>
+ * Comments and default expressions will be intact.
+ */
+ public static ProjectionColumn ofForwarded(Column column) {
+ String name = column.getName();
+ return new ProjectionColumn(column, name, name, Collections.singletonList(name));
}
- public static ProjectionColumn of(
+ /**
+ * This projection is created with a simple $id$ AS $new_id$ expression. <br>
+ * Just like column {@code new_name} in {@code id, name AS new_name, age + 1 AS new_age}. <br>
+ * Comments and default expressions will be intact.
+ */
+ public static ProjectionColumn ofAliased(Column column, String newName) {
+ String originalName = column.getName();
+ return new ProjectionColumn(
+ column.copy(newName),
+ originalName,
+ originalName,
+ Collections.singletonList(originalName));
+ }
+
+ /**
+ * This projection is created with a complex calculation expression. <br>
+ * Just like column {@code new_age} in {@code id, name AS new_name, age + 1 AS new_age}. <br>
+ * No comments nor default expressions will be kept.
+ */
+ public static ProjectionColumn ofCalculated(
String columnName,
DataType dataType,
String expression,
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 6d697652e..16416daff 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
@@ -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.types.DataType;
+import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
import org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
import org.apache.flink.cdc.runtime.parser.metadata.TransformSchemaFactory;
@@ -72,12 +73,12 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
+import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;
@@ -279,112 +280,119 @@ public class TransformParser {
.collect(
Collectors.toMap(
RelDataTypeField::getName, RelDataTypeField::getType));
-
- Map<String, DataType> rawDataTypeMap =
- columns.stream().collect(Collectors.toMap(Column::getName, Column::getType));
-
- Map<String, Boolean> isNotNullMap =
- columns.stream()
- .collect(
- Collectors.toMap(
- Column::getName, column -> !column.getType().isNullable()));
-
+ Map<String, Column> originalColumnMap =
+ columns.stream().collect(Collectors.toMap(Column::getName, column -> column));
List<ProjectionColumn> projectionColumns = new ArrayList<>();
+ Map<String, Integer> addedProjectionColumnNames = new HashMap<>();
for (SqlNode sqlNode : sqlSelect.getSelectList()) {
+ ProjectionColumn projectionColumn;
+
+ // A projection column could be <EXPR> AS <IDENTIFIER>...
if (sqlNode instanceof SqlBasicCall) {
SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
- if (SqlKind.AS.equals(sqlBasicCall.getOperator().kind)) {
- Optional<SqlNode> transformOptional = Optional.empty();
- String columnName;
- List<SqlNode> operandList = sqlBasicCall.getOperandList();
- if (operandList.size() == 2) {
- transformOptional = Optional.of(operandList.get(0));
- SqlNode sqlNode1 = operandList.get(1);
- if (sqlNode1 instanceof SqlIdentifier) {
- SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode1;
- columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
- } else {
- columnName = null;
- }
- } else {
- columnName = null;
- }
- if (isMetadataColumn(columnName)) {
- continue;
- }
- ProjectionColumn projectionColumn =
- transformOptional.isPresent()
- ? ProjectionColumn.of(
- columnName,
- DataTypeConverter.convertCalciteRelDataTypeToDataType(
- relDataTypeMap.get(columnName)),
- transformOptional.get().toString(),
- JaninoCompiler.translateSqlNodeToJaninoExpression(
- transformOptional.get(), udfDescriptors),
- parseColumnNameList(transformOptional.get()))
- : ProjectionColumn.of(
- columnName,
- DataTypeConverter.convertCalciteRelDataTypeToDataType(
- relDataTypeMap.get(columnName)));
- boolean hasReplacedDuplicateColumn = false;
- for (int i = 0; i < projectionColumns.size(); i++) {
- if (projectionColumns.get(i).getColumnName().equals(columnName)
- && !projectionColumns.get(i).isValidTransformedProjectionColumn()) {
- hasReplacedDuplicateColumn = true;
- projectionColumns.set(i, projectionColumn);
- break;
- }
- }
- if (!hasReplacedDuplicateColumn) {
- projectionColumns.add(projectionColumn);
- }
+ List<SqlNode> operandList = sqlBasicCall.getOperandList();
+ Preconditions.checkArgument(
+ SqlKind.AS.equals(sqlBasicCall.getOperator().kind)
+ && operandList.size() == 2
+ && operandList.get(1) instanceof SqlIdentifier,
+ "Unrecognized projection expression: "
+ + sqlBasicCall
+ + ". Should be <EXPR> AS <IDENTIFIER>");
+
+ // It's the identifier node for aliased column.
+ SqlIdentifier aliasNode = (SqlIdentifier) operandList.get(1);
+ String columnName = aliasNode.names.get(aliasNode.names.size() - 1);
+
+ Preconditions.checkArgument(
+ !isMetadataColumn(columnName),
+ "Column name %s is reserved and shading it is not allowed.",
+ columnName);
+
+ // This is the actual expression node of this projection column.
+ SqlNode exprNode = operandList.get(0);
+
+ if (exprNode instanceof SqlIdentifier) {
+ // This is a simple column rename like col_a AS col_b. Simply forward it to
+ // avoid losing metadata info like comments and default expressions.
+ SqlIdentifier identifierExprNode = (SqlIdentifier) exprNode;
+ String originalName =
+ identifierExprNode.names.get(identifierExprNode.names.size() - 1);
+ projectionColumn =
+ resolveProjectionColumnFromIdentifier(
+ relDataTypeMap, originalColumnMap, originalName, columnName);
} else {
- throw new ParseException(
- "Unrecognized projection expression: "
- + sqlBasicCall
- + ". Should be <EXPR> AS <IDENTIFIER>");
+ projectionColumn =
+ ProjectionColumn.ofCalculated(
+ columnName,
+ DataTypeConverter.convertCalciteRelDataTypeToDataType(
+ relDataTypeMap.get(columnName)),
+ exprNode.toString(),
+ JaninoCompiler.translateSqlNodeToJaninoExpression(
+ exprNode, udfDescriptors),
+ parseColumnNameList(exprNode));
}
- } else if (sqlNode instanceof SqlIdentifier) {
+ }
+ // ... or an existing column's name identifier.
+ else if (sqlNode instanceof SqlIdentifier) {
SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode;
String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
- DataType columnType;
- if (rawDataTypeMap.containsKey(columnName)) {
- columnType = rawDataTypeMap.get(columnName);
- } else if (relDataTypeMap.containsKey(columnName)) {
- columnType =
- DataTypeConverter.convertCalciteRelDataTypeToDataType(
- relDataTypeMap.get(columnName));
- } else {
- throw new RuntimeException(
- String.format("Failed to deduce column %s type", columnName));
- }
- if (isMetadataColumn(columnName)) {
- projectionColumns.add(
- ProjectionColumn.of(
- columnName,
- // Metadata columns should never be null
- columnType.notNull(),
- columnName,
- columnName,
- Arrays.asList(columnName)));
- } else {
- // Calcite translated column type doesn't keep nullability.
- // Appending it manually to circumvent this problem.
- projectionColumns.add(
- ProjectionColumn.of(
- columnName,
- isNotNullMap.get(columnName)
- ? columnType.notNull()
- : columnType.nullable()));
- }
+ projectionColumn =
+ resolveProjectionColumnFromIdentifier(
+ relDataTypeMap, originalColumnMap, columnName, columnName);
} else {
throw new ParseException("Unrecognized projection: " + sqlNode.toString());
}
+ // Projection columns comes later could override previous ones.
+ String projectionColumnName = projectionColumn.getColumnName();
+ if (addedProjectionColumnNames.containsKey(projectionColumnName)) {
+ // If we already have one column with identical name, replace it
+ projectionColumns.set(
+ addedProjectionColumnNames.get(projectionColumnName), projectionColumn);
+ } else {
+ // Otherwise, append it at the end. Don't forget to set the index!
+ projectionColumns.add(projectionColumn);
+ addedProjectionColumnNames.put(projectionColumnName, projectionColumns.size() - 1);
+ }
}
return projectionColumns;
}
+ /**
+ * Create a projection column from a simple identifier node (could be an upstream physical
+ * column or a metadata column).
+ */
+ public static ProjectionColumn resolveProjectionColumnFromIdentifier(
+ Map<String, RelDataType> relDataTypeMap,
+ Map<String, Column> originalColumnMap,
+ String identifier,
+ String projectedColumnName) {
+ if (isMetadataColumn(identifier)) {
+ // For a metadata column, we simply generate a projection column with the same
+ return ProjectionColumn.ofCalculated(
+ projectedColumnName,
+ // Metadata columns should never be null
+ DataTypeConverter.convertCalciteRelDataTypeToDataType(
+ relDataTypeMap.get(projectedColumnName))
+ .notNull(),
+ identifier,
+ identifier,
+ Collections.singletonList(identifier));
+ }
+
+ Preconditions.checkArgument(
+ originalColumnMap.containsKey(identifier),
+ "Referenced column %s is not present in original table.",
+ identifier);
+
+ Column column = originalColumnMap.get(identifier);
+ if (Objects.equals(identifier, projectedColumnName)) {
+ return ProjectionColumn.ofForwarded(column);
+ } else {
+ return ProjectionColumn.ofAliased(column, projectedColumnName);
+ }
+ }
+
public static String translateFilterExpressionToJaninoExpression(
String filterExpression, List<UserDefinedFunctionDescriptor> udfDescriptors) {
if (isNullOrWhitespaceOnly(filterExpression)) {
@@ -430,8 +438,6 @@ public class TransformParser {
String columnName = sqlNode.toString();
if (isMetadataColumn(columnName) && !columnNames.contains(columnName)) {
columnNames.add(columnName);
- } else {
- continue;
}
} else {
throw new ParseException("Unrecognized projection: " + sqlNode.toString());
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 9e0e9843c..faf50a879 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
@@ -92,7 +92,7 @@ public class PostTransformOperatorTest {
.physicalColumn("sid", DataTypes.INT())
.physicalColumn("name", DataTypes.STRING())
.physicalColumn("name_upper", DataTypes.STRING())
- .physicalColumn("tbname", DataTypes.STRING())
+ .physicalColumn("tbname", DataTypes.STRING().notNull())
.primaryKey("sid")
.build();
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UnifiedTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UnifiedTransformOperatorTest.java
index 43a364e51..d83e54e36 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UnifiedTransformOperatorTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UnifiedTransformOperatorTest.java
@@ -942,9 +942,11 @@ public class UnifiedTransformOperatorTest {
.physicalColumn("name", DataTypes.STRING().notNull())
.physicalColumn("age", DataTypes.INT().notNull())
.physicalColumn("computed", DataTypes.INT())
- .physicalColumn("metaColNameSpaceName", DataTypes.STRING())
- .physicalColumn("metaColSchemaName", DataTypes.STRING())
- .physicalColumn("metaColNameTableName", DataTypes.STRING())
+ .physicalColumn(
+ "metaColNameSpaceName", DataTypes.STRING().notNull())
+ .physicalColumn("metaColSchemaName", DataTypes.STRING().notNull())
+ .physicalColumn(
+ "metaColNameTableName", DataTypes.STRING().notNull())
.physicalColumn("metaColSchemaNameUpper", DataTypes.STRING())
.physicalColumn("metaColStr1", DataTypes.STRING())
.physicalColumn("metaColStr2", DataTypes.STRING())
@@ -1033,4 +1035,120 @@ public class UnifiedTransformOperatorTest {
.runTests()
.destroyHarness();
}
+
+ @Test
+ public void testTransformWithCommentsAndExpressions() throws Exception {
+ TableId tableId = TableId.tableId("my_company", "my_branch", "data_changes");
+ UnifiedTransformTestCase.of(
+ tableId,
+ "*, id + age as computed",
+ "id > 100",
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .physicalColumn("computed", DataTypes.INT())
+ .primaryKey("id")
+ .build())
+ .initializeHarness()
+ .insertSource(1001, "Alice", 17)
+ .insertPreTransformed(1001, "Alice", 17)
+ .insertPostTransformed(1001, "Alice", 17, 1018)
+ .runTests()
+ .destroyHarness();
+ UnifiedTransformTestCase.of(
+ tableId,
+ "id, age, id + age as computed",
+ "id > 100",
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .physicalColumn("computed", DataTypes.INT())
+ .primaryKey("id")
+ .build())
+ .initializeHarness()
+ .insertSource(1001, "Alice", 17)
+ .insertPreTransformed(1001, 17)
+ .insertPostTransformed(1001, 17, 1018)
+ .runTests()
+ .destroyHarness();
+
+ UnifiedTransformTestCase.of(
+ tableId,
+ "'extras' AS extras, age + 1 AS new_age_incremented, age AS new_age, name AS new_name, age, name, id",
+ "id > 100",
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .physicalColumn(
+ "description",
+ DataTypes.STRING(),
+ "description column",
+ "nothing really important")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn("extras", DataTypes.STRING())
+ .physicalColumn("new_age_incremented", DataTypes.INT())
+ .physicalColumn("new_age", DataTypes.INT(), "age column", "17")
+ .physicalColumn(
+ "new_name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn("age", DataTypes.INT(), "age column", "17")
+ .physicalColumn(
+ "name", DataTypes.STRING(), "name column", "John Smith")
+ .physicalColumn(
+ "id", DataTypes.INT(), "id column", "AUTO_INCREMENT()")
+ .primaryKey("id")
+ .build())
+ .initializeHarness()
+ .insertSource(1001, "Alice", 17, "Whatever")
+ .insertPreTransformed(1001, "Alice", 17)
+ .insertPostTransformed("extras", 18, 17, "Alice", 17, "Alice", 1001)
+ .runTests()
+ .destroyHarness();
+ }
}
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 86ebefedc..f17f27c9d 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
@@ -327,11 +327,11 @@ public class TransformParserTest {
List<String> expected =
Arrays.asList(
- "ProjectionColumn{column=`id` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
+ "ProjectionColumn{column=`id` INT 'id', expression='id', scriptExpression='id', originalColumnNames=[id], transformExpressionKey=null}",
"ProjectionColumn{column=`name` STRING, expression='UPPER(`TB`.`name`)', scriptExpression='upper(name)', originalColumnNames=[name], transformExpressionKey=null}",
"ProjectionColumn{column=`newage` INT, expression='`TB`.`age` + 1', scriptExpression='age + 1', originalColumnNames=[age], transformExpressionKey=null}",
"ProjectionColumn{column=`bmi` DOUBLE, expression='`TB`.`weight` / (`TB`.`height` * `TB`.`height`)', scriptExpression='weight / height * height', originalColumnNames=[weight, height, height], transformExpressionKey=null}");
- Assertions.assertThat(result.toString()).isEqualTo("[" + String.join(", ", expected) + "]");
+ Assertions.assertThat(result).hasToString("[" + String.join(", ", expected) + "]");
List<ProjectionColumn> metadataResult =
TransformParser.generateProjectionColumns(
@@ -341,24 +341,27 @@ public class TransformParserTest {
List<String> metadataExpected =
Arrays.asList(
- "ProjectionColumn{column=`id` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
- "ProjectionColumn{column=`name` STRING, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
- "ProjectionColumn{column=`age` INT, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
- "ProjectionColumn{column=`address` STRING, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
- "ProjectionColumn{column=`weight` DOUBLE, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
- "ProjectionColumn{column=`height` DOUBLE, expression='null', scriptExpression='null', originalColumnNames=null, transformExpressionKey=null}",
+ "ProjectionColumn{column=`id` INT 'id', expression='id', scriptExpression='id', originalColumnNames=[id], transformExpressionKey=null}",
+ "ProjectionColumn{column=`name` STRING 'string', expression='name', scriptExpression='name', originalColumnNames=[name], transformExpressionKey=null}",
+ "ProjectionColumn{column=`age` INT 'age', expression='age', scriptExpression='age', originalColumnNames=[age], transformExpressionKey=null}",
+ "ProjectionColumn{column=`address` STRING 'address', expression='address', scriptExpression='address', originalColumnNames=[address], transformExpressionKey=null}",
+ "ProjectionColumn{column=`weight` DOUBLE 'weight', expression='weight', scriptExpression='weight', originalColumnNames=[weight], transformExpressionKey=null}",
+ "ProjectionColumn{column=`height` DOUBLE 'height', expression='height', scriptExpression='height', originalColumnNames=[height], transformExpressionKey=null}",
"ProjectionColumn{column=`__namespace_name__` STRING NOT NULL, expression='__namespace_name__', scriptExpression='__namespace_name__', originalColumnNames=[__namespace_name__], transformExpressionKey=null}",
"ProjectionColumn{column=`__schema_name__` STRING NOT NULL, expression='__schema_name__', scriptExpression='__schema_name__', originalColumnNames=[__schema_name__], transformExpressionKey=null}",
"ProjectionColumn{column=`__table_name__` STRING NOT NULL, expression='__table_name__', scriptExpression='__table_name__', originalColumnNames=[__table_name__], transformExpressionKey=null}");
- Assertions.assertThat(metadataResult.toString())
- .isEqualTo("[" + String.join(", ", metadataExpected) + "]");
+ Assertions.assertThat(metadataResult)
+ .map(ProjectionColumn::toString)
+ .containsExactlyElementsOf(metadataExpected);
// calculated columns must use AS to provide an alias name
Assertions.assertThatThrownBy(
() ->
TransformParser.generateProjectionColumns(
"id, 1 + 1", testColumns, Collections.emptyList()))
- .isExactlyInstanceOf(ParseException.class);
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Unrecognized projection expression: 1 + 1. Should be <EXPR> AS <IDENTIFIER>");
}
@Test