Definition

Transform module helps users delete and expand data columns based on the data columns in the table.
What's more, it also helps users filter some unnecessary data during the synchronization process.

Parameters

To describe a transform rule, the following parameters can be used:

Parameter	Meaning	Optional/Required
source-table	Source table id, supports regular expressions	required
projection	Projection rule, supports syntax similar to the select clause in SQL	optional
filter	Filter rule, supports syntax similar to the where clause in SQL	optional
primary-keys	Sink table primary keys, separated by commas	optional
partition-keys	Sink table partition keys, separated by commas	optional
table-options	used to the configure table creation statement when automatically creating tables	optional
table-options.delimiter	delimiter for table-options key-value pairs, default is ,	optional
converter-after-transform	used to add a converter to change DataChangeEvent after transform	optional
description	Transform rule description	optional

Multiple rules can be declared in one single pipeline YAML file.

converter-after-transform

converter-after-transform is used to change the DataChangeEvent after other transform. The available values of this options are as follows.

• SOFT_DELETE: The delete event will be converted as an insert event. This converter should be used together with the metadata __data_event_type__. Then you can implement the soft delete.

For example, the following transform will not delete data when the delete event happens. Instead it will update the column op_type to -D in sink and transform it to an insert record.

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

Metadata Fields

Fields definition

There are some hidden columns used to access metadata information. They will only take effect when explicitly referenced in the transform rules.

Field	Data Type	Description
namespace_name	String	Name of the namespace that contains the row.
schema_name	String	Name of the schema that contains the row.
table_name	String	Name of the table that contains the row.
data_event_type	String	Operation type of data change event.

Besides these fields, pipeline connectors could parse more metadata and put them in the meta map of the DataChangeEvent. These metadata could be accessed in the transform module. For example, MySQL pipeline connector could parse op_ts and use it in the transform module.

source:
  type: mysql
  hostname: localhost
  port: 3306
  username: testuser
  password: testpwd
  tables: testdb.customer
  server-id: 5400-5404
  server-time-zone: UTC
  metadata.list: op_ts
  
transform:
  - source-table: testdb.customer
    projection: \*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type, op_ts AS opts
  
sink:
  type: values

Metadata relationship

Type	Namespace	SchemaName	Table
JDBC	Catalog	Schema	Table
Debezium	Catalog	Schema	Table
MySQL	Database	-	Table
Postgres	Database	Schema	Table
Oracle	-	Schema	Table
Microsoft SQL Server	Database	Schema	Table
StarRocks	Database	-	Table
Doris	Database	-	Table

Functions

Flink CDC uses Calcite to parse expressions and Janino script to evaluate expressions with function call.

Comparison Functions

