Flink CDC Transform module - Realtime Compute for Apache Flink

Transform rule parameters

The Transform module supports the direct manipulation of data columns. You can delete or extend existing columns and filter unwanted data during data synchronization. Use the following parameters to define transform rules:

Parameter	Description	Required	Notes
`source-table`	Specifies the upstream table to transform.	Yes	Supports regular expressions.
`projection`	Specifies the projection rule for the upstream table. This determines all data columns after the transformation.	No	The syntax is similar to the SQL SELECT statement. If this parameter is not specified, no columns are appended or deleted. For more information, see Data screening. For information about available built-in functions, see Flink CDC built-in functions.
`filter`	The rule for row filtering.	No	The syntax is similar to the SQL WHERE statement. If this parameter is not specified, no rows are filtered.
`primary-keys`	Specifies the primary key columns after the transformation.	No	If this parameter is not specified, the primary key definition from the original schema is retained. Primary keys are separated by commas (`,`). Important By default, you cannot delete primary key columns from the upstream. Add the `transform.allow.trimming.pk-columns` parameter to the `pipeline:` configuration to reduce the number of upstream primary key columns. When you define custom primary key columns, ensure that the incoming data from the upstream table complies with the PRIMARY KEY constraint. We recommend that you include the primary key columns of the upstream table in the custom primary key columns to prevent data disorder issues during cross-partition writes.
`partition-keys`	Specifies the list of partition key columns after the transformation.	No	If this parameter is not specified, the partition key definition from the original schema is retained. Partition keys are separated by commas (`,`). Important When you define custom partition key columns, ensure that the incoming data from the upstream table complies with the PRIMARY KEY constraint to prevent data disorder issues during cross-partition writes.
`table-options`	Passes extra configuration information to the sink.	No	A list of optional properties, such as the number of buckets for a Paimon sink or comments. Configuration items are separated by commas (`,`). The key and value of each item are separated by an equal sign (`=`). Example configuration: `key1=value1,key2=value2`
`description`	The description of the transform rule.	No	None.
`converter-after-transform`	A converter that performs additional processing on data after the transformation is complete.	No	For more information, see Converter after Transform.

Notes

