[
https://issues.apache.org/jira/browse/FLINK-19271?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-19271:
-----------------------------------
Labels: auto-deprioritized-major stale-minor (was:
auto-deprioritized-major)
I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help
the community manage its development. I see this issues has been marked as
Minor but is unassigned and neither itself nor its Sub-Tasks have been updated
for 180 days. I have gone ahead and marked it "stale-minor". If this ticket is
still Minor, please either assign yourself or give an update. Afterwards,
please remove the label or in 7 days the issue will be deprioritized.
> wrong HOP_PROCTIME output when materialize proctime
> ---------------------------------------------------
>
> Key: FLINK-19271
> URL: https://issues.apache.org/jira/browse/FLINK-19271
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API, Table SQL / Planner, Table SQL / Runtime
> Reporter: Leonard Xu
> Priority: Minor
> Labels: auto-deprioritized-major, stale-minor
>
> The HOP_PROCTIME is bigger than HOP_END in the following case.
> The reason is we materialize the process time(HOP_PROCTIME) in the downstream
> of
> `WindowAggregate` rather than internal of `WindowAggregate`, this lead the
> HOP_PROCTIME is bigger than HOP_END forever.
> And I believe this problem exists in TUMBLE_PROCTIME and SESSION_PROCTIME too,
> We should use the `HOP_END - 1` as the HOP_PROCTIME when the proctime need
> materialization.
> *WindowAggregateITCase*
> {code:java}
> @Test
> def testEventTimeSlidingWindowProcTime(): Unit = {
> val stream = failingDataSource(data)
> .assignTimestampsAndWatermarks(
> new TimestampAndWatermarkWithOffset
> [(Long, Int, Double, Float, BigDecimal, String, String)](10L))
> val table = stream.toTable(tEnv,
> 'a, 'int, 'double, 'float, 'bigdec, 'string, 'name,
> 'proctime.proctime())
> tEnv.registerTable("T1", table)
> val sql =
> """
> |SELECT
> | HOP_START(proctime, INTERVAL '0.004' SECOND, INTERVAL '0.005'
> SECOND),
> | HOP_END(proctime, INTERVAL '0.004' SECOND, INTERVAL '0.005'
> SECOND),
> | HOP_PROCTIME(proctime, INTERVAL '0.004' SECOND, INTERVAL '0.005'
> SECOND)
> |FROM T1
> |GROUP BY `string`, HOP(proctime, INTERVAL '0.004' SECOND, INTERVAL
> '0.005' SECOND)
> """.stripMargin
> val sink = new TestingAppendSink
> tEnv.sqlQuery(sql).toAppendStream[Row].addSink(sink)
> env.execute()
> val expected = Seq(
> "2020-09-17T07:13:40.348, 2020-09-17T07:13:40.353,
> 2020-09-17T07:13:43.479",
> "2020-09-17T07:13:40.348, 2020-09-17T07:13:40.353,
> 2020-09-17T07:13:43.479",
> "2020-09-17T07:13:40.352, 2020-09-17T07:13:40.357,
> 2020-09-17T07:13:44.030")
> assertEquals(expected.sorted, sink.getAppendResults.sorted)
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