Function	Janino Code	Description
value1 = value2	valueEquals(value1, value2)	Returns TRUE if value1 is equal to value2; returns FALSE if value1 or value2 is NULL.
value1 <> value2	!valueEquals(value1, value2)	Returns TRUE if value1 is not equal to value2; returns FALSE if value1 or value2 is NULL.
value1 > value2	greaterThan(value1, value2)	Returns TRUE if value1 is greater than value2; returns FALSE if value1 or value2 is NULL.
value1 >= value2	greaterThanOrEqual(value1, value2)	Returns TRUE if value1 is greater than or equal to value2; returns FALSE if value1 or value2 is NULL.
value1 < value2	lessThan(value1, value2)	Returns TRUE if value1 is less than value2; returns FALSE if value1 or value2 is NULL.
value1 <= value2	lessThanOrEqual(value1, value2)	Returns TRUE if value1 is less than or equal to value2; returns FALSE if value1 or value2 is NULL.
value IS NULL	isNull(value)	Returns TRUE if value is NULL.
value IS NOT NULL	isNotNull(value)	Returns TRUE if value is not NULL.
value1 IS DISTINCT FROM value2	isDistinctFrom(value1, value2)	Returns TRUE if value1 and value2 are distinct. NULL values are compared as values and never return NULL.
value1 IS NOT DISTINCT FROM value2	isNotDistinctFrom(value1, value2)	Returns TRUE if value1 and value2 are not distinct. NULL values are compared as values and never return NULL.
value1 BETWEEN value2 AND value3	betweenAsymmetric(value1, value2, value3)	Returns TRUE if value1 is greater than or equal to value2 and less than or equal to value3.
value1 NOT BETWEEN value2 AND value3	notBetweenAsymmetric(value1, value2, value3)	Returns TRUE if value1 is less than value2 or greater than value3.
string1 LIKE string2	like(string1, string2)	Returns TRUE if string1 matches Java regular expression string2 using substring matching.
string1 NOT LIKE string2	notLike(string1, string2)	Returns TRUE if string1 does not match Java regular expression string2 using substring matching.
string1 LIKE string2 ESCAPE string3	like(string1, string2, string3)	Returns TRUE if string1 matches SQL LIKE pattern string2. % matches zero or more characters, _ matches one character, and string3 escapes wildcard characters. Returns NULL if any argument is NULL.
string1 NOT LIKE string2 ESCAPE string3	notLike(string1, string2, string3)	Returns TRUE if string1 does not match SQL LIKE pattern string2. Returns NULL if any argument is NULL.
string1 SIMILAR TO string2	similarTo(string1, string2)	Returns TRUE if string1 matches SQL SIMILAR TO pattern string2. Returns NULL if any argument is NULL.
string1 NOT SIMILAR TO string2	notSimilarTo(string1, string2)	Returns TRUE if string1 does not match SQL SIMILAR TO pattern string2. Returns NULL if any argument is NULL.
string1 SIMILAR TO string2 ESCAPE string3	similarTo(string1, string2, string3)	Returns TRUE if string1 matches SQL SIMILAR TO pattern string2 using string3 as the escape character. Returns NULL if any argument is NULL.
string1 NOT SIMILAR TO string2 ESCAPE string3	notSimilarTo(string1, string2, string3)	Returns TRUE if string1 does not match SQL SIMILAR TO pattern string2 using string3 as the escape character. Returns NULL if any argument is NULL.
value1 IN (value2 [, value3]* )	in(value1, value2 [, value3]*)	Returns TRUE if value1 exists in the given list (value2, value3, …).
value1 NOT IN (value2 [, value3]* )	notIn(value1, value2 [, value3]*)	Returns TRUE if value1 does not exist in the given list (value2, value3, …).

Logical Functions

Logical functions follow SQL three-valued logic for nullable BOOLEAN values. AND and OR short-circuit the right operand when the left operand determines the result. Generated Janino code may use native operators, conditional expressions, or lazy function calls depending on operand nullability.

Function	Janino Code	Description
boolean1 OR boolean2	short-circuit OR expression	Returns TRUE if either value is TRUE; returns NULL if neither value is TRUE and at least one value is NULL.
boolean1 AND boolean2	short-circuit AND expression	Returns FALSE if either value is FALSE; returns NULL if neither value is FALSE and at least one value is NULL.
NOT boolean	not(boolean)	Returns TRUE if boolean is FALSE; returns FALSE if boolean is TRUE; returns NULL if boolean is NULL.
boolean IS FALSE	isFalse(boolean)	Returns TRUE if boolean is FALSE; returns FALSE if boolean is TRUE or NULL.
boolean IS NOT FALSE	isNotFalse(boolean)	Returns TRUE if boolean is TRUE or NULL; returns FALSE if boolean is FALSE.
boolean IS TRUE	isTrue(boolean)	Returns TRUE if boolean is TRUE; returns FALSE if boolean is FALSE or NULL.
boolean IS NOT TRUE	isNotTrue(boolean)	Returns TRUE if boolean is FALSE or NULL; returns FALSE if boolean is TRUE.
boolean IS UNKNOWN	isNull(boolean)	Returns TRUE if boolean is NULL.
boolean IS NOT UNKNOWN	isNotNull(boolean)	Returns TRUE if boolean is not NULL.

Arithmetic Functions

