[FLINK-30841] Incorrect calc merge generate wrong plan

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.16.1
• Fix Version/s: 1.17.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: lincoln lee
• Reporter: lincoln lee
• Created: 31/Jan/23 07:00
• Updated: 03/Feb/23 13:38
• Resolved: 03/Feb/23 13:38

Description

currently we have a `FlinkCalcMergeRuleTest`, take one test as example:

  @Test
  def testCalcMergeWithNonDeterministicExpr1(): Unit = {
    val sqlQuery = "SELECT a, a1 FROM (SELECT a, random_udf(a) AS a1 FROM MyTable) t WHERE a1 > 10"
    util.verifyRelPlan(sqlQuery)
  }

the current final optimized plan will be wrong:

Calc(select=[a, random_udf(b) AS a1], where=[(random_udf(b) > 10)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])

the merged calc contains two `random_udf` call, users may encounter the result satisfied by where predicate (>10) but the selected column <= 10, that's counter-intuitive for users

the expected plan is:

Calc(select=[a, a1], where=[(a1 > 10)])
+- Calc(select=[a, random_udf(b) AS a1])
   +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])

Issue Links

• links to: GitHub Pull Request #21799