[FLINK-8002] [table] Fix join window boundary for LESS_THAN and GREAT…

…ER_THAN predicates. This closes apache#4962.
nicusX · Nov 8, 2017 · 989c779 · 989c779
1 parent 8c5b547
commit 989c779
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 4 deletions.
diff --git a/...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/WindowJoinUtil.scala b/...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/WindowJoinUtil.scala
@@ -346,10 +346,14 @@ object WindowJoinUtil {
       leftLiteral.get - rightLiteral.get
     }
     val boundary = timePred.pred.getKind match {
-      case SqlKind.LESS_THAN =>
+      case SqlKind.LESS_THAN if timePred.leftInputOnLeftSide =>
         tmpTimeOffset - 1
-      case SqlKind.GREATER_THAN =>
+      case SqlKind.LESS_THAN if !timePred.leftInputOnLeftSide =>
         tmpTimeOffset + 1
+      case SqlKind.GREATER_THAN if timePred.leftInputOnLeftSide =>
+        tmpTimeOffset + 1
+      case SqlKind.GREATER_THAN if !timePred.leftInputOnLeftSide =>
+        tmpTimeOffset - 1
       case _ =>
         tmpTimeOffset
     }

diff --git a/...libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala b/...libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
@@ -379,20 +379,54 @@ class JoinTest extends TableTestBase {
       "rowtime")
 
     verifyTimeBoundary(
-      "t1.c - interval '2' second >= t2.c + interval '1' second -" +
-        "interval '10' second and " +
+      "t1.c >= t2.c - interval '1' second and " +
+        "t1.c <= t2.c + interval '10' second",
+      -1000,
+      10000,
+      "rowtime")
+
+    verifyTimeBoundary(
+      "t1.c - interval '2' second >= t2.c + interval '1' second - interval '10' second and " +
         "t1.c <= t2.c + interval '10' second",
       -7000,
       10000,
       "rowtime")
 
+    verifyTimeBoundary(
+      "t2.c + interval '1' second - interval '10' second <= t1.c - interval '2' second and " +
+        "t2.c + interval '10' second >= t1.c",
+      -7000,
+      10000,
+      "rowtime")
+
     verifyTimeBoundary(
       "t1.c >= t2.c - interval '10' second and " +
         "t1.c <= t2.c - interval '5' second",
       -10000,
       -5000,
       "rowtime")
 
+    verifyTimeBoundary(
+      "t2.c - interval '10' second <= t1.c and " +
+        "t2.c - interval '5' second >= t1.c",
+      -10000,
+      -5000,
+      "rowtime")
+
+    verifyTimeBoundary(
+      "t1.c > t2.c - interval '2' second and " +
+        "t1.c < t2.c + interval '2' second",
+      -1999,
+      1999,
+      "rowtime")
+
+    verifyTimeBoundary(
+      "t2.c > t1.c - interval '2' second and " +
+        "t2.c < t1.c + interval '2' second",
+      -1999,
+      1999,
+      "rowtime")
+
     verifyTimeBoundary(
       "t1.c = t2.c",
       0,