[FLINK-39089][SQL] Enhanced projection field alias display

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RelExplainUtil.scala

@@ -969,4 +969,43 @@ object RelExplainUtil {
     sb.append("]")
     sb.toString
   }
+
+  /**
+   * Converts projection fields to a more readable string format with field names instead of
+   * positional indices like $0, $1, etc.
+   *
+   * @param projects
+   *   the list of projection expressions
+   * @param inputFieldNames
+   *   the list of input field names
+   * @param outputFieldNames
+   *   the list of output field names
+   * @return
+   *   a formatted string showing field names and their sources
+   */
+  def projectFieldsToString(
+      projects: util.List[RexNode],
+      inputFieldNames: util.List[String],
+      outputFieldNames: util.List[String]): String = {
+    val result = new StringBuilder
+    for (i <- 0 until projects.size()) {
+      if (i > 0) result.append(", ")
+      val project = projects.get(i)
+      val outputName = outputFieldNames.get(i)
+
+      project match {
+        case inputRef: RexInputRef =>
+          val inputName = inputFieldNames.get(inputRef.getIndex)
+          if (inputName.equals(outputName)) {
+            result.append(outputName)
+          } else {
+            result.append(s"$inputName AS $outputName")
+          }
+        case _ =>
+          // For complex expressions, show both expression and output name
+          result.append(s"${project.toString} AS $outputName")
+      }
+    }
+    result.toString
+  }
 }

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RelTreeWriterImpl.scala

@@ -24,9 +24,10 @@ import org.apache.flink.table.planner.plan.nodes.physical.FlinkPhysicalRel
 import org.apache.flink.table.planner.plan.nodes.physical.stream._
 
 import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.{Aggregate, Correlate, Join, TableScan}
+import org.apache.calcite.rel.core.{Aggregate, Correlate, Join, Project, TableScan}
 import org.apache.calcite.rel.externalize.RelWriterImpl
 import org.apache.calcite.rel.hint.Hintable
+import org.apache.calcite.rex.RexNode
 import org.apache.calcite.sql.SqlExplainLevel
 import org.apache.calcite.util.Pair
 
@@ -97,6 +98,20 @@ class RelTreeWriterImpl(
 
     val printValues = new util.ArrayList[Pair[String, AnyRef]]()
     if (explainLevel != SqlExplainLevel.NO_ATTRIBUTES) {
+      // Enhanced handling for Project nodes to show readable field names
+      val isProjectNode = rel match {
+        case project: Project =>
+          val projectList = project.getProjects
+          val inputFieldNames = project.getInput.getRowType.getFieldNames
+          val outputFieldNames = project.getRowType.getFieldNames
+          val projectStr =
+            RelExplainUtil.projectFieldsToString(projectList, inputFieldNames, outputFieldNames)
+          printValues.add(Pair.of("select", projectStr))
+          true
+        case _ =>
+          false
+      }
+
       if (withAdvice) {
         if (depth == 0) {
           applyAdvice(rel)
@@ -112,7 +127,11 @@ class RelTreeWriterImpl(
           printValues.add(Pair.of("advice", adviceIds.mkString(", ")))
         }
       }
-      printValues.addAll(values)
+      // For Project nodes, don't add default values to avoid duplication
+      // Our custom select=[...] format replaces the default field definitions
+      if (!isProjectNode) {
+        printValues.addAll(values)
+      }
     }
 
     if (withChangelogTraits) rel match {

flink-table/flink-table-planner/src/test/resources/org/apache/flink/connector/file/table/FileSystemTableSourceTest.xml

@@ -23,7 +23,7 @@ limitations under the License.
     <Resource name="ast">
       <![CDATA[
 LogicalSink(table=[default_catalog.default_database.MySink], fields=[a, b, c])
-+- LogicalProject(a=[$0], b=[$1], c=[$2])
++- LogicalProject(select=[a, b, c])
    +- LogicalFilter(condition=[>($0, 10)])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
 ]]>
@@ -43,8 +43,8 @@ Sink(table=[default_catalog.default_database.MySink], fields=[a, b, c])
     <Resource name="ast">
       <![CDATA[
 LogicalSink(table=[default_catalog.default_database.MySink], targetColumns=[[0],[1],[2]], fields=[a, b, filemeta])
-+- LogicalProject(a=[$0], b=[$1], filemeta=[$3])
-   +- LogicalProject(a=[$0], b=[$1], c=[$2], filemeta=[CAST($3):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"])
++- LogicalProject(select=[a, b, filemeta])
+   +- LogicalProject(select=[a, b, c]):VARCHAR(2147483647) CHARACTER SET "UTF-16LE"])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTableWithMeta, metadata=[file.path]]])
 ]]>
     </Resource>

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/api/batch/ExplainTest.xml

