Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/4625#discussion_r143401988
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala
---
@@ -102,5 +117,154 @@ class JoinITCase extends StreamingWithStateTestBase {
env.execute()
}
+ /** test rowtime inner join **/
+ @Test
+ def testRowTimeInnerJoin(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ env.setStateBackend(getStateBackend)
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ StreamITCase.clear
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 join T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.rt BETWEEN t2.rt - INTERVAL '5' SECOND AND
+ | t2.rt + INTERVAL '6' SECOND
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String, Long)]
--- End diff --
Yes, you understood the problem correctly. Without PR #4732, join keys are
mapped to `Tuple` which do not support null fields unless the field type
supports them (which is the case for String but not other primitives). With
#4732 keys are mapped to `Row` which supports null fields but treats `null ==
null`. Therefore, we need to add these predicates into the code-gen'd
conditions that correctly evaluate the predicates according to three-value
logic.
After #4732 all types will support null value keys.
---