Function	Janino Code	Description
numeric1 + numeric2	numeric1 + numeric2	Returns NUMERIC1 plus NUMERIC2.
numeric1 - numeric2	numeric1 - numeric2	Returns NUMERIC1 minus NUMERIC2.
numeric1 * numeric2	numeric1 * numeric2	Returns NUMERIC1 multiplied by NUMERIC2.
numeric1 / numeric2	numeric1 / numeric2	Returns NUMERIC1 divided by NUMERIC2.
numeric1 % numeric2	numeric1 % numeric2	Returns the remainder (modulus) of numeric1 divided by numeric2.
ABS(numeric)	abs(numeric)	Returns the absolute value of numeric.
CEIL(numeric) CEILING(numeric)	ceil(numeric)	Rounds numeric up, and returns the smallest number that is greater than or equal to numeric.
FLOOR(numeric)	floor(numeric)	Rounds numeric down, and returns the largest number that is less than or equal to numeric.
ROUND(numeric, int)	round(numeric)	Returns a number rounded to INT decimal places for NUMERIC.
UUID()	uuid()	Returns an UUID (Universally Unique Identifier) string (e.g., "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 4122 type 4 (pseudo randomly generated) UUID.

String Functions

Function	Janino Code	Description
string1 || string2	concat(string1, string2)	Returns the concatenation of STRING1 and STRING2.
CHAR_LENGTH(string)	charLength(string)	Returns the number of characters in STRING.
UPPER(string)	upper(string)	Returns string in uppercase.
LOWER(string)	lower(string)	Returns string in lowercase.
TRIM(string1)	trim('BOTH',string1)	Returns a string that removes whitespaces at both sides.
LTRIM(string[, trimString])	ltrim(string[, trimString])	Returns a string with leading characters in trimString removed. Whitespace is removed by default.
RTRIM(string[, trimString])	rtrim(string[, trimString])	Returns a string with trailing characters in trimString removed. Whitespace is removed by default.
BTRIM(string[, trimString])	btrim(string[, trimString])	Returns a string with leading and trailing characters in trimString removed. Whitespace is removed by default.
REGEXP_REPLACE(string1, string2, string3)	regexpReplace(string1, string2, string3)	Returns a string from STRING1 with all the substrings that match a regular expression STRING2 consecutively being replaced with STRING3. E.g., 'foobar'.regexpReplace('oo|ar', '') returns "fb".
REGEXP_EXTRACT(string, regex[, extractIndex])	regexpExtract(string, regex[, extractIndex])	Returns the substring captured by the regular expression group extractIndex. extractIndex defaults to 0, where 0 means the whole match. Returns NULL for NULL input, no match, invalid regex, or invalid group index.
REGEXP_EXTRACT_ALL(string, regex[, extractIndex])	regexpExtractAll(string, regex[, extractIndex])	Returns ARRAY<STRING> with all substrings captured by group extractIndex. extractIndex defaults to 1, and 0 means the whole match. Returns an empty array if there is no match, and NULL for NULL input, invalid regex, or invalid group index.
REGEXP_COUNT(string, regex)	regexpCount(string, regex)	Returns the number of non-overlapping substrings that match regex. Returns 0 if there is no match, and NULL for NULL input or invalid regex.
REGEXP_INSTR(string, regex)	regexpInstr(string, regex)	Returns the 1-based position of the first substring that matches regex. Returns 0 if there is no match, and NULL for NULL input or invalid regex.
REGEXP_SUBSTR(string, regex)	regexpSubstr(string, regex)	Returns the first substring that matches regex. Returns NULL for NULL input, no match, or invalid regex.
SUBSTR(string, integer1[, integer2])	substr(string,integer1,integer2)	Returns a substring of STRING starting from position integer1 with length integer2 (to the end by default).
SUBSTRING(string FROM integer1 [ FOR integer2 ])	substring(string,integer1,integer2)	Returns a substring of STRING starting from position integer1 with length integer2 (to the end by default).
OVERLAY(string1 PLACING string2 FROM integer1 [FOR integer2])	overlay(string1, string2, integer1[, integer2])	Replaces a substring of STRING1 with STRING2 from position integer1. The replaced length defaults to the length of STRING2. Character and binary strings are supported.
POSITION(string1 IN string2 [ FROM integer ])	position(string1, string2[, integer])	Returns the position of the first occurrence of STRING1 in STRING2, optionally starting from integer. The first position is 1. Returns 0 if not found. Character and binary strings are supported.
LOCATE(string1, string2[, integer])	locate(string1, string2[, integer])	Returns the position of the first occurrence of STRING1 in STRING2, optionally starting from integer. The first position is 1. Returns 0 if not found.
INSTR(string1, string2)	instr(string1, string2)	Returns the position of the first occurrence of STRING2 in STRING1. The first position is 1. Returns 0 if not found.
CONCAT(string1, string2,…)	concat(string1, string2,…)	Returns a string that concatenates string1, string2, …. E.g., CONCAT('AA', 'BB', 'CC') returns 'AABBCC'.
CONCAT_WS(separator, string1, string2,...)	concatWs(separator, string1, string2,...)	Returns a string that concatenates string1, string2, ... with a separator. Null string arguments are skipped.
LPAD(string1, integer, string2)	lpad(string1, integer, string2)	Returns STRING1 left-padded with STRING2 to a length of integer characters. If STRING1 is longer, it is shortened to integer characters.
RPAD(string1, integer, string2)	rpad(string1, integer, string2)	Returns STRING1 right-padded with STRING2 to a length of integer characters. If STRING1 is longer, it is shortened to integer characters.
REPLACE(string1, string2, string3)	replace(string1, string2, string3)	Returns STRING1 with all occurrences of STRING2 replaced by STRING3.
REPEAT(string, integer)	repeat(string, integer)	Returns a string that repeats STRING integer times.
LEFT(string, integer)	left(string, integer)	Returns the leftmost integer characters from STRING.
RIGHT(string, integer)	right(string, integer)	Returns the rightmost integer characters from STRING.
STARTSWITH(string1, string2)	startswith(string1, string2)	Returns whether STRING1 starts with STRING2. Character and binary strings are supported.
ENDSWITH(string1, string2)	endswith(string1, string2)	Returns whether STRING1 ends with STRING2. Character and binary strings are supported.
TO_BASE64(string | binary)	toBase64(string | binary)	Encodes a character or binary string to a base64 string.
FROM_BASE64(string)	fromBase64(string)	Decodes a base64 string as UTF-8 and assumes the decoded bytes are valid UTF-8. Use FROM_BASE64_BINARY for arbitrary binary content.
FROM_BASE64_BINARY(string)	fromBase64Binary(string)	Decodes a base64 string to VARBINARY and preserves the original bytes.

Temporal Functions

Function	Janino Code	Description
LOCALTIME	localtime()	Returns the current SQL time in the local time zone, the return type is TIME(0).
LOCALTIMESTAMP	localtimestamp()	Returns the current SQL timestamp in local time zone, the return type is TIMESTAMP(3).
CURRENT_TIME	currentTime()	Returns the current SQL time in the local time zone, this is a synonym of LOCAL_TIME.
CURRENT_DATE	currentDate()	Returns the current SQL date in the local time zone.
CURRENT_TIMESTAMP	currentTimestamp()	Returns the current SQL timestamp in the local time zone, the return type is TIMESTAMP_LTZ(3).
NOW()	now()	Returns the current SQL timestamp in the local time zone, this is a synonym of CURRENT_TIMESTAMP.
EXTRACT(timeintervalunit FROM temporal)	extract(timeintervalunit, temporal)	Extracts a time interval unit from a temporal value. Supported units are YEAR, QUARTER, MONTH, WEEK, DAY, DOY, DOW, HOUR, MINUTE, and SECOND.
YEAR / QUARTER / MONTH / WEEK(temporal)	extract(timeintervalunit, temporal)	Extracts the year, quarter, month, or ISO week from a temporal value.
DAYOFYEAR / DAYOFMONTH / DAYOFWEEK(temporal)	extract(timeintervalunit, temporal)	Extracts the day of year, day of month, or day of week from a temporal value. DAYOFWEEK uses Sunday as 1.
HOUR / MINUTE / SECOND(temporal)	extract(timeintervalunit, temporal)	Extracts the hour, minute, or second from a temporal value.
temporal +/- INTERVAL 'value' unit	temporalPlusMonths / temporalPlusMillis	Adds or subtracts a year-month or day-time interval literal for DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ values. INTERVAL + temporal and compound intervals such as INTERVAL '1-2' YEAR TO MONTH and INTERVAL '1 02:03:04' DAY TO SECOND are supported.
DATE_FORMAT(timestamp, string)	dateFormat(timestamp, string)	Converts timestamp to a value of string in the format specified by the format string. The format string is compatible with Java's SimpleDateFormat.
DATE_FORMAT(date, string)	dateFormat(date, string)	Converts given date to a value of string in the format specified by the format string. The format string is compatible with Java's SimpleDateFormat.
DATE_FORMAT(time, string)	dateFormat(time, string)	Converts given time to a value of string in the format specified by the format string. The format string is compatible with Java's SimpleDateFormat.
DATE_FORMAT_TZ(timestamp, format, timezone)	dateFormatTz(timestamp, format, timezone)	Formats a timestamp or datetime value as a string using the given pattern and the specified time zone. The timezone argument can be a time zone ID (for example, 'UTC', 'Asia/Shanghai') or an offset such as '+08:00'.
TIMESTAMPADD(timeintervalunit, interval, timepoint)	timestampadd(timeintervalunit, interval, timepoint)	Returns the timestamp of timepoint2 after timepoint added interval. The unit for the interval is given by the first argument, which should be one of the following values: SECOND, MINUTE, HOUR, DAY, MONTH, or YEAR.
TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2)	timestampDiff(timepointunit, timepoint1, timepoint2)	Returns the (signed) number of timepointunit between timepoint1 and timepoint2. The unit for the interval is given by the first argument, which should be one of the following values: SECOND, MINUTE, HOUR, DAY, MONTH, or YEAR.
TO_DATE(string1[, string2])	toDate(string1[, string2])	Converts a date string string1 with format string2 (by default 'yyyy-MM-dd') to a date. The format argument applies only to this string overload.
TO_DATE(timestamp)	toDate(timestamp)	Converts a TIMESTAMP, TIMESTAMP WITH TIME ZONE (TIMESTAMP_TZ), or TIMESTAMP_LTZ value to a DATE using that value's calendar date. A format pattern is not supported for this overload; use TO_DATE(string1[, string2]) to parse formatted strings.
TO_TIMESTAMP(string1[, string2])	toTimestamp(string1[, string2])	Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') to a timestamp, without time zone.
TO_TIMESTAMP_LTZ(string1[, string2])	toTimestampLtz(string1[, string2])	Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') to a timestamp, with local time zone.
FROM_UNIXTIME(numeric[, string])	fromUnixtime(NUMERIC[, STRING])	Returns a representation of the numeric argument as a value in string format (default is ‘yyyy-MM-dd HH:mm:ss’). numeric is an internal timestamp value representing seconds since ‘1970-01-01 00:00:00’ UTC, such as produced by the UNIX_TIMESTAMP() function. The return value is expressed in the session time zone (specified in TableConfig). E.g., FROM_UNIXTIME(44) returns ‘1970-01-01 00:00:44’ if in UTC time zone, but returns ‘1970-01-01 09:00:44’ if in ‘Asia/Tokyo’ time zone.
UNIX_TIMESTAMP()	unixTimestamp()	Gets current Unix timestamp in seconds. This function is not deterministic which means the value would be recalculated for each record.
UNIX_TIMESTAMP(string1[, string2])	unixTimestamp(STRING1[, STRING2])	Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config. If a time zone is specified in the date time string and parsed by UTC+X format such as "yyyy-MM-dd HH:mm:ss.SSS X", this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.
DATE_ADD(date, int)	dateAdd(date, int)	Add N days to given date data.

