gustavodemorais commented on code in PR #27935:
URL: https://github.com/apache/flink/pull/27935#discussion_r3092588028
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java:
##########
@@ -1112,35 +1113,61 @@ public class ProcessTableFunctionTestPrograms {
public static final TableTestProgram PROCESS_LATE_EVENTS =
TableTestProgram.of(
"process-late-events",
- "test that late events are dropped in both input
and when registering timers")
+ "test that late events enter PTF (eval and timer
registration)")
.setupTemporarySystemFunction("f",
LateTimersFunction.class)
.setupTableSource(TIMED_SOURCE_LATE_EVENTS)
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
.consumedValues(
"+I[Bob, {Processing input row
+I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null},
1970-01-01T00:00:00Z]",
- "+I[Bob, {Registering timer t for
0 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer bob
for 0 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
"+I[Bob, {Registering timer for 0
at time 0 watermark null}, 1970-01-01T00:00:00Z]",
"+I[Alice, {Processing input row
+I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1},
1970-01-01T00:00:00.001Z]",
- "+I[Alice, {Registering timer t
for 0 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
- "+I[Alice, {Registering timer for
0 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer
alice for 1 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
"+I[Bob, {Timer null fired at time
0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Bob, {Registering timer again
for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Alice, {Timer null fired at
time 0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Alice, {Registering timer
again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Bob, {Timer t fired at time 0
watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Bob, {Registering timer again
for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Alice, {Timer t fired at time
0 watermark 0}, 1970-01-01T00:00:00Z]",
- "+I[Alice, {Registering timer
again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Timer bob fired at time
0 watermark 0}, 1970-01-01T00:00:00Z]",
"+I[Bob, {Processing input row
+I[Bob, 2, 1970-01-01T00:01:39.999Z] at time 99999 watermark 0},
1970-01-01T00:01:39.999Z]",
- "+I[Bob, {Registering timer t for
0 at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]",
- "+I[Bob, {Registering timer for 0
at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]")
+ "+I[Bob, {Registering timer bob
for 0 at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]",
+ "+I[Bob, {Registering timer for 0
at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]",
+ "+I[Bob, {Timer null fired at time
0 watermark 99998}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Timer bob fired at time
0 watermark 99998}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Timer alice fired at
time 1 watermark 99998}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer
alice-again for 0 at time 1 watermark 99998}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Timer alice-again
fired at time 0 watermark 99998}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Processing input row
+I[Bob, 3, 1970-01-01T00:00:00.003Z] at time 3 watermark 99998},
1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Registering timer bob
for 0 at time 3 watermark 99998}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Registering timer for 0
at time 3 watermark 99998}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Processing input row
+I[Bob, 4, 1970-01-01T00:00:00.004Z] at time 4 watermark 99998},
1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Registering timer bob
for 0 at time 4 watermark 99998}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Registering timer for 0
at time 4 watermark 99998}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Timer null fired at time
0 watermark 9223372036854775807}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Timer bob fired at time
0 watermark 9223372036854775807}, 1970-01-01T00:00:00Z]")
.build())
.runSql(
"INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
.build();
+ public static final TableTestProgram PROCESS_ROW_LATE_EVENTS =
+ TableTestProgram.of(
+ "process-row-late-events",
+ "test that late events enter a row-level PTF")
Review Comment:
```suggestion
"test that late events enter a PTF with row
semantics")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
