[FLINK-30687][table] Fix wrong result of agg with fiter which references first input column

This closes #26303.

Author: lincoln-lil

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala

@@ -330,7 +330,7 @@ class AggsHandlerCodeGenerator(
       aggIndex: Int,
       aggName: String): Option[Expression] = {
 
-    if (filterArg > 0) {
+    if (filterArg >= 0) {
       val filterType = inputFieldTypes(filterArg)
       if (!filterType.isInstanceOf[BooleanType]) {
         throw new TableException(

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/agg/AggTestBase.scala

@@ -67,6 +67,7 @@ abstract class AggTestBase(isBatchMode: Boolean) {
   val aggInfo1: AggregateInfo = {
     val aggInfo = mock(classOf[AggregateInfo])
     val call = mock(classOf[AggregateCall])
+    updateFilter(call, -1)
     when(aggInfo.agg).thenReturn(call)
     when(call.getName).thenReturn("avg1")
     when(call.hasFilter).thenReturn(false)
@@ -82,6 +83,7 @@ abstract class AggTestBase(isBatchMode: Boolean) {
   val aggInfo2: AggregateInfo = {
     val aggInfo = mock(classOf[AggregateInfo])
     val call = mock(classOf[AggregateCall])
+    updateFilter(call, -1)
     when(aggInfo.agg).thenReturn(call)
     when(call.getName).thenReturn("avg2")
     when(call.hasFilter).thenReturn(false)
@@ -98,6 +100,7 @@ abstract class AggTestBase(isBatchMode: Boolean) {
   val aggInfo3: AggregateInfo = {
     val aggInfo = mock(classOf[AggregateInfo])
     val call = mock(classOf[AggregateCall])
+    updateFilter(call, -1)
     when(aggInfo.agg).thenReturn(call)
     when(call.getName).thenReturn("avg3")
     when(call.hasFilter).thenReturn(false)
@@ -118,4 +121,10 @@ abstract class AggTestBase(isBatchMode: Boolean) {
   val classLoader: ClassLoader = Thread.currentThread().getContextClassLoader
   val context: ExecutionContext = mock(classOf[ExecutionContext])
   when(context.getRuntimeContext).thenReturn(mock(classOf[RuntimeContext]))
+
+  private def updateFilter(call: AggregateCall, v: Int): Unit = {
+    val field = call.getClass.getField("filterArg")
+    field.setAccessible(true)
+    field.set(call, v)
+  }
 }

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateITCaseBase.scala

@@ -1224,6 +1224,11 @@ abstract class AggregateITCaseBase(testName: String) extends BatchTestBase {
     checkResult(sql, Seq(row("11, 11"), row("12, 12"), row("null, null")))
   }
 
+  @Test
+  def testAggFilterReferenceFirstColumn(): Unit = {
+    checkResult("select count(*) filter (where a < 10) from Table3", Seq(row(9)))
+  }
+
   // TODO support csv
 //  @Test
 //  def testMultiGroupBys(): Unit = {

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

@@ -1644,6 +1644,26 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
     assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
   }
 
+  @TestTemplate
+  def testAggFilterReferenceFirstColumn(): Unit = {
+    val t = failingDataSource(TestData.tupleData3).toTable(tEnv).as("a", "b", "c")
+    tEnv.createTemporaryView("MyTable", t)
+
+    val sqlQuery =
+      s"""
+         |SELECT
+         |  COUNT(*) filter (where a < 10)
+         |FROM MyTable
+       """.stripMargin
+
+    val sink = new TestingRetractSink
+    val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
+    result.addSink(sink).setParallelism(1)
+    env.execute()
+    val expected = List("9")
+    assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
+  }
+
   @TestTemplate
   def testPruneUselessAggCall(): Unit = {
     val data = new mutable.MutableList[(Int, Long, String)]