Conditional Functions

Function	Janino Code	Description
CASE value WHEN value1_1 [, value1_2]* THEN RESULT1 (WHEN value2_1 [, value2_2 ]* THEN result_2)* (ELSE result_z) END	Nested ternary expression	Returns resultX when the first time value is contained in (valueX_1, valueX_2, …). When no value matches, returns result_z if it is provided and returns NULL otherwise.
CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END	Nested ternary expression	Returns resultX when the first conditionX is met. When no condition is met, returns result_z if it is provided and returns NULL otherwise.
COALESCE(value1 [, value2]*)	coalesce(Object... objects)	Returns the first argument that is not NULL.If all arguments are NULL, it returns NULL as well. The return type is the least restrictive, common type of all of its arguments. The return type is nullable if all arguments are nullable as well.
IF(condition, true_value, false_value)	condition ? true_value : false_value	Returns the true_value if condition is met, otherwise false_value. E.g., IF(5 > 3, 5, 3) returns 5.
IFNULL(value, replacement)	ifNull(value, replacement)	Returns replacement when value is NULL; otherwise, returns value. The arguments must have a common type. The result can be NULL only when replacement can be NULL.
NULLIF(value1, value2)	nullIf(value1, value2)	Returns NULL when value1 equals value2; otherwise, returns value1. Its return type is the nullable type of value1.

