[FLINK-33689] JsonObjectAggFunction can't retract previous data which is invalid when enabling local global agg

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.18.0
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Shuai Xu
• Created: 29/Nov/23 07:16
• Updated: 16/Jan/24 02:26

Description

Run the test as following and enable LocalGlobal in sql/AggregateITCase . 

def testGroupJsonObjectAggWithRetract(): Unit = {
  val data = new mutable.MutableList[(Long, String, Long)]
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  data.+=((2L, "Hallo", 2L))
  val sql =
    s"""
       |select
       |   JSON_OBJECTAGG(key k value v)
       |from (select
       |             cast(SUM(a) as string) as k,t as v
       |       from
       |             Table6
       |       group by t)
       |""".stripMargin
  val t = failingDataSource(data).toTable(tEnv, 'a, 'c, 't)
  tEnv.createTemporaryView("Table6", t)
  val sink = new TestingRetractSink
  tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink).setParallelism(1)
  env.execute()
  val expected =
    List(
      "{\"30\":2}"
    )
  assertThat(sink.getRetractResults).isEqualTo(expected)
} 

The result is as following.

List({"14":2,"30":2}) 

However,  {"14":2}  should be retracted.

Issue Links

• links to: GitHub Pull Request #23827