[FLINK-40151][runtime] Support missing YAML Transform predicates and nullable BOOLEAN logical evaluation

[haruki-830]

Summary

This commit improves YAML Transform predicate support and nullable BOOLEAN logical evaluation. It adds support for several previously unsupported SQL predicates, and fixes runtime failures when logical expressions evaluate nullable BOOLEAN operands.

Key Changes

Additional Predicate Support

• Added support for IS UNKNOWN / IS NOT UNKNOWN.
• Added support for IS DISTINCT FROM / IS NOT DISTINCT FROM.
• Added support for LIKE ... ESCAPE.
• Added support for SIMILAR TO / NOT SIMILAR TO.

Nullable BOOLEAN Logical Evaluation

• Added null-safe handling for AND, OR, and NOT with nullable BOOLEAN operands.
• Added null-safe handling for IS TRUE / IS NOT TRUE.
• Added null-safe handling for IS FALSE / IS NOT FALSE.
• Kept boolean condition consumers null-safe so UNKNOWN does not cause Java unboxing failures.

Test Coverage

• Added unit tests for distinct predicates, nullable BOOLEAN logical functions, legacy comparison behavior, legacy two-argument LIKE, LIKE ... ESCAPE, and SIMILAR TO.
• Updated parser tests for generated Janino expressions.
• Updated YAML transform specs for nullable BOOLEAN logic and the newly supported predicates.

JIRA Reference
https://issues.apache.org/jira/browse/FLINK-40151

[haruki-830]

@lvyanquan Could u PTAL when u have time? Thanks!

[lvyanquan]

Thanks for the work here. The changes split into three buckets with very different compatibility impact. I'd suggest applying buckets 1–2 unconditionally and gating bucket 3 behind an explicit option.

1. New operators (currently unparseable) — keep always-on

SQL: IS [NOT] DISTINCT FROM, IS [NOT] UNKNOWN, SIMILAR TO / NOT SIMILAR TO, LIKE ... ESCAPE '<char>'.

These fail validation on the current release, so no existing pipeline can depend on them. Safe to enable unconditionally.

2. Crash → 3-valued logic, non-NULL behavior unchanged — keep always-on

SQL: AND, OR, NOT, IS [NOT] TRUE, IS [NOT] FALSE where the operand is a NULL boolean (nullable BOOLEAN column or CAST(NULL AS BOOLEAN)).

These previously threw an NPE on NULL; results for non-NULL operands are identical. Pure crash fix, safe to enable unconditionally.

3. Silent result changes — must be gated behind an explicit option

SQL:

• Comparison / set predicates: =, <>, >, >=, <, <=, IN, NOT IN, BETWEEN, NOT BETWEEN. On a NULL operand the result changes from false/true to NULL. For negated forms (<>, NOT IN, NOT BETWEEN) this is true → NULL, which drops rows in WHERE and flips CASE / IF branches.
• LIKE / NOT LIKE: pattern engine changes from Java regex (Pattern.find(), substring match) to SQL wildcards (%, _, whole-string match). This changes results for every row, not just NULL — e.g. 'xabcy' LIKE 'abc' goes true → false, col LIKE 'A%' goes false → true.

This is a backward-incompatible change to user-facing semantics, and the silent nature makes it dangerous: after the upgrade the YAML is unchanged, the job stays healthy, and no error or warning is emitted, yet the output data changes. For a CDC sync, altered filter results change which INSERT / UPDATE / DELETE events reach the sink, so the sink silently diverges from the source with no signal to detect or repair it. It therefore must not be the default on a minor/patch release.

Recommendation

Introduce pipeline.transform.expression.semantics (enum LEGACY | FLINK_SQL, default LEGACY). Keep the current behavior under LEGACY; apply bucket 3 only under FLINK_SQL. Buckets 1 and 2 remain always-on.

Also, to ease review, consider splitting this so buckets 1–2 and bucket 3 land as separate commits (or separate PRs) — the crash/parse fixes can then be merged quickly, while the semantic change is reviewed on its own.

[haruki-830]

Also, to ease review, consider splitting this so buckets 1–2 and bucket 3 land as separate commits (or separate PRs) — the crash/parse fixes can then be merged quickly, while the semantic change is reviewed on its own.

Thanks for the detailed review and the compatibility analysis. I agree with the split.

I will separate this work into two PRs:

1. The first PR will only include the always-on changes with no silent behavior change:
   • newly supported operators that are currently unparseable, such as IS [NOT] DISTINCT FROM, IS [NOT] UNKNOWN, SIMILAR TO / NOT SIMILAR TO, and LIKE ... ESCAPE
   • NULL boolean crash fixes for AND, OR, NOT, IS [NOT] TRUE, and IS [NOT] FALSE
2. The second PR will handle the backward-incompatible semantic changes behind an explicit option:
   • comparison predicates, IN, BETWEEN, and legacy LIKE / NOT LIKE
   • add pipeline.transform.expression.semantics with LEGACY as the default and FLINK_SQL as the opt-in SQL-compatible behavior

This should allow the parse/crash fixes to move forward independently, while keeping the silent result-changing semantics isolated and explicitly gated.

[lvyanquan]

Could we add an end-to-end compilation test to JaninoCompilerTest?

The existing tests only verify the generated expression string or invoke
LogicalFunctions directly. They do not compile the generated expression with
Janino, so they cannot detect that Janino 3.1.10 does not support lambda
expressions.

[haruki-830]

Addressed in the latest commit.

[lvyanquan]

Please update both the English and Chinese transform documentation to reflect these changes

[lvyanquan]

Also cc’ing @yuxiqian.