NULLIF compares numeric operands across numeric types. For example, NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT)) returns NULL. This differs from the current type-sensitive equality behavior of the Transform = operator, which returns FALSE for the same mixed-type comparison.

Casting Functions

You can use CAST( <EXPR> AS <T> ) syntax to convert any valid expression <EXPR> to a specific type <T>. Possible conversion paths are:

TRY_CAST( <EXPR> AS <T> ) supports the same conversion paths as CAST. It returns NULL when a value cannot be converted. An unsupported source-to-target type path is rejected during pipeline validation or initialization. Configuration errors, UDF failures, and internal errors are not converted to NULL.

Source Type	Target Type	Notes
ANY	STRING	All types can be cast to STRING.
NUMERIC, STRING	BOOLEAN	Any non-zero numerics will be evaluated to TRUE.
NUMERIC	BYTE	Value must be in the range of Byte (-128 ~ 127).
NUMERIC	SHORT	Value must be in the range of Short (-32768 ~ 32767).
NUMERIC	INTEGER	Value must be in the range of Integer (-2147483648 ~ 2147483647).
NUMERIC	LONG	Value must be in the range of Long (-9223372036854775808 ~ 9223372036854775807).
NUMERIC	FLOAT	Value must be in the range of Float (1.40239846e-45f ~ 3.40282347e+38f).
NUMERIC	DOUBLE	Value must be in the range of Double (4.94065645841246544e-324 ~ 1.79769313486231570e+308)
NUMERIC	DECIMAL	Value must be in the range of BigDecimal(10, 0).
STRING, TIMESTAMP_TZ, TIMESTAMP_LTZ	TIMESTAMP	String type value must be a valid ISO_LOCAL_DATE_TIME string.

