[
https://issues.apache.org/jira/browse/FLINK-33213?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17785297#comment-17785297
]
Sergey Nuyanzin edited comment on FLINK-33213 at 11/12/23 5:08 PM:
-------------------------------------------------------------------
Merged as
[29e49482a82e8c1cd404e42c3aae0944188d956e|https://github.com/apache/flink/commit/29e49482a82e8c1cd404e42c3aae0944188d956e]
was (Author: sergey nuyanzin):
Merged as
[29e49482a82e8c1cd404e42c3aae0944188d956e|29e49482a82e8c1cd404e42c3aae0944188d956e]
> Flink SQL MinMax aggregations without retract messages when `where` condition
> exist
> -----------------------------------------------------------------------------------
>
> Key: FLINK-33213
> URL: https://issues.apache.org/jira/browse/FLINK-33213
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.12.7, 1.13.6, 1.14.6, 1.15.4, 1.16.2, 1.17.1
> Reporter: Sergey Paryshev
> Assignee: Sergey Paryshev
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.19.0
>
>
> Currently Flink didn't take account `where` condition when calculate
> SqlMonotonicity for Calc/FlinkLogicalCalc/StreamPhysicalCalc.
>
> To reproduce bug (put into
> org.apache.flink.table.planner.runtime.stream.sql.AggregateITCase):
> {code:java}
> @Test
> def testMaxRetract(): Unit = {
> env.setParallelism(1)
> val data = new mutable.MutableList[(Int, Int)]
> data.+=((1, 10))
> data.+=((1, 10))
> data.+=((2, 5))
> data.+=((1, 10))
> val t = failingDataSource(data).toTable(tEnv, 'id, 'price)
> tEnv.createTemporaryView("T", t)
> val sql =
> """
> |SELECT MAX(price) FROM(
> | SELECT id, count(*) as c, price FROM T GROUP BY id, price)
> |WHERE c > 0 and c < 3""".stripMargin
> val sink = new TestingRetractSink
> tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink)
> env.execute()
> val expected = List("5")
> assertEquals(expected.sorted, sink.getRetractResults.sorted)
> } {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
