This is an automated email from the ASF dual-hosted git repository.
lvyanquan 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 85b83fc04 [FLINK-40151][runtime] Support missing YAML Transform
predicates and nullable BOOLEAN logical evaluation (#4474)
85b83fc04 is described below
commit 85b83fc0444c9f6762839c187d17785ffc69bb71
Author: haruki <[email protected]>
AuthorDate: Wed Jul 22 09:03:03 2026 +0800
[FLINK-40151][runtime] Support missing YAML Transform predicates and
nullable BOOLEAN logical evaluation (#4474)
---
docs/content.zh/docs/core-concept/transform.md | 40 ++++--
docs/content/docs/core-concept/transform.md | 38 +++--
.../src/test/resources/specs/comparison.yaml | 91 ++++++++++++
.../src/test/resources/specs/logical.yaml | 23 ++-
.../functions/impl/ComparisonFunctions.java | 11 ++
.../runtime/functions/impl/LogicalFunctions.java | 74 ++++++++++
.../runtime/functions/impl/StringFunctions.java | 34 +++++
.../transform/TransformFilterProcessor.java | 3 +-
.../flink/cdc/runtime/parser/JaninoCompiler.java | 159 +++++++++++++++++++--
.../flink/cdc/runtime/parser/TransformParser.java | 153 +++++++++++++++++++-
.../parser/metadata/TransformSqlOperatorTable.java | 7 +
.../functions/impl/ComparisonFunctionsTest.java | 78 ++++++++++
.../functions/impl/StringFunctionsTest.java | 60 ++++++++
.../cdc/runtime/parser/JaninoCompilerTest.java | 103 +++++++++++++
.../cdc/runtime/parser/TransformParserTest.java | 84 ++++++++---
15 files changed, 896 insertions(+), 62 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index 8dc5ceae0..dc405f777 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -120,26 +120,38 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/)
来解析表达式并且
| value1 >= value2 | greaterThanOrEqual(value1, value2)
| 如果 value1 大于等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 < value2 | lessThan(value1, value2)
| 如果 value1 小于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 <= value2 | lessThanOrEqual(value1, value2)
| 如果 value1 小于等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
-| value IS NULL | null == value
| 如果 value 为 NULL，返回 TRUE。 |
-| value IS NOT NULL | null != value
| 如果 value 不为 NULL，返回 TRUE。 |
+| value IS NULL | isNull(value)
| 如果 value 为 NULL，返回 TRUE。 |
+| value IS NOT NULL | isNotNull(value)
| 如果 value 不为 NULL，返回 TRUE。 |
+| value1 IS DISTINCT FROM value2 | isDistinctFrom(value1, value2)
| 如果 value1 和 value2 是不同的值，返回 TRUE。NULL 会作为普通值参与比较，且不会返回 NULL。 |
+| value1 IS NOT DISTINCT FROM value2 | isNotDistinctFrom(value1, value2)
| 如果 value1 和 value2 不是不同的值，返回 TRUE。NULL 会作为普通值参与比较，且不会返回 NULL。 |
| value1 BETWEEN value2 AND value3 | betweenAsymmetric(value1, value2,
value3) | 如果 value1 大于等于 value2 且小于等于 value3，返回 TRUE。 |
| value1 NOT BETWEEN value2 AND value3 | notBetweenAsymmetric(value1, value2,
value3) | 如果 value1 小于 value2 或大于 value3，返回 TRUE。 |
-| string1 LIKE string2 | like(string1, string2)
| 如果 string1 匹配模式 string2，返回 TRUE。 |
-| string1 NOT LIKE string2 | notLike(string1, string2)
| 如果 string1 不匹配模式 string2，返回 TRUE。 |
+| string1 LIKE string2 | like(string1, string2)
| 如果 string1 按 Java 正则表达式和子串匹配语义匹配 string2，返回 TRUE。 |
+| string1 NOT LIKE string2 | notLike(string1, string2)
| 如果 string1 按 Java 正则表达式和子串匹配语义不匹配 string2，返回 TRUE。 |
+| string1 LIKE string2 ESCAPE string3 | like(string1, string2, string3)
| 如果 string1 匹配 SQL LIKE 模式 string2，返回 TRUE。`%` 匹配任意长度字符串，`_`
匹配单个字符，string3 用于转义通配符。如果任一参数为 NULL，返回 NULL。 |
+| string1 NOT LIKE string2 ESCAPE string3 | notLike(string1, string2, string3)
| 如果 string1 不匹配 SQL LIKE 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
+| string1 SIMILAR TO string2 | similarTo(string1, string2)
| 如果 string1 匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
+| string1 NOT SIMILAR TO string2 | notSimilarTo(string1, string2)
| 如果 string1 不匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。
|
+| string1 SIMILAR TO string2 ESCAPE string3 | similarTo(string1, string2,
string3) | 如果 string1 使用 string3 作为转义字符匹配 SQL SIMILAR TO 模式 string2，返回
TRUE。如果任一参数为 NULL，返回 NULL。 |
+| string1 NOT SIMILAR TO string2 ESCAPE string3 | notSimilarTo(string1,
string2, string3) | 如果 string1 使用 string3 作为转义字符不匹配 SQL SIMILAR TO 模式
string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| value1 IN (value2 [, value3]* ) | in(value1, value2 [, value3]*)
| 如果 value1 存在于给定列表 (value2, value3, …) 中，返回 TRUE。 |
| value1 NOT IN (value2 [, value3]* ) | notIn(value1, value2 [, value3]*)
| 如果 value1 不存在于给定列表 (value2, value3, …) 中，返回 TRUE。 |
## 逻辑函数
-| 函数 | Janino 代码 | 描述
|
-|-----------------------|--------------------------------|-------------------------------------------|
-| boolean1 OR boolean2 | boolean1 || boolean2 | 如果 BOOLEAN1 为 TRUE
或 BOOLEAN2 为 TRUE，返回 TRUE。 |
-| boolean1 AND boolean2 | boolean1 && boolean2 | 如果 BOOLEAN1 和
BOOLEAN2 都为 TRUE，返回 TRUE。 |
-| NOT boolean | !boolean | 如果 boolean 为
FALSE，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE。 |
-| boolean IS FALSE | false == boolean | 如果 boolean 为
FALSE，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE。 |
-| boolean IS NOT FALSE | true == boolean | 如果 BOOLEAN 为
TRUE，返回 TRUE；如果 BOOLEAN 为 FALSE，返回 FALSE。 |
-| boolean IS TRUE | true == boolean | 如果 BOOLEAN 为
TRUE，返回 TRUE；如果 BOOLEAN 为 FALSE，返回 FALSE。 |
-| boolean IS NOT TRUE | false == boolean | 如果 boolean 为
FALSE，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE。 |
+逻辑函数遵循 SQL 的三值逻辑来处理可为 NULL 的 BOOLEAN 值。`AND` 和 `OR`
会在左操作数已经决定结果时短路，不再计算右操作数。生成的 Janino 代码会根据操作数可空性使用原生运算符、条件表达式或惰性函数调用。
+
+| 函数 | Janino 代码 | 描述
|
+|------------------------|----------------------|--------------------------------------------------------------------|
+| boolean1 OR boolean2 | 短路 OR 表达式 | 如果任一值为 TRUE，返回 TRUE；如果没有 TRUE 且至少一个值为
NULL，返回 NULL。 |
+| boolean1 AND boolean2 | 短路 AND 表达式 | 如果任一值为 FALSE，返回 FALSE；如果没有 FALSE
且至少一个值为 NULL，返回 NULL。 |
+| NOT boolean | not(boolean) | 如果 boolean 为 FALSE，返回
TRUE；如果 boolean 为 TRUE，返回 FALSE；如果为 NULL，返回 NULL。 |
+| boolean IS FALSE | isFalse(boolean) | 如果 boolean 为 FALSE，返回
TRUE；如果 boolean 为 TRUE 或 NULL，返回 FALSE。 |
+| boolean IS NOT FALSE | isNotFalse(boolean) | 如果 boolean 为 TRUE 或
NULL，返回 TRUE；如果 boolean 为 FALSE，返回 FALSE。 |
+| boolean IS TRUE | isTrue(boolean) | 如果 boolean 为 TRUE，返回
TRUE；如果 boolean 为 FALSE 或 NULL，返回 FALSE。 |
+| boolean IS NOT TRUE | isNotTrue(boolean) | 如果 boolean 为 FALSE 或
NULL，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE。 |
+| boolean IS UNKNOWN | isNull(boolean) | 如果 boolean 为 NULL，返回 TRUE。
|
+| boolean IS NOT UNKNOWN | isNotNull(boolean) | 如果 boolean 不为 NULL，返回
TRUE。 |
## 数学函数
@@ -542,4 +554,4 @@ pipeline:
|---------------|--------|----------|----------------------------------------------------------------------------------------------------------------------|
| openai.model | STRING | 必填 |
要调用的模型名称，例如："text-embedding-3-small"，可用选项有
"text-embedding-3-small"、"text-embedding-3-large"、"text-embedding-ada-002"。 |
| openai.host | STRING | 必填 |
要连接的模型服务器地址，例如：`http://langchain4j.dev/demo/openai/v1`。
|
-| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
\ No newline at end of file
+| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index 4a6d40b61..eb7629057 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -121,26 +121,38 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to
parse expressions and [
| 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 | null == value
| Returns TRUE if value is NULL.
|
-| value IS NOT NULL | null != value
| Returns TRUE if value is not 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 pattern string2.
|
-| string1 NOT LIKE string2 | notLike(string1, string2)
| Returns TRUE if string1 does not match pattern string2.
|
+| 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
-| Function | Janino Code | Description
|
-|-----------------------|--------------------------------|---------------------------------------------------------------------|
-| boolean1 OR boolean2 | boolean1 || boolean2 | Returns TRUE if
BOOLEAN1 is TRUE or BOOLEAN2 is TRUE. |
-| boolean1 AND boolean2 | boolean1 && boolean2 | Returns TRUE if
BOOLEAN1 and BOOLEAN2 are both TRUE. |
-| NOT boolean | !boolean | Returns TRUE if
boolean is FALSE; returns FALSE if boolean is TRUE. |
-| boolean IS FALSE | false == boolean | Returns TRUE if
boolean is FALSE; returns FALSE if boolean is TRUE. |
-| boolean IS NOT FALSE | true == boolean | Returns TRUE if
BOOLEAN is TRUE; returns FALSE if BOOLEAN is FALSE. |
-| boolean IS TRUE | true == boolean | Returns TRUE if
BOOLEAN is TRUE; returns FALSE if BOOLEAN is FALSE. |
-| boolean IS NOT TRUE | false == boolean | Returns TRUE if
boolean is FALSE; returns FALSE if boolean is TRUE. |
+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
diff --git a/flink-cdc-composer/src/test/resources/specs/comparison.yaml
b/flink-cdc-composer/src/test/resources/specs/comparison.yaml
index c022f0780..85d7e80b4 100644
--- a/flink-cdc-composer/src/test/resources/specs/comparison.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/comparison.yaml
@@ -165,6 +165,26 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, 天地玄黄宇宙洪荒, true, true,
true, false], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, true,
false, true, false], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, true, false, true,
false], after=[], op=DELETE, meta=()}
+- do: IsUnknown Op
+ projection: id_, bool_, bool_ IS UNKNOWN AS comp_1, bool_ IS NOT UNKNOWN AS
comp_2, (bool_ AND CAST(NULL AS BOOLEAN)) IS UNKNOWN AS comp_3
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`bool_` BOOLEAN 'George' 'false',`comp_1` BOOLEAN,`comp_2`
BOOLEAN,`comp_3` BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, false,
true, true], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, false, true, true],
after=[-1, false, false, true, false], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, false, false, true,
false], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, true,
false, true], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, true, false, true],
after=[], op=DELETE, meta=()}
+- do: Distinct From Op
+ projection: id_, int_ IS DISTINCT FROM 4 AS comp_1, int_ IS NOT DISTINCT
FROM 4 AS comp_2, CAST(NULL AS INT) IS DISTINCT FROM int_ AS comp_3, CAST(NULL
AS INT) IS NOT DISTINCT FROM int_ AS comp_4
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, false, true,
true, false], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, false, true, true, false],
after=[-1, true, false, true, false], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, true, false, true,
false], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, true, false,
false, true], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, true, false, false, true],
after=[], op=DELETE, meta=()}
- do: Between Op
projection: |-
id_
@@ -235,6 +255,77 @@
ignore: FLINK-38905
projection: id_, char_ NOT LIKE 'A.*' AS comp_1, varchar_ NOT LIKE '.*rro'
AS comp_2, string_ NOT LIKE 'From [A-Z] to [A-Z] is Lie' AS comp_3
primary-key: id_
+- do: String Like Escape Op
+ projection: id_, 'A%' LIKE 'A$%' ESCAPE '$' AS comp_1, 'A_' LIKE 'A$_'
ESCAPE '$' AS comp_2, char_ LIKE 'A$%' ESCAPE '$' AS comp_3
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, true,
false], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, true, false],
after=[-1, true, true, false], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, true, true, false],
after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, true, true,
null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, true, true, null],
after=[], op=DELETE, meta=()}
+- do: String Not Like Escape Op
+ projection: |-
+ id_
+ 'A%' NOT LIKE 'A$%' ESCAPE '$' AS comp_1
+ 'A_' NOT LIKE 'A$_' ESCAPE '$' AS comp_2
+ char_ NOT LIKE 'A%' ESCAPE '$' AS comp_3
+ CAST(NULL AS STRING) NOT LIKE 'A%' ESCAPE '$' AS comp_4
+ char_ NOT LIKE CAST(NULL AS STRING) ESCAPE '$' AS comp_5
+ char_ NOT LIKE 'A%' ESCAPE CAST(NULL AS STRING) AS comp_6
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN,`comp_5` BOOLEAN,`comp_6` BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, false, false,
false, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, false, false, false, null,
null, null], after=[-1, false, false, true, null, null, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, false, false, true, null,
null, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, false, false,
null, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, false, false, null, null,
null, null], after=[], op=DELETE, meta=()}
+- do: String Similar To Escape Op
+ projection: |-
+ id_
+ 'A%' SIMILAR TO 'A$%' ESCAPE '$' AS comp_1
+ 'A_' SIMILAR TO 'A$_' ESCAPE '$' AS comp_2
+ char_ SIMILAR TO 'A%' ESCAPE '$' AS comp_3
+ CAST(NULL AS STRING) SIMILAR TO 'A%' ESCAPE '$' AS comp_4
+ char_ SIMILAR TO CAST(NULL AS STRING) ESCAPE '$' AS comp_5
+ char_ SIMILAR TO 'A%' ESCAPE CAST(NULL AS STRING) AS comp_6
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN,`comp_5` BOOLEAN,`comp_6` BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, true,
true, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, true, true, null,
null, null], after=[-1, true, true, false, null, null, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, true, true, false, null,
null, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, true, true,
null, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, true, true, null, null,
null, null], after=[], op=DELETE, meta=()}
+- do: String Not Similar To Escape Op
+ projection: |-
+ id_
+ 'A%' NOT SIMILAR TO 'A$%' ESCAPE '$' AS comp_1
+ 'A_' NOT SIMILAR TO 'A$_' ESCAPE '$' AS comp_2
+ char_ NOT SIMILAR TO 'A%' ESCAPE '$' AS comp_3
+ CAST(NULL AS STRING) NOT SIMILAR TO 'A%' ESCAPE '$' AS comp_4
+ char_ NOT SIMILAR TO CAST(NULL AS STRING) ESCAPE '$' AS comp_5
+ char_ NOT SIMILAR TO 'A%' ESCAPE CAST(NULL AS STRING) AS comp_6
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN,`comp_5` BOOLEAN,`comp_6` BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, false, false,
false, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, false, false, false, null,
null, null], after=[-1, false, false, true, null, null, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, false, false, true, null,
null, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, false, false,
null, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, false, false, null, null,
null, null], after=[], op=DELETE, meta=()}
+- do: String Similar To Op
+ projection: id_, char_ SIMILAR TO 'A%' AS comp_1, varchar_ NOT SIMILAR TO
'%rro' AS comp_2, string_ SIMILAR TO 'From (A|B) to (Y|Z) is Lie' AS comp_3,
char_ NOT SIMILAR TO 'A%' AS comp_4
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, false,
true, false], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, false, true, false],
after=[-1, false, true, false, true], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, false, true, false,
true], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null],
after=[], op=DELETE, meta=()}
- do: In Op
projection: |-
id_
diff --git a/flink-cdc-composer/src/test/resources/specs/logical.yaml
b/flink-cdc-composer/src/test/resources/specs/logical.yaml
index 4f09aa823..d5f80818c 100644
--- a/flink-cdc-composer/src/test/resources/specs/logical.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/logical.yaml
@@ -45,4 +45,25 @@
bool_ IS FALSE AS comp_8
bool_ IS NOT FALSE AS comp_9
primary-key: id_
- expect-error: 'java.lang.RuntimeException: Failed to evaluate projection
expression ``TB`.`bool_` OR FALSE` for column `comp_2` in table `foo.bar.baz`.'
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`bool_` BOOLEAN 'George' 'false',`comp_1` BOOLEAN,`comp_2`
BOOLEAN,`comp_3` BOOLEAN,`comp_4` BOOLEAN,`comp_5` BOOLEAN,`comp_6`
BOOLEAN,`comp_7` BOOLEAN,`comp_8` BOOLEAN,`comp_9` BOOLEAN}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, true,
true, true, false, false, true, false, false, true], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, true, true, true,
false, false, true, false, false, true], after=[-1, false, true, false, false,
false, true, false, true, true, false], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, false, true, false,
false, false, true, false, true, true, false], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, true,
null, null, false, null, false, true, false, true], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, true, null, null,
false, null, false, true, false, true], after=[], op=DELETE, meta=()}
+- do: Logical Operators Short-Circuit
+ projection: |-
+ id_
+ TRUE OR 1 / CAST(0 AS INT) > 0 AS comp_1
+ FALSE AND 1 / CAST(0 AS INT) > 0 AS comp_2
+ CAST(TRUE AS BOOLEAN) OR 1 / CAST(0 AS INT) > 0 AS comp_3
+ CAST(FALSE AS BOOLEAN) AND 1 / CAST(0 AS INT) > 0 AS comp_4
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` BOOLEAN,`comp_2` BOOLEAN,`comp_3` BOOLEAN,`comp_4`
BOOLEAN}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, true, false,
true, false], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, true, false, true, false],
after=[-1, true, false, true, false], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, true, false, true,
false], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, true, false,
true, false], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, true, false, true, false],
after=[], op=DELETE, meta=()}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctions.java
index 41605ca8b..e318e45e1 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctions.java
@@ -27,6 +27,17 @@ public class ComparisonFunctions {
return (object1 != null && object2 != null) && object1.equals(object2);
}
+ public static boolean isDistinctFrom(Object object1, Object object2) {
+ if (object1 == null || object2 == null) {
+ return object1 != object2;
+ }
+ return !object1.equals(object2);
+ }
+
+ public static boolean isNotDistinctFrom(Object object1, Object object2) {
+ return !isDistinctFrom(object1, object2);
+ }
+
private static int universalCompares(Object lhs, Object rhs) {
Class<?> leftClass = lhs.getClass();
Class<?> rightClass = rhs.getClass();
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java
index 80ae5ccf2..07366bd34 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java
@@ -17,9 +17,83 @@
package org.apache.flink.cdc.runtime.functions.impl;
+import java.util.function.Supplier;
+
/** Logical built-in functions. */
public class LogicalFunctions {
+ public static Boolean and(Boolean left, Boolean right) {
+ if (Boolean.FALSE.equals(left) || Boolean.FALSE.equals(right)) {
+ return false;
+ }
+ if (left == null || right == null) {
+ return null;
+ }
+ return true;
+ }
+
+ public static Boolean or(Boolean left, Boolean right) {
+ if (Boolean.TRUE.equals(left) || Boolean.TRUE.equals(right)) {
+ return true;
+ }
+ if (left == null || right == null) {
+ return null;
+ }
+ return false;
+ }
+
+ public static Boolean and(Boolean left, Supplier<Boolean> rightSupplier) {
+ if (Boolean.FALSE.equals(left)) {
+ return false;
+ }
+ Boolean right = rightSupplier.get();
+ return and(left, right);
+ }
+
+ public static Boolean or(Boolean left, Supplier<Boolean> rightSupplier) {
+ if (Boolean.TRUE.equals(left)) {
+ return true;
+ }
+ Boolean right = rightSupplier.get();
+ return or(left, right);
+ }
+
+ public static Boolean not(Boolean value) {
+ return value == null ? null : !value;
+ }
+
+ public static boolean isNull(Object value) {
+ return value == null;
+ }
+
+ public static boolean isNotNull(Object value) {
+ return value != null;
+ }
+
+ public static boolean isTrue(Boolean value) {
+ return Boolean.TRUE.equals(value);
+ }
+
+ public static boolean isNotTrue(Boolean value) {
+ return !Boolean.TRUE.equals(value);
+ }
+
+ public static boolean isFalse(Boolean value) {
+ return Boolean.FALSE.equals(value);
+ }
+
+ public static boolean isNotFalse(Boolean value) {
+ return !Boolean.FALSE.equals(value);
+ }
+
+ public static boolean isUnknown(Boolean value) {
+ return value == null;
+ }
+
+ public static boolean isNotUnknown(Boolean value) {
+ return value != null;
+ }
+
public static Object coalesce(Object... objects) {
for (Object item : objects) {
if (item != null) {
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
index 39194de2d..7e29f63e7 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.runtime.functions.impl;
import org.apache.flink.cdc.common.types.variant.BinaryVariantInternalBuilder;
import org.apache.flink.cdc.common.types.variant.Variant;
+import org.apache.calcite.runtime.SqlFunctions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,10 +70,43 @@ public class StringFunctions {
return Pattern.compile(regex).matcher(str).find();
}
+ public static Boolean like(String str, String pattern, String escape) {
+ if (str == null || pattern == null || escape == null) {
+ return null;
+ }
+ return SqlFunctions.like(str, pattern, escape);
+ }
+
public static boolean notLike(String str, String regex) {
return !like(str, regex);
}
+ public static Boolean notLike(String str, String pattern, String escape) {
+ return LogicalFunctions.not(like(str, pattern, escape));
+ }
+
+ public static Boolean similarTo(String str, String pattern) {
+ if (str == null || pattern == null) {
+ return null;
+ }
+ return SqlFunctions.similar(str, pattern);
+ }
+
+ public static Boolean similarTo(String str, String pattern, String escape)
{
+ if (str == null || pattern == null || escape == null) {
+ return null;
+ }
+ return SqlFunctions.similar(str, pattern, escape);
+ }
+
+ public static Boolean notSimilarTo(String str, String pattern) {
+ return LogicalFunctions.not(similarTo(str, pattern));
+ }
+
+ public static Boolean notSimilarTo(String str, String pattern, String
escape) {
+ return LogicalFunctions.not(similarTo(str, pattern, escape));
+ }
+
public static String substr(String str, int beginIndex) {
return substring(str, beginIndex);
}
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 77dc5cfe6..56ce510c1 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
@@ -115,7 +115,8 @@ public class TransformFilterProcessor {
}
try {
- return (Boolean)
expressionEvaluator.evaluate(generateParams(preRow, postRow, context));
+ return Boolean.TRUE.equals(
+ expressionEvaluator.evaluate(generateParams(preRow,
postRow, context)));
} catch (InvocationTargetException e) {
throw new RuntimeException(
String.format(
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
index 14c483d70..909c628fd 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
@@ -28,6 +28,7 @@ import org.apache.flink.cdc.common.types.DataTypeRoot;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.common.utils.StringUtils;
import
org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
+import org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns;
import org.apache.calcite.sql.SqlBasicCall;
import org.apache.calcite.sql.SqlBasicTypeNameSpec;
@@ -230,7 +231,8 @@ public class JaninoCompiler {
sqlCaseRvalueTemp =
new Java.ConditionalExpression(
Location.NOWHERE,
- whenAtoms.get(i),
+ generateFunctionOperation(
+ "isTrue", new Java.Rvalue[]
{whenAtoms.get(i)}),
thenAtoms.get(i),
sqlCaseRvalueTemp);
}
@@ -258,30 +260,41 @@ public class JaninoCompiler {
Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) {
switch (sqlBasicCall.getKind()) {
case AND:
- return generateBinaryOperation(context, sqlBasicCall, atoms,
"&&");
+ return generateLogicalBinaryOperation(context, sqlBasicCall,
atoms, true);
case OR:
- return generateBinaryOperation(context, sqlBasicCall, atoms,
"||");
+ return generateLogicalBinaryOperation(context, sqlBasicCall,
atoms, false);
case NOT:
- return generateUnaryOperation(context, "!", atoms[0]);
+ return generateFunctionOperation("not", atoms);
case EQUALS:
return generateEqualsOperation(context, sqlBasicCall, atoms);
case NOT_EQUALS:
return generateUnaryOperation(
context, "!", generateEqualsOperation(context,
sqlBasicCall, atoms));
+ case IS_DISTINCT_FROM:
+ case IS_NOT_DISTINCT_FROM:
+ return generateOtherFunctionOperation(context, sqlBasicCall,
atoms);
case IS_NULL:
- return generateUnaryOperation(context, "null == ", atoms[0]);
+ return generateFunctionOperation("isNull", atoms);
case IS_NOT_NULL:
- return generateUnaryOperation(context, "null != ", atoms[0]);
+ return generateFunctionOperation("isNotNull", atoms);
case IS_FALSE:
+ return generateFunctionOperation("isFalse", atoms);
case IS_NOT_TRUE:
- return generateUnaryOperation(context, "false == ", atoms[0]);
+ return generateFunctionOperation("isNotTrue", atoms);
case IS_TRUE:
+ return generateFunctionOperation("isTrue", atoms);
case IS_NOT_FALSE:
- return generateUnaryOperation(context, "true == ", atoms[0]);
+ return generateFunctionOperation("isNotFalse", atoms);
+ case IS_UNKNOWN:
+ if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IS
NOT UNKNOWN")) {
+ return generateFunctionOperation("isNotUnknown", atoms);
+ }
+ return generateFunctionOperation("isUnknown", atoms);
case BETWEEN:
case IN:
case NOT_IN:
case LIKE:
+ case SIMILAR:
case CEIL:
case FLOOR:
case TRIM:
@@ -322,6 +335,131 @@ public class JaninoCompiler {
return new Java.UnaryOperation(Location.NOWHERE, operator, atom);
}
+ private static Java.Rvalue generateFunctionOperation(String functionName,
Java.Rvalue[] atoms) {
+ return new Java.MethodInvocation(Location.NOWHERE, null, functionName,
atoms);
+ }
+
+ private static Java.Rvalue generateLazyBinaryFunctionOperation(
+ Context context, SqlBasicCall sqlBasicCall, String functionName,
Java.Rvalue[] atoms) {
+ if (atoms.length != 2) {
+ throw new ParseException("Unrecognized expression: " +
sqlBasicCall.toString());
+ }
+ Java.Rvalue rightOperandSupplier =
+ new Java.AmbiguousName(
+ Location.NOWHERE,
+ new String[] {
+ "new java.util.function.Supplier<Boolean>() {
public Boolean get() { return "
+ + atoms[1]
+ + "; } }"
+ });
+ return generateFunctionOperation(
+ functionName, new Java.Rvalue[] {atoms[0],
rightOperandSupplier});
+ }
+
+ private static Java.Rvalue generateLogicalBinaryOperation(
+ Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms,
boolean isAnd) {
+ if (atoms.length != 2) {
+ throw new ParseException("Unrecognized expression: " +
sqlBasicCall.toString());
+ }
+ boolean leftNullable = isExpressionNullable(context,
sqlBasicCall.getOperandList().get(0));
+ boolean rightNullable = isExpressionNullable(context,
sqlBasicCall.getOperandList().get(1));
+ if (!leftNullable && !rightNullable) {
+ return generateBinaryOperation(context, sqlBasicCall, atoms, isAnd
? "&&" : "||");
+ }
+ if (!leftNullable) {
+ return generateLeftNonNullableLogicalOperation(atoms, isAnd);
+ }
+ // TODO: This nullable-left path still relies on JIT escape analysis
to optimize
+ // per-evaluation Supplier allocation. Introduce statement-level
codegen
+ // (for example, GeneratedExpression/ScriptEvaluator) to avoid the
potential cost.
+ return generateLazyBinaryFunctionOperation(
+ context, sqlBasicCall, isAnd ? "and" : "or", atoms);
+ }
+
+ private static Java.Rvalue generateLeftNonNullableLogicalOperation(
+ Java.Rvalue[] atoms, boolean isAnd) {
+ if (isAnd) {
+ return new Java.ConditionalExpression(
+ Location.NOWHERE,
+ atoms[0],
+ atoms[1],
+ new Java.AmbiguousName(Location.NOWHERE, new String[]
{"Boolean.FALSE"}));
+ }
+ return new Java.ConditionalExpression(
+ Location.NOWHERE,
+ atoms[0],
+ new Java.AmbiguousName(Location.NOWHERE, new String[]
{"Boolean.TRUE"}),
+ atoms[1]);
+ }
+
+ private static boolean isExpressionNullable(Context context, SqlNode
sqlNode) {
+ if (sqlNode instanceof SqlIdentifier) {
+ return isIdentifierNullable(context, (SqlIdentifier) sqlNode);
+ }
+ if (sqlNode instanceof SqlLiteral) {
+ return ((SqlLiteral) sqlNode).getValue() == null;
+ }
+ if (sqlNode instanceof SqlBasicCall) {
+ return isBasicCallNullable(context, (SqlBasicCall) sqlNode);
+ }
+ return true;
+ }
+
+ private static boolean isIdentifierNullable(Context context, SqlIdentifier
sqlIdentifier) {
+ String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size()
- 1);
+ for (Column column : context.columns) {
+ if (column.getName().equals(columnName)) {
+ return column.getType().isNullable();
+ }
+ }
+ for (SupportedMetadataColumn metadataColumn :
context.supportedMetadataColumns) {
+ if (metadataColumn.getName().equals(columnName)) {
+ return metadataColumn.getType().isNullable();
+ }
+ }
+ return MetadataColumns.METADATA_COLUMNS.stream()
+ .filter(column -> column.f0.equals(columnName))
+ .findFirst()
+ .map(column -> column.f1.isNullable())
+ .orElse(true);
+ }
+
+ private static boolean isBasicCallNullable(Context context, SqlBasicCall
sqlBasicCall) {
+ switch (sqlBasicCall.getKind()) {
+ case AND:
+ case OR:
+ return sqlBasicCall.getOperandList().stream()
+ .anyMatch(operand -> isExpressionNullable(context,
operand));
+ case NOT:
+ return isExpressionNullable(context,
sqlBasicCall.getOperandList().get(0));
+ case IS_NULL:
+ case IS_NOT_NULL:
+ case IS_FALSE:
+ case IS_NOT_TRUE:
+ case IS_TRUE:
+ case IS_NOT_FALSE:
+ case IS_UNKNOWN:
+ case IS_DISTINCT_FROM:
+ case IS_NOT_DISTINCT_FROM:
+ case EQUALS:
+ case NOT_EQUALS:
+ case LESS_THAN:
+ case GREATER_THAN:
+ case LESS_THAN_OR_EQUAL:
+ case GREATER_THAN_OR_EQUAL:
+ case BETWEEN:
+ case IN:
+ case NOT_IN:
+ return false;
+ case LIKE:
+ case SIMILAR:
+ return sqlBasicCall.getOperandList().stream()
+ .anyMatch(operand -> isExpressionNullable(context,
operand));
+ default:
+ return true;
+ }
+ }
+
private static final Map<String, String> decimalArithmeticHandlers =
Map.of(
"+", "plus",
@@ -517,7 +655,10 @@ public class JaninoCompiler {
if (operationName.equals("IF")) {
if (atoms.length == 3) {
return new Java.ConditionalExpression(
- Location.NOWHERE, atoms[0], atoms[1], atoms[2]);
+ Location.NOWHERE,
+ generateFunctionOperation("isTrue", new Java.Rvalue[]
{atoms[0]}),
+ atoms[1],
+ atoms[2]);
} else {
throw new ParseException("Unrecognized expression: " +
sqlBasicCall);
}
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 5ee715301..4912ef5a5 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
@@ -62,6 +62,7 @@ import org.apache.calcite.sql.type.InferTypes;
import org.apache.calcite.sql.type.OperandTypes;
import org.apache.calcite.sql.type.SqlReturnTypeInference;
import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.util.SqlOperatorTables;
import org.apache.calcite.sql.validate.SqlConformanceEnum;
import org.apache.calcite.sql.validate.SqlValidator;
@@ -285,13 +286,15 @@ public class TransformParser {
}
expandWildcard(sqlSelect, columns);
- RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors,
supportedMetadataColumns);
- RelDataType[] relDataTypes =
- relNode.getRowType().getFieldList().stream()
- .map(RelDataTypeField::getType)
- .toArray(RelDataType[]::new);
Map<String, Column> originalColumnMap =
columns.stream().collect(Collectors.toMap(Column::getName,
column -> column));
+ RelDataType[] relDataTypes =
+ deduceProjectionRelDataTypes(
+ columns,
+ originalColumnMap,
+ sqlSelect,
+ udfDescriptors,
+ supportedMetadataColumns);
List<ProjectionColumn> projectionColumns = new ArrayList<>();
Map<String, Integer> addedProjectionColumnNames = new HashMap<>();
@@ -387,6 +390,146 @@ public class TransformParser {
return projectionColumns;
}
+ private static RelDataType[] deduceProjectionRelDataTypes(
+ List<Column> columns,
+ Map<String, Column> originalColumnMap,
+ SqlSelect sqlSelect,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ try {
+ RelNode relNode =
+ sqlToRel(columns, sqlSelect, udfDescriptors,
supportedMetadataColumns);
+ return relNode.getRowType().getFieldList().stream()
+ .map(RelDataTypeField::getType)
+ .toArray(RelDataType[]::new);
+ } catch (RuntimeException e) {
+ try {
+ // Keep Calcite as the primary type inference path. This
fallback only covers
+ // transform predicates that Janino can evaluate but Calcite
may fail to convert
+ // while building projection columns.
+ SqlTypeFactoryImpl typeFactory = new
SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+ List<RelDataType> relDataTypes = new ArrayList<>();
+ for (SqlNode sqlNode : sqlSelect.getSelectList()) {
+ relDataTypes.add(
+ deduceProjectionRelDataType(
+ typeFactory,
+ columns,
+ originalColumnMap,
+ unwrapAsExpression(sqlNode),
+ udfDescriptors,
+ supportedMetadataColumns));
+ }
+ return relDataTypes.toArray(new RelDataType[0]);
+ } catch (RuntimeException fallbackException) {
+ e.addSuppressed(fallbackException);
+ throw e;
+ }
+ }
+ }
+
+ private static SqlNode unwrapAsExpression(SqlNode sqlNode) {
+ if (sqlNode instanceof SqlBasicCall) {
+ SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
+ if (SqlKind.AS.equals(sqlBasicCall.getOperator().kind)
+ && !sqlBasicCall.getOperandList().isEmpty()) {
+ return sqlBasicCall.getOperandList().get(0);
+ }
+ }
+ return sqlNode;
+ }
+
+ private static RelDataType deduceProjectionRelDataType(
+ RelDataTypeFactory typeFactory,
+ List<Column> columns,
+ Map<String, Column> originalColumnMap,
+ SqlNode exprNode,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ if (exprNode instanceof SqlIdentifier) {
+ String columnName =
+ ((SqlIdentifier) exprNode)
+ .names.get(((SqlIdentifier) exprNode).names.size()
- 1);
+ return toRelDataType(
+ typeFactory,
+ findIdentifierDataType(
+ originalColumnMap, columnName,
supportedMetadataColumns));
+ }
+ if (requiresBooleanTypeFallback(exprNode)) {
+ validateReferencedColumns(originalColumnMap, exprNode,
supportedMetadataColumns);
+ return typeFactory.createTypeWithNullability(
+ typeFactory.createSqlType(SqlTypeName.BOOLEAN), true);
+ }
+ return toRelDataType(
+ typeFactory,
+ deduceSubExpressionType(
+ columns, exprNode, udfDescriptors,
supportedMetadataColumns));
+ }
+
+ private static DataType findIdentifierDataType(
+ Map<String, Column> originalColumnMap,
+ String columnName,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ if (originalColumnMap.containsKey(columnName)) {
+ return originalColumnMap.get(columnName).getType();
+ }
+ for (SupportedMetadataColumn metadataColumn :
supportedMetadataColumns) {
+ if (metadataColumn.getName().equals(columnName)) {
+ return metadataColumn.getType();
+ }
+ }
+ return METADATA_COLUMNS.stream()
+ .filter(column -> column.f0.equals(columnName))
+ .findFirst()
+ .map(column -> column.f1)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ "Referenced column "
+ + columnName
+ + " is not present in original
table."));
+ }
+
+ private static RelDataType toRelDataType(RelDataTypeFactory typeFactory,
DataType dataType) {
+ return CalciteDataTypeConverter.convertCalciteRelDataType(
+ typeFactory,
+
Collections.singletonList(Column.physicalColumn("__tmp", dataType)))
+ .getFieldList()
+ .get(0)
+ .getType();
+ }
+
+ private static void validateReferencedColumns(
+ Map<String, Column> originalColumnMap,
+ SqlNode exprNode,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ for (String columnName : parseColumnNameList(exprNode)) {
+ findIdentifierDataType(originalColumnMap, columnName,
supportedMetadataColumns);
+ }
+ }
+
+ private static boolean requiresBooleanTypeFallback(SqlNode sqlNode) {
+ if (!(sqlNode instanceof SqlBasicCall)) {
+ return false;
+ }
+ SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
+ switch (sqlBasicCall.getKind()) {
+ case AND:
+ case OR:
+ case NOT:
+ return sqlBasicCall.getOperandList().stream()
+
.anyMatch(TransformParser::requiresBooleanTypeFallback);
+ case IS_DISTINCT_FROM:
+ case IS_NOT_DISTINCT_FROM:
+ case IS_UNKNOWN:
+ case SIMILAR:
+ return true;
+ case LIKE:
+ return sqlBasicCall.getOperandList().size() == 3;
+ default:
+ return false;
+ }
+ }
+
/**
* Create a projection column from a simple identifier node (could be an
upstream physical
* column or a metadata column).
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
index eef0755c1..7a94e0586 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
@@ -89,6 +89,9 @@ public class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
// --------------------
public static final SqlBinaryOperator EQUALS = SqlStdOperatorTable.EQUALS;
public static final SqlBinaryOperator NOT_EQUALS =
SqlStdOperatorTable.NOT_EQUALS;
+ public static final SqlBinaryOperator IS_DISTINCT_FROM =
SqlStdOperatorTable.IS_DISTINCT_FROM;
+ public static final SqlBinaryOperator IS_NOT_DISTINCT_FROM =
+ SqlStdOperatorTable.IS_NOT_DISTINCT_FROM;
public static final SqlBinaryOperator GREATER_THAN =
SqlStdOperatorTable.GREATER_THAN;
public static final SqlBinaryOperator GREATER_THAN_OR_EQUAL =
SqlStdOperatorTable.GREATER_THAN_OR_EQUAL;
@@ -98,12 +101,16 @@ public class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
public static final SqlPostfixOperator IS_NULL =
SqlStdOperatorTable.IS_NULL;
public static final SqlPostfixOperator IS_NOT_NULL =
SqlStdOperatorTable.IS_NOT_NULL;
+ public static final SqlPostfixOperator IS_UNKNOWN =
SqlStdOperatorTable.IS_UNKNOWN;
+ public static final SqlPostfixOperator IS_NOT_UNKNOWN =
SqlStdOperatorTable.IS_NOT_UNKNOWN;
public static final SqlBetweenOperator BETWEEN =
SqlStdOperatorTable.BETWEEN;
public static final SqlBetweenOperator NOT_BETWEEN =
SqlStdOperatorTable.NOT_BETWEEN;
public static final SqlSpecialOperator LIKE = SqlStdOperatorTable.LIKE;
public static final SqlSpecialOperator NOT_LIKE =
SqlStdOperatorTable.NOT_LIKE;
+ public static final SqlSpecialOperator SIMILAR_TO =
SqlStdOperatorTable.SIMILAR_TO;
+ public static final SqlSpecialOperator NOT_SIMILAR_TO =
SqlStdOperatorTable.NOT_SIMILAR_TO;
public static final SqlBinaryOperator IN = SqlStdOperatorTable.IN;
public static final SqlBinaryOperator NOT_IN = SqlStdOperatorTable.NOT_IN;
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctionsTest.java
new file mode 100644
index 000000000..a7a1c3f39
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/ComparisonFunctionsTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.functions.impl;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link ComparisonFunctions}. */
+class ComparisonFunctionsTest {
+
+ @Test
+ void testLegacyNullComparisonBehavior() {
+ assertThat(ComparisonFunctions.valueEquals(null, 1)).isFalse();
+ assertThat(ComparisonFunctions.valueEquals(1, null)).isFalse();
+ assertThat(ComparisonFunctions.lessThan(null, 1)).isFalse();
+ assertThat(ComparisonFunctions.lessThanOrEqual(1, null)).isFalse();
+ assertThat(ComparisonFunctions.greaterThan(null, 1)).isFalse();
+ assertThat(ComparisonFunctions.greaterThanOrEqual(1, null)).isFalse();
+ }
+
+ @Test
+ void testThreeValuedLogicalOperators() {
+ assertThat(LogicalFunctions.and(true, (Boolean) null)).isNull();
+ assertThat(LogicalFunctions.and(false, (Boolean) null)).isFalse();
+ assertThat(LogicalFunctions.or(false, (Boolean) null)).isNull();
+ assertThat(LogicalFunctions.or(true, (Boolean) null)).isTrue();
+ assertThat(LogicalFunctions.not(null)).isNull();
+ }
+
+ @Test
+ void testLogicalOperatorsShortCircuitRightOperand() {
+ assertThat(LogicalFunctions.and(false, () ->
failIfEvaluated())).isFalse();
+ assertThat(LogicalFunctions.or(true, () ->
failIfEvaluated())).isTrue();
+ }
+
+ @Test
+ void testLegacyBetweenNullBehavior() {
+ assertThat(ComparisonFunctions.betweenAsymmetric((Integer) null, 1,
3)).isFalse();
+ assertThat(ComparisonFunctions.notBetweenAsymmetric((Integer) null, 1,
3)).isTrue();
+ }
+
+ @Test
+ void testLegacyInNullBehavior() {
+ assertThat(ComparisonFunctions.in(1, 1, null)).isTrue();
+ assertThat(ComparisonFunctions.in(1, 2, null)).isFalse();
+ assertThat(ComparisonFunctions.in(null, 1, 2)).isFalse();
+ assertThat(ComparisonFunctions.notIn(1, 2, null)).isTrue();
+ }
+
+ @Test
+ void testDistinctFromNeverReturnsUnknown() {
+ assertThat(ComparisonFunctions.isDistinctFrom(null, null)).isFalse();
+ assertThat(ComparisonFunctions.isDistinctFrom(null, 1)).isTrue();
+ assertThat(ComparisonFunctions.isDistinctFrom(1, 1)).isFalse();
+ assertThat(ComparisonFunctions.isNotDistinctFrom(null, null)).isTrue();
+ assertThat(ComparisonFunctions.isNotDistinctFrom(null, 1)).isFalse();
+ }
+
+ private static Boolean failIfEvaluated() {
+ throw new AssertionError("Right operand should not be evaluated.");
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
new file mode 100644
index 000000000..420b6713f
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.functions.impl;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link StringFunctions}. */
+class StringFunctionsTest {
+
+ @Test
+ void testLegacyLikeUsesJavaRegex() {
+ assertThat(StringFunctions.like("Alice", "A.*")).isTrue();
+ assertThat(StringFunctions.like("xabcy", "abc")).isTrue();
+ assertThat(StringFunctions.like("Alice", "A%")).isFalse();
+ assertThat(StringFunctions.notLike("Alice", "A.*")).isFalse();
+ }
+
+ @Test
+ void testLikeEscape() {
+ assertThat(StringFunctions.like("A%", "A$%", "$")).isTrue();
+ assertThat(StringFunctions.like("A_", "A$_", "$")).isTrue();
+ assertThat(StringFunctions.like("Alice", "A$%", "$")).isFalse();
+ }
+
+ @Test
+ void testLikeEscapeNullReturnsUnknown() {
+ assertThat(StringFunctions.like("Alice", "A%", null)).isNull();
+ }
+
+ @Test
+ void testSimilarTo() {
+ assertThat(StringFunctions.similarTo("Alice", "(A|B)%")).isTrue();
+ assertThat(StringFunctions.similarTo("Carol", "(A|B)%")).isFalse();
+ assertThat(StringFunctions.notSimilarTo("Alice", "(A|B)%")).isFalse();
+ }
+
+ @Test
+ void testSimilarToNullReturnsUnknown() {
+ assertThat(StringFunctions.similarTo(null, "(A|B)%")).isNull();
+ assertThat(StringFunctions.similarTo("Alice", null)).isNull();
+ assertThat(StringFunctions.notSimilarTo(null, "(A|B)%")).isNull();
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java
index 413b0dd6f..8298d384a 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.cdc.runtime.parser;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.variant.BinaryVariantInternalBuilder;
import org.apache.flink.cdc.common.types.variant.Variant;
import org.apache.flink.cdc.common.types.variant.VariantTypeException;
@@ -39,6 +42,7 @@ import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.stream.Stream;
/** Unit tests for the {@link JaninoCompiler}. */
@@ -214,6 +218,75 @@ class JaninoCompilerTest {
.isEqualTo(BinaryVariantInternalBuilder.parseJson(duplicatedNameJsonStr, true));
}
+ @Test
+ void testTranslatedLogicalExpressionCompilesWithJanino() throws
InvocationTargetException {
+ List<Column> columns =
+ List.of(
+ Column.physicalColumn("id", DataTypes.INT()),
+ Column.physicalColumn("uid", DataTypes.INT()));
+ Map<String, String> columnNameMap = Map.of("id", "$0", "uid", "$1");
+
+ ExpressionEvaluator andEvaluator =
+ compileTranslatedFilterExpression("id = 1 and 1 / uid > 0",
columns, columnNameMap);
+ Assertions.assertThat(andEvaluator.evaluate(new Object[] {2,
0})).isEqualTo(false);
+
+ ExpressionEvaluator orEvaluator =
+ compileTranslatedFilterExpression("id = 1 or 1 / uid > 0",
columns, columnNameMap);
+ Assertions.assertThat(orEvaluator.evaluate(new Object[] {1,
0})).isEqualTo(true);
+
+ List<Column> booleanColumns =
+ List.of(
+ Column.physicalColumn("left_bool",
DataTypes.BOOLEAN().notNull()),
+ Column.physicalColumn("right_bool",
DataTypes.BOOLEAN().notNull()),
+ Column.physicalColumn("divisor", DataTypes.INT()));
+ Map<String, String> booleanColumnNameMap =
+ Map.of("left_bool", "$0", "right_bool", "$1", "divisor", "$2");
+ List<String> booleanColumnNames = List.of("$0", "$1", "$2");
+ List<Class<?>> booleanColumnTypes = List.of(Boolean.class,
Boolean.class, Integer.class);
+
+ ExpressionEvaluator nativeAndEvaluator =
+ compileTranslatedFilterExpression(
+ "left_bool and right_bool",
+ booleanColumns,
+ booleanColumnNameMap,
+ booleanColumnNames,
+ booleanColumnTypes);
+ Assertions.assertThat(nativeAndEvaluator.evaluate(new Object[] {true,
true, 0}))
+ .isEqualTo(true);
+
+ ExpressionEvaluator conditionalAndEvaluator =
+ compileTranslatedFilterExpression(
+ "left_bool and 1 / divisor > 0",
+ booleanColumns,
+ booleanColumnNameMap,
+ booleanColumnNames,
+ booleanColumnTypes);
+ Assertions.assertThat(conditionalAndEvaluator.evaluate(new Object[]
{false, true, 0}))
+ .isEqualTo(false);
+
+ ExpressionEvaluator conditionalOrEvaluator =
+ compileTranslatedFilterExpression(
+ "left_bool or 1 / divisor > 0",
+ booleanColumns,
+ booleanColumnNameMap,
+ booleanColumnNames,
+ booleanColumnTypes);
+ Assertions.assertThat(conditionalOrEvaluator.evaluate(new Object[]
{true, false, 0}))
+ .isEqualTo(true);
+
+ List<Column> nullableBooleanColumns =
+ List.of(Column.physicalColumn("nullable_bool",
DataTypes.BOOLEAN()));
+ Map<String, String> nullableBooleanColumnNameMap =
Map.of("nullable_bool", "$0");
+ ExpressionEvaluator nullableOrEvaluator =
+ compileTranslatedFilterExpression(
+ "nullable_bool or false",
+ nullableBooleanColumns,
+ nullableBooleanColumnNameMap,
+ List.of("$0"),
+ List.of(Boolean.class));
+ Assertions.assertThat(nullableOrEvaluator.evaluate(new Object[]
{null})).isNull();
+ }
+
@Test
void testLargeNumericLiterals() {
// Test parsing integer literals
@@ -293,4 +366,34 @@ class JaninoCompilerTest {
}
});
}
+
+ private static ExpressionEvaluator compileTranslatedFilterExpression(
+ String expression, List<Column> columns, Map<String, String>
columnNameMap) {
+ return compileTranslatedFilterExpression(
+ expression,
+ columns,
+ columnNameMap,
+ List.of("$0", "$1"),
+ List.of(Integer.class, Integer.class));
+ }
+
+ private static ExpressionEvaluator compileTranslatedFilterExpression(
+ String expression,
+ List<Column> columns,
+ Map<String, String> columnNameMap,
+ List<String> columnNames,
+ List<Class<?>> columnTypes) {
+ String janinoExpression =
+ TransformParser.translateFilterExpressionToJaninoExpression(
+ expression,
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0],
+ columnNameMap);
+ return JaninoCompiler.compileExpression(
+ JaninoCompiler.loadSystemFunction(janinoExpression),
+ columnNames,
+ columnTypes,
+ Boolean.class);
+ }
}
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 a522f4f34..9943ed069 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
@@ -172,21 +172,25 @@ class TransformParserTest {
@Test
void testTranslateFilterToJaninoExpression() {
- testFilterExpression("id is not null", "null != id");
- testFilterExpression("id is null", "null == id");
+ testFilterExpression("id is not null", "isNotNull(id)");
+ testFilterExpression("id is null", "isNull(id)");
testFilterExpression("id = 1 and uid = 2", "valueEquals(id, 1) &&
valueEquals(uid, 2)");
testFilterExpression("id = 1 or id = 2", "valueEquals(id, 1) ||
valueEquals(id, 2)");
- testFilterExpression("not (id = 1)", "!valueEquals(id, 1)");
+ testFilterExpression("not (id = 1)", "not(valueEquals(id, 1))");
testFilterExpression("id = '1'", "valueEquals(id, \"1\")");
testFilterExpression("id <> '1'", "!valueEquals(id, \"1\")");
+ testFilterExpression("id is distinct from '1'", "isDistinctFrom(id,
\"1\")");
+ testFilterExpression("id is not distinct from '1'",
"isNotDistinctFrom(id, \"1\")");
testFilterExpression("d between d1 and d2", "betweenAsymmetric(d, d1,
d2)");
testFilterExpression("d not between d1 and d2",
"notBetweenAsymmetric(d, d1, d2)");
testFilterExpression("d in (d1, d2)", "in(d, d1, d2)");
testFilterExpression("d not in (d1, d2)", "notIn(d, d1, d2)");
- testFilterExpression("id is false", "false == id");
- testFilterExpression("id is not false", "true == id");
- testFilterExpression("id is true", "true == id");
- testFilterExpression("id is not true", "false == id");
+ testFilterExpression("id is false", "isFalse(id)");
+ testFilterExpression("id is not false", "isNotFalse(id)");
+ testFilterExpression("id is true", "isTrue(id)");
+ testFilterExpression("id is not true", "isNotTrue(id)");
+ testFilterExpression("id is unknown", "isNull(id)");
+ testFilterExpression("id is not unknown", "isNotNull(id)");
testFilterExpression("a || b", "concat(a, b)");
testFilterExpression("CHAR_LENGTH(id)", "charLength(id)");
testFilterExpression("trim(id)", "trim(\"BOTH\", \" \", id)");
@@ -198,6 +202,9 @@ class TransformParserTest {
testFilterExpression("SUBSTR(a,1)", "substr(a, 1)");
testFilterExpression("id like '^[a-zA-Z]'", "like(id, \"^[a-zA-Z]\")");
testFilterExpression("id not like '^[a-zA-Z]'", "notLike(id,
\"^[a-zA-Z]\")");
+ testFilterExpression("id like 'A$%' escape '$'", "like(id, \"A$%\",
\"$\")");
+ testFilterExpression("id similar to '(A|B)%'", "similarTo(id,
\"(A|B)%\")");
+ testFilterExpression("id not similar to '(A|B)%'", "notSimilarTo(id,
\"(A|B)%\")");
testFilterExpression("abs(2)", "abs(2)");
testFilterExpression("ceil(2)", "ceil(2)");
testFilterExpression("ceiling(2)", "ceil(2)");
@@ -329,7 +336,7 @@ class TransformParserTest {
"TIMESTAMPADD(YEAR, 1, dt)", "timestampadd(\"YEAR\", 1, dt,
__time_zone__)");
testFilterExpression(
"timestampadd(year, 1, dt)", "timestampadd(\"YEAR\", 1, dt,
__time_zone__)");
- testFilterExpression("IF(a>b,a,b)", "greaterThan(a, b) ? a : b");
+ testFilterExpression("IF(a>b,a,b)", "isTrue(greaterThan(a, b)) ? a :
b");
testFilterExpression("NULLIF(a,b)", "nullif(a, b)");
testFilterExpression("COALESCE(a,b,c)", "coalesce(a, b, c)");
testFilterExpression("id + 2", "id + 2");
@@ -348,19 +355,19 @@ class TransformParserTest {
testFilterExpression("upper(lower(id))", "upper(lower(id))");
testFilterExpression(
"abs(uniq_id) > 10 and id is not null",
- "greaterThan(abs(uniq_id), 10) && null != id");
+ "greaterThan(abs(uniq_id), 10) && isNotNull(id)");
testFilterExpression(
"case id when 1 then 'a' when 2 then 'b' else 'c' end",
- "(valueEquals(id, 1) ? \"a\" : valueEquals(id, 2) ? \"b\" :
\"c\")");
+ "(isTrue(valueEquals(id, 1)) ? \"a\" : isTrue(valueEquals(id,
2)) ? \"b\" : \"c\")");
testFilterExpression(
"case when id = 1 then 'a' when id = 2 then 'b' else 'c' end",
- "(valueEquals(id, 1) ? \"a\" : valueEquals(id, 2) ? \"b\" :
\"c\")");
+ "(isTrue(valueEquals(id, 1)) ? \"a\" : isTrue(valueEquals(id,
2)) ? \"b\" : \"c\")");
testFilterExpression(
"case id when 1 then 'a' when 2 then 'b' else 'c' end",
- "(valueEquals(id, 1) ? \"a\" : valueEquals(id, 2) ? \"b\" :
\"c\")");
+ "(isTrue(valueEquals(id, 1)) ? \"a\" : isTrue(valueEquals(id,
2)) ? \"b\" : \"c\")");
testFilterExpression(
"case when id = 1 then 'a' when id = 2 then 'b' else 'c' end",
- "(valueEquals(id, 1) ? \"a\" : valueEquals(id, 2) ? \"b\" :
\"c\")");
+ "(isTrue(valueEquals(id, 1)) ? \"a\" : isTrue(valueEquals(id,
2)) ? \"b\" : \"c\")");
testFilterExpression("cast(id||'0' as int)", "castToInteger(concat(id,
\"0\"))");
testFilterExpression("cast(1 as string)", "castToString(1)");
testFilterExpression("cast(1 as boolean)", "castToBoolean(1)");
@@ -391,6 +398,38 @@ class TransformParserTest {
testFilterExpression("try_parse_json(jsonStr)",
"tryParseJson(jsonStr)");
}
+ @Test
+ void testTranslateLogicalFilterToJaninoExpressionByNullability() {
+ List<Column> columns =
+ List.of(
+ Column.physicalColumn("left_bool",
DataTypes.BOOLEAN().notNull()),
+ Column.physicalColumn("right_bool",
DataTypes.BOOLEAN().notNull()),
+ Column.physicalColumn("nullable_bool",
DataTypes.BOOLEAN()));
+
+ testFilterExpressionWithColumns(
+ "left_bool and right_bool", "left_bool && right_bool",
columns);
+ testFilterExpressionWithColumns(
+ "left_bool or right_bool", "left_bool || right_bool", columns);
+ testFilterExpressionWithColumns(
+ "left_bool and nullable_bool",
+ "left_bool ? nullable_bool : Boolean.FALSE",
+ columns);
+ testFilterExpressionWithColumns(
+ "left_bool or nullable_bool", "left_bool ? Boolean.TRUE :
nullable_bool", columns);
+ testFilterExpressionWithColumns(
+ "nullable_bool and right_bool",
+ lazyLogicalFunction("and", "nullable_bool", "right_bool"),
+ columns);
+ testFilterExpressionWithColumns(
+ "nullable_bool or right_bool",
+ lazyLogicalFunction("or", "nullable_bool", "right_bool"),
+ columns);
+ testFilterExpressionWithColumns(
+ "nullable_bool or false",
+ lazyLogicalFunction("or", "nullable_bool", "false"),
+ columns);
+ }
+
@Test
public void testTranslateItemAccessToJaninoExpression() {
// Test collection access functions (ARRAY, MAP) with proper column
schema
@@ -596,12 +635,12 @@ class TransformParserTest {
"ProjectionColumn{column=`id` INT 'id',
expression='id', scriptExpression='$0', originalColumnNames=[id],
columnNameMap={id=$0}}",
"ProjectionColumn{column=`name2` STRING,
expression='UPPER(`TB`.`name`)', scriptExpression='upper($0)',
originalColumnNames=[name], columnNameMap={name=$0}}",
"ProjectionColumn{column=`sex2` STRING,
expression='UPPER(`TB`.`sex`)', scriptExpression='upper($0)',
originalColumnNames=[sex], columnNameMap={sex=$0}}",
- "ProjectionColumn{column=`address2` BINARY(50),
expression='CASE WHEN `TB`.`address` IS NOT NULL THEN `TB`.`address` ELSE
`TB`.`address` END', scriptExpression='(null != $0 ? $0 : $0)',
originalColumnNames=[address, address, address], columnNameMap={address=$0}}",
- "ProjectionColumn{column=`phone2` VARBINARY(50),
expression='CASE WHEN `TB`.`phone` IS NOT NULL THEN `TB`.`phone` ELSE
`TB`.`phone` END', scriptExpression='(null != $0 ? $0 : $0)',
originalColumnNames=[phone, phone, phone], columnNameMap={phone=$0}}",
- "ProjectionColumn{column=`deposit2` DECIMAL(10, 2),
expression='CASE WHEN `TB`.`deposit` IS NOT NULL THEN `TB`.`deposit` ELSE
`TB`.`deposit` END', scriptExpression='(null != $0 ? $0 : $0)',
originalColumnNames=[deposit, deposit, deposit], columnNameMap={deposit=$0}}",
- "ProjectionColumn{column=`birthday2` TIMESTAMP(3),
expression='CASE WHEN `TB`.`birthday` IS NOT NULL THEN `TB`.`birthday` ELSE
`TB`.`birthday` END', scriptExpression='(null != $0 ? $0 : $0)',
originalColumnNames=[birthday, birthday, birthday],
columnNameMap={birthday=$0}}",
- "ProjectionColumn{column=`birthday_ltz2`
TIMESTAMP_LTZ(3), expression='CASE WHEN `TB`.`birthday_ltz` IS NOT NULL THEN
`TB`.`birthday_ltz` ELSE `TB`.`birthday_ltz` END', scriptExpression='(null !=
$0 ? $0 : $0)', originalColumnNames=[birthday_ltz, birthday_ltz, birthday_ltz],
columnNameMap={birthday_ltz=$0}}",
- "ProjectionColumn{column=`update_time2` TIME(3),
expression='CASE WHEN `TB`.`update_time` IS NOT NULL THEN `TB`.`update_time`
ELSE `TB`.`update_time` END', scriptExpression='(null != $0 ? $0 : $0)',
originalColumnNames=[update_time, update_time, update_time],
columnNameMap={update_time=$0}}");
+ "ProjectionColumn{column=`address2` BINARY(50),
expression='CASE WHEN `TB`.`address` IS NOT NULL THEN `TB`.`address` ELSE
`TB`.`address` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)',
originalColumnNames=[address, address, address], columnNameMap={address=$0}}",
+ "ProjectionColumn{column=`phone2` VARBINARY(50),
expression='CASE WHEN `TB`.`phone` IS NOT NULL THEN `TB`.`phone` ELSE
`TB`.`phone` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)',
originalColumnNames=[phone, phone, phone], columnNameMap={phone=$0}}",
+ "ProjectionColumn{column=`deposit2` DECIMAL(10, 2),
expression='CASE WHEN `TB`.`deposit` IS NOT NULL THEN `TB`.`deposit` ELSE
`TB`.`deposit` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)',
originalColumnNames=[deposit, deposit, deposit], columnNameMap={deposit=$0}}",
+ "ProjectionColumn{column=`birthday2` TIMESTAMP(3),
expression='CASE WHEN `TB`.`birthday` IS NOT NULL THEN `TB`.`birthday` ELSE
`TB`.`birthday` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)',
originalColumnNames=[birthday, birthday, birthday],
columnNameMap={birthday=$0}}",
+ "ProjectionColumn{column=`birthday_ltz2`
TIMESTAMP_LTZ(3), expression='CASE WHEN `TB`.`birthday_ltz` IS NOT NULL THEN
`TB`.`birthday_ltz` ELSE `TB`.`birthday_ltz` END',
scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)',
originalColumnNames=[birthday_ltz, birthday_ltz, birthday_ltz],
columnNameMap={birthday_ltz=$0}}",
+ "ProjectionColumn{column=`update_time2` TIME(3),
expression='CASE WHEN `TB`.`update_time` IS NOT NULL THEN `TB`.`update_time`
ELSE `TB`.`update_time` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 :
$0)', originalColumnNames=[update_time, update_time, update_time],
columnNameMap={update_time=$0}}");
Assertions.assertThat(result).hasToString("[" + String.join(", ",
expected) + "]");
}
@@ -887,6 +926,13 @@ class TransformParserTest {
Assertions.assertThat(janinoExpression).isEqualTo(expressionExpect);
}
+ private static String lazyLogicalFunction(
+ String functionName, String leftOperand, String rightOperand) {
+ return String.format(
+ "%s(%s, new java.util.function.Supplier<Boolean>() { public
Boolean get() { return %s; } })",
+ functionName, leftOperand, rightOperand);
+ }
+
private void testFilterExpressionWithUdf(String expression, String
expressionExpect) {
testFilterExpressionWithUdf(
expression, expressionExpect, DUMMY_COLUMNS,
Collections.emptyMap());
