[FLINK-31830] Coalesce on nested fields with different nullabilities will get wrong plan

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.14.6
• Component/s: Table SQL / Planner
• Assignee: Jane Chan
• Reporter: lincoln lee
• Created: 18/Apr/23 02:06
• Updated: 01/Jul/24 08:20

Description

A test case similar to FLINK-31829, only changes the nullable field `a.np` to not null, will get a wrong plan in 1.14.x (reported from the community user):

  @Test
  def testCoalesceOnNestedColumns(): Unit = {
    val tEnv = util.tableEnv
    val tableDescriptor = TableDescriptor.forConnector("datagen")
        .schema(Schema.newBuilder
            .column("id", DataTypes.INT.notNull)
            .column("a", DataTypes.ROW(DataTypes.FIELD("np", DataTypes.INT.notNull())).nullable)
            .build)
        .build
    tEnv.createTemporaryTable("t1", tableDescriptor)
    tEnv.createTemporaryTable("t2", tableDescriptor)
    val res = tEnv.executeSql("EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np is null")
    res.print()
}  

== Abstract Syntax Tree ==
LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])
+- LogicalFilter(condition=[OR(IS NULL($1), IS NULL(CAST($1.np):INTEGER))])
   +- LogicalJoin(condition=[=($0, $2)], joinType=[left])
      :- LogicalTableScan(table=[[default_catalog, default_database, t1]])
      +- LogicalTableScan(table=[[default_catalog, default_database, t2]])

the top project in the ast is wrong: `LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])`, the `c1=[CAST($1.np):INTEGER]` relate to `COALESCE(a.a.np, b.a.np) c1` is incorrect,
but this works fine when using sql ddl to create tables

  @Test
  def testCoalesceOnNestedColumns2(): Unit = {
    val tEnv = util.tableEnv
    tEnv.executeSql(
      s"""
         |create temporary table t1 (
         |  id int not null,
         |  a row<np int not null>
         |) with (
         | 'connector' = 'datagen'
         |)
         |""".stripMargin)
    tEnv.executeSql(
      s"""
         |create temporary table t2 (
         |  id int not null,
         |  a row<np int not null>
         |) with (
         | 'connector' = 'datagen'
         |)
         |""".stripMargin)
    val res = tEnv.executeSql(
      "EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np is null")
    res.print()
  }

from 1.15, the coalesce will be a new builtin function, and the ast looks correct in version 1.15+, while before 1.15 it was rewritten as `case when`

Issue Links

• duplicates: FLINK-32464 AssertionError when converting between Table and SQL with selection and type cast
• is related to: FLINK-30282 Logical type ROW lost inner field's nullability after convert to RelDataType
• is related to: FLINK-31829 Conversion to relational algebra failed to preserve datatypes' nullabilities
• is related to: FLINK-33217 Flink SQL: UNNEST fails with on LEFT JOIN with NOT NULL type in array
• is related to: FLINK-33597 Can not use a nested column for a join condition
• relates to: FLINK-20539 Type mismatch when using ROW in computed column
• mentioned in: Page Loading...