[GitHub] pnowojski commented on a change in pull request #6871: [FLINK-10583][FLINK-10584][table] Add State Retention to temporal joins.

[GitBox <gi...@apache.org>]

pnowojski 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_r237507423
 
 

 ##########
 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 = {
 
 Review comment:
   nit: rename `eventtime` to `rowtime` and also we are not using nomenclature "read" or "write" side, but either "probe/build" or "left/right".

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services