Variant Functions

Function	Janino Code	Description
PARSE_JSON(string[, allowDuplicateKeys])	parseJson(string[, allowDuplicateKeys])	Parse a JSON string into a Variant. If the JSON string is invalid, an error will be thrown. To return NULL instead of an error, use the TRY_PARSE_JSON function. If there are duplicate keys in the input JSON string, when allowDuplicateKeys is true, the parser will keep the last occurrence of all fields with the same key, otherwise it will throw an error. The default value of allowDuplicateKeys is false.
TRY_PARSE_JSON(string[, allowDuplicateKeys])	tryParseJson(string[, allowDuplicateKeys])	Parse a JSON string into a Variant if possible. If the JSON string is invalid, return NULL. To throw an error instead of returning NULL, use the PARSE_JSON function. If there are duplicate keys in the input JSON string, when allowDuplicateKeys is true, the parser will keep the last occurrence of all fields with the same key, otherwise it will return NULL. The default value of allowDuplicateKeys is false.

Struct Functions

Struct functions are used to access elements in ARRAY, MAP, ROW, and VARIANT types using subscript syntax.

Function	Janino Code	Description
ARRAY[value, ...]	array(value, ...)	Creates an array from the given values. The values must have a common type.
MAP[key, value, ...]	map(key, value, ...)	Creates a map from key-value pairs. Keys and values must have common types respectively.
ROW(value, ...)	row(value, ...)	Creates a row from the given values. Field names are generated as f0, f1, and so on.
CARDINALITY(arrayOrMap)	cardinality(arrayOrMap)	Returns the number of elements in an array or map. Returns NULL if the input is NULL.
ARRAY_CONTAINS(array, value)	arrayContains(array, value)	Returns whether the array contains the given value.
ARRAY_POSITION(array, value)	arrayPosition(array, value)	Returns the 1-based position of the first matching value, or 0 if it is not found.
ELEMENT(array)	element(array)	Returns the element of a single-element array. Returns NULL for an empty array and throws an exception if the array has more than one element.
array[index]	itemAccess(array, index)	Returns the element at position index in the array. Index is 1-based (SQL standard). Returns NULL if the index is out of bounds or if the array is NULL.
map[key]	itemAccess(map, key)	Returns the value associated with key in the map. Returns NULL if the key does not exist or if the map is NULL.
row[index]	itemAccess(row, index)	Returns the field at position index in the row. Index is 1-based. The index must be a constant (not a computed expression) since the return type must be statically determined.
variant[index]	itemAccess(variant, index)	Returns the element at position index in the variant (if it's an array). Index is 1-based (SQL standard). Returns NULL if the index is out of bounds or if the variant is not an array.
variant[key]	itemAccess(variant, key)	Returns the value associated with key in the variant (if it's an object). Returns NULL if the key does not exist or if the variant is not an object.

Example

Add computed columns

Evaluation expressions can be used to generate new columns. For example, if we want to append two computed columns based on the table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name, localtimestamp as new_timestamp
    description: append calculated columns based on source table

Reference metadata columns

We may reference metadata column in projection expressions. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ identifier_name
    description: access metadata columns from source table

Use wildcard character to project all fields

A wildcard character (*) can be used to reference all fields in a table. For example, given two tables web_order and app_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: \*, UPPER(product_name) as product_name
    description: project fields with wildcard character from source table
  - source-table: mydb.app_order
    projection: UPPER(product_name) as product_name, *
    description: project fields with wildcard character from source table

Notice: When * character presents at the beginning of expressions, an escaping backslash is required.

Add filter rule

Use reference columns when adding filtering rules to the table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    filter: id > 10 AND order_id > 100
    description: filtering rows from source table

Computed columns can be used in filtering conditions, too. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(province) as new_province 
    filter: new_province = 'SHANGHAI'
    description: filtering rows based on computed columns

Reassign primary key

We can reassign the primary key in transform rules. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id
    primary-keys: order_id
    description: reassign primary key example

Composite primary keys are also supported:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    primary-keys: order_id, product_name
    description: reassign composite primary keys example

Notice that primary key columns will be attributed as NOT NULL in the downstream table, so you should ensure that no NULL value will be assigned to these columns.

Reassign partition key

We can reassign the partition key in transform rules. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    partition-keys: product_name
    description: reassign partition key example

Specify table creation configuration

Extra options can be defined in a transform rule, and will be applied when creating downstream tables. Given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    table-options: comment=web order
    description: auto creating table options example

Tips: The format of table-options is key1=value1,key2=value2. If option values contain commas or other special characters, you can specify a custom delimiter using table-options.delimiter (such as ;, |, $, etc.):

transform:
  - source-table: mydb.web_order
    table-options: sequence.field=gxsj,jjsj;file-index.bloom-filter.columns=jjdbh
    table-options.delimiter: ";"

Classification mapping

If a table hits ultiple transform rules, only the first matched transform rule will apply. For example, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id
    filter: id > 1001
  - source-table: mydb.\.*
    projection: \*, 'fallback' AS FALLBACK

Here, though mydb.web_order matches the second rule (mydb.\.*), it will not fall through the next rule as it has been handled in the first rule.

User-defined Functions

User-defined functions (UDFs) can be used in transform rules.

Classes could be used as a UDF if:

• implements org.apache.flink.cdc.common.udf.UserDefinedFunction interface
• has a public constructor with no parameters
• has at least one public method named eval

It may also:

• overrides getReturnType method to indicate its return CDC type
• overrides open and close method to do some initialization and cleanup work

For example, this is a valid UDF class:

public class AddOneFunctionClass implements UserDefinedFunction {
    
    public Object eval(Integer num) {
        return num + 1;
    }
    
    @Override
    public DataType getReturnType() {
        return DataTypes.INT();
    }
    
    @Override
    public void open() throws Exception {
        // ...
    }

    @Override
    public void close() throws Exception {
        // ...
    }
}

To ease the migration from Flink SQL to Flink CDC, a Flink ScalarFunction could also be used as a transform UDF, with some limitations:

• ScalarFunction which has a constructor with parameters is not supported.
• Flink-style type hint in ScalarFunction will be ignored.
• open / close lifecycle hooks will not be invoked.

UDF classes could be registered by adding a user-defined-function block:

pipeline:
  user-defined-function:
    - name: addone
      classpath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass
    - name: format
      classpath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass

Notice that given classpath must be fully-qualified, and corresponding jar files must be included in Flink /lib folder, or be passed with flink-cdc.sh --jar option.

UDF Options

You can pass extra options to UDFs by adding an options block. These options will be available in the open method through UserDefinedFunctionContext.configuration():

pipeline:
  user-defined-function:
    - name: query_redis
      classpath: com.example.flink.cdc.udf.RedisQueryFunction
      options:
        hostname: localhost
        port: "6379"
        cache.enabled: "true"

And in your UDF implementation, you can access these options by defining ConfigOption instances:

import org.apache.flink.cdc.common.configuration.ConfigOption;
import org.apache.flink.cdc.common.configuration.ConfigOptions;

public class RedisQueryFunction implements UserDefinedFunction {
    private static final ConfigOption<String> HOSTNAME =
        ConfigOptions.key("hostname").stringType().noDefaultValue();
    private static final ConfigOption<Integer> PORT =
        ConfigOptions.key("port").intType().defaultValue(6379);

    private String hostname;
    private int port;
    
    @Override
    public void open(UserDefinedFunctionContext context) throws Exception {
        hostname = context.configuration().get(HOSTNAME);
        port = context.configuration().get(PORT);
        // Initialize your connection here...
    }
    
    public Object eval(String key) {
        // Query Redis using hostname and port...
    }
}

The options field is optional. If not specified, an empty configuration will be passed to the UDF.

After being correctly registered, UDFs could be used in both projection and filter expressions, just like built-in functions:

transform:
  - source-table: db.\.*
    projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
    filter: inc(id) < 100

For AI model functions and configuration, see AI Model.

Python UDFs

Flink CDC supports defining Python UDFs inline in the pipeline YAML. The distribution includes the flink-cdc-python module, which embeds Python on each TaskManager through Pemja.

transform:
  - source-table: db.users
    projection: ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE) AS DOUBLED

pipeline:
  user-defined-function:
    - name: py_normalize
      python-code: |
        def eval(s: str) -> str:
            return None if s is None else s.strip().lower()
      python-executable: /usr/bin/python3
    - name: py_double
      python-code: |
        def eval(x: int) -> int:
            return None if x is None else x * 2
      python-files:
        - /opt/flink/python-deps
        - /opt/flink/python-deps.zip

Each entry must contain one top-level function named eval with a return type annotation. Supported annotations are bool, bytes, float, int, and str, mapped to Flink CDC BOOLEAN, BYTES, DOUBLE, BIGINT, and STRING, respectively.

The following runtime requirements apply:

• Every TaskManager must have Python and pemja==0.5.7 installed. python-executable defaults to the first python3 on PATH.
• python-files is optional and accepts a YAML list of existing directories or .zip archives. These paths must be available on every TaskManager. Zip archives are extracted before being added to Python's import path.
• python-code cannot be combined with classpath or options. Use one UDF entry for each Python function.

Embedding AI Model

Embedding AI Model can be used in transform rules. To use Embedding AI Model, you need to download the jar of built-in model, and then add --jar {$BUILT_IN_MODEL_PATH} to your flink-cdc.sh command.

How to define a Embedding AI Model:

pipeline:
  model:
    - model-name: CHAT
      class-name: OpenAIChatModel
      openai.model: text-embedding-3-small
      openai.host: https://xxxx
      openai.apikey: abcd1234
      openai.chat.prompt: please summary this
    - model-name: GET_EMBEDDING
      class-name: OpenAIEmbeddingModel
      openai.model: text-embedding-3-small
      openai.host: https://xxxx
      openai.apikey: abcd1234

Note:

• model-name is a common required parameter for all support models, which represent the function name called in projection or filter.
• class-name is a common required parameter for all support models, available values can be found in All Support models.
• openai.model, openai.host, openai.apiKey and openai.chat.prompt is option parameters that defined in specific model.

How to use a Embedding AI Model:

transform:
  - source-table: db.\.*
    projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb, CHAT(page) as summary"
    filter: inc(id) < 100
pipeline:
  model:
    - model-name: CHAT
      class-name: OpenAIChatModel
      openai.model: gpt-4o-mini
      openai.host: http://langchain4j.dev/demo/openai/v1
      openai.apikey: demo
      openai.chat.prompt: please summary this
    - model-name: GET_EMBEDDING
      class-name: OpenAIEmbeddingModel
      openai.model: text-embedding-3-small
      openai.host: http://langchain4j.dev/demo/openai/v1
      openai.apikey: demo

Here, GET_EMBEDDING is defined though model-name in pipeline.

All Support models

The following built-in models are provided:

OpenAIChatModel

parameter	type	optional/required	meaning
openai.model	STRING	required	Name of model to be called, for example: "gpt-4o-mini", Available options are "gpt-4o-mini", "gpt-4o", "gpt-4-32k", "gpt-3.5-turbo".
openai.host	STRING	required	Host of the Model server to be connected, for example: http://langchain4j.dev/demo/openai/v1.
openai.apikey	STRING	required	Api Key for verification of the Model server, for example, "demo".
openai.chat.prompt	STRING	optional	Prompt for chatting with OpenAI, for example: "Please summary this ".

OpenAIEmbeddingModel

parameter	type	optional/required	meaning
openai.model	STRING	required	Name of model to be called, for example: "text-embedding-3-small", Available options are "text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002".
openai.host	STRING	required	Host of the Model server to be connected, for example: http://langchain4j.dev/demo/openai/v1.
openai.apikey	STRING	required	Api Key for verification of the Model server, for example, "demo".