After you modify a statement in the transform module, you must restart the job from a stateless initial state.
Typically, `projection` and `filter` statements do not need to be enclosed in quotation marks.
```
transform:
  - projection: a, b, c
    # Equivalent to
  - projection: "a, b, c"
```
However, if a projection expression starts with a special character, such as * or ', the entire expression might not be parsed as a valid YAML string literal. In this case, you must manually enclose the entire expression in single (') or double (") quotation marks, or use a backslash (\) to escape the character.
```
transform:
  - projection: *, 42      # Not valid YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```
Starting from Ververica Runtime (VVR) 11.6, you can purge the primary key settings inferred from the upstream and convert the table to one without a primary key. The syntax is as follows:
```
transform:
  - source-table: db.tbl
    primary-keys: '' # Clears the PRIMARY KEY setting for the db.tbl table
```

Data screening

The Transform module uses SQL-like syntax to define data screening (projection) rules. These rules allow you to select specific columns, add computed columns, and add metadata columns.

Column pruning

To extract specific columns from the source table and synchronize them to the downstream, you can specify the desired columns in the projection rule. Columns that are not specified are not sent to the downstream.

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 is pruned

Important

Pruning columns can cause the upstream and downstream table schemas to become inconsistent if the upstream table schema changes.

Wildcard characters

To send all columns from the source table, including any new columns that are appended later, to the downstream without modification, you can use the asterisk (*) wildcard character in the projection rule.

Note

If a projection rule does not use a wildcard character (*), the resulting schema is fixed. The schema remains consistent with the version specified in the projection rule, and schema evolution cannot be synchronized.

For example, *, 'extras' AS extras indicates that an extra column is appended after the columns from the upstream schema. Upstream schema evolution is still synchronized to the downstream.

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

Computed columns

You can use the <Expression> AS <ColName> syntax in a projection rule to add a computed column. The expression is evaluated for each record from the upstream, and the result populates the corresponding column.

Note

The expression for a computed column cannot reference the value of another computed column, even if the referenced column appears before it. For example, a, b AS c, c AS d is not a valid expression.

For example, when the upstream table `db.tbl` sends the data record [+I, id = 1], the record is transformed into the data row [+I, id = 1, inc_id = 2] and then sent to the downstream.

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

Metadata columns

When you write projection rules, you can use the following predefined metadata columns in the same way as regular data columns:

Important

Do not define regular data columns with the same names as metadata columns.

The following metadata columns apply to all connectors.

Metadata column name	Data type	Description
`__namespace_name__`	String	The namespace name of the source table for this change record.
`__schema_name__`	String	The schema name of the source table for this change record.
`__table_name__`	String	The table name of the source table for this change record.
`__data_event_type__`	String	The operation type for this change record (`+I`, `-U`, `+U`, or `-D`). Important A CDC event always packages the Update Before and Update After records for an update into a single event. Therefore, the content of `__data_event_type__` is `-U` and `+U` respectively within the same update event. Do not use it as a primary key.

For example, you can write the fully qualified name of the upstream table into a computed column and send it to the downstream.

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

The following table shows the mapping relationships between database connectors and the Namespace, Schema, and Table names.

Database type	Namespace name	Schema name	Table name
MySQL	-	Database	Table
Kafka	-	-	Topic
SLS	-	Project	LogStore
MongoDB	-	Database	Collection
Paimon	-	Database	Table
Hologres	-	Schema	Table
StarRocks	-	Database	Table
Doris	-	Database	Table
Postgres	Database Note This is effective only when the `table-id.include-database` parameter is enabled.	Schema	Table

Data filtering

The Transform module uses SQL-like syntax to define row filtering rules.

The filter rule must be an expression that evaluates to a BOOLEAN value. It can reference any column from the source table and any computed columns.

If a change record matches a transform rule with a filter, and the filter expression evaluates to FALSE, the data row is not sent to the downstream.

Note

If you use a computed column in a projection rule to overwrite an existing upstream column, the filter expression references the computed column.

For example, the following transform rule is valid:

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

The id referenced in the filter expression is the computed column that was converted to the VARCHAR type.

Advanced configuration rules

Reference non-pruned columns and metadata

Example 1: Filter based on a pruned column

Assume the upstream table schema is [INT a, INT b, BOOLEAN c]. To output columns a and b and keep only the rows where c is true, you can use the following configuration rule:

transform:
  - source-table: ...
    projection: a, b
    filter: c

Example 2: Filter based on metadata columns

You can use metadata columns directly as filter conditions without explicitly defining them in the projection. For example, you can filter out change data of the DELETE type.

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

Overwrite existing columns

In the projection, you can define a field with the same name as an upstream column to overwrite that column's value or type. This method ensures that schema evolution is synchronized correctly.

Example: Force type conversion

Assume the upstream table schema is [INT a, INT b, BOOLEAN c]. To keep the column names unchanged but force the conversion of column a to a string type, you can use the following configuration:

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

The downstream table schema becomes [STRING a, INT b, BOOLEAN c]. The original column a is overwritten by the new definition.

Reuse computed columns in filter conditions

A `filter` can directly reference the alias of a computed column defined in a `projection`.

Example: Reference a computed result

You can define a new column d in the projection and use it directly in the filter.

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

This configuration has the same effect as the following one but is more readable:

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Converter after Transform

converter-after-transform is used to process data changes after all transform rules have been applied. You can specify multiple converters by separating them with commas (,). The converters are applied in the order they are listed. The following configuration values are supported.

Converter name	Feature	Supported versions
SOFT_DELETE	Converts delete changes to inserts.	VVR 8.0.11 and later.
FIELD_NAME_LOWER_CASE	Converts all table field names to lowercase.	VVR 11.1 and later.

Soft delete

The SOFT_DELETE converter works with the __data_event_type__ metadata column to implement logical deletes. This means that data is not physically deleted in the downstream system. For example, the following transform configuration implements a logical delete. A delete operation is converted into an insert, and the op_type for the corresponding data is updated to -D to indicate the deletion.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE