test: add QTT tests for GRACE PERIOD

confluentinc · Jun 16, 2021 · 8bda8a3 · 8bda8a3
1 parent bdf34b6
commit 8bda8a3
Showing 1 changed file with 75 additions and 0 deletions.
diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json
@@ -855,6 +855,81 @@
         ]
       }
     },
+    {
+      "name": "stream stream inner join with out of order and custom grace period",
+      "format": ["AVRO", "JSON", "PROTOBUF"],
+      "statements": [
+        "CREATE STREAM LEFT_STREAM (ID BIGINT KEY, L1 varchar) WITH (kafka_topic='left_topic', value_format='{FORMAT}');",
+        "CREATE STREAM RIGHT_STREAM (ID BIGINT KEY, L2 varchar) WITH (kafka_topic='right_topic', value_format='{FORMAT}');",
+        "CREATE STREAM INNER_JOIN as SELECT t.id, l1, l2 FROM LEFT_STREAM t join RIGHT_STREAM tt WITHIN 1 minute GRACE PERIOD 1 minute on t.id = tt.id;"
+      ],
+      "inputs": [
+        {"topic": "left_topic", "key": 0, "value": {"L1": "A"}, "timestamp": 0},
+        {"topic": "right_topic", "key": 0, "value": {"L2": "a"}, "timestamp": 60000},
+        {"topic": "left_topic", "key": 1, "value": {"L1": "B"}, "timestamp": 330000},
+        {"topic": "left_topic", "key": 2, "value": {"L1": "C"}, "timestamp": 90000},
+        {"topic": "right_topic", "key": 2, "value": {"L2": "c"}, "timestamp": 90000},
+        {"topic": "left_topic", "key": 3, "value": {"L1": "D"}, "timestamp": 60000},
+        {"topic": "right_topic", "key": 3, "value": {"L2": "d"}, "timestamp": 60000}
+      ],
+      "outputs": [
+        {"topic": "INNER_JOIN", "key": 0, "value": {"L1": "A", "L2": "a"}, "timestamp": 60000},
+        {"topic": "INNER_JOIN", "key": 2, "value": {"L1": "C", "L2": "c"}, "timestamp": 90000}
+      ]
+    },
+    {
+      "name": "stream stream left join with out of order and custom grace period",
+      "format": ["AVRO", "JSON"],
+      "statements": [
+        "CREATE STREAM LEFT_STREAM (ID BIGINT KEY, L1 varchar) WITH (kafka_topic='left_topic', value_format='{FORMAT}');",
+        "CREATE STREAM RIGHT_STREAM (ID BIGINT KEY, L2 varchar) WITH (kafka_topic='right_topic', value_format='{FORMAT}');",
+        "CREATE STREAM LEFT_JOIN as SELECT t.id, l1, l2 FROM LEFT_STREAM t left join RIGHT_STREAM tt WITHIN 1 minute GRACE PERIOD 1 minute on t.id = tt.id;"
+      ],
+      "inputs": [
+        {"topic": "left_topic", "key": 0, "value": {"L1": "A"}, "timestamp": 0},
+        {"topic": "right_topic", "key": 0, "value": {"L2": "a"}, "timestamp": 60000},
+        {"topic": "left_topic", "key": 1, "value": {"L1": "B"}, "timestamp": 330000},
+        {"topic": "left_topic", "key": 2, "value": {"L1": "C"}, "timestamp": 90000},
+        {"topic": "right_topic", "key": 2, "value": {"L2": "c"}, "timestamp": 90000},
+        {"topic": "left_topic", "key": 3, "value": {"L1": "D"}, "timestamp": 60000},
+        {"topic": "right_topic", "key": 3, "value": {"L2": "d"}, "timestamp": 60000}
+      ],
+      "outputs": [
+        {"topic": "LEFT_JOIN", "key": 0, "value": {"L1": "A", "L2": null}, "timestamp": 0},
+        {"topic": "LEFT_JOIN", "key": 0, "value": {"L1": "A", "L2": "a"}, "timestamp": 60000},
+        {"topic": "LEFT_JOIN", "key": 1, "value": {"L1": "B", "L2": null}, "timestamp": 330000},
+        {"topic": "LEFT_JOIN", "key": 2, "value": {"L1": "C", "L2": null}, "timestamp": 90000},
+        {"topic": "LEFT_JOIN", "key": 2, "value": {"L1": "C", "L2": "c"}, "timestamp": 90000},
+        {"topic": "LEFT_JOIN", "key": 3, "value": {"L1": "D", "L2": null}, "timestamp": 60000}
+      ]
+    },
+    {
+      "name": "stream stream full outer join with out of order and custom grace period",
+      "format": ["AVRO", "JSON"],
+      "statements": [
+        "CREATE STREAM LEFT_STREAM (ID BIGINT KEY, L1 varchar) WITH (kafka_topic='left_topic', value_format='{FORMAT}');",
+        "CREATE STREAM RIGHT_STREAM (ID BIGINT KEY, L2 varchar) WITH (kafka_topic='right_topic', value_format='{FORMAT}');",
+        "CREATE STREAM OUTER_JOIN as SELECT ROWKEY as ID, t.id, tt.id, l1, l2 FROM LEFT_STREAM t full outer join RIGHT_STREAM tt WITHIN 1 minute GRACE PERIOD 1 minute on t.id = tt.id;"
+      ],
+      "inputs": [
+        {"topic": "left_topic", "key": 0, "value": {"L1": "A"}, "timestamp": 0},
+        {"topic": "right_topic", "key": 0, "value": {"L2": "a"}, "timestamp": 60000},
+        {"topic": "right_topic", "key": 1, "value": {"L2": "b"}, "timestamp": 330000},
+        {"topic": "right_topic", "key": 2, "value": {"L2": "c"}, "timestamp": 90000},
+        {"topic": "left_topic", "key": 2, "value": {"L1": "C"}, "timestamp": 90000},
+        {"topic": "right_topic", "key": 3, "value": {"L2": "d"}, "timestamp": 60000},
+        {"topic": "left_topic", "key": 3, "value": {"L1": "D"}, "timestamp": 60000}
+      ],
+      "outputs": [
+        {"topic": "OUTER_JOIN", "key": 0, "value": {"T_ID": 0, "TT_ID": null, "L1": "A", "L2": null}, "timestamp": 0},
+        {"topic": "OUTER_JOIN", "key": 0, "value": {"T_ID": 0, "TT_ID": 0, "L1": "A", "L2": "a"}, "timestamp": 60000},
+        {"topic": "OUTER_JOIN", "key": 1, "value": {"T_ID": null, "TT_ID": 1, "L1": null, "L2": "b"}, "timestamp": 330000},
+        {"topic": "OUTER_JOIN", "key": 2, "value": {"T_ID": null, "TT_ID": 2, "L1": null, "L2": "c"}, "timestamp": 90000},
+        {"topic": "OUTER_JOIN", "key": 2, "value": {"T_ID": 2, "TT_ID": 2, "L1": "C", "L2": "c"}, "timestamp": 90000},
+        {"topic": "OUTER_JOIN", "key": 3, "value": {"T_ID": null, "TT_ID": 3, "L1": null, "L2": "d"}, "timestamp": 60000},
+        {"topic": "OUTER_JOIN", "key": 3, "value": {"T_ID": 3, "TT_ID": null, "L1": "D", "L2": null}, "timestamp": 60000}
+      ]
+    },
     {
       "name": "stream stream outer join",
       "format": ["AVRO", "JSON"],