[FLINK-39715] [table-planner] Fix IndexOutOfBoundsException in FlinkExpandConversionRule for global aggregate after sort

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/FlinkExpandConversionRule.scala

@@ -30,6 +30,8 @@ import org.apache.calcite.plan.volcano.AbstractConverter
 import org.apache.calcite.rel.{RelCollation, RelCollations, RelCollationTraitDef, RelNode}
 import org.apache.calcite.rel.RelDistribution.Type._
 
+import scala.collection.JavaConverters._
+
 /** Rule which converts an [[AbstractConverter]] to a RelNode which satisfies the target traits. */
 class FlinkExpandConversionRule(flinkConvention: Convention)
   extends RelOptRule(
@@ -68,6 +70,9 @@ class FlinkExpandConversionRule(flinkConvention: Convention)
       val toCollation = requiredTraits.getTrait(RelCollationTraitDef.INSTANCE)
       transformedNode = satisfyCollation(flinkConvention, transformedNode, toCollation)
     }
+    if (transformedNode == null) {
+      return
+    }
     checkSatisfyRequiredTrait(transformedNode, requiredTraits)
     call.transformTo(transformedNode)
   }
@@ -85,6 +90,9 @@ class FlinkExpandConversionRule(flinkConvention: Convention)
             // collation. So it is necessary to check whether collation satisfy requirement.
             val requiredCollation = requiredTraits.getTrait(RelCollationTraitDef.INSTANCE)
             val finalRel = satisfyCollation(flinkConvention, newRel, requiredCollation)
+            if (finalRel == null) {
+              return
+            }
             checkSatisfyRequiredTrait(finalRel, requiredTraits)
             call.transformTo(finalRel)
           case _ => // do nothing
@@ -151,6 +159,15 @@ object FlinkExpandConversionRule {
       requiredCollation: RelCollation): RelNode = {
     val fromCollation = node.getTraitSet.getTrait(RelCollationTraitDef.INSTANCE)
     if (!fromCollation.satisfies(requiredCollation)) {
+      val fieldCount = node.getRowType.getFieldCount
+      val isValidCollation =
+        requiredCollation.getFieldCollations.asScala.forall(_.getFieldIndex < fieldCount)
+      if (!isValidCollation) {
+        // An out-of-bounds collation index can never be satisfied by sorting the current node's
+        // row type since the field does not exist, so skipping this path cannot cause a
+        // CannotPlanException for any valid query.
+        return null
+      }
       val traitSet = node.getTraitSet.replace(requiredCollation).replace(flinkConvention)
       val sortCollation = RelCollationTraitDef.INSTANCE.canonize(requiredCollation)
       flinkConvention match {

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/FlinkExpandConversionRuleTest.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+class FlinkExpandConversionRuleTest extends TableTestBase {
+
+    private BatchTableTestUtil util;
+
+    @BeforeEach
+    void setup() {
+        util = batchTestUtil(TableConfig.getDefault());
+        util.tableEnv().getConfig().set("parallelism.default", "1");
+
+        util.tableEnv()
+                .executeSql(
+                        "CREATE TABLE MyTable ("
+                                + "  a INT,"
+                                + "  b STRING"
+                                + ") WITH ("
+                                + "  'connector' = 'values',"
+                                + "  'bounded' = 'true'"
+                                + ")");
+    }
+
+    @Test
+    void testOrderByWithGlobalAggregate() {
+        Table src = util.tableEnv().from("MyTable");
+        Table result = src.orderBy($("b").asc()).select($("a").max());
+
+        util.verifyRelPlan(result);
+    }
+}

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/FlinkExpandConversionRuleTest.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" ?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to you under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<Root>
+  <TestCase name="testOrderByWithGlobalAggregate">
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(EXPR$0=[$0])
++- LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+   +- LogicalSort(sort0=[$1], dir0=[ASC])
+      +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized rel plan">
+      <![CDATA[
+HashAggregate(isMerge=[false], select=[MAX(a) AS EXPR$0])
++- Sort(orderBy=[b ASC])
+   +- Exchange(distribution=[single])
+      +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b])
+]]>
+    </Resource>
+  </TestCase>
+</Root>