[FLINK-38565][table] Reduce TRY_CAST chain if source and target types are same

Author: snuyanzin

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkConvertletTable.java

@@ -101,7 +101,7 @@ private RexNode convertTryCast(SqlRexContext cx, final SqlCall call) {
         }
         type = typeFactory.createTypeWithNullability(type, true);
 
-        if (SqlUtil.isNullLiteral(leftNode, false)) {
+        if (SqlUtil.isNullLiteral(leftNode, false) || type.equals(valueRex.getType())) {
             final SqlValidatorImpl validator = (SqlValidatorImpl) cx.getValidator();
             validator.setValidatedNodeType(leftNode, type);
             return cx.convertExpression(leftNode);

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/CalcTest.xml

@@ -532,6 +532,74 @@ LogicalProject(EXPR$0=[ROW(1, _UTF-16LE'Hi', $0)])
       <![CDATA[
 Calc(select=[ROW(1, 'Hi', a) AS EXPR$0])
 +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testRepeatedTryCast">
+    <Resource name="sql">
+      <![CDATA[SELECT TRY_CAST(TRY_CAST(a AS STRING) AS STRING) FROM MyTable]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(EXPR$0=[TRY_CAST($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized exec plan">
+      <![CDATA[
+Calc(select=[TRY_CAST(a AS VARCHAR(2147483647)) AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testRepeatedTryCastAfterCast">
+    <Resource name="sql">
+      <![CDATA[SELECT TRY_CAST(CAST(a AS STRING) AS STRING) FROM MyTable]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(EXPR$0=[CAST($0):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized exec plan">
+      <![CDATA[
+Calc(select=[CAST(a AS VARCHAR(2147483647)) AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testRepeatedTryCastSameType">
+    <Resource name="sql">
+      <![CDATA[SELECT TRY_CAST(a AS BIGINT) FROM MyTable]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(EXPR$0=[$0])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized exec plan">
+      <![CDATA[
+Calc(select=[a AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testRepeatedTryCastDifferentType">
+    <Resource name="sql">
+      <![CDATA[SELECT TRY_CAST(TRY_CAST(a AS STRING) AS INTEGER) FROM MyTable]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(EXPR$0=[TRY_CAST(TRY_CAST($0))])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized exec plan">
+      <![CDATA[
+Calc(select=[TRY_CAST(TRY_CAST(a AS VARCHAR(2147483647)) AS INTEGER) AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
 ]]>
     </Resource>
   </TestCase>

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala

@@ -230,4 +230,28 @@ class CalcTest extends TableTestBase {
                            |SELECT my_row = ROW(1, 'str') from src
                            |""".stripMargin)
   }
+
+  @Test
+  def testRepeatedTryCast(): Unit = {
+    val sqlQuery = "SELECT TRY_CAST(TRY_CAST(a AS STRING) AS STRING) FROM MyTable"
+    util.verifyExecPlan(sqlQuery)
+  }
+
+  @Test
+  def testRepeatedTryCastAfterCast(): Unit = {
+    val sqlQuery = "SELECT TRY_CAST(CAST(a AS STRING) AS STRING) FROM MyTable"
+    util.verifyExecPlan(sqlQuery)
+  }
+
+  @Test
+  def testRepeatedTryCastSameType(): Unit = {
+    val sqlQuery = "SELECT TRY_CAST(a AS BIGINT) FROM MyTable"
+    util.verifyExecPlan(sqlQuery)
+  }
+
+  @Test
+  def testRepeatedTryCastDifferentType(): Unit = {
+    val sqlQuery = "SELECT TRY_CAST(TRY_CAST(a AS STRING) AS INTEGER) FROM MyTable"
+    util.verifyExecPlan(sqlQuery)
+  }
 }