[FLINK-35804][table-planner] Fix incorrect calc merge during decorrelate phase

[zhaorongsheng]

What is the purpose of the change

Like the same issue in FLINK-30841.

Take one test as example:

@Test
def testCalcMergeWithNonDeterministicExpr3(): Unit =
{ 
  val sqlUdtfQuery = "SELECT a, b, len FROM MyTable, LATERAL TABLE (length_udtf(c)) AS T(len)" 
  val sqlView1Query = "SELECT a, b, len " + s"FROM ($sqlUdtfQuery) t JOIN MyTable_Join t2 " + "ON t.a = t2.d" 
  val view1 = util.tableEnv.sqlQuery(sqlView1Query) 
  util.tableEnv.createTemporaryView("View1", view1) 
  val sqlView2Query = "SELECT random_udf(b) AS r FROM View1" 
  val view2 = util.tableEnv.sqlQuery(sqlView2Query) 
  util.tableEnv.createTemporaryView("View2", view2) 
  val sqlQuery = "SELECT r FROM View2 WHERE r > 10" 
  util.verifyRelPlan(sqlQuery) 
}

optimized plan will be wrong:

Calc(select=[random_udf(b) AS r])
+- Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, b, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, b], where=[>(random_udf(b), 10)])
: +- Correlate(invocation=[length_udtf($cor0.c)], correlate=[table(length_udtf($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(BIGINT a, INTEGER b, VARCHAR(2147483647) c, INTEGER EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[d]])
+- Calc(select=[d])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable_Join, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])

the expected plan is:

Calc(select=[r], where=[>(r, 10)])
+- Calc(select=[random_udf(b) AS r])
+-Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, b, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[hash[a]])
:+-Calc(select=[a, b])
:+-Correlate(invocation=[length_udtf($cor0.c)], correlate=[table(length_udtf($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(BIGINT a, INTEGER b, VARCHAR(2147483647) c, INTEGER EXPR$0)], joinType=[INNER])
:+-LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+-Exchange(distribution=[hash[d]])
+-Calc(select=[d])
+-LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])

Brief change log

change rule related to this problem in RelDecorrelator

Verifying this change

Add new plan test in CalcTest

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[flinkbot]

CI report:

• e22e881 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[zhaorongsheng]

@lincoln-lil @godfreyhe Would you mind reviewing this pr? Thanks~

[zhaorongsheng]

@flinkbot Could someone review this pr? Thanks~

[lincoln-lil]

@zhaorongsheng Thanks for fixing this! The commit msg can be "Fix incorrect calc merge during decorrelate phase" since it happens in RelDecorrelator.

[lincoln-lil]

We can use a minimal case to reproduce the error, e.g.,

  @Test
  def testCalcMergeWithCorrelate(): Unit = {
    util.addTemporarySystemFunction("str_split", new StringSplit())
    val sqlQuery =
      """
        |
        |SELECT a, r FROM (
        | SELECT a, random_udf(b) r FROM (
        |  select a, b, c1 FROM MyTable, LATERAL TABLE(str_split(c)) AS T(c1)
        | ) t
        |)
        |WHERE r > 10
        |""".stripMargin
    util.verifyRelPlan(sqlQuery)
  }

[zhaorongsheng]

OK. The code and commit msg has been updated. Could you please check it again? Thanks~

[lincoln-lil]

LGTM!

[lincoln-lil]

@zhaorongsheng Thanks for the updating! Just one minor comment before merging.

[lincoln-lil]

@zhaorongsheng Could you also cherry pick this patch into release-1.20 branch?

[lincoln-lil]

@zhaorongsheng Could you also cherry pick this patch into release-1.20 branch?

I've created the backport pr #25084 to catch up 1.20's rc1 building.

[zhaorongsheng]

@zhaorongsheng Could you also cherry pick this patch into release-1.20 branch?

I've created the backport pr #25084 to catch up 1.20's rc1 building.

OK!