[flink] branch release-1.11 updated: [FLINK-19108][table] Stop expanding the identifiers with scope aliased by the system with 'EXPR$' prefix

[lz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.11 by this push:
     new f2cc139  [FLINK-19108][table] Stop expanding the identifiers with scope aliased by the system with 'EXPR$' prefix
f2cc139 is described below

commit f2cc139d99e2ac08182c7203f63f83168a821c44
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Fri Sep 4 13:50:14 2020 +0800

    [FLINK-19108][table] Stop expanding the identifiers with scope aliased by the system with 'EXPR$' prefix
    
    This closes #13293
---
 .../apache/flink/table/planner/utils/Expander.java |  7 +++++-
 .../planner/plan/common/ViewsExpandingTest.scala   | 29 ++++++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/Expander.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/Expander.java
index 8a3e403..43a5bac 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/Expander.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/Expander.java
@@ -94,7 +94,12 @@ public class Expander {
 			}
 
 			@Override public Void visit(SqlIdentifier identifier) {
-				identifiers.putIfAbsent(identifier.getParserPosition(), identifier);
+				// See SqlUtil#deriveAliasFromOrdinal, there is no good solution
+				// to distinguish between system alias (EXPR${number}) and user defines,
+				// and we stop expanding all of them.
+				if (!identifier.names.get(0).startsWith("EXPR$")) {
+					identifiers.putIfAbsent(identifier.getParserPosition(), identifier);
+				}
 				return null;
 			}
 		});
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/common/ViewsExpandingTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/common/ViewsExpandingTest.scala
index 666de8d..de7b6b8 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/common/ViewsExpandingTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/common/ViewsExpandingTest.scala
@@ -172,6 +172,35 @@ class ViewsExpandingTest(tableTestUtil: TableTestBase => TableTestUtil) extends
       is("SELECT `default_catalog`.`default_database`.`func`(1, 2, 'abc')"))
   }
 
+  @Test
+  def testExpandQueryWithSystemAlias(): Unit = {
+    val tableUtil = tableTestUtil(this)
+    val tableEnv = tableUtil.tableEnv
+    tableEnv.createTemporaryView("source",
+      tableEnv.fromValues("danny#21", "julian#55", "fabian#30").as("f0"))
+    val createView =
+      """
+        |create view tmp_view as
+        |select * from (
+        |  select f0,
+        |  row_number() over (partition by f0 order by f0 desc) as rowNum
+        |  from source)
+        |  where rowNum = 1
+        |""".stripMargin
+    tableEnv.executeSql(createView)
+    val objectID = ObjectIdentifier.of(tableEnv.getCurrentCatalog,
+      tableEnv.getCurrentDatabase, "tmp_view")
+    val view: CatalogBaseTable = tableEnv.getCatalog(objectID.getCatalogName)
+      .get().getTable(objectID.toObjectPath)
+    assertThat(view.asInstanceOf[CatalogView].getExpandedQuery,
+      is("SELECT *\n"
+        + "FROM (SELECT `source`.`f0`, "
+        + "ROW_NUMBER() "
+        + "OVER (PARTITION BY `source`.`f0` ORDER BY `source`.`f0` DESC) AS `rowNum`\n"
+        + "FROM `default_catalog`.`default_database`.`source`)\n"
+        + "WHERE `rowNum` = 1"))
+  }
+
   private def createSqlView(originTable: String): CatalogView = {
       new CatalogViewImpl(
         s"select * as c from $originTable",