@@ -19,14 +19,14 @@ limitations under the License.
   <TestCase name="testExplainMultipleInput[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], EXPR$1=[$1], d=[$2], EXPR$10=[$3])
+LogicalProject(select=[a, EXPR$1, d, EXPR$10])
 +- LogicalFilter(condition=[=($0, $2)])
    +- LogicalJoin(condition=[true], joinType=[inner])
       :- LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
-      :  +- LogicalProject(a=[$0], b=[$1])
+      :  +- LogicalProject(select=[a, b])
       :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
       +- LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
-         +- LogicalProject(d=[$0], e=[$1])
+         +- LogicalProject(select=[d, e])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 
 == Optimized Physical Plan ==
@@ -58,14 +58,14 @@ MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[
   <TestCase name="testExplainMultipleInput[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], EXPR$1=[$1], d=[$2], EXPR$10=[$3])
+LogicalProject(select=[a, EXPR$1, d, EXPR$10])
 +- LogicalFilter(condition=[=($0, $2)])
    +- LogicalJoin(condition=[true], joinType=[inner])
       :- LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
-      :  +- LogicalProject(a=[$0], b=[$1])
+      :  +- LogicalProject(select=[a, b])
       :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
       +- LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
-         +- LogicalProject(d=[$0], e=[$1])
+         +- LogicalProject(select=[d, e])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 
 == Optimized Physical Plan ==
@@ -97,9 +97,9 @@ MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[
   <TestCase name="testExplainWithAgg[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(EXPR$0=[$1])
+LogicalProject(select=[EXPR$0])
 +- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()])
-   +- LogicalProject(a=[$0])
+   +- LogicalProject(select=[a])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -123,9 +123,9 @@ Calc(select=[EXPR$0])
   <TestCase name="testExplainWithAgg[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(EXPR$0=[$1])
+LogicalProject(select=[EXPR$0])
 +- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()])
-   +- LogicalProject(a=[$0])
+   +- LogicalProject(select=[a])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -149,7 +149,7 @@ Calc(select=[EXPR$0])
   <TestCase name="testExplainWithDataStreamScan[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -163,7 +163,7 @@ BoundedStreamScan(table=[[default_catalog, default_database, MyTable1]], fields=
   <TestCase name="testExplainWithDataStreamScan[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -177,7 +177,7 @@ BoundedStreamScan(table=[[default_catalog, default_database, MyTable1]], fields=
   <TestCase name="testExplainWithFilter[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalFilter(condition=[=(MOD($0, 2), 0)])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
@@ -194,7 +194,7 @@ Calc(select=[a, b, c], where=[(MOD(a, 2) = 0)])
   <TestCase name="testExplainWithFilter[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalFilter(condition=[=(MOD($0, 2), 0)])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
@@ -211,7 +211,7 @@ Calc(select=[a, b, c], where=[(MOD(a, 2) = 0)])
   <TestCase name="testExplainWithJoin[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2], e=[$4], f=[$5])
+LogicalProject(select=[a, b, c, e, f])
 +- LogicalFilter(condition=[=($0, $3)])
    +- LogicalJoin(condition=[true], joinType=[inner])
       :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
@@ -238,7 +238,7 @@ Calc(select=[a, b, c, e, f])
   <TestCase name="testExplainWithJoin[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2], e=[$4], f=[$5])
+LogicalProject(select=[a, b, c, e, f])
 +- LogicalFilter(condition=[=($0, $3)])
    +- LogicalJoin(condition=[true], joinType=[inner])
       :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
@@ -266,17 +266,17 @@ Calc(select=[a, b, c, e, f])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSink(table=[default_catalog.default_database.sink1], fields=[a, cnt])
-+- LogicalProject(a=[$0], cnt=[$1])
++- LogicalProject(select=[a, cnt])
    +- LogicalFilter(condition=[>($1, 10)])
       +- LogicalAggregate(group=[{0}], cnt=[COUNT()])
