[
https://issues.apache.org/jira/browse/FLINK-40799?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sergey Nuyanzin reassigned FLINK-40799:
---------------------------------------
Assignee: Chris Johnson (was: Sergey Nuyanzin)
> 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
> Assignee: Chris Johnson
> Priority: Major
>
> 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)
