Chris Johnson created FLINK-40799:
-------------------------------------
Summary: ClassCastException in FlinkMarkChangelogNormalizeProgram
when an upsert source is read twice with the same boolean column filter
Key: FLINK-40799
URL: https://issues.apache.org/jira/browse/FLINK-40799
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 2.3.0
Reporter: Chris Johnson
Planning fails with a ClassCastException when a query reads the same upsert
source more than once and every reader filters on the same bare boolean column.
The query is valid SQL. 2.3.0 is affected; 2.2.1 is not.
*Reproduce*
{code:sql}
CREATE TABLE foo (
id INT,
flag BOOLEAN,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'values',
'changelog-mode' = 'I,UA,D'
);
EXPLAIN
SELECT id FROM foo WHERE flag
UNION ALL
SELECT id FROM foo WHERE flag;
{code}
The 'values' connector is the planner test connector (flink-table-planner
test-jar). 'I,UA,D' makes it an upsert source, so the plan gets a
ChangelogNormalize. Any upsert source should behave the same, for example
upsert-kafka.
A self join fails the same way:
{code:sql}
EXPLAIN
SELECT f1.id FROM foo AS f1 JOIN foo AS f2 ON f1.id = f2.id
WHERE f1.flag AND f2.flag;
{code}
Verified with TableEnvironment.explainSql against the released Maven Central
artifacts: both queries fail on 2.3.0 and plan successfully on 2.2.1.
*Actual*
{noformat}
java.lang.ClassCastException: class org.apache.calcite.rex.RexInputRef cannot
be cast to class org.apache.calcite.rex.RexCall
at
org.apache.flink.table.planner.plan.rules.physical.stream.FlinkMarkChangelogNormalizeProgram.calculateCommonCondition(FlinkMarkChangelogNormalizeProgram.java:183)
at
org.apache.flink.table.planner.plan.rules.physical.stream.FlinkMarkChangelogNormalizeProgram.optimize(FlinkMarkChangelogNormalizeProgram.java:158)
at
org.apache.flink.table.planner.plan.rules.physical.stream.FlinkMarkChangelogNormalizeProgram.optimize(FlinkMarkChangelogNormalizeProgram.java:135)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:59)
{noformat}
*Cause*
FlinkMarkChangelogNormalizeProgram looks for filter conditions shared by all
ChangelogNormalize nodes on the same source. Since FLINK-38693 it ORs the
conditions together and calls RexUtil.pullFactors, then casts the result:
{code:java}
final RexCall factors = (RexCall) RexUtil.pullFactors(rexBuilder, or);
{code}
https://github.com/apache/flink/blob/654f0edc88676df691b8a59f5e44186602b5820f/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java#L183
pullFactors does not always return a RexCall. When every reader has the same
single boolean column as its filter, pullFactors(OR(flag, flag)) returns the
RexInputRef for flag, and the cast fails. Writing the filter as flag = TRUE or
flag IS TRUE does not help, because both are simplified to flag before this
point.
The cast was added in FLINK-38693 (commit
e5f53f272dcdac4249e2433091fbacf0e1df2daf). Before that, in 2.2.x,
calculateCommonCondition used a set intersection and did not have this problem.
*What does not trigger it*
- The source is read only once.
- Only one reader has a filter (the method returns early when conditions are
null).
- The shared filter is a call, for example WHERE id > 0 on both sides.
- The source is append only, so there is no ChangelogNormalize.
*Workaround*
Wrap the boolean in at least one reader, for example WHERE COALESCE(flag,
FALSE).
*Suggested fix*
Check the type before casting and return no common condition when it is not a
RexCall, the same as the existing early return:
{code:java}
final RexNode pulled = RexUtil.pullFactors(rexBuilder, or);
if (!(pulled instanceof RexCall)) {
return List.of();
}
final RexCall factors = (RexCall) pulled;
{code}
Returning List.of(pulled) as the common filter may be better, since the whole
condition is shared in this case. I am not sure which is preferred.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
