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 2025/04/23 09:14:38 UTC
(flink-cdc) branch master updated: [FLINK-37104][cdc-transform] Refactor transform module and cleanup code
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 4bc736643 [FLINK-37104][cdc-transform] Refactor transform module and cleanup code
4bc736643 is described below
commit 4bc736643c9d047c7747101c9fe2cba1086a87b4
Author: yuxiqian <34...@users.noreply.github.com>
AuthorDate: Wed Apr 23 17:14:33 2025 +0800
[FLINK-37104][cdc-transform] Refactor transform module and cleanup code
This closes #3986.
---
.../flink/cdc/common/event/DataChangeEvent.java | 29 +
.../flink/cdc/common/utils/SchemaMergingUtils.java | 134 ++++
.../apache/flink/cdc/common/utils/SchemaUtils.java | 16 +
.../cdc/composer/flink/FlinkPipelineComposer.java | 3 +-
.../flink/translator/TransformTranslator.java | 55 +-
.../flink/FlinkPipelineComposerITCase.java | 13 -
.../flink/FlinkPipelineTransformITCase.java | 51 +-
.../transform/BatchPreTransformOperator.java | 362 ---------
.../transform/PostTransformChangeInfo.java | 111 ++-
.../operators/transform/PostTransformOperator.java | 883 +++++++++------------
.../transform/PostTransformOperatorBuilder.java | 93 +++
.../operators/transform/PostTransformer.java | 20 +-
.../operators/transform/PreTransformOperator.java | 99 +--
.../transform/PreTransformOperatorBuilder.java | 89 +++
.../transform/ProjectionColumnProcessor.java | 108 +--
.../operators/transform/TransformContext.java | 86 ++
.../operators/transform/TransformFilter.java | 16 +
.../transform/TransformFilterProcessor.java | 123 +--
.../operators/transform/TransformProjection.java | 24 +-
.../transform/TransformProjectionProcessor.java | 222 ++----
.../runtime/operators/transform/TransformRule.java | 4 +-
.../transform/UserDefinedFunctionDescriptor.java | 10 +-
.../flink/cdc/runtime/parser/TransformParser.java | 102 +--
.../cdc/runtime/parser/TransformParserTest.java | 19 -
24 files changed, 1177 insertions(+), 1495 deletions(-)
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
index 6185be3c4..92b2504aa 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
@@ -79,6 +79,19 @@ public class DataChangeEvent implements ChangeEvent, Serializable {
return op;
}
+ public String opTypeString(boolean isAfter) {
+ switch (op) {
+ case INSERT:
+ return "+I";
+ case UPDATE:
+ return isAfter ? "+U" : "-U";
+ case DELETE:
+ return "-D";
+ default:
+ throw new UnsupportedOperationException("Unknown operation type: " + op);
+ }
+ }
+
public Map<String, String> meta() {
return meta;
}
@@ -168,6 +181,22 @@ public class DataChangeEvent implements ChangeEvent, Serializable {
dataChangeEvent.meta);
}
+ /**
+ * Updates the after of a {@link DataChangeEvent} instance that describes the event with meta
+ * info.
+ */
+ public static DataChangeEvent projectRecords(
+ DataChangeEvent dataChangeEvent,
+ RecordData projectedBefore,
+ RecordData projectedAfter) {
+ return new DataChangeEvent(
+ dataChangeEvent.tableId,
+ projectedBefore,
+ projectedAfter,
+ dataChangeEvent.op,
+ dataChangeEvent.meta);
+ }
+
/** Updates the {@link TableId} info of current data change event. */
public static DataChangeEvent route(DataChangeEvent dataChangeEvent, TableId tableId) {
return new DataChangeEvent(
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
index dd93d5ca3..1851f8935 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
@@ -58,6 +58,7 @@ import org.apache.flink.cdc.common.types.VarBinaryType;
import org.apache.flink.cdc.common.types.VarCharType;
import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ArrayListMultimap;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
import org.apache.flink.shaded.guava31.com.google.common.collect.Streams;
import org.apache.flink.shaded.guava31.com.google.common.io.BaseEncoding;
@@ -308,6 +309,139 @@ public class SchemaMergingUtils {
return coercedRow;
}
+ /**
+ * Try to merge given {@link Schema}s and ensure they're identical. The only difference allowed
+ * is nullability, string and varchar precision, default value, and comments.
+ */
+ public static Schema strictlyMergeSchemas(List<Schema> schemas) {
+ Preconditions.checkArgument(
+ !schemas.isEmpty(), "Trying to merge transformed schemas %s, but got empty list");
+ if (schemas.size() == 1) {
+ return schemas.get(0);
+ }
+
+ List<List<String>> primaryKeys =
+ schemas.stream()
+ .map(Schema::primaryKeys)
+ .filter(p -> !p.isEmpty())
+ .distinct()
+ .collect(Collectors.toList());
+ List<List<String>> partitionKeys =
+ schemas.stream()
+ .map(Schema::partitionKeys)
+ .filter(p -> !p.isEmpty())
+ .distinct()
+ .collect(Collectors.toList());
+ List<Map<String, String>> options =
+ schemas.stream()
+ .map(Schema::options)
+ .filter(p -> !p.isEmpty())
+ .distinct()
+ .collect(Collectors.toList());
+ List<List<String>> columnNames =
+ schemas.stream()
+ .map(Schema::getColumnNames)
+ .distinct()
+ .collect(Collectors.toList());
+
+ Preconditions.checkArgument(
+ primaryKeys.size() <= 1,
+ "Trying to merge transformed schemas %s, but got more than one primary key configurations: %s",
+ schemas,
+ primaryKeys);
+ Preconditions.checkArgument(
+ partitionKeys.size() <= 1,
+ "Trying to merge transformed schemas %s, but got more than one partition key configurations: %s",
+ schemas,
+ partitionKeys);
+ Preconditions.checkArgument(
+ options.size() <= 1,
+ "Trying to merge transformed schemas %s, but got more than one option configurations: %s",
+ schemas,
+ options);
+ Preconditions.checkArgument(
+ columnNames.size() == 1,
+ "Trying to merge transformed schemas %s, but got more than one column name views: %s",
+ schemas,
+ columnNames);
+
+ int arity = columnNames.get(0).size();
+
+ ArrayListMultimap<Integer, DataType> toBeMergedColumnTypes =
+ ArrayListMultimap.create(arity, 1);
+ for (Schema schema : schemas) {
+ List<DataType> columnTypes = schema.getColumnDataTypes();
+ for (int colIndex = 0; colIndex < columnTypes.size(); colIndex++) {
+ toBeMergedColumnTypes.put(colIndex, columnTypes.get(colIndex));
+ }
+ }
+
+ List<String> mergedColumnNames = columnNames.iterator().next();
+ List<DataType> mergedColumnTypes = new ArrayList<>(arity);
+ for (int i = 0; i < arity; i++) {
+ mergedColumnTypes.add(strictlyMergeDataTypes(toBeMergedColumnTypes.get(i)));
+ }
+
+ List<Column> mergedColumns = new ArrayList<>();
+ for (int i = 0; i < mergedColumnNames.size(); i++) {
+ mergedColumns.add(
+ Column.physicalColumn(mergedColumnNames.get(i), mergedColumnTypes.get(i)));
+ }
+
+ return Schema.newBuilder()
+ .primaryKey(primaryKeys.isEmpty() ? Collections.emptyList() : primaryKeys.get(0))
+ .partitionKey(
+ partitionKeys.isEmpty() ? Collections.emptyList() : partitionKeys.get(0))
+ .options(options.isEmpty() ? Collections.emptyMap() : options.get(0))
+ .setColumns(mergedColumns)
+ .build();
+ }
+
+ private static DataType strictlyMergeDataTypes(List<DataType> dataTypes) {
+ Preconditions.checkArgument(
+ !dataTypes.isEmpty(),
+ "Trying to merge transformed data types %s, but got empty list");
+
+ List<DataType> simpleMergeTypes =
+ dataTypes.stream().distinct().collect(Collectors.toList());
+ if (simpleMergeTypes.size() == 1) {
+ return simpleMergeTypes.get(0);
+ }
+
+ List<DataTypeRoot> typeRoots =
+ dataTypes.stream()
+ .map(DataType::getTypeRoot)
+ .distinct()
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ typeRoots.size() == 1,
+ "Trying to merge types %s, but got more than one type root: %s",
+ dataTypes,
+ typeRoots);
+
+ // Decay types to the most
+ DataType type = dataTypes.get(0);
+
+ if (type.is(DataTypeRoot.CHAR)) {
+ return DataTypes.CHAR(CharType.MAX_LENGTH);
+ } else if (type.is(DataTypeRoot.VARCHAR)) {
+ return DataTypes.STRING();
+ } else if (type.is(DataTypeRoot.BINARY)) {
+ return DataTypes.BINARY(BinaryType.MAX_LENGTH);
+ } else if (type.is(DataTypeRoot.VARBINARY)) {
+ return DataTypes.VARBINARY(VarBinaryType.MAX_LENGTH);
+ } else if (type.is(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)) {
+ return DataTypes.TIMESTAMP(TimestampType.MAX_PRECISION);
+ } else if (type.is(DataTypeRoot.TIMESTAMP_WITH_TIME_ZONE)) {
+ return DataTypes.TIMESTAMP_TZ(ZonedTimestampType.MAX_PRECISION);
+ } else if (type.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+ return DataTypes.TIMESTAMP_LTZ(LocalZonedTimestampType.MAX_PRECISION);
+ } else {
+ throw new IllegalArgumentException(
+ "Unable to merge data types with different precision: " + dataTypes);
+ }
+ }
+
@VisibleForTesting
static boolean isDataTypeCompatible(@Nullable DataType currentType, DataType upcomingType) {
// If two types are identical, they're compatible of course.
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java
index eeb85faf8..131a783f5 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java
@@ -41,11 +41,13 @@ import javax.annotation.CheckReturnValue;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -389,6 +391,20 @@ public class SchemaUtils {
}));
}
+ @CheckReturnValue
+ public static Schema ensurePkNonNull(Schema schema) {
+ Set<String> pkColumns = new HashSet<>(schema.primaryKeys());
+ List<Column> columns =
+ schema.getColumns().stream()
+ .map(
+ col ->
+ pkColumns.contains(col.getName())
+ ? col.copy(col.getType().notNull())
+ : col)
+ .collect(Collectors.toList());
+ return schema.copy(columns);
+ }
+
// Schema merging related utility methods have been moved to SchemaMergingUtils class.
// The following methods have been deprecated and should not be used.
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 44ed8f0ff..c0189833c 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
@@ -155,8 +155,7 @@ public class FlinkPipelineComposer implements PipelineComposer {
pipelineDef.getUdfs(),
pipelineDef.getModels(),
dataSource.supportedMetadataColumns(),
- isParallelMetadataSource,
- isBatchMode);
+ !isParallelMetadataSource && !isBatchMode);
// PreTransform ---> PostTransform
stream =
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 3d966ce08..e81bf99e5 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
@@ -23,9 +23,10 @@ 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;
-import org.apache.flink.cdc.runtime.operators.transform.BatchPreTransformOperator;
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
+import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperatorBuilder;
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
+import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperatorBuilder;
import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -50,18 +51,10 @@ public class TransformTranslator {
List<UdfDef> udfFunctions,
List<ModelDef> models,
SupportedMetadataColumn[] supportedMetadataColumns,
- boolean canContainDistributedTables,
- boolean isBatchMode) {
+ boolean shouldStoreSchemasInState) {
if (transforms.isEmpty()) {
return input;
}
- if (isBatchMode) {
- return input.transform(
- "BatchTransform:Schema",
- new EventTypeInfo(),
- generatePreBatchTransform(
- transforms, udfFunctions, models, supportedMetadataColumns));
- }
return input.transform(
"Transform:Schema",
new EventTypeInfo(),
@@ -70,7 +63,7 @@ public class TransformTranslator {
udfFunctions,
models,
supportedMetadataColumns,
- canContainDistributedTables));
+ shouldStoreSchemasInState));
}
private PreTransformOperator generatePreTransform(
@@ -78,10 +71,9 @@ public class TransformTranslator {
List<UdfDef> udfFunctions,
List<ModelDef> models,
SupportedMetadataColumn[] supportedMetadataColumns,
- boolean canContainDistributedTables) {
+ boolean shouldStoreSchemasInState) {
- PreTransformOperator.Builder preTransformFunctionBuilder =
- PreTransformOperator.newBuilder();
+ PreTransformOperatorBuilder preTransformFunctionBuilder = PreTransformOperator.newBuilder();
for (TransformDef transform : transforms) {
preTransformFunctionBuilder.addTransform(
transform.getSourceTable(),
@@ -101,42 +93,11 @@ public class TransformTranslator {
.collect(Collectors.toList()))
.addUdfFunctions(
models.stream().map(this::modelToUDFTuple).collect(Collectors.toList()))
- .canContainDistributedTables(canContainDistributedTables);
+ .shouldStoreSchemasInState(shouldStoreSchemasInState);
return preTransformFunctionBuilder.build();
}
- private BatchPreTransformOperator generatePreBatchTransform(
- List<TransformDef> transforms,
- List<UdfDef> udfFunctions,
- List<ModelDef> models,
- SupportedMetadataColumn[] supportedMetadataColumns) {
-
- BatchPreTransformOperator.Builder preBatchTransformFunctionBuilder =
- BatchPreTransformOperator.newBuilder();
- for (TransformDef transform : transforms) {
- preBatchTransformFunctionBuilder.addTransform(
- transform.getSourceTable(),
- transform.getProjection(),
- transform.getFilter(),
- transform.getPrimaryKeys(),
- transform.getPartitionKeys(),
- transform.getTableOptions(),
- transform.getPostTransformConverter(),
- supportedMetadataColumns);
- }
-
- preBatchTransformFunctionBuilder
- .addUdfFunctions(
- udfFunctions.stream()
- .map(this::udfDefToUDFTuple)
- .collect(Collectors.toList()))
- .addUdfFunctions(
- models.stream().map(this::modelToUDFTuple).collect(Collectors.toList()));
-
- return preBatchTransformFunctionBuilder.build();
- }
-
public DataStream<Event> translatePostTransform(
DataStream<Event> input,
List<TransformDef> transforms,
@@ -148,7 +109,7 @@ public class TransformTranslator {
return input;
}
- PostTransformOperator.Builder postTransformFunctionBuilder =
+ PostTransformOperatorBuilder postTransformFunctionBuilder =
PostTransformOperator.newBuilder();
for (TransformDef transform : transforms) {
if (transform.isValidProjection() || transform.isValidFilter()) {
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 09a595379..13cf54e7f 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
@@ -1085,19 +1085,6 @@ class FlinkPipelineComposerITCase {
.primaryKey("id")
.build();
- // Create test dataset:
- // Create table 1 [id, name, age]
- // Create table 2 [id, name, age, description]
- // Table 1: +I[1, Alice, 18]
- // Table 1: +I[2, Bob, 20]
- // Table 1: -U[2, Bob, 20] +U[2, Bob, 30]
- // Table 2: +I[3, Charlie, 15, student]
- // Table 2: +I[4, Donald, 25, student]
- // Table 2: -D[4, Donald, 25, student]
- // Rename column for table 1: name -> last_name
- // Add column for table 2: gender
- // Table 1: +I[5, Eliza, 24]
- // Table 2: +I[6, Frank, 30, student, male]
List<Event> events = new ArrayList<>();
BinaryRecordDataGenerator table1dataGenerator =
new BinaryRecordDataGenerator(
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 29fa06488..e1df23a11 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
@@ -1002,7 +1002,7 @@ class FlinkPipelineTransformITCase {
sourceDef,
sinkDef,
Collections.emptyList(),
- new ArrayList<>(Arrays.asList(transformDef)),
+ Collections.singletonList(transformDef),
Collections.emptyList(),
pipelineConfig);
@@ -2030,8 +2030,7 @@ class FlinkPipelineTransformITCase {
execution.execute();
// Check the order and content of all received events
- String[] outputEvents = outCaptor.toString().trim().split("\n");
- return outputEvents;
+ return outCaptor.toString().trim().split("\n");
} finally {
outCaptor.reset();
}
@@ -2571,6 +2570,52 @@ class FlinkPipelineTransformITCase {
Collections.emptyList());
}
+ @Test
+ void testShadeOriginalColumnsWithDifferentType() throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.TRANSFORM_TABLE);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+ Configuration sinkConfig = new Configuration();
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.singletonList(
+ new TransformDef(
+ "\\.*.\\.*.\\.*",
+ "*, 0.5 + CAST(col1 AS DOUBLE) AS col1",
+ "col1 > 1.5",
+ null,
+ null,
+ null,
+ null,
+ null)),
+ Collections.emptyList(),
+ pipelineConfig);
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+ String[] outputEvents = outCaptor.toString().trim().split("\n");
+ assertThat(outputEvents)
+ .containsExactly(
+ "CreateTableEvent{tableId=default_namespace.default_schema.table1, schema=columns={`col1` DOUBLE NOT NULL,`col2` STRING}, primaryKeys=col1, options=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[2.5, 2], op=INSERT, meta=({op_ts=2})}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[], after=[3.5, 3], 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.5, ], after=[2.5, x], op=UPDATE, meta=({op_ts=5})}");
+ }
+
private List<Event> generateFloorCeilAndRoundEvents(TableId tableId) {
List<Event> events = new ArrayList<>();
Schema schema =
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/BatchPreTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/BatchPreTransformOperator.java
deleted file mode 100644
index 9ba810863..000000000
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/BatchPreTransformOperator.java
+++ /dev/null
@@ -1,362 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.cdc.runtime.operators.transform;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
-import org.apache.flink.cdc.common.event.ChangeEvent;
-import org.apache.flink.cdc.common.event.CreateTableEvent;
-import org.apache.flink.cdc.common.event.DataChangeEvent;
-import org.apache.flink.cdc.common.event.Event;
-import org.apache.flink.cdc.common.event.SchemaChangeEvent;
-import org.apache.flink.cdc.common.event.TableId;
-import org.apache.flink.cdc.common.schema.Column;
-import org.apache.flink.cdc.common.schema.Schema;
-import org.apache.flink.cdc.common.schema.Selectors;
-import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
-import org.apache.flink.cdc.common.utils.Preconditions;
-import org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException;
-import org.apache.flink.cdc.runtime.parser.TransformParser;
-import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.tasks.StreamTask;
-
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-
-/**
- * A data process function that filters out columns which aren't (directly & indirectly) referenced
- * in batch mode. There is no need to consider the task failover scenario in batch mode, as it
- * should be re - emitted from the source.
- */
-public class BatchPreTransformOperator extends AbstractStreamOperator<Event>
- implements OneInputStreamOperator<Event, Event>, Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private final List<TransformRule> transformRules;
- private transient List<PreTransformer> transforms;
- private final Map<TableId, PreTransformChangeInfo> preTransformChangeInfoMap;
- private final List<Tuple2<Selectors, SchemaMetadataTransform>> schemaMetadataTransformers;
- private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
- private List<UserDefinedFunctionDescriptor> udfDescriptors;
- private Map<TableId, PreTransformProcessor> preTransformProcessorMap;
- private Map<TableId, Boolean> hasAsteriskMap;
-
- public static BatchPreTransformOperator.Builder newBuilder() {
- return new BatchPreTransformOperator.Builder();
- }
-
- /** Builder of {@link BatchPreTransformOperator}. */
- public static class Builder {
- private final List<TransformRule> transformRules = new ArrayList<>();
-
- private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
- new ArrayList<>();
-
- public BatchPreTransformOperator.Builder addTransform(
- String tableInclusions, @Nullable String projection, @Nullable String filter) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- "",
- "",
- "",
- null,
- new SupportedMetadataColumn[0]));
- return this;
- }
-
- public BatchPreTransformOperator.Builder addTransform(
- String tableInclusions,
- @Nullable String projection,
- @Nullable String filter,
- String primaryKey,
- String partitionKey,
- String tableOption,
- @Nullable String postTransformConverter,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- primaryKey,
- partitionKey,
- tableOption,
- postTransformConverter,
- supportedMetadataColumns));
- return this;
- }
-
- public BatchPreTransformOperator.Builder addUdfFunctions(
- List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
- this.udfFunctions.addAll(udfFunctions);
- return this;
- }
-
- public BatchPreTransformOperator build() {
- return new BatchPreTransformOperator(transformRules, udfFunctions);
- }
- }
-
- private BatchPreTransformOperator(
- List<TransformRule> transformRules,
- List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
- this.preTransformChangeInfoMap = new ConcurrentHashMap<>();
- this.preTransformProcessorMap = new ConcurrentHashMap<>();
- this.schemaMetadataTransformers = new ArrayList<>();
- this.chainingStrategy = ChainingStrategy.ALWAYS;
-
- this.transformRules = transformRules;
- this.udfFunctions = udfFunctions;
- }
-
- @Override
- public void setup(
- StreamTask<?, ?> containingTask,
- StreamConfig config,
- Output<StreamRecord<Event>> output) {
- super.setup(containingTask, config, output);
- this.udfDescriptors =
- this.udfFunctions.stream()
- .map(udf -> new UserDefinedFunctionDescriptor(udf.f0, udf.f1, udf.f2))
- .collect(Collectors.toList());
-
- // Initialize data fields in advance because they might be accessed in
- // `::initializeState` function when restoring from a previous state.
- this.transforms = new ArrayList<>();
- for (TransformRule transformRule : transformRules) {
- String tableInclusions = transformRule.getTableInclusions();
- String projection = transformRule.getProjection();
- String filter = transformRule.getFilter();
- String primaryKeys = transformRule.getPrimaryKey();
- String partitionKeys = transformRule.getPartitionKey();
- String tableOptions = transformRule.getTableOption();
- Selectors selectors =
- new Selectors.SelectorsBuilder().includeTables(tableInclusions).build();
- transforms.add(
- new PreTransformer(
- selectors,
- TransformProjection.of(projection).orElse(null),
- TransformFilter.of(filter, udfDescriptors).orElse(null)));
- schemaMetadataTransformers.add(
- new Tuple2<>(
- selectors,
- new SchemaMetadataTransform(primaryKeys, partitionKeys, tableOptions)));
- }
- this.preTransformProcessorMap = new ConcurrentHashMap<>();
- this.hasAsteriskMap = new ConcurrentHashMap<>();
- }
-
- @Override
- public void finish() throws Exception {
- super.finish();
- clearOperator();
- }
-
- @Override
- public void close() throws Exception {
- super.close();
- clearOperator();
- }
-
- @Override
- public void processElement(StreamRecord<Event> element) throws Exception {
- Event event = element.getValue();
-
- try {
- processEvent(event);
- } catch (Exception e) {
- TableId tableId = null;
- Schema schemaBefore = null;
- Schema schemaAfter = null;
-
- if (event instanceof ChangeEvent) {
- tableId = ((ChangeEvent) event).tableId();
- PreTransformChangeInfo info = preTransformChangeInfoMap.get(tableId);
- if (info != null) {
- schemaBefore = info.getSourceSchema();
- schemaAfter = info.getPreTransformedSchema();
- }
- }
-
- throw new TransformException(
- "batch-pre-transform", event, tableId, schemaBefore, schemaAfter, e);
- }
- }
-
- private void processEvent(Event event) {
- if (event instanceof CreateTableEvent) {
- CreateTableEvent createTableEvent = (CreateTableEvent) event;
- preTransformProcessorMap.remove(createTableEvent.tableId());
- output.collect(new StreamRecord<>(cacheCreateTable(createTableEvent)));
- } else if (event instanceof DataChangeEvent) {
- output.collect(new StreamRecord<>(processDataChangeEvent(((DataChangeEvent) event))));
- }
- }
-
- private SchemaChangeEvent cacheCreateTable(CreateTableEvent event) {
- TableId tableId = event.tableId();
- Schema originalSchema = event.getSchema();
- event = transformCreateTableEvent(event);
- Schema newSchema = (event).getSchema();
- preTransformChangeInfoMap.put(
- tableId, PreTransformChangeInfo.of(tableId, originalSchema, newSchema));
- return event;
- }
-
- private CreateTableEvent transformCreateTableEvent(CreateTableEvent createTableEvent) {
- TableId tableId = createTableEvent.tableId();
- for (Tuple2<Selectors, SchemaMetadataTransform> transform : schemaMetadataTransformers) {
- Selectors selectors = transform.f0;
- if (selectors.isMatch(tableId)) {
- createTableEvent =
- new CreateTableEvent(
- tableId,
- transformSchemaMetaData(
- createTableEvent.getSchema(), transform.f1));
- }
- }
-
- cachePreTransformProcessor(tableId, createTableEvent.getSchema());
- if (preTransformProcessorMap.containsKey(tableId)) {
- return preTransformProcessorMap
- .get(tableId)
- .preTransformCreateTableEvent(createTableEvent);
- }
- return createTableEvent;
- }
-
- private void cachePreTransformProcessor(TableId tableId, Schema tableSchema) {
- LinkedHashSet<Column> referencedColumnsSet = new LinkedHashSet<>();
- boolean hasMatchTransform = false;
- for (PreTransformer transform : transforms) {
- if (!transform.getSelectors().isMatch(tableId)) {
- continue;
- }
- processProjectionTransform(tableId, tableSchema, referencedColumnsSet, transform);
- hasMatchTransform = true;
- }
- if (!hasMatchTransform) {
- processProjectionTransform(tableId, tableSchema, referencedColumnsSet, null);
- }
- }
-
- public void processProjectionTransform(
- TableId tableId,
- Schema tableSchema,
- LinkedHashSet<Column> referencedColumnsSet,
- @Nullable PreTransformer transform) {
- // If this TableId isn't presented in any transform block, it should behave like a "*"
- // projection and should be regarded as asterisk-ful.
- if (transform == null) {
- referencedColumnsSet.addAll(tableSchema.getColumns());
- hasAsteriskMap.put(tableId, true);
- } else {
- TransformProjection transformProjection = transform.getProjection().get();
- boolean hasAsterisk = TransformParser.hasAsterisk(transformProjection.getProjection());
- if (hasAsterisk) {
- referencedColumnsSet.addAll(tableSchema.getColumns());
- hasAsteriskMap.put(tableId, true);
- } else {
- TransformFilter transformFilter = transform.getFilter().orElse(null);
- List<Column> referencedColumns =
- TransformParser.generateReferencedColumns(
- transformProjection.getProjection(),
- transformFilter != null ? transformFilter.getExpression() : null,
- tableSchema.getColumns());
- // update referenced columns of other projections of the same tableId, if any
- referencedColumnsSet.addAll(referencedColumns);
- hasAsteriskMap.putIfAbsent(tableId, false);
- }
- }
-
- PreTransformChangeInfo tableChangeInfo =
- PreTransformChangeInfo.of(
- tableId,
- tableSchema,
- tableSchema.copy(new ArrayList<>(referencedColumnsSet)));
- preTransformProcessorMap.put(tableId, new PreTransformProcessor(tableChangeInfo));
- }
-
- private Schema transformSchemaMetaData(
- Schema schema, SchemaMetadataTransform schemaMetadataTransform) {
- Schema.Builder schemaBuilder = Schema.newBuilder().setColumns(schema.getColumns());
- if (!schemaMetadataTransform.getPrimaryKeys().isEmpty()) {
- schemaBuilder.primaryKey(schemaMetadataTransform.getPrimaryKeys());
- } else {
- schemaBuilder.primaryKey(schema.primaryKeys());
- }
- if (!schemaMetadataTransform.getPartitionKeys().isEmpty()) {
- schemaBuilder.partitionKey(schemaMetadataTransform.getPartitionKeys());
- } else {
- schemaBuilder.partitionKey(schema.partitionKeys());
- }
- if (!schemaMetadataTransform.getOptions().isEmpty()) {
- schemaBuilder.options(schemaMetadataTransform.getOptions());
- } else {
- schemaBuilder.options(schema.options());
- }
- return schemaBuilder.build();
- }
-
- private DataChangeEvent processDataChangeEvent(DataChangeEvent dataChangeEvent) {
- if (!transforms.isEmpty()) {
- TableId tableId = dataChangeEvent.tableId();
- PreTransformProcessor processor = preTransformProcessorMap.get(tableId);
- Preconditions.checkArgument(
- processor != null,
- "Transform operator receives a data change event from table %s without a full schema view. "
- + "This might happen if source with distributed tables doesn't emit CreateTableEvent first after fail-over. "
- + "This is likely a bug, please consider filing an issue.",
- tableId);
-
- BinaryRecordData before = (BinaryRecordData) dataChangeEvent.before();
- BinaryRecordData after = (BinaryRecordData) dataChangeEvent.after();
- if (before != null) {
- BinaryRecordData projectedBefore = processor.processFillDataField(before);
- dataChangeEvent = DataChangeEvent.projectBefore(dataChangeEvent, projectedBefore);
- }
- if (after != null) {
- BinaryRecordData projectedAfter = processor.processFillDataField(after);
- dataChangeEvent = DataChangeEvent.projectAfter(dataChangeEvent, projectedAfter);
- }
- }
- return dataChangeEvent;
- }
-
- private void clearOperator() {
- this.transforms = null;
- this.preTransformProcessorMap = null;
- }
-}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformChangeInfo.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformChangeInfo.java
index 989af2e75..b91dbc741 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformChangeInfo.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformChangeInfo.java
@@ -24,33 +24,78 @@ import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/**
* PostTransformChangeInfo caches pre-transformed / pre-transformed schema, schema field getters,
* and binary record data generator for post-transform schema.
*/
public class PostTransformChangeInfo {
- private TableId tableId;
- private Schema preTransformedSchema;
- private Schema postTransformedSchema;
- private RecordData.FieldGetter[] preTransformedFieldGetters;
- private RecordData.FieldGetter[] postTransformedFieldGetters;
- private BinaryRecordDataGenerator recordDataGenerator;
-
- public PostTransformChangeInfo(
+
+ private final TableId tableId;
+
+ private final Schema preTransformedSchema;
+ private final Schema postTransformedSchema;
+ private final Map<String, Integer> preTransformedSchemaFieldNameToIndexMap;
+
+ private final RecordData.FieldGetter[] preTransformedFieldGetters;
+ private final RecordData.FieldGetter[] postTransformedFieldGetters;
+ private final BinaryRecordDataGenerator postTransformedRecordDataGenerator;
+ private final Map<String, Integer> postTransformedSchemaFieldNameToIndexMap;
+
+ public static PostTransformChangeInfo of(
+ TableId tableId, Schema preTransformedSchema, Schema postTransformedSchema) {
+
+ List<RecordData.FieldGetter> preTransformedFieldGetters =
+ SchemaUtils.createFieldGetters(preTransformedSchema.getColumns());
+
+ List<RecordData.FieldGetter> postTransformedFieldGetters =
+ SchemaUtils.createFieldGetters(postTransformedSchema.getColumns());
+
+ BinaryRecordDataGenerator postTransformedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ DataTypeConverter.toRowType(postTransformedSchema.getColumns()));
+
+ return new PostTransformChangeInfo(
+ tableId,
+ preTransformedSchema,
+ preTransformedFieldGetters.toArray(new RecordData.FieldGetter[0]),
+ postTransformedSchema,
+ postTransformedFieldGetters.toArray(new RecordData.FieldGetter[0]),
+ postTransformedRecordDataGenerator);
+ }
+
+ private PostTransformChangeInfo(
TableId tableId,
- Schema postTransformedSchema,
- RecordData.FieldGetter[] postTransformedFieldGetters,
Schema preTransformedSchema,
RecordData.FieldGetter[] preTransformedFieldGetters,
- BinaryRecordDataGenerator recordDataGenerator) {
+ Schema postTransformedSchema,
+ RecordData.FieldGetter[] postTransformedFieldGetters,
+ BinaryRecordDataGenerator postTransformedRecordDataGenerator) {
+
this.tableId = tableId;
- this.postTransformedSchema = postTransformedSchema;
- this.postTransformedFieldGetters = postTransformedFieldGetters;
+
this.preTransformedSchema = preTransformedSchema;
this.preTransformedFieldGetters = preTransformedFieldGetters;
- this.recordDataGenerator = recordDataGenerator;
+ this.preTransformedSchemaFieldNameToIndexMap = new HashMap<>();
+ for (int i = 0; i < preTransformedSchema.getColumns().size(); i++) {
+ preTransformedSchemaFieldNameToIndexMap.put(
+ preTransformedSchema.getColumns().get(i).getName(), i);
+ }
+
+ this.postTransformedSchema = postTransformedSchema;
+ this.postTransformedFieldGetters = postTransformedFieldGetters;
+ this.postTransformedRecordDataGenerator = postTransformedRecordDataGenerator;
+ this.postTransformedSchemaFieldNameToIndexMap = new HashMap<>();
+
+ for (int i = 0; i < postTransformedSchema.getColumns().size(); i++) {
+ postTransformedSchemaFieldNameToIndexMap.put(
+ postTransformedSchema.getColumns().get(i).getName(), i);
+ }
}
public String getName() {
@@ -73,45 +118,27 @@ public class PostTransformChangeInfo {
return tableId;
}
+ public Schema getPreTransformedSchema() {
+ return preTransformedSchema;
+ }
+
public Schema getPostTransformedSchema() {
return postTransformedSchema;
}
- public Schema getPreTransformedSchema() {
- return preTransformedSchema;
+ public @Nullable Integer getPreTransformedSchemaFieldIndex(String fieldName) {
+ return preTransformedSchemaFieldNameToIndexMap.get(fieldName);
}
- public RecordData.FieldGetter[] getPostTransformedFieldGetters() {
- return postTransformedFieldGetters;
+ public @Nullable Integer getPostTransformedSchemaFieldIndex(String fieldName) {
+ return postTransformedSchemaFieldNameToIndexMap.get(fieldName);
}
public RecordData.FieldGetter[] getPreTransformedFieldGetters() {
return preTransformedFieldGetters;
}
- public BinaryRecordDataGenerator getRecordDataGenerator() {
- return recordDataGenerator;
- }
-
- public static PostTransformChangeInfo of(
- TableId tableId, Schema postTransformedSchema, Schema preTransformedSchema) {
-
- List<RecordData.FieldGetter> postTransformedFieldGetters =
- SchemaUtils.createFieldGetters(postTransformedSchema.getColumns());
-
- List<RecordData.FieldGetter> preTransformedFieldGetters =
- SchemaUtils.createFieldGetters(preTransformedSchema.getColumns());
-
- BinaryRecordDataGenerator postTransformedRecordDataGenerator =
- new BinaryRecordDataGenerator(
- DataTypeConverter.toRowType(postTransformedSchema.getColumns()));
-
- return new PostTransformChangeInfo(
- tableId,
- postTransformedSchema,
- postTransformedFieldGetters.toArray(new RecordData.FieldGetter[0]),
- preTransformedSchema,
- preTransformedFieldGetters.toArray(new RecordData.FieldGetter[0]),
- postTransformedRecordDataGenerator);
+ public BinaryRecordDataGenerator getPostTransformedRecordDataGenerator() {
+ return postTransformedRecordDataGenerator;
}
}
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 6ba1ec747..a9661db2c 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
@@ -26,42 +26,38 @@ import org.apache.flink.cdc.common.event.ChangeEvent;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
-import org.apache.flink.cdc.common.event.OperationType;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
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.SchemaMergingUtils;
import org.apache.flink.cdc.common.utils.SchemaUtils;
-import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverter;
import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverters;
import org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException;
import org.apache.flink.cdc.runtime.parser.TransformParser;
-import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.HashBasedTable;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Table;
import javax.annotation.Nullable;
import java.io.Serializable;
-import java.lang.reflect.InvocationTargetException;
-import java.time.ZoneId;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.common.utils.Preconditions.checkNotNull;
/**
* A data process function that performs column filtering, calculated column evaluation & final
@@ -74,596 +70,473 @@ public class PostTransformOperator extends AbstractStreamOperator<Event>
private final String timezone;
private final List<TransformRule> transformRules;
- private transient List<PostTransformer> transforms;
-
- /** keep the relationship of TableId and table information. */
- private final Map<TableId, PostTransformChangeInfo> postTransformChangeInfoMap;
-
- private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
- private List<UserDefinedFunctionDescriptor> udfDescriptors;
- private transient Map<String, Object> udfFunctionInstances;
-
- private transient Map<Tuple2<TableId, String>, TransformProjectionProcessor>
- transformProjectionProcessorMap;
- private transient Map<Tuple2<TableId, TransformFilter>, TransformFilterProcessor>
- transformFilterProcessorMap;
private final Map<TableId, Boolean> hasAsteriskMap;
private final Map<TableId, List<String>> projectedColumnsMap;
+ private final Map<TableId, PostTransformChangeInfo> postTransformInfoMap;
- public static PostTransformOperator.Builder newBuilder() {
- return new PostTransformOperator.Builder();
- }
-
- /** Builder of {@link PostTransformOperator}. */
- public static class Builder {
- private final List<TransformRule> transformRules = new ArrayList<>();
- private String timezone;
- private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
- new ArrayList<>();
-
- public PostTransformOperator.Builder addTransform(
- String tableInclusions,
- @Nullable String projection,
- @Nullable String filter,
- String primaryKey,
- String partitionKey,
- String tableOptions,
- String postTransformConverter,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- primaryKey,
- partitionKey,
- tableOptions,
- postTransformConverter,
- supportedMetadataColumns));
- return this;
- }
+ // Tuple3 items are: function name, class path, and extra options.
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
- public PostTransformOperator.Builder addTransform(
- String tableInclusions, @Nullable String projection, @Nullable String filter) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- "",
- "",
- "",
- null,
- new SupportedMetadataColumn[0]));
- return this;
- }
+ private transient List<PostTransformer> transformers;
+ private transient List<UserDefinedFunctionDescriptor> udfDescriptors;
+ private transient List<Object> udfFunctionInstances;
- public PostTransformOperator.Builder addTimezone(String timezone) {
- if (PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue().equals(timezone)) {
- this.timezone = ZoneId.systemDefault().toString();
- } else {
- this.timezone = timezone;
- }
- return this;
- }
+ // Querying a TransformProjectionProcessor with an upstream TableId and effective
+ // post-transformer.
+ private transient Table<TableId, PostTransformer, TransformProjectionProcessor>
+ projectionProcessors;
+ private transient Table<TableId, PostTransformer, TransformFilterProcessor> filterProcessors;
- public PostTransformOperator.Builder addUdfFunctions(
- List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
- this.udfFunctions.addAll(udfFunctions);
- return this;
- }
-
- public PostTransformOperator build() {
- return new PostTransformOperator(transformRules, timezone, udfFunctions);
- }
+ public static PostTransformOperatorBuilder newBuilder() {
+ return new PostTransformOperatorBuilder();
}
- private PostTransformOperator(
+ PostTransformOperator(
List<TransformRule> transformRules,
String timezone,
List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
- this.transformRules = transformRules;
this.timezone = timezone;
- this.postTransformChangeInfoMap = new ConcurrentHashMap<>();
- this.transformFilterProcessorMap = new ConcurrentHashMap<>();
- this.transformProjectionProcessorMap = new ConcurrentHashMap<>();
- this.udfFunctions = udfFunctions;
- this.udfFunctionInstances = new ConcurrentHashMap<>();
+ this.transformRules = transformRules;
this.hasAsteriskMap = new HashMap<>();
this.projectedColumnsMap = new HashMap<>();
- }
-
- @Override
- public void setup(
- StreamTask<?, ?> containingTask,
- StreamConfig config,
- Output<StreamRecord<Event>> output) {
- super.setup(containingTask, config, output);
- udfDescriptors =
- udfFunctions.stream()
- .map(udf -> new UserDefinedFunctionDescriptor(udf.f0, udf.f1, udf.f2))
- .collect(Collectors.toList());
+ this.postTransformInfoMap = new ConcurrentHashMap<>();
+ this.udfFunctions = udfFunctions;
}
@Override
public void open() throws Exception {
super.open();
- transforms =
- transformRules.stream()
- .map(
- transformRule -> {
- String tableInclusions = transformRule.getTableInclusions();
- String projection = transformRule.getProjection();
- String filterExpression = transformRule.getFilter();
-
- Selectors selectors =
- new Selectors.SelectorsBuilder()
- .includeTables(tableInclusions)
- .build();
- return new PostTransformer(
- selectors,
- TransformProjection.of(projection).orElse(null),
- TransformFilter.of(filterExpression, udfDescriptors)
- .orElse(null),
- PostTransformConverters.of(
- transformRule.getPostTransformConverter()),
- transformRule.getSupportedMetadataColumns());
- })
- .collect(Collectors.toList());
- this.transformProjectionProcessorMap = new ConcurrentHashMap<>();
- this.transformFilterProcessorMap = new ConcurrentHashMap<>();
- this.udfFunctionInstances = new ConcurrentHashMap<>();
- udfDescriptors.forEach(
- udf -> {
- try {
- Class<?> clazz = Class.forName(udf.getClasspath());
- udfFunctionInstances.put(udf.getName(), clazz.newInstance());
- } catch (ClassNotFoundException
- | InstantiationException
- | IllegalAccessException e) {
- throw new RuntimeException("Failed to instantiate UDF function " + udf);
- }
- });
+
+ // Initialize multi-key lookup tables
+ this.projectionProcessors = HashBasedTable.create();
+ this.filterProcessors = HashBasedTable.create();
+
+ // Be sure to initialize UDF related fields before creating transformers
initializeUdf();
- }
- @Override
- public void finish() throws Exception {
- super.finish();
- clearOperator();
+ this.transformers = createTransformers();
}
@Override
public void close() throws Exception {
super.close();
- clearOperator();
-
- // Clean up UDF instances
+ TransformExpressionCompiler.cleanUp();
destroyUdf();
- udfFunctionInstances.clear();
}
@Override
public void processElement(StreamRecord<Event> element) throws Exception {
- Event event = element.getValue();
-
try {
- processEvent(event);
+ processElementInternal(element);
} catch (Exception e) {
+ Event event = element.getValue();
TableId tableId = null;
Schema schemaBefore = null;
Schema schemaAfter = null;
if (event instanceof ChangeEvent) {
tableId = ((ChangeEvent) event).tableId();
-
- PostTransformChangeInfo info = postTransformChangeInfoMap.get(tableId);
+ PostTransformChangeInfo info = postTransformInfoMap.get(tableId);
if (info != null) {
schemaBefore = info.getPreTransformedSchema();
schemaAfter = info.getPostTransformedSchema();
}
}
+
throw new TransformException(
"post-transform", event, tableId, schemaBefore, schemaAfter, e);
}
}
- private void processEvent(Event event) throws Exception {
- if (event instanceof SchemaChangeEvent) {
- SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
- transformProjectionProcessorMap
- .keySet()
- .removeIf(e -> Objects.equals(e.f0, schemaChangeEvent.tableId()));
- transformFilterProcessorMap
- .keySet()
- .removeIf(e -> Objects.equals(e.f0, schemaChangeEvent.tableId()));
- cacheSchema(schemaChangeEvent).ifPresent(e -> output.collect(new StreamRecord<>(e)));
- } else if (event instanceof DataChangeEvent) {
- Optional<DataChangeEvent> dataChangeEventOptional =
- processDataChangeEvent(((DataChangeEvent) event));
- if (dataChangeEventOptional.isPresent()) {
- output.collect(new StreamRecord<>(dataChangeEventOptional.get()));
- }
+ private void processElementInternal(StreamRecord<Event> element) {
+ Event event = element.getValue();
+ if (event == null) {
+ return;
}
- }
- private Optional<SchemaChangeEvent> cacheSchema(SchemaChangeEvent event) throws Exception {
- TableId tableId = event.tableId();
- List<String> columnNamesBeforeChange = Collections.emptyList();
+ // Reject processing non-schema or data change events.
+ if (!(event instanceof ChangeEvent)) {
+ throw new UnsupportedOperationException("Unexpected stream record event: " + event);
+ }
- if (event instanceof CreateTableEvent) {
- CreateTableEvent createTableEvent = (CreateTableEvent) event;
- Set<String> projectedColumnsSet =
- transforms.stream()
- .filter(t -> t.getSelectors().isMatch(tableId))
- .flatMap(
- rule ->
- TransformParser.generateProjectionColumns(
- rule.getProjection()
- .map(TransformProjection::getProjection)
- .orElse(null),
- createTableEvent.getSchema().getColumns(),
- udfDescriptors,
- rule.getSupportedMetadataColumns())
- .stream())
- .map(ProjectionColumn::getColumnName)
- .collect(Collectors.toSet());
-
- boolean notTransformed =
- transforms.stream().noneMatch(t -> t.getSelectors().isMatch(tableId));
-
- if (notTransformed) {
- // If this TableId isn't presented in any transform block, it should behave like a
- // "*" projection and should be regarded as asterisk-ful.
- hasAsteriskMap.put(tableId, true);
- } else {
- boolean hasAsterisk =
- transforms.stream()
- .filter(t -> t.getSelectors().isMatch(tableId))
- .anyMatch(
- t ->
- TransformParser.hasAsterisk(
- t.getProjection()
- .map(
- TransformProjection
- ::getProjection)
- .orElse(null)));
-
- hasAsteriskMap.put(tableId, hasAsterisk);
- }
- projectedColumnsMap.put(
- tableId,
- createTableEvent.getSchema().getColumnNames().stream()
- .filter(projectedColumnsSet::contains)
- .collect(Collectors.toList()));
- } else {
- columnNamesBeforeChange =
- getPostTransformChangeInfo(tableId).getPreTransformedSchema().getColumnNames();
+ ChangeEvent changeEvent = (ChangeEvent) event;
+ TableId tableId = changeEvent.tableId();
+ List<PostTransformer> transformers = getEffectiveTransformers(tableId);
+
+ // Short-circuit if there's no effective transformers.
+ if (transformers.isEmpty()) {
+ output.collect(element);
+ return;
}
- Schema schema;
if (event instanceof CreateTableEvent) {
- CreateTableEvent createTableEvent = (CreateTableEvent) event;
- schema = createTableEvent.getSchema();
+ processCreateTableEvent((CreateTableEvent) event, transformers)
+ .map(StreamRecord::new)
+ .ifPresent(output::collect);
+ invalidateCache(tableId);
+ } else if (event instanceof SchemaChangeEvent) {
+ processSchemaChangeEvent((SchemaChangeEvent) event, transformers)
+ .map(StreamRecord::new)
+ .ifPresent(output::collect);
+ invalidateCache(tableId);
+ } else if (event instanceof DataChangeEvent) {
+ processDataChangeEvent((DataChangeEvent) event, transformers)
+ .map(StreamRecord::new)
+ .ifPresent(output::collect);
} else {
- schema =
- SchemaUtils.applySchemaChangeEvent(
- getPostTransformChangeInfo(tableId).getPreTransformedSchema(), event);
+ throw new UnsupportedOperationException("Unexpected stream record event: " + event);
}
+ }
- Schema projectedSchema = transformSchema(tableId, schema);
- postTransformChangeInfoMap.put(
- tableId, PostTransformChangeInfo.of(tableId, projectedSchema, schema));
+ // -------------------
+ // Key methods for processing upstream events.
+ // -------------------
- if (event instanceof CreateTableEvent) {
- return Optional.of(new CreateTableEvent(tableId, projectedSchema));
- } else if (hasAsteriskMap.getOrDefault(tableId, true)) {
+ /**
+ * Apply effective transform rules to {@link CreateTableEvent}s based on effective transformers.
+ */
+ private Optional<Event> processCreateTableEvent(
+ CreateTableEvent event, List<PostTransformer> effectiveTransformers) {
+ TableId tableId = event.tableId();
+ Schema preSchema = event.getSchema();
+
+ // Apply transform rules and verify we can get a deterministic post schema
+ List<Schema> schemas =
+ effectiveTransformers.stream()
+ .map(trans -> transformSchema(preSchema, trans))
+ .collect(Collectors.toList());
+
+ Schema postSchema =
+ SchemaUtils.ensurePkNonNull(SchemaMergingUtils.strictlyMergeSchemas(schemas));
+
+ // Update transform info map
+ postTransformInfoMap.put(
+ tableId, PostTransformChangeInfo.of(tableId, preSchema, postSchema));
+
+ // Update "if-table-has-been–wildcard–matched" map
+ boolean wildcardMatched =
+ effectiveTransformers.stream()
+ .map(PostTransformer::getProjection)
+ .flatMap(this::optionalToStream)
+ .map(TransformProjection::getProjection)
+ .anyMatch(TransformParser::hasAsterisk);
+ hasAsteriskMap.put(tableId, wildcardMatched);
+ projectedColumnsMap.put(
+ tableId,
+ preSchema.getColumnNames().stream()
+ .filter(postSchema.getColumnNames()::contains)
+ .collect(Collectors.toList()));
+
+ return Optional.of(new CreateTableEvent(tableId, postSchema));
+ }
+
+ /**
+ * Apply effective transform rules to other {@link SchemaChangeEvent}s based on effective
+ * transformers and existing {@link PostTransformChangeInfo}.
+ */
+ private Optional<Event> processSchemaChangeEvent(
+ SchemaChangeEvent event, List<PostTransformer> effectiveTransformers) {
+ TableId tableId = event.tableId();
+ PostTransformChangeInfo info = checkNotNull(postTransformInfoMap.get(tableId));
+
+ // Apply schema change event to the pre-transformed schema
+ Schema prevPreSchema = info.getPreTransformedSchema();
+ Schema nextPreSchema = SchemaUtils.applySchemaChangeEvent(prevPreSchema, event);
+
+ // Apply transform rules and verify we can get a deterministic post schema
+ List<Schema> schemas =
+ effectiveTransformers.stream()
+ .map(trans -> transformSchema(nextPreSchema, trans))
+ .collect(Collectors.toList());
+
+ Schema nextPostSchema =
+ SchemaUtils.ensurePkNonNull(SchemaMergingUtils.strictlyMergeSchemas(schemas));
+
+ // Update transform info map
+ postTransformInfoMap.put(
+ tableId, PostTransformChangeInfo.of(tableId, nextPreSchema, nextPostSchema));
+
+ // Prepare transformed schema change events
+ Schema prevPostSchema = info.getPostTransformedSchema();
+ List<String> columnNamesBeforeChange = prevPostSchema.getColumnNames();
+
+ if (hasAsteriskMap.getOrDefault(tableId, true)) {
// See comments in PreTransformOperator#cacheChangeSchema method.
- return SchemaUtils.transformSchemaChangeEvent(true, columnNamesBeforeChange, event);
+ return SchemaUtils.transformSchemaChangeEvent(true, columnNamesBeforeChange, event)
+ .map(Event.class::cast);
} else {
return SchemaUtils.transformSchemaChangeEvent(
- false, projectedColumnsMap.get(tableId), event);
+ false, projectedColumnsMap.get(tableId), event)
+ .map(Event.class::cast);
}
}
- private PostTransformChangeInfo getPostTransformChangeInfo(TableId tableId) {
- PostTransformChangeInfo tableInfo = postTransformChangeInfoMap.get(tableId);
- if (tableInfo == null) {
- throw new RuntimeException(
- "Schema for " + tableId + " not found. This shouldn't happen.");
- }
- return tableInfo;
- }
+ /** Apply projection rules to given {@link DataChangeEvent}. */
+ private Optional<Event> processDataChangeEvent(
+ DataChangeEvent event, List<PostTransformer> effectiveTransformers) {
+ TableId tableId = event.tableId();
+ PostTransformChangeInfo info = checkNotNull(postTransformInfoMap.get(tableId));
+
+ // Prepare transform context
+ TransformContext context = new TransformContext();
+ context.epochTime = System.currentTimeMillis();
+ context.meta = event.meta();
+
+ String beforeOp = event.opTypeString(false);
+ String afterOp = event.opTypeString(true);
+
+ for (PostTransformer transformer : effectiveTransformers) {
+ TransformProjectionProcessor projectionProcessor =
+ getProjectionProcessor(tableId, transformer);
+ TransformFilterProcessor filterProcessor = getFilterProcessor(tableId, transformer);
+
+ RecordData beforeRow = null;
+ RecordData afterRow = null;
+ boolean filterPassed = true;
+
+ if (event.before() != null) {
+ context.opType = beforeOp;
+ Tuple2<BinaryRecordData, Boolean> result =
+ transformRecord(
+ event.before(),
+ info,
+ projectionProcessor,
+ filterProcessor,
+ context);
+ beforeRow = result.f0;
+ filterPassed = result.f1;
+ }
- private Schema transformSchema(TableId tableId, Schema schema) {
- List<Schema> newSchemas = new ArrayList<>();
- for (PostTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
- if (selectors.isMatch(tableId) && transform.getProjection().isPresent()) {
- TransformProjection transformProjection =
- TransformProjection.of(transform.getProjection().get().getProjection())
- .get();
- if (transformProjection.isValid()) {
- if (!transformProjectionProcessorMap.containsKey(
- Tuple2.of(tableId, transformProjection.getProjection()))) {
- transformProjectionProcessorMap.put(
- Tuple2.of(tableId, transformProjection.getProjection()),
- TransformProjectionProcessor.of(
- transformProjection,
- timezone,
- udfDescriptors,
- getUdfFunctionInstances(),
- transform.getSupportedMetadataColumns()));
- }
- TransformProjectionProcessor postTransformProcessor =
- transformProjectionProcessorMap.get(
- Tuple2.of(tableId, transformProjection.getProjection()));
- // update the columns of projection and add the column of projection into Schema
- newSchemas.add(
- postTransformProcessor.processSchema(
- schema, transform.getSupportedMetadataColumns()));
+ if (event.after() != null) {
+ context.opType = afterOp;
+ Tuple2<BinaryRecordData, Boolean> result =
+ transformRecord(
+ event.after(), info, projectionProcessor, filterProcessor, context);
+ afterRow = result.f0;
+ filterPassed = result.f1;
+ }
+
+ if (filterPassed) {
+ DataChangeEvent finalEvent =
+ DataChangeEvent.projectRecords(event, beforeRow, afterRow);
+ if (transformer.getPostTransformConverter().isPresent()) {
+ return transformer
+ .getPostTransformConverter()
+ .get()
+ .convert(finalEvent)
+ .map(Event.class::cast);
+ } else {
+ return Optional.of(finalEvent);
}
}
}
- if (newSchemas.isEmpty()) {
- return schema;
- }
- return SchemaMergingUtils.getCommonSchema(newSchemas);
+ // Return original event if no transform predicate is satisfied/.
+ return Optional.empty();
}
- private List<Object> getUdfFunctionInstances() {
- return udfDescriptors.stream()
- .map(e -> udfFunctionInstances.get(e.getName()))
- .collect(Collectors.toList());
+ /**
+ * Generates transformed version of schema based on upstream schema and effective transformer.
+ */
+ private Schema transformSchema(Schema preSchema, PostTransformer transformer) {
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ transformer
+ .getProjection()
+ .map(TransformProjection::getProjection)
+ .orElse(null),
+ preSchema.getColumns(),
+ udfDescriptors,
+ transformer.getSupportedMetadataColumns());
+ return preSchema.copy(
+ projectionColumns.stream()
+ .map(ProjectionColumn::getColumn)
+ .collect(Collectors.toList()));
}
- private Optional<DataChangeEvent> processDataChangeEvent(DataChangeEvent dataChangeEvent)
- throws Exception {
- TableId tableId = dataChangeEvent.tableId();
- PostTransformChangeInfo tableInfo = getPostTransformChangeInfo(tableId);
- List<Optional<DataChangeEvent>> transformedDataChangeEventOptionalList = new ArrayList<>();
- long epochTime = System.currentTimeMillis();
- for (PostTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
-
- if (selectors.isMatch(tableId)) {
- Optional<DataChangeEvent> dataChangeEventOptional = Optional.of(dataChangeEvent);
- Optional<TransformProjection> transformProjectionOptional =
- transform.getProjection();
- Optional<TransformFilter> transformFilterOptional = transform.getFilter();
-
- if (transformFilterOptional.isPresent()
- && transformFilterOptional.get().isValid()) {
- TransformFilter transformFilter = transformFilterOptional.get();
- if (!transformFilterProcessorMap.containsKey(
- Tuple2.of(tableId, transformFilter))) {
- transformFilterProcessorMap.put(
- Tuple2.of(tableId, transformFilter),
- TransformFilterProcessor.of(
- tableInfo,
- transformFilter,
- timezone,
- udfDescriptors,
- getUdfFunctionInstances(),
- transform.getSupportedMetadataColumns()));
- }
- TransformFilterProcessor transformFilterProcessor =
- transformFilterProcessorMap.get(Tuple2.of(tableId, transformFilter));
- dataChangeEventOptional =
- processFilter(
- transformFilterProcessor,
- dataChangeEventOptional.get(),
- epochTime);
- }
- if (dataChangeEventOptional.isPresent()
- && transformProjectionOptional.isPresent()
- && transformProjectionOptional.get().isValid()) {
- TransformProjection transformProjection = transformProjectionOptional.get();
- if (!transformProjectionProcessorMap.containsKey(
- Tuple2.of(tableId, transformProjection.getProjection()))) {
- transformProjectionProcessorMap.put(
- Tuple2.of(tableId, transformProjection.getProjection()),
- TransformProjectionProcessor.of(
- tableInfo,
- transformProjection,
- timezone,
- udfDescriptors,
- getUdfFunctionInstances(),
- transform.getSupportedMetadataColumns()));
- } else if (!transformProjectionProcessorMap
- .get(Tuple2.of(tableId, transformProjection.getProjection()))
- .hasTableInfo()) {
- TransformProjectionProcessor transformProjectionProcessorWithoutTableInfo =
- transformProjectionProcessorMap.get(
- Tuple2.of(tableId, transformProjection.getProjection()));
- transformProjectionProcessorMap.put(
- Tuple2.of(tableId, transformProjection.getProjection()),
- TransformProjectionProcessor.of(
- tableInfo,
- transformProjectionProcessorWithoutTableInfo
- .getTransformProjection(),
- timezone,
- udfDescriptors,
- getUdfFunctionInstances(),
- transform.getSupportedMetadataColumns()));
- }
- TransformProjectionProcessor postTransformProcessor =
- transformProjectionProcessorMap.get(
- Tuple2.of(tableId, transformProjection.getProjection()));
- dataChangeEventOptional =
- processProjection(
- postTransformProcessor,
- dataChangeEventOptional.get(),
- epochTime);
- }
- if (dataChangeEventOptional.isPresent()
- && transform.getPostTransformConverter().isPresent()) {
- dataChangeEventOptional =
- convertDataChangeEvent(
- dataChangeEventOptional.get(),
- transform.getPostTransformConverter().get());
- }
- transformedDataChangeEventOptionalList.add(dataChangeEventOptional);
- }
+ /** Projects given {@link RecordData} based on given processor. */
+ private Tuple2<BinaryRecordData, Boolean> transformRecord(
+ RecordData recordData,
+ PostTransformChangeInfo info,
+ @Nullable TransformProjectionProcessor projectionProcessor,
+ @Nullable TransformFilterProcessor filterProcessor,
+ TransformContext context) {
+ RecordData.FieldGetter[] preFieldGetters = info.getPreTransformedFieldGetters();
+ Schema preSchema = info.getPreTransformedSchema();
+ Schema postSchema = info.getPostTransformedSchema();
+ BinaryRecordDataGenerator postGenerator = info.getPostTransformedRecordDataGenerator();
+
+ Object[] preRow = new Object[preFieldGetters.length];
+ for (int i = 0; i < preFieldGetters.length; i++) {
+ preRow[i] =
+ DataTypeConverter.convertToOriginal(
+ preFieldGetters[i].getFieldOrNull(recordData),
+ preSchema.getColumnDataTypes().get(i));
}
- if (transformedDataChangeEventOptionalList.isEmpty()) {
- return processPostProjection(tableInfo, dataChangeEvent);
- } else {
- for (Optional<DataChangeEvent> dataChangeEventOptional :
- transformedDataChangeEventOptionalList) {
- if (dataChangeEventOptional.isPresent()) {
- return processPostProjection(tableInfo, dataChangeEventOptional.get());
- }
- }
- return Optional.empty();
- }
- }
+ Object[] postRow =
+ projectionProcessor != null ? projectionProcessor.project(preRow, context) : preRow;
- private Optional<DataChangeEvent> convertDataChangeEvent(
- DataChangeEvent dataChangeEvent, PostTransformConverter postTransformConverter) {
- return postTransformConverter.convert(dataChangeEvent);
- }
+ // Filter predicate test might refer to both PreTransformed only columns (that have been
+ // eliminated from transform result) and PostTransformed only columns (that do not exist
+ // until expression evaluation finishes). So we need pass both rows to FilterProcessor.
+ boolean filterPassed =
+ filterProcessor == null || filterProcessor.test(preRow, postRow, context);
- private Optional<DataChangeEvent> processFilter(
- TransformFilterProcessor transformFilterProcessor,
- DataChangeEvent dataChangeEvent,
- long epochTime)
- 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(), '+'), meta)) {
- return Optional.of(dataChangeEvent);
- } else {
- return Optional.empty();
- }
- } else if (before != null) {
- if (transformFilterProcessor.process(
- before, epochTime, opTypeToRowKind(dataChangeEvent.op(), '-'), meta)) {
- return Optional.of(dataChangeEvent);
- } else {
- return Optional.empty();
- }
+ Object[] postRowBinary = new Object[postSchema.getColumnCount()];
+ for (int i = 0; i < postRow.length; i++) {
+ postRowBinary[i] =
+ DataTypeConverter.convert(postRow[i], postSchema.getColumnDataTypes().get(i));
}
- return Optional.empty();
+ return Tuple2.of(postGenerator.generate(postRowBinary), filterPassed);
}
- private Optional<DataChangeEvent> processProjection(
- TransformProjectionProcessor postTransformProcessor,
- DataChangeEvent dataChangeEvent,
- long epochTime) {
- BinaryRecordData before = (BinaryRecordData) dataChangeEvent.before();
- BinaryRecordData after = (BinaryRecordData) dataChangeEvent.after();
- if (before != null) {
- BinaryRecordData projectedBefore =
- postTransformProcessor.processData(
- before,
- epochTime,
- opTypeToRowKind(dataChangeEvent.op(), '-'),
- dataChangeEvent.meta());
- dataChangeEvent = DataChangeEvent.projectBefore(dataChangeEvent, projectedBefore);
- }
- if (after != null) {
- BinaryRecordData projectedAfter =
- postTransformProcessor.processData(
- after,
- epochTime,
- opTypeToRowKind(dataChangeEvent.op(), '+'),
- dataChangeEvent.meta());
- dataChangeEvent = DataChangeEvent.projectAfter(dataChangeEvent, projectedAfter);
- }
- return Optional.of(dataChangeEvent);
+ // -------------------
+ // Convenience methods for coping with transient fields.
+ // -------------------
+
+ /** Obtain effective transformers based on given {@link TableId}. */
+ private List<PostTransformer> getEffectiveTransformers(TableId tableId) {
+ return transformers.stream()
+ .filter(trans -> trans.getSelectors().isMatch(tableId))
+ .collect(Collectors.toList());
}
- private Optional<DataChangeEvent> processPostProjection(
- PostTransformChangeInfo tableInfo, DataChangeEvent dataChangeEvent) throws Exception {
- BinaryRecordData before = (BinaryRecordData) dataChangeEvent.before();
- BinaryRecordData after = (BinaryRecordData) dataChangeEvent.after();
- if (before != null) {
- BinaryRecordData projectedBefore = projectRecord(tableInfo, before);
- dataChangeEvent = DataChangeEvent.projectBefore(dataChangeEvent, projectedBefore);
- }
- if (after != null) {
- BinaryRecordData projectedAfter = projectRecord(tableInfo, after);
- dataChangeEvent = DataChangeEvent.projectAfter(dataChangeEvent, projectedAfter);
+ /**
+ * Get the unique {@link TransformProjectionProcessor} based on provided {@link TableId} and
+ * {@link PostTransformer}.
+ */
+ private TransformProjectionProcessor getProjectionProcessor(
+ TableId tableId, PostTransformer postTransformer) {
+ if (!projectionProcessors.contains(tableId, postTransformer)) {
+ PostTransformChangeInfo changeInfo = postTransformInfoMap.get(tableId);
+ projectionProcessors.put(
+ tableId,
+ postTransformer,
+ new TransformProjectionProcessor(
+ changeInfo,
+ postTransformer
+ .getProjection()
+ .map(TransformProjection::getProjection)
+ .orElse(null),
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ postTransformer.getSupportedMetadataColumns()));
}
- return Optional.of(dataChangeEvent);
+ return projectionProcessors.get(tableId, postTransformer);
}
- private BinaryRecordData projectRecord(
- PostTransformChangeInfo tableInfo, BinaryRecordData recordData) {
- List<Object> valueList = new ArrayList<>();
- RecordData.FieldGetter[] fieldGetters = tableInfo.getPostTransformedFieldGetters();
-
- for (RecordData.FieldGetter fieldGetter : fieldGetters) {
- valueList.add(fieldGetter.getFieldOrNull(recordData));
+ /**
+ * Get the unique {@link TransformFilterProcessor} based on provided {@link TableId} and {@link
+ * PostTransformer}.
+ */
+ private TransformFilterProcessor getFilterProcessor(
+ TableId tableId, PostTransformer postTransformer) {
+ if (!filterProcessors.contains(tableId, postTransformer)) {
+ if (!postTransformer.getFilter().isPresent()) {
+ filterProcessors.put(tableId, postTransformer, TransformFilterProcessor.ofNoOp());
+ } else {
+ PostTransformChangeInfo changeInfo = postTransformInfoMap.get(tableId);
+ filterProcessors.put(
+ tableId,
+ postTransformer,
+ TransformFilterProcessor.of(
+ changeInfo,
+ postTransformer.getFilter().orElse(null),
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ postTransformer.getSupportedMetadataColumns()));
+ }
}
+ return filterProcessors.get(tableId, postTransformer);
+ }
- return tableInfo
- .getRecordDataGenerator()
- .generate(valueList.toArray(new Object[valueList.size()]));
+ /**
+ * Flush caches saved for given {@link TableId}. Be sure to invalidate caches after its schema
+ * has been changed!
+ */
+ private void invalidateCache(TableId tableId) {
+ projectionProcessors.row(tableId).clear();
+ filterProcessors.row(tableId).clear();
}
- private void clearOperator() {
- this.transforms = null;
- this.transformProjectionProcessorMap = null;
- this.transformFilterProcessorMap = null;
- TransformExpressionCompiler.cleanUp();
+ private List<PostTransformer> createTransformers() {
+ List<PostTransformer> list = new ArrayList<>();
+ for (TransformRule rule : transformRules) {
+ String projection = rule.getProjection();
+ String filterExpression = rule.getFilter();
+ String tableInclusions = rule.getTableInclusions();
+ Selectors selectors =
+ new Selectors.SelectorsBuilder().includeTables(tableInclusions).build();
+ PostTransformer apply =
+ new PostTransformer(
+ selectors,
+ TransformProjection.of(projection).orElse(null),
+ TransformFilter.of(filterExpression, udfDescriptors).orElse(null),
+ PostTransformConverters.of(rule.getPostTransformConverter())
+ .orElse(null),
+ rule.getSupportedMetadataColumns());
+ list.add(apply);
+ }
+ return list;
}
private void initializeUdf() {
- udfDescriptors.forEach(
- udf -> {
- try {
- if (udf.isCdcPipelineUdf()) {
- // We use reflection to invoke UDF methods since we may add more methods
- // into UserDefinedFunction interface, thus the provided UDF classes
- // might not be compatible with the interface definition in CDC common.
- Object udfInstance = udfFunctionInstances.get(udf.getName());
- UserDefinedFunctionContext userDefinedFunctionContext =
- () -> Configuration.fromMap(udf.getParameters());
- udfInstance
- .getClass()
- .getMethod("open", UserDefinedFunctionContext.class)
- .invoke(udfInstance, userDefinedFunctionContext);
- } else {
- // Do nothing, Flink-style UDF lifecycle hooks are not supported
- }
- } catch (InvocationTargetException
- | NoSuchMethodException
- | IllegalAccessException ex) {
- throw new RuntimeException("Failed to initialize UDF " + udf, ex);
- }
- });
+ this.udfDescriptors =
+ udfFunctions.stream()
+ .map(UserDefinedFunctionDescriptor::new)
+ .collect(Collectors.toList());
+ this.udfFunctionInstances = new ArrayList<>();
+
+ for (UserDefinedFunctionDescriptor udf : udfDescriptors) {
+ try {
+ Class<?> clazz = Class.forName(udf.getClasspath());
+ Object udfInstance = clazz.getDeclaredConstructor().newInstance();
+ udfFunctionInstances.add(udfInstance);
+
+ if (udf.isCdcPipelineUdf()) {
+ // We use reflection to invoke UDF methods since we may add more methods
+ // into UserDefinedFunction interface, thus the provided UDF classes
+ // might not be compatible with the interface definition in CDC common.
+ UserDefinedFunctionContext userDefinedFunctionContext =
+ () -> Configuration.fromMap(udf.getParameters());
+ udfInstance
+ .getClass()
+ .getMethod("open", UserDefinedFunctionContext.class)
+ .invoke(udfInstance, userDefinedFunctionContext);
+ }
+ // Do nothing for Flink-style UDF since their lifecycle hooks are not supported
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to instantiate UDF function " + udf, e);
+ }
+ }
}
private void destroyUdf() {
- udfDescriptors.forEach(
- udf -> {
- try {
- if (udf.isCdcPipelineUdf()) {
- // We use reflection to invoke UDF methods since we may add more methods
- // into UserDefinedFunction interface, thus the provided UDF classes
- // might not be compatible with the interface definition in CDC common.
- Object udfInstance = udfFunctionInstances.get(udf.getName());
- udfInstance.getClass().getMethod("close").invoke(udfInstance);
- } else {
- // Do nothing, Flink-style UDF lifecycle hooks are not supported
- }
- } catch (InvocationTargetException
- | NoSuchMethodException
- | IllegalAccessException ex) {
- throw new RuntimeException("Failed to destroy UDF " + udf, ex);
- }
- });
+ if (udfDescriptors == null || udfFunctionInstances == null) {
+ return;
+ }
+ for (int i = 0; i < udfDescriptors.size(); i++) {
+ UserDefinedFunctionDescriptor udf = udfDescriptors.get(i);
+ try {
+ if (udf.isCdcPipelineUdf()) {
+ Object udfInstance = udfFunctionInstances.get(i);
+ udfInstance.getClass().getMethod("close").invoke(udfInstance);
+ }
+ // Do nothing for Flink-style UDF since their lifecycle hooks are not supported
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to destroy UDF " + udf, e);
+ }
+ }
+ udfDescriptors.clear();
+ udfFunctionInstances.clear();
}
- private String opTypeToRowKind(OperationType opType, char beforeOrAfter) {
- return String.format("%c%c", beforeOrAfter, opType.name().charAt(0));
+ /** Backport of {@code Optional#stream} before Java 11. */
+ @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+ private <T> Stream<T> optionalToStream(Optional<T> optional) {
+ return optional.map(Stream::of).orElseGet(Stream::empty);
}
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
new file mode 100644
index 000000000..f28bc6ce5
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.transform;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+
+import javax.annotation.Nullable;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Builder of {@link PostTransformOperator}. */
+public class PostTransformOperatorBuilder {
+ private final List<TransformRule> transformRules = new ArrayList<>();
+ private String timezone;
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
+ new ArrayList<>();
+
+ public PostTransformOperatorBuilder addTransform(
+ String tableInclusions,
+ @Nullable String projection,
+ @Nullable String filter,
+ String primaryKey,
+ String partitionKey,
+ String tableOptions,
+ String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ primaryKey,
+ partitionKey,
+ tableOptions,
+ postTransformConverter,
+ supportedMetadataColumns));
+ return this;
+ }
+
+ public PostTransformOperatorBuilder addTransform(
+ String tableInclusions, @Nullable String projection, @Nullable String filter) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ "",
+ "",
+ "",
+ null,
+ new SupportedMetadataColumn[0]));
+ return this;
+ }
+
+ public PostTransformOperatorBuilder addTimezone(String timezone) {
+ if (PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue().equals(timezone)) {
+ this.timezone = ZoneId.systemDefault().toString();
+ } else {
+ this.timezone = timezone;
+ }
+ return this;
+ }
+
+ public PostTransformOperatorBuilder addUdfFunctions(
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ this.udfFunctions.addAll(udfFunctions);
+ return this;
+ }
+
+ public PostTransformOperator build() {
+ return new PostTransformOperator(transformRules, timezone, udfFunctions);
+ }
+}
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 c448e9903..c6d8b06ab 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
@@ -28,21 +28,21 @@ 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;
- private final Optional<PostTransformConverter> postTransformConverter;
+ private final @Nullable TransformProjection projection;
+ private final @Nullable TransformFilter filter;
+ private final @Nullable PostTransformConverter postTransformConverter;
+ private final SupportedMetadataColumn[] supportedMetadataColumns;
public PostTransformer(
Selectors selectors,
@Nullable TransformProjection projection,
@Nullable TransformFilter filter,
- Optional<PostTransformConverter> postTransformConverter,
+ @Nullable 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.projection = projection;
+ this.filter = filter;
this.postTransformConverter = postTransformConverter;
this.supportedMetadataColumns = supportedMetadataColumns;
}
@@ -52,15 +52,15 @@ public class PostTransformer {
}
public Optional<TransformProjection> getProjection() {
- return projection;
+ return Optional.ofNullable(projection);
}
public Optional<TransformFilter> getFilter() {
- return filter;
+ return Optional.ofNullable(filter);
}
public Optional<PostTransformConverter> getPostTransformConverter() {
- return postTransformConverter;
+ return Optional.ofNullable(postTransformConverter);
}
public SupportedMetadataColumn[] getSupportedMetadataColumns() {
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 b1ea8af8c..97293ab5e 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
@@ -34,7 +34,6 @@ import org.apache.flink.cdc.common.event.TruncateTableEvent;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.schema.Selectors;
-import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException;
@@ -71,92 +70,29 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
private static final long serialVersionUID = 1L;
- private final List<TransformRule> transformRules;
- private transient List<PreTransformer> transforms;
- private final Map<TableId, PreTransformChangeInfo> preTransformChangeInfoMap;
-
/** All tables which have been sent {@link CreateTableEvent} to downstream. */
private final Set<TableId> alreadySentCreateTableEvents;
+ private final List<TransformRule> transformRules;
+ private final Map<TableId, PreTransformChangeInfo> preTransformChangeInfoMap;
private final List<Tuple2<Selectors, SchemaMetadataTransform>> schemaMetadataTransformers;
- private transient ListState<byte[]> state;
private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
- private List<UserDefinedFunctionDescriptor> udfDescriptors;
- private Map<TableId, PreTransformProcessor> preTransformProcessorMap;
- private Map<TableId, Boolean> hasAsteriskMap;
- private final boolean canContainDistributedTables;
+ private final boolean shouldStoreSchemasInState;
- public static PreTransformOperator.Builder newBuilder() {
- return new PreTransformOperator.Builder();
- }
-
- /** Builder of {@link PreTransformOperator}. */
- public static class Builder {
- private final List<TransformRule> transformRules = new ArrayList<>();
- private boolean canContainDistributedTables;
-
- private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
- new ArrayList<>();
-
- public PreTransformOperator.Builder addTransform(
- String tableInclusions, @Nullable String projection, @Nullable String filter) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- "",
- "",
- "",
- null,
- new SupportedMetadataColumn[0]));
- return this;
- }
-
- public PreTransformOperator.Builder addTransform(
- String tableInclusions,
- @Nullable String projection,
- @Nullable String filter,
- String primaryKey,
- String partitionKey,
- String tableOption,
- @Nullable String postTransformConverter,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- transformRules.add(
- new TransformRule(
- tableInclusions,
- projection,
- filter,
- primaryKey,
- partitionKey,
- tableOption,
- postTransformConverter,
- supportedMetadataColumns));
- return this;
- }
-
- public PreTransformOperator.Builder addUdfFunctions(
- List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
- this.udfFunctions.addAll(udfFunctions);
- return this;
- }
-
- public PreTransformOperator.Builder canContainDistributedTables(
- boolean canContainDistributedTables) {
- this.canContainDistributedTables = canContainDistributedTables;
- return this;
- }
+ private transient ListState<byte[]> state;
+ private transient List<PreTransformer> transforms;
+ private transient List<UserDefinedFunctionDescriptor> udfDescriptors;
+ private transient Map<TableId, PreTransformProcessor> preTransformProcessorMap;
+ private transient Map<TableId, Boolean> hasAsteriskMap;
- public PreTransformOperator build() {
- return new PreTransformOperator(
- transformRules, udfFunctions, canContainDistributedTables);
- }
+ public static PreTransformOperatorBuilder newBuilder() {
+ return new PreTransformOperatorBuilder();
}
- private PreTransformOperator(
+ PreTransformOperator(
List<TransformRule> transformRules,
List<Tuple3<String, String, Map<String, String>>> udfFunctions,
- boolean canContainDistributedTables) {
+ boolean shouldStoreSchemasInState) {
this.preTransformChangeInfoMap = new ConcurrentHashMap<>();
this.alreadySentCreateTableEvents = new HashSet<>();
this.preTransformProcessorMap = new ConcurrentHashMap<>();
@@ -165,7 +101,7 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
this.transformRules = transformRules;
this.udfFunctions = udfFunctions;
- this.canContainDistributedTables = canContainDistributedTables;
+ this.shouldStoreSchemasInState = shouldStoreSchemasInState;
}
@Override
@@ -208,10 +144,9 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- if (canContainDistributedTables) {
- // In distributed mode, we don't have a globally consistent schema for each partition.
- // It's not meaningful to persist them to state. Instead, we rely on each source
- // partition to send fresh CreateTableEvent to instantiate each event flow.
+ if (!shouldStoreSchemasInState) {
+ // Skip schema persistency if we're in the distributed schema mode or the batch
+ // execution mode.
return;
}
OperatorStateStore stateStore = context.getOperatorStateStore();
@@ -240,7 +175,7 @@ public class PreTransformOperator extends AbstractStreamOperator<Event>
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
- if (canContainDistributedTables) {
+ if (!shouldStoreSchemasInState) {
// Same reason in this#initializeState.
return;
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorBuilder.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorBuilder.java
new file mode 100644
index 000000000..e458b6416
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorBuilder.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.transform;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Builder of {@link PreTransformOperator}. */
+public class PreTransformOperatorBuilder {
+ private final List<TransformRule> transformRules = new ArrayList<>();
+ private boolean shouldStoreSchemasInState;
+
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
+ new ArrayList<>();
+
+ public PreTransformOperatorBuilder addTransform(
+ String tableInclusions, @Nullable String projection, @Nullable String filter) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ "",
+ "",
+ "",
+ null,
+ new SupportedMetadataColumn[0]));
+ return this;
+ }
+
+ public PreTransformOperatorBuilder addTransform(
+ String tableInclusions,
+ @Nullable String projection,
+ @Nullable String filter,
+ String primaryKey,
+ String partitionKey,
+ String tableOption,
+ @Nullable String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ primaryKey,
+ partitionKey,
+ tableOption,
+ postTransformConverter,
+ supportedMetadataColumns));
+ return this;
+ }
+
+ public PreTransformOperatorBuilder addUdfFunctions(
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ this.udfFunctions.addAll(udfFunctions);
+ return this;
+ }
+
+ public PreTransformOperatorBuilder shouldStoreSchemasInState(
+ boolean shouldStoreSchemasInState) {
+ this.shouldStoreSchemasInState = shouldStoreSchemasInState;
+ return this;
+ }
+
+ public PreTransformOperator build() {
+ return new PreTransformOperator(transformRules, udfFunctions, shouldStoreSchemasInState);
+ }
+}
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 3551f3b70..8db8efc31 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
@@ -17,25 +17,20 @@
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;
import org.codehaus.janino.ExpressionEvaluator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
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.operators.transform.TransformContext.lookupObjectByName;
import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METADATA_COLUMNS;
/**
@@ -43,16 +38,14 @@ import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METAD
* computed columns.
*/
public class ProjectionColumnProcessor {
- private static final Logger LOG = LoggerFactory.getLogger(ProjectionColumnProcessor.class);
- private PostTransformChangeInfo tableInfo;
- private ProjectionColumn projectionColumn;
- 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;
+ private final PostTransformChangeInfo tableInfo;
+ private final ProjectionColumn projectionColumn;
+ private final String timezone;
+ private final TransformExpressionKey transformExpressionKey;
+ private final Map<String, SupportedMetadataColumn> supportedMetadataColumns;
+ private final List<Object> udfFunctionInstances;
+ private final ExpressionEvaluator expressionEvaluator;
public ProjectionColumnProcessor(
PostTransformChangeInfo tableInfo,
@@ -60,11 +53,10 @@ public class ProjectionColumnProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
final List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
this.tableInfo = tableInfo;
this.projectionColumn = projectionColumn;
this.timezone = timezone;
- this.udfDescriptors = udfDescriptors;
this.supportedMetadataColumns = supportedMetadataColumns;
this.transformExpressionKey = generateTransformExpressionKey();
this.expressionEvaluator =
@@ -79,7 +71,7 @@ public class ProjectionColumnProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
return new ProjectionColumnProcessor(
tableInfo,
projectionColumn,
@@ -89,14 +81,10 @@ public class ProjectionColumnProcessor {
supportedMetadataColumns);
}
- public ProjectionColumn getProjectionColumn() {
- return projectionColumn;
- }
-
- public Object evaluate(
- BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
+ public Object evaluate(Object[] rowData, TransformContext context) {
try {
- return expressionEvaluator.evaluate(generateParams(record, epochTime, opType, meta));
+ Object[] params = generateParams(rowData, context);
+ return expressionEvaluator.evaluate(params);
} catch (InvocationTargetException e) {
throw new RuntimeException(
String.format(
@@ -110,63 +98,26 @@ public class ProjectionColumnProcessor {
}
}
- private Object[] generateParams(
- BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
+ private Object[] generateParams(Object[] rowData, TransformContext context) {
List<Object> params = new ArrayList<>();
- List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
// 1 - Add referenced columns
- RecordData.FieldGetter[] fieldGetters = tableInfo.getPreTransformedFieldGetters();
LinkedHashSet<String> originalColumnNames =
new LinkedHashSet<>(projectionColumn.getOriginalColumnNames());
- for (String originalColumnName : originalColumnNames) {
- switch (originalColumnName) {
- case MetadataColumns.DEFAULT_NAMESPACE_NAME:
- params.add(tableInfo.getNamespace());
- continue;
- case MetadataColumns.DEFAULT_SCHEMA_NAME:
- params.add(tableInfo.getSchemaName());
- continue;
- case MetadataColumns.DEFAULT_TABLE_NAME:
- params.add(tableInfo.getTableName());
- continue;
- case MetadataColumns.DEFAULT_DATA_EVENT_TYPE:
- params.add(opType);
- 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);
- if (column.getName().equals(originalColumnName)) {
- params.add(
- DataTypeConverter.convertToOriginal(
- fieldGetters[i].getFieldOrNull(record), column.getType()));
- argumentFound = true;
- break;
- }
- }
- if (!argumentFound) {
- throw new IllegalArgumentException(
- "Failed to evaluate argument " + originalColumnName);
- }
+ for (String columnName : originalColumnNames) {
+ params.add(
+ lookupObjectByName(
+ columnName,
+ tableInfo,
+ supportedMetadataColumns,
+ rowData,
+ null,
+ context));
}
// 2 - Add time-sensitive function arguments
params.add(timezone);
- params.add(epochTime);
+ params.add(context.epochTime);
// 3 - Add UDF function instances
params.addAll(udfFunctionInstances);
@@ -176,6 +127,7 @@ public class ProjectionColumnProcessor {
private TransformExpressionKey generateTransformExpressionKey() {
List<String> argumentNames = new ArrayList<>();
List<Class<?>> paramTypes = new ArrayList<>();
+
List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
String scriptExpression = projectionColumn.getScriptExpression();
Map<String, String> columnNameMap = projectionColumn.getColumnNameMap();
@@ -198,18 +150,20 @@ public class ProjectionColumnProcessor {
argumentNames.add(columnNameMap.get(col.f0));
paramTypes.add(col.f2);
});
- Stream.of(supportedMetadataColumns)
- .filter(col -> col.getName().equals(originalColumnName))
+
+ supportedMetadataColumns.entrySet().stream()
+ .filter(col -> col.getValue().getName().equals(originalColumnName))
.findFirst()
.ifPresent(
col -> {
- argumentNames.add(columnNameMap.get(col.getName()));
- paramTypes.add(col.getJavaClass());
+ argumentNames.add(columnNameMap.get(col.getValue().getName()));
+ paramTypes.add(col.getValue().getJavaClass());
});
}
argumentNames.add(JaninoCompiler.DEFAULT_TIME_ZONE);
paramTypes.add(String.class);
+
argumentNames.add(JaninoCompiler.DEFAULT_EPOCH_TIME);
paramTypes.add(Long.class);
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformContext.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformContext.java
new file mode 100644
index 000000000..61449739e
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformContext.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.operators.transform;
+
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/** Contextual information during Post-transform phase. */
+public class TransformContext {
+
+ public long epochTime;
+
+ public String opType;
+
+ public Map<String, String> meta;
+
+ /**
+ * Retrieve a corresponding object based on identifier name. The lookup order would be: <br>
+ * 1. Built-in metadata column names; <br>
+ * 2. Source-provided metadata column names; <br>
+ * 3. Calculated column names (which may shade original columns); <br>
+ * 4. Existing upstream column names. <br>
+ * If given name is nowhere to be found, an exception will be thrown.
+ */
+ public static Object lookupObjectByName(
+ String name,
+ PostTransformChangeInfo tableInfo,
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns,
+ Object[] preRow,
+ @Nullable Object[] postRow,
+ TransformContext context) {
+ switch (name) {
+ case MetadataColumns.DEFAULT_NAMESPACE_NAME:
+ return tableInfo.getNamespace();
+ case MetadataColumns.DEFAULT_SCHEMA_NAME:
+ return tableInfo.getSchemaName();
+ case MetadataColumns.DEFAULT_TABLE_NAME:
+ return tableInfo.getTableName();
+ case MetadataColumns.DEFAULT_DATA_EVENT_TYPE:
+ return context.opType;
+ }
+
+ // or source-provided metadata column
+ if (supportedMetadataColumns.containsKey(name)) {
+ return supportedMetadataColumns.get(name).read(context.meta);
+ }
+
+ if (postRow != null) {
+ // or a column that is presented in post-transform schema
+ @Nullable
+ Integer indexInPostTransformedSchema =
+ tableInfo.getPostTransformedSchemaFieldIndex(name);
+ if (indexInPostTransformedSchema != null) {
+ return postRow[indexInPostTransformedSchema];
+ }
+ }
+
+ // or a pre-transformed column that has been projected out
+ @Nullable
+ Integer indexInPreTransformedSchema = tableInfo.getPreTransformedSchemaFieldIndex(name);
+ if (indexInPreTransformedSchema != null) {
+ return preRow[indexInPreTransformedSchema];
+ }
+
+ throw new RuntimeException("Failed to lookup column name: " + name);
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java
index 585faf4b2..3feaa0822 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java
@@ -94,4 +94,20 @@ public class TransformFilter implements Serializable {
public boolean isValid() {
return !columnNames.isEmpty();
}
+
+ @Override
+ public String toString() {
+ return "TransformFilter{"
+ + "expression='"
+ + expression
+ + '\''
+ + ", scriptExpression='"
+ + scriptExpression
+ + '\''
+ + ", columnNames="
+ + columnNames
+ + ", columnNameMap="
+ + columnNameMap
+ + '}';
+ }
}
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 b0dd29724..483b07973 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
@@ -18,54 +18,66 @@
package org.apache.flink.cdc.runtime.operators.transform;
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;
import org.codehaus.janino.ExpressionEvaluator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import static org.apache.flink.cdc.runtime.operators.transform.TransformContext.lookupObjectByName;
import static org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns.METADATA_COLUMNS;
/** The processor of the transform filter. It processes the data change event of matched table. */
public class TransformFilterProcessor {
- private static final Logger LOG = LoggerFactory.getLogger(TransformFilterProcessor.class);
- private PostTransformChangeInfo tableInfo;
- private TransformFilter transformFilter;
- private String timezone;
- private TransformExpressionKey transformExpressionKey;
- private final transient List<Object> udfFunctionInstances;
- private transient ExpressionEvaluator expressionEvaluator;
+
+ private final boolean isNoOp;
+ private final PostTransformChangeInfo tableInfo;
+ private final TransformFilter transformFilter;
+ private final String timezone;
+ private final List<Object> udfFunctionInstances;
private final Map<String, SupportedMetadataColumn> supportedMetadataColumns;
- public TransformFilterProcessor(
+ private final TransformExpressionKey transformExpressionKey;
+ private final ExpressionEvaluator expressionEvaluator;
+
+ protected TransformFilterProcessor(
+ boolean isNoOp,
PostTransformChangeInfo tableInfo,
TransformFilter transformFilter,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
+ this.isNoOp = isNoOp;
this.tableInfo = tableInfo;
this.transformFilter = transformFilter;
this.timezone = timezone;
- this.supportedMetadataColumns = supportedMetadataColumns;
- this.transformExpressionKey = generateTransformExpressionKey();
this.udfFunctionInstances = udfFunctionInstances;
- this.expressionEvaluator =
- TransformExpressionCompiler.compileExpression(
- transformExpressionKey, udfDescriptors);
+ this.supportedMetadataColumns = supportedMetadataColumns;
+
+ if (isNoOp) {
+ this.transformExpressionKey = null;
+ this.expressionEvaluator = null;
+ } else {
+ this.transformExpressionKey = generateTransformExpressionKey();
+ this.expressionEvaluator =
+ TransformExpressionCompiler.compileExpression(
+ transformExpressionKey, udfDescriptors);
+ }
+ }
+
+ public static TransformFilterProcessor ofNoOp() {
+ return new TransformFilterProcessor(true, null, null, null, null, null, null);
}
public static TransformFilterProcessor of(
@@ -81,6 +93,7 @@ public class TransformFilterProcessor {
supportedMetadataColumn.getName(), supportedMetadataColumn);
}
return new TransformFilterProcessor(
+ false,
tableInfo,
transformFilter,
timezone,
@@ -89,11 +102,13 @@ public class TransformFilterProcessor {
supportedMetadataColumnsMap);
}
- public boolean process(
- BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
+ public boolean test(Object[] preRow, Object[] postRow, TransformContext context) {
+ if (isNoOp) {
+ return true;
+ }
+
try {
- return (Boolean)
- expressionEvaluator.evaluate(generateParams(record, epochTime, opType, meta));
+ return (Boolean) expressionEvaluator.evaluate(generateParams(preRow, postRow, context));
} catch (InvocationTargetException e) {
throw new RuntimeException(
String.format(
@@ -107,15 +122,27 @@ public class TransformFilterProcessor {
}
private Tuple2<List<String>, List<Class<?>>> generateArguments(boolean mapColumnNames) {
+ List<String> rawArgNames = new ArrayList<>();
List<String> argNames = new ArrayList<>();
List<Class<?>> argTypes = new ArrayList<>();
String expression = transformFilter.getExpression();
- List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
+
+ // Post-transformed columns comes in priority
+ List<Column> columns = new ArrayList<>(tableInfo.getPostTransformedSchema().getColumns());
+ {
+ Set<String> existingColumnNames =
+ new HashSet<>(tableInfo.getPostTransformedSchema().getColumnNames());
+ tableInfo.getPreTransformedSchema().getColumns().stream()
+ .filter(col -> !existingColumnNames.contains(col.getName()))
+ .forEach(columns::add);
+ }
+
Map<String, String> columnNameMap = transformFilter.getColumnNameMap();
LinkedHashSet<String> columnNames = new LinkedHashSet<>(transformFilter.getColumnNames());
for (String columnName : columnNames) {
for (Column column : columns) {
if (column.getName().equals(columnName)) {
+ rawArgNames.add(columnName);
argNames.add(mapColumnNames ? columnNameMap.get(columnName) : columnName);
argTypes.add(DataTypeConverter.convertOriginalClass(column.getType()));
break;
@@ -125,7 +152,8 @@ public class TransformFilterProcessor {
METADATA_COLUMNS.forEach(
col -> {
- if (expression.contains(col.f0) && !argNames.contains(col.f0)) {
+ if (expression.contains(col.f0) && !rawArgNames.contains(col.f0)) {
+ rawArgNames.add(col.f0);
argNames.add(mapColumnNames ? columnNameMap.get(col.f0) : col.f0);
argTypes.add(col.f2);
}
@@ -135,7 +163,8 @@ public class TransformFilterProcessor {
.keySet()
.forEach(
colName -> {
- if (expression.contains(colName) && !argNames.contains(colName)) {
+ if (expression.contains(colName) && !rawArgNames.contains(colName)) {
+ rawArgNames.add(colName);
argNames.add(mapColumnNames ? columnNameMap.get(colName) : colName);
argTypes.add(supportedMetadataColumns.get(colName).getJavaClass());
}
@@ -143,49 +172,25 @@ public class TransformFilterProcessor {
return Tuple2.of(argNames, argTypes);
}
- private Object[] generateParams(
- BinaryRecordData record, long epochTime, String opType, Map<String, String> meta) {
+ private Object[] generateParams(Object[] preRow, Object[] postRow, TransformContext context) {
List<Object> params = new ArrayList<>();
- List<Column> columns = tableInfo.getPreTransformedSchema().getColumns();
// 1 - Add referenced columns
Tuple2<List<String>, List<Class<?>>> args = generateArguments(false);
- RecordData.FieldGetter[] fieldGetters = tableInfo.getPreTransformedFieldGetters();
for (String columnName : args.f0) {
- switch (columnName) {
- case MetadataColumns.DEFAULT_NAMESPACE_NAME:
- params.add(tableInfo.getNamespace());
- continue;
- case MetadataColumns.DEFAULT_SCHEMA_NAME:
- params.add(tableInfo.getSchemaName());
- continue;
- case MetadataColumns.DEFAULT_TABLE_NAME:
- params.add(tableInfo.getTableName());
- continue;
- case MetadataColumns.DEFAULT_DATA_EVENT_TYPE:
- 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)) {
- params.add(
- DataTypeConverter.convertToOriginal(
- fieldGetters[i].getFieldOrNull(record), column.getType()));
- break;
- }
- }
+ params.add(
+ lookupObjectByName(
+ columnName,
+ tableInfo,
+ supportedMetadataColumns,
+ preRow,
+ postRow,
+ context));
}
// 2 - Add time-sensitive function arguments
params.add(timezone);
- params.add(epochTime);
+ params.add(context.epochTime);
// 3 - Add UDF function instances
params.addAll(udfFunctionInstances);
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjection.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjection.java
index 3d15e78aa..c318cd051 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjection.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjection.java
@@ -37,13 +37,8 @@ import java.util.Optional;
*/
public class TransformProjection implements Serializable {
private static final long serialVersionUID = 1L;
- private String projection;
- private List<ProjectionColumn> projectionColumns;
-
- public TransformProjection(String projection) {
- this.projection = projection;
- this.projectionColumns = new ArrayList<>();
- }
+ private final String projection;
+ private final List<ProjectionColumn> projectionColumns;
public TransformProjection(String projection, List<ProjectionColumn> projectionColumns) {
this.projection = projection;
@@ -58,10 +53,6 @@ public class TransformProjection implements Serializable {
return projectionColumns;
}
- public void setProjectionColumns(List<ProjectionColumn> projectionColumns) {
- this.projectionColumns = projectionColumns;
- }
-
public boolean isValid() {
return !StringUtils.isNullOrWhitespaceOnly(projection);
}
@@ -72,4 +63,15 @@ public class TransformProjection implements Serializable {
}
return Optional.of(new TransformProjection(projection, new ArrayList<>()));
}
+
+ @Override
+ public String toString() {
+ return "TransformProjection{"
+ + "projection='"
+ + getProjection()
+ + '\''
+ + ", projectionColumns="
+ + getProjectionColumns()
+ + '}';
+ }
}
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 8ac1cbe56..231e72875 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
@@ -17,24 +17,16 @@
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.schema.Schema;
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.parser.TransformParser;
-import org.apache.flink.cdc.runtime.typeutils.DataTypeConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -51,189 +43,73 @@ import java.util.stream.Collectors;
*/
public class TransformProjectionProcessor {
private static final Logger LOG = LoggerFactory.getLogger(TransformProjectionProcessor.class);
- private final PostTransformChangeInfo postTransformChangeInfo;
- private final TransformProjection transformProjection;
+
+ private final PostTransformChangeInfo changeInfo;
+ private final String projectionExpression;
private final String timezone;
- private final List<ProjectionColumnProcessor> cachedProjectionColumnProcessors;
private final List<UserDefinedFunctionDescriptor> udfDescriptors;
- private final transient List<Object> udfFunctionInstances;
+ private final List<Object> udfFunctionInstances;
+ private final List<ProjectionColumnProcessor> columnProcessors;
+ private final SupportedMetadataColumn[] supportedMetadataColumns;
+ private final Map<String, SupportedMetadataColumn> supportedMetadataColumnsMap;
public TransformProjectionProcessor(
- PostTransformChangeInfo postTransformChangeInfo,
- TransformProjection transformProjection,
+ PostTransformChangeInfo changeInfo,
+ String projectionExpression,
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- final List<Object> udfFunctionInstances,
+ List<Object> udfFunctionInstances,
SupportedMetadataColumn[] supportedMetadataColumns) {
- this.postTransformChangeInfo = postTransformChangeInfo;
- this.transformProjection = transformProjection;
+ this.changeInfo = changeInfo;
+ this.projectionExpression = projectionExpression;
this.timezone = timezone;
this.udfDescriptors = udfDescriptors;
this.udfFunctionInstances = udfFunctionInstances;
+ this.supportedMetadataColumns = supportedMetadataColumns;
- // Create cached projection column processors after setting all other fields.
- this.cachedProjectionColumnProcessors =
- cacheProjectionColumnProcessors(
- postTransformChangeInfo, transformProjection, supportedMetadataColumns);
- }
-
- public boolean hasTableInfo() {
- return this.postTransformChangeInfo != null;
- }
-
- public TransformProjection getTransformProjection() {
- return transformProjection;
- }
-
- public static TransformProjectionProcessor of(
- PostTransformChangeInfo tableInfo,
- TransformProjection transformProjection,
- String timezone,
- List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- return new TransformProjectionProcessor(
- tableInfo,
- transformProjection,
- timezone,
- udfDescriptors,
- udfFunctionInstances,
- supportedMetadataColumns);
+ // Construct a mapping table ad-hoc to accelerate looking-up
+ Map<String, SupportedMetadataColumn> supportedMetadataColumnsMap = new HashMap<>();
+ for (SupportedMetadataColumn supportedMetadataColumn : supportedMetadataColumns) {
+ supportedMetadataColumnsMap.put(
+ supportedMetadataColumn.getName(), supportedMetadataColumn);
+ }
+ this.supportedMetadataColumnsMap = supportedMetadataColumnsMap;
+ this.columnProcessors = createProjectionColumnProcessors();
}
- public static TransformProjectionProcessor of(
- TransformProjection transformProjection,
- String timezone,
- List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- return new TransformProjectionProcessor(
- null,
- transformProjection,
- timezone,
- udfDescriptors,
- udfFunctionInstances,
- supportedMetadataColumns);
+ public Object[] project(Object[] rowData, TransformContext context) {
+ return columnProcessors.stream()
+ .map(processor -> processor.evaluate(rowData, context))
+ .toArray();
}
- public static TransformProjectionProcessor of(
- TransformProjection transformProjection,
- List<UserDefinedFunctionDescriptor> udfDescriptors,
- List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- return new TransformProjectionProcessor(
- null,
- transformProjection,
- null,
- udfDescriptors,
- udfFunctionInstances,
- supportedMetadataColumns);
- }
+ private List<ProjectionColumnProcessor> createProjectionColumnProcessors() {
+ Preconditions.checkNotNull(
+ changeInfo,
+ "Projection column processors could only be created if changeInfo is available.");
- public Schema processSchema(Schema schema, SupportedMetadataColumn[] supportedMetadataColumns) {
List<ProjectionColumn> projectionColumns =
TransformParser.generateProjectionColumns(
- transformProjection.getProjection(),
- schema.getColumns(),
+ projectionExpression,
+ changeInfo.getPreTransformedSchema().getColumns(),
udfDescriptors,
supportedMetadataColumns);
- transformProjection.setProjectionColumns(projectionColumns);
- Set<String> primaryKeys = new HashSet<>(schema.primaryKeys());
- return schema.copy(
- projectionColumns.stream()
- .map(ProjectionColumn::getColumn)
- .map(column -> setPkNonNull(primaryKeys, column))
- .collect(Collectors.toList()));
- }
-
- private static Column setPkNonNull(Set<String> primaryKeys, Column column) {
- if (primaryKeys.contains(column.getName())) {
- return column.copy(column.getType().notNull());
- } else {
- return column;
- }
- }
-
- public BinaryRecordData processData(
- BinaryRecordData payload, long epochTime, String opType, Map<String, String> meta) {
- List<Object> valueList = new ArrayList<>();
- List<Column> columns = postTransformChangeInfo.getPostTransformedSchema().getColumns();
-
- for (int i = 0; i < columns.size(); i++) {
- ProjectionColumnProcessor projectionColumnProcessor =
- cachedProjectionColumnProcessors.get(i);
- if (projectionColumnProcessor != null) {
- ProjectionColumn projectionColumn = projectionColumnProcessor.getProjectionColumn();
- valueList.add(
- DataTypeConverter.convert(
- projectionColumnProcessor.evaluate(
- payload, epochTime, opType, meta),
- projectionColumn.getDataType()));
- } else {
- Column column = columns.get(i);
- valueList.add(
- getValueFromBinaryRecordData(
- column.getName(),
- column.getType(),
- payload,
- postTransformChangeInfo.getPreTransformedSchema().getColumns(),
- postTransformChangeInfo.getPreTransformedFieldGetters()));
- }
- }
-
- return postTransformChangeInfo
- .getRecordDataGenerator()
- .generate(valueList.toArray(new Object[0]));
- }
- private Object getValueFromBinaryRecordData(
- String columnName,
- DataType expectedType,
- BinaryRecordData binaryRecordData,
- List<Column> columns,
- RecordData.FieldGetter[] fieldGetters) {
- for (int i = 0; i < columns.size(); i++) {
- if (columnName.equals(columns.get(i).getName())) {
- return DataTypeConverter.convert(
- fieldGetters[i].getFieldOrNull(binaryRecordData), expectedType);
- }
- }
- return null;
- }
-
- private List<ProjectionColumnProcessor> cacheProjectionColumnProcessors(
- PostTransformChangeInfo tableInfo,
- TransformProjection transformProjection,
- SupportedMetadataColumn[] supportedMetadataColumns) {
- List<ProjectionColumnProcessor> cachedProjectionColumnProcessors = new ArrayList<>();
- if (!hasTableInfo()) {
- return cachedProjectionColumnProcessors;
- }
-
- for (Column column : tableInfo.getPostTransformedSchema().getColumns()) {
- ProjectionColumn matchedProjectionColumn = null;
- for (ProjectionColumn projectionColumn : transformProjection.getProjectionColumns()) {
- if (column.getName().equals(projectionColumn.getColumnName())
- && projectionColumn.isValidTransformedProjectionColumn()) {
- matchedProjectionColumn = projectionColumn;
- break;
- }
- }
-
- cachedProjectionColumnProcessors.add(
- Optional.ofNullable(matchedProjectionColumn)
- .map(
- col ->
- ProjectionColumnProcessor.of(
- tableInfo,
- col,
- timezone,
- udfDescriptors,
- udfFunctionInstances,
- supportedMetadataColumns))
- .orElse(null));
- }
-
- return cachedProjectionColumnProcessors;
+ List<ProjectionColumnProcessor> columnProcessors =
+ projectionColumns.stream()
+ .map(
+ column ->
+ ProjectionColumnProcessor.of(
+ changeInfo,
+ column,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumnsMap))
+ .collect(Collectors.toList());
+
+ LOG.info("Successfully created projection column processors cache.");
+ LOG.info("Cached results: {}", columnProcessors);
+ return columnProcessors;
}
}
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 d4459e5dd..5c444fe42 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
@@ -24,8 +24,6 @@ import javax.annotation.Nullable;
import java.io.Serializable;
-import static org.apache.flink.cdc.runtime.parser.TransformParser.normalizeFilter;
-
/** A rule defining pre-transformations where filtered rows and irrelevant columns are removed. */
public class TransformRule implements Serializable {
private static final long serialVersionUID = 1L;
@@ -50,7 +48,7 @@ public class TransformRule implements Serializable {
SupportedMetadataColumn[] supportedMetadataColumns) {
this.tableInclusions = tableInclusions;
this.projection = StringUtils.isNullOrWhitespaceOnly(projection) ? "*" : projection;
- this.filter = normalizeFilter(projection, filter);
+ this.filter = filter;
this.primaryKey = primaryKey;
this.partitionKey = partitionKey;
this.tableOption = tableOption;
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
index 9acdad351..4e792e871 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cdc.common.annotation.Internal;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.udf.UserDefinedFunction;
@@ -46,6 +47,11 @@ public class UserDefinedFunctionDescriptor implements Serializable {
this(name, classpath, new HashMap<>());
}
+ public UserDefinedFunctionDescriptor(
+ Tuple3<String, String, Map<String, String>> descriptorTuple) {
+ this(descriptorTuple.f0, descriptorTuple.f1, descriptorTuple.f2);
+ }
+
public UserDefinedFunctionDescriptor(
String name, String classpath, Map<String, String> parameters) {
this.name = name;
@@ -60,7 +66,9 @@ public class UserDefinedFunctionDescriptor implements Serializable {
// into UserDefinedFunction interface, thus the provided UDF classes
// might not be compatible with the interface definition in CDC common.
returnTypeHint =
- (DataType) clazz.getMethod("getReturnType").invoke(clazz.newInstance());
+ (DataType)
+ clazz.getMethod("getReturnType")
+ .invoke(clazz.getConstructor().newInstance());
} else {
returnTypeHint = null;
}
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 1fc3101ae..65c8dfd9c 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
@@ -83,7 +83,6 @@ import java.util.Objects;
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;
@@ -128,6 +127,8 @@ public class TransformParser {
Class<?> clazz = Class.forName(udf.getClasspath());
SqlReturnTypeInference returnTypeInference;
ScalarFunction function = ScalarFunctionImpl.create(clazz, "eval");
+ Preconditions.checkNotNull(
+ function, "UDF function must provide at least one `eval` method.");
if (udf.getReturnTypeHint() != null) {
// This UDF has return type hint annotation
returnTypeInference =
@@ -186,7 +187,7 @@ public class TransformParser {
}
public static SqlSelect parseSelect(String statement) {
- SqlNode sqlNode = null;
+ SqlNode sqlNode;
try {
sqlNode = getCalciteParser(statement).parseQuery();
} catch (SqlParseException e) {
@@ -285,17 +286,19 @@ public class TransformParser {
expandWildcard(sqlSelect, columns);
RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors, supportedMetadataColumns);
- Map<String, RelDataType> relDataTypeMap =
+ RelDataType[] relDataTypes =
relNode.getRowType().getFieldList().stream()
- .collect(
- Collectors.toMap(
- RelDataTypeField::getName, RelDataTypeField::getType));
+ .map(RelDataTypeField::getType)
+ .toArray(RelDataType[]::new);
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()) {
+ SqlNodeList selectExpressionList = sqlSelect.getSelectList();
+ for (int i = 0; i < selectExpressionList.size(); i++) {
+ SqlNode sqlNode = selectExpressionList.get(i);
+ RelDataType relDataType = relDataTypes[i];
ProjectionColumn projectionColumn;
// A projection column could be <EXPR> AS <IDENTIFIER>...
@@ -330,7 +333,7 @@ public class TransformParser {
identifierExprNode.names.get(identifierExprNode.names.size() - 1);
projectionColumn =
resolveProjectionColumnFromIdentifier(
- relDataTypeMap,
+ relDataType,
originalColumnMap,
originalName,
columnName,
@@ -342,7 +345,7 @@ public class TransformParser {
ProjectionColumn.ofCalculated(
columnName,
DataTypeConverter.convertCalciteRelDataTypeToDataType(
- relDataTypeMap.get(columnName)),
+ relDataType),
exprNode.toString(),
JaninoCompiler.translateSqlNodeToJaninoExpression(
exprNode, udfDescriptors, columnNameMap),
@@ -356,7 +359,7 @@ public class TransformParser {
String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
projectionColumn =
resolveProjectionColumnFromIdentifier(
- relDataTypeMap,
+ relDataType,
originalColumnMap,
columnName,
columnName,
@@ -384,7 +387,7 @@ public class TransformParser {
* column or a metadata column).
*/
public static ProjectionColumn resolveProjectionColumnFromIdentifier(
- Map<String, RelDataType> relDataTypeMap,
+ RelDataType relDataType,
Map<String, Column> originalColumnMap,
String identifier,
String projectedColumnName,
@@ -396,9 +399,7 @@ public class TransformParser {
return ProjectionColumn.ofCalculated(
projectedColumnName,
// Metadata columns should never be null
- DataTypeConverter.convertCalciteRelDataTypeToDataType(
- relDataTypeMap.get(projectedColumnName))
- .notNull(),
+ DataTypeConverter.convertCalciteRelDataTypeToDataType(relDataType).notNull(),
identifier,
columnNameMap.get(identifier),
Collections.singletonList(identifier),
@@ -462,7 +463,7 @@ public class TransformParser {
}
columnNames.add(columnName);
} else {
- throw new ParseException("Unrecognized projection: " + sqlBasicCall.toString());
+ throw new ParseException("Unrecognized projection: " + sqlBasicCall);
}
} else if (sqlNode instanceof SqlIdentifier) {
String columnName = sqlNode.toString();
@@ -561,77 +562,6 @@ public class TransformParser {
return parseSelect(statement.toString());
}
- public static SqlNode rewriteExpression(SqlNode sqlNode, Map<String, SqlNode> replaceMap) {
- if (sqlNode instanceof SqlCall) {
- SqlCall sqlCall = (SqlCall) sqlNode;
-
- List<SqlNode> operands = sqlCall.getOperandList();
- IntStream.range(0, sqlCall.operandCount())
- .forEach(
- i ->
- sqlCall.setOperand(
- i, rewriteExpression(operands.get(i), replaceMap)));
- return sqlCall;
- } else if (sqlNode instanceof SqlIdentifier) {
- SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode;
- if (sqlIdentifier.names.size() == 1) {
- String name = sqlIdentifier.names.get(0);
- if (replaceMap.containsKey(name)) {
- return replaceMap.get(name);
- }
- }
- return sqlIdentifier;
- } else if (sqlNode instanceof SqlNodeList) {
- SqlNodeList sqlNodeList = (SqlNodeList) sqlNode;
- IntStream.range(0, sqlNodeList.size())
- .forEach(
- i ->
- sqlNodeList.set(
- i, rewriteExpression(sqlNodeList.get(i), replaceMap)));
- return sqlNodeList;
- } else {
- return sqlNode;
- }
- }
-
- // Filter expression might hold reference to a calculated column, which causes confusion about
- // the sequence of projection and filtering operations. This function rewrites filtering about
- // calculated columns to circumvent this problem.
- public static String normalizeFilter(String projection, String filter) {
- if (isNullOrWhitespaceOnly(projection) || isNullOrWhitespaceOnly(filter)) {
- return filter;
- }
-
- SqlSelect sqlSelect = parseProjectionExpression(projection);
- if (sqlSelect.getSelectList().isEmpty()) {
- return filter;
- }
-
- Map<String, SqlNode> calculatedExpression = new HashMap<>();
- for (SqlNode sqlNode : sqlSelect.getSelectList()) {
- if (sqlNode instanceof SqlBasicCall) {
- SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
- if (SqlKind.AS.equals(sqlBasicCall.getOperator().kind)) {
- List<SqlNode> operandList = sqlBasicCall.getOperandList();
- if (operandList.size() == 2) {
- SqlIdentifier alias = (SqlIdentifier) operandList.get(1);
- String name = alias.names.get(alias.names.size() - 1);
- SqlNode expression = operandList.get(0);
- calculatedExpression.put(name, expression);
- }
- }
- }
- }
-
- SqlNode sqlFilter = parseFilterExpression(filter).getWhere();
- sqlFilter = rewriteExpression(sqlFilter, calculatedExpression);
- if (sqlFilter != null) {
- return sqlFilter.toString();
- } else {
- return filter;
- }
- }
-
public static boolean hasAsterisk(@Nullable String projection) {
if (isNullOrWhitespaceOnly(projection)) {
// Providing an empty projection expression is equivalent to writing `*` explicitly.
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 351e75ecf..0063d6561 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
@@ -572,25 +572,6 @@ class TransformParserTest {
.isExactlyInstanceOf(ParseException.class);
}
- @Test
- void testNormalizeFilter() {
- Assertions.assertThat(TransformParser.normalizeFilter("a, b, c, d", "a > 0 and b > 0"))
- .isEqualTo("`a` > 0 AND `b` > 0");
- Assertions.assertThat(TransformParser.normalizeFilter("a, b, c, d", null)).isNull();
- Assertions.assertThat(
- TransformParser.normalizeFilter(
- "abs(a) as cal_a, char_length(b) as cal_b, c, d",
- "a > 4 and cal_a > 8 and cal_b < 17 and c != d"))
- .isEqualTo("`a` > 4 AND ABS(`a`) > 8 AND CHAR_LENGTH(`b`) < 17 AND `c` <> `d`");
-
- Assertions.assertThat(
- TransformParser.normalizeFilter(
- "x, y, z, 1 - x as u, 1 - y as v, 1 - z as w",
- "concat(u, concat(v, concat(w, x), y), z) != 10"))
- .isEqualTo(
- "`concat`(1 - `x`, `concat`(1 - `y`, `concat`(1 - `z`, `x`), `y`), `z`) <> 10");
- }
-
@Test
void testTranslateUdfFilterToJaninoExpression() {
testFilterExpressionWithUdf(