[FLINK-21302][table-planner-blink] Fix NPE when use row_number() in o…

…verAgg

This closes #14880

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

@@ -380,7 +380,7 @@ object AggregateUtil extends Enumeration {
       .zipWithIndex
       .map { case (call, index) =>
         val argIndexes = call.getAggregation match {
-          case _: SqlRankFunction => orderKeyIndexes
+          case _: SqlRankFunction => if (orderKeyIndexes != null) orderKeyIndexes else Array[Int]()
           case _ => call.getArgList.map(_.intValue()).toArray
         }
         transformToAggregateInfo(

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/OverAggregateITCase.scala

@@ -55,6 +55,36 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
     env.getCheckpointConfig.enableUnalignedCheckpoints(false)
   }
 
+  @Test
+  def testRowNumberOnOver(): Unit = {
+    val t = failingDataSource(TestData.tupleData5)
+      .toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
+    tEnv.registerTable("MyTable", t)
+    val sqlQuery = "SELECT a, ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime()) FROM MyTable"
+
+    val sink = new TestingAppendSink
+    tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink)
+    env.execute()
+
+    val expected = List(
+      "1,1",
+      "2,1",
+      "2,2",
+      "3,1",
+      "3,2",
+      "3,3",
+      "4,1",
+      "4,2",
+      "4,3",
+      "4,4",
+      "5,1",
+      "5,2",
+      "5,3",
+      "5,4",
+      "5,5")
+    assertEquals(expected, sink.getAppendResults)
+  }
+
   @Test
   def testProcTimeBoundedPartitionedRowsOver(): Unit = {
     val t = failingDataSource(TestData.tupleData5)