[FLINK-37610] Check if orderBy keys exists before accessing

- If ORDER BY clause is not specified in an Over agg, it would lead to
  IndexOutOfBoundsException
- This commit adds a check to ensure orderby fields exists before accessing them

Author: bvarghese1

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -310,18 +310,26 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
           .newBuilder()
           .addContainedKind(ModifyKind.INSERT)
 
-        // All aggregates are computed over the same window and order by is supported for only 1 field
-        val orderKeyIndex =
-          over.logicWindow.groups.get(0).orderKeys.getFieldCollations.get(0).getFieldIndex
-        val orderKeyType = over.logicWindow.getRowType.getFieldList.get(orderKeyIndex).getType
         if (
-          !FlinkTypeFactory.isRowtimeIndicatorType(orderKeyType)
-          && !FlinkTypeFactory.isProctimeIndicatorType(orderKeyType)
+          !over.logicWindow.groups.isEmpty && !over.logicWindow.groups
+            .get(0)
+            .orderKeys
+            .getFieldCollations
+            .isEmpty
         ) {
-          // Only non row-time/proc-time sort can support UPDATES
-          builder.addContainedKind(ModifyKind.UPDATE)
-          builder.addContainedKind(ModifyKind.DELETE)
-          overRequiredTrait = ModifyKindSetTrait.ALL_CHANGES
+          // All aggregates are computed over the same window and order by is supported for only 1 field
+          val orderKeyIndex =
+            over.logicWindow.groups.get(0).orderKeys.getFieldCollations.get(0).getFieldIndex
+          val orderKeyType = over.logicWindow.getRowType.getFieldList.get(orderKeyIndex).getType
+          if (
+            !FlinkTypeFactory.isRowtimeIndicatorType(orderKeyType)
+            && !FlinkTypeFactory.isProctimeIndicatorType(orderKeyType)
+          ) {
+            // Only non row-time/proc-time sort can support UPDATES
+            builder.addContainedKind(ModifyKind.UPDATE)
+            builder.addContainedKind(ModifyKind.DELETE)
+            overRequiredTrait = ModifyKindSetTrait.ALL_CHANGES
+          }
         }
         val children = visitChildren(over, overRequiredTrait)
         val providedTrait = new ModifyKindSetTrait(builder.build())

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

@@ -526,4 +526,27 @@ Calc(select=[c, w0$o0 AS cnt1, CASE((w0$o0 > 0), w0$o1, null:INTEGER) AS cnt2])
 ]]>
     </Resource>
   </TestCase>
+	<TestCase name="testWithoutOrderByClause">
+		<Resource name="sql">
+			<![CDATA[
+SELECT c,
+    COUNT(a) OVER (PARTITION BY c) AS cnt1
+FROM MyTable
+      ]]>
+		</Resource>
+		<Resource name="ast">
+			<![CDATA[
+LogicalProject(c=[$2], cnt1=[COUNT($0) OVER (PARTITION BY $2)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+		</Resource>
+		<Resource name="optimized exec plan">
+			<![CDATA[
+Calc(select=[c, w0$o0 AS $1])
++- OverAggregate(partitionBy=[c], orderBy=[], window=[ RANG BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], select=[a, b, c, proctime, rowtime, COUNT(a) AS w0$o0])
+   +- Exchange(distribution=[hash[c]])
+      +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
+]]>
+		</Resource>
+	</TestCase>
 </Root>

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

@@ -467,4 +467,17 @@ class OverAggregateTest extends TableTestBase {
                            |)
                            |""".stripMargin)
   }
+
+  @Test
+  def testWithoutOrderByClause(): Unit = {
+    val sql =
+      """
+        |SELECT c,
+        |    COUNT(a) OVER (PARTITION BY c) AS cnt1
+        |FROM MyTable
+      """.stripMargin
+
+    util.verifyExecPlan(sql)
+  }
+
 }