-         +- LogicalProject(a=[$0])
+         +- LogicalProject(select=[a])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 LogicalSink(table=[default_catalog.default_database.sink2], fields=[a, cnt])
-+- LogicalProject(a=[$0], cnt=[$1])
++- LogicalProject(select=[a, cnt])
    +- LogicalFilter(condition=[<($1, 10)])
       +- LogicalAggregate(group=[{0}], cnt=[COUNT()])
-         +- LogicalProject(a=[$0])
+         +- LogicalProject(select=[a])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -317,17 +317,17 @@ Sink(table=[default_catalog.default_database.sink2], fields=[a, cnt])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSink(table=[default_catalog.default_database.sink1], fields=[a, cnt])
-+- LogicalProject(a=[$0], cnt=[$1])
++- LogicalProject(select=[a, cnt])
    +- LogicalFilter(condition=[>($1, 10)])
       +- LogicalAggregate(group=[{0}], cnt=[COUNT()])
-         +- LogicalProject(a=[$0])
+         +- LogicalProject(select=[a])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 LogicalSink(table=[default_catalog.default_database.sink2], fields=[a, cnt])
-+- LogicalProject(a=[$0], cnt=[$1])
++- LogicalProject(select=[a, cnt])
    +- LogicalFilter(condition=[<($1, 10)])
       +- LogicalAggregate(group=[{0}], cnt=[COUNT()])
-         +- LogicalProject(a=[$0])
+         +- LogicalProject(select=[a])
             +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -368,7 +368,7 @@ Sink(table=[default_catalog.default_database.sink2], fields=[a, cnt])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSink(table=[default_catalog.default_database.sink], fields=[a, b, c])
-+- LogicalProject(a=[$0], b=[$1], c=[$2])
++- LogicalProject(select=[a, b, c])
    +- LogicalFilter(condition=[>($0, 10)])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
@@ -388,7 +388,7 @@ Sink(table=[default_catalog.default_database.sink], fields=[a, b, c])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSink(table=[default_catalog.default_database.sink], fields=[a, b, c])
-+- LogicalProject(a=[$0], b=[$1], c=[$2])
++- LogicalProject(select=[a, b, c])
    +- LogicalFilter(condition=[>($0, 10)])
       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
@@ -408,7 +408,7 @@ Sink(table=[default_catalog.default_database.sink], fields=[a, b, c])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSort(sort0=[$0], dir0=[ASC-nulls-first], fetch=[5])
-+- LogicalProject(a=[$0], b=[$1], c=[$2])
++- LogicalProject(select=[a, b, c])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -429,7 +429,7 @@ SortLimit(orderBy=[a ASC], offset=[0], fetch=[5], global=[true])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalSort(sort0=[$0], dir0=[ASC-nulls-first], fetch=[5])
-+- LogicalProject(a=[$0], b=[$1], c=[$2])
++- LogicalProject(select=[a, b, c])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
 
 == Optimized Physical Plan ==
@@ -449,7 +449,7 @@ SortLimit(orderBy=[a ASC], offset=[0], fetch=[5], global=[true])
   <TestCase name="testExplainWithTableSourceScan[extended=false]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
 
 == Optimized Physical Plan ==
@@ -463,7 +463,7 @@ TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a,
   <TestCase name="testExplainWithTableSourceScan[extended=true]">
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(a=[$0], b=[$1], c=[$2])
+LogicalProject(select=[a, b, c])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
 
 == Optimized Physical Plan ==
@@ -478,9 +478,9 @@ TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a,
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalUnion(all=[true])
-:- LogicalProject(a=[$0], b=[$1], c=[$2])
+:- LogicalProject(select=[a, b, c])
 :  +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
-+- LogicalProject(d=[$0], e=[$1], f=[$2])
++- LogicalProject(select=[d, e, f])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 
 == Optimized Physical Plan ==
@@ -499,9 +499,9 @@ Union(all=[true], union=[a, b, c])
     <Resource name="explain">
       <![CDATA[== Abstract Syntax Tree ==
 LogicalUnion(all=[true])
-:- LogicalProject(a=[$0], b=[$1], c=[$2])
+:- LogicalProject(select=[a, b, c])
 :  +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
-+- LogicalProject(d=[$0], e=[$1], f=[$2])
++- LogicalProject(select=[d, e, f])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 
 == Optimized Physical Plan ==