[
https://issues.apache.org/jira/browse/FLINK-10583?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16703272#comment-16703272
]
ASF GitHub Bot commented on FLINK-10583:
----------------------------------------
kl0u commented on a change in pull request #6871:
[FLINK-10583][FLINK-10584][table] Add State Retention to temporal joins.
URL: https://github.com/apache/flink/pull/6871#discussion_r237509459
##########
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/TemporalJoinHarnessTest.scala
##########
@@ -530,6 +530,214 @@ class TemporalJoinHarnessTest extends HarnessTestBase {
0)
}
+ // ---------------------- Event time TTL tests ----------------------
+
+ @Test
+ def testEventTimeScenarioWithoutAdvancingProcessingTime(): Unit = {
+ // min=2ms max=4ms
+ val testHarness = createTestHarness(new
OrdersRatesRowtimeTemporalJoinInfo())
+
+ testHarness.open()
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ testHarness.setProcessingTime(1L)
+
+ // process without conversion rates
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 1L)))
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 114L, 0L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 1L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(2L))
+ testHarness.processWatermark2(new Watermark(2L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 4L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 4L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(5L))
+ testHarness.processWatermark2(new Watermark(5L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ testHarness.close()
+ }
+
+ @Test
+ def testEventTimeCleanupShouldSucceed(): Unit = {
+ // min=2ms max=4ms
+ val testHarness = createTestHarness(new
OrdersRatesRowtimeTemporalJoinInfo())
+
+ testHarness.open()
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ testHarness.setProcessingTime(1L)
+
+ // process without conversion rates
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 1L)))
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 114L, 0L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 1L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(2L))
+ testHarness.processWatermark2(new Watermark(2L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ // this should clean-up the state
+ testHarness.setProcessingTime(5L)
+
+ assert(testHarness.numKeyedStateEntries() == 1) // this is the index
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 4L))) //
this should succeed
+
+ testHarness.processWatermark1(new Watermark(5L))
+ testHarness.processWatermark2(new Watermark(5L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ testHarness.close()
+ }
+
+ @Test
+ def testEventTimeCleanupTimerGettingOverwrittenFromReadSide(): Unit = {
+ // min=2ms max=4ms
+ val testHarness = createTestHarness(new
OrdersRatesRowtimeTemporalJoinInfo())
+
+ testHarness.open()
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ testHarness.setProcessingTime(1L)
+
+ // process without conversion rates
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 1L)))
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 114L, 0L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 1L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(2L))
+ testHarness.processWatermark2(new Watermark(2L))
+
+ // this should update the clean-up timer to 8
+ testHarness.setProcessingTime(4L)
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 4L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 4L, "Euro", 114L,
0L)))
+
+ // this should now do nothing (also it does not update the timer as 5 +
2ms (min) < 8)
+ testHarness.setProcessingTime(5L)
+
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 5L)))
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 5L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(5L))
+ testHarness.processWatermark2(new Watermark(5L))
+
+ // this should now clean up the state
+ testHarness.setProcessingTime(8L)
+
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 7L))) //
this should find no match
+
+ testHarness.processWatermark1(new Watermark(10L))
+ testHarness.processWatermark2(new Watermark(10L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ testHarness.close()
+ }
+
+ @Test
+ def testEventTimeCleanupTimerGettingOverwrittenFromWriteSide(): Unit = {
+ // min=2ms max=4ms
+ val testHarness = createTestHarness(new
OrdersRatesRowtimeTemporalJoinInfo())
+
+ testHarness.open()
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ testHarness.setProcessingTime(1L)
+
+ // process without conversion rates
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 1L)))
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 114L, 0L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 1L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(2L))
+ testHarness.processWatermark2(new Watermark(2L))
+
+ // this should update the clean-up timer to 8
+ testHarness.setProcessingTime(4L)
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 117L, 4L)))
+
+ // this should now do nothing
+ testHarness.setProcessingTime(5L)
+
+ // so this should be joined with the "old" value
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 3L)))
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 3L, "Euro", 114L,
0L)))
+
+ testHarness.processWatermark1(new Watermark(5L))
+ testHarness.processWatermark2(new Watermark(5L))
+
+ // this should now clean up the state
+ testHarness.setProcessingTime(8L)
+
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 7L))) //
this should find no match
+
+ testHarness.processWatermark1(new Watermark(10L))
+ testHarness.processWatermark2(new Watermark(10L))
+
+ verify(expectedOutput, testHarness.getOutput)
+
+ testHarness.close()
+ }
+
+ @Test
+ def testEventTimeCleanupGetOverwrittenAfterEvaluation(): Unit = {
+ // min=2ms max=4ms
+ val testHarness = createTestHarness(new
OrdersRatesRowtimeTemporalJoinInfo())
+
+ testHarness.open()
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ testHarness.setProcessingTime(1L)
+
+ // process without conversion rates
+ testHarness.processElement1(new StreamRecord(CRow(2L, "Euro", 1L)))
+ testHarness.processElement2(new StreamRecord(CRow("Euro", 114L, 0L)))
+
+ expectedOutput.add(new StreamRecord(CRow(2L, "Euro", 1L, "Euro", 114L,
0L)))
+
+ testHarness.setProcessingTime(4L)
+
+ // this should trigger an evaluation, which should also update the
clean-up timer to 8
Review comment:
For this I think either @twalthr or @fhueske have to answer. I just assumed
that given that we access the state, we should assume that the key is still
active, so we should keep the state.
If this is a wrong assumption feel free to point it out and I will update
the PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add support for state retention to the Processing Time versioned joins.
> -----------------------------------------------------------------------
>
> Key: FLINK-10583
> URL: https://issues.apache.org/jira/browse/FLINK-10583
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Affects Versions: 1.7.0
> Reporter: Kostas Kloudas
> Assignee: Kostas Kloudas
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
