[BP-2.3][FLINK-39436][FLINK-39437][table] Backport of two PTF fixes

docs/content.zh/docs/dev/table/functions/ptfs.md

@@ -1021,6 +1021,18 @@ class TimerFunction extends ProcessTableFunction<String> {
 {{< /tab >}}
 {{< /tabs >}}
 
+### Handling of Late Records
+
+A late record is a record with a time attribute value that is less than or equal to the current
+watermark. PTFs handle late records just like non-late records by calling the `eval()` method. If
+the `on_time` argument is specified, the late timestamp is preserved in the output. This behavior is
+the same for PTFs with row and set semantics.
+
+Registering a timer for a time that is less than or equal to the current watermark is allowed.
+If registered from within `eval()`, the timer fires on the next watermark advance. If registered
+from within `onTimer()`, the timer fires immediately after the current timer finishes. Note that
+unconditionally re-registering a past-time timer from within `onTimer()` causes an infinite loop.
+
 ### Efficiency and Design Principles
 
 Registering too many timers might affect performance. An ever-growing timer state can happen

docs/content/docs/dev/table/functions/ptfs.md

@@ -1022,6 +1022,18 @@ class TimerFunction extends ProcessTableFunction<String> {
 {{< /tab >}}
 {{< /tabs >}}
 
+### Handling of Late Records
+
+A late record is a record with a time attribute value that is less than or equal to the current
+watermark. PTFs handle late records just like non-late records by calling the `eval()` method. If 
+the `on_time` argument is specified, the late timestamp is preserved in the output. This behavior is
+the same for PTFs with row and set semantics.
+
+Registering a timer for a time that is less than or equal to the current watermark is allowed.
+If registered from within `eval()`, the timer fires on the next watermark advance. If registered
+from within `onTimer()`, the timer fires immediately after the current timer finishes. Note that
+unconditionally re-registering a past-time timer from within `onTimer()` causes an infinite loop.
+
 ### Efficiency and Design Principles
 
 Registering too many timers might affect performance. An ever-growing timer state can happen

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java

@@ -403,6 +403,19 @@
  * }
  * }</pre>
  *
+ * <h2>Handling of Late Records</h2>
+ *
+ * <p>A late record is a record with a time attribute value that is less than or equal to the
+ * current watermark. PTFs handle late records just like non-late records by calling the {@code
+ * eval()} method. If the {@code on_time} argument is specified, the late timestamp is preserved in
+ * the output. This behavior is the same for PTFs with row and set semantics.
+ *
+ * <p>Registering a timer for a time that is less than or equal to the current watermark is allowed.
+ * If registered from within {@code eval()}, the timer fires on the next watermark advance. If
+ * registered from within {@code onTimer()}, the timer fires immediately after the current timer
+ * finishes. Note that unconditionally re-registering a past-time timer from within {@code
+ * onTimer()} causes an infinite loop.
+ *
  * <h2>Efficiency and Design Principles</h2>
  *
  * <p>Registering too many timers might affect performance. An ever-growing timer state can happen
@@ -660,6 +673,12 @@ public interface TimeContext<TimeType> {
          * timer only fires if a watermark was received from all inputs and the timestamp is smaller
          * or equal to the minimum of all received watermarks.
          *
+         * <p>If the timestamp of the registered timer is already less than or equal to the current
+         * watermark, the timer fires on the next watermark advance if registered from within {@code
+         * eval()}, or immediately after the current timer finishes if registered from within {@code
+         * onTimer()}. Note that unconditionally re-registering a past-time timer from within {@code
+         * onTimer()} causes an infinite loop.
+         *
          * <p>Timers can be named for distinguishing them in the {@code onTimer()} method.
          * Registering a timer under the same name twice will replace an existing timer.
          *
@@ -680,6 +699,12 @@ public interface TimeContext<TimeType> {
          * timer only fires if a watermark was received from all inputs and the timestamp is smaller
          * or equal to the minimum of all received watermarks.
          *
+         * <p>If the timestamp of the registered timer is already less than or equal to the current
+         * watermark, the timer fires on the next watermark advance if registered from within {@code
+         * eval()}, or immediately after the current timer finishes if registered from within {@code
+         * onTimer()}. Note that unconditionally re-registering a past-time timer from within {@code
+         * onTimer()} causes an infinite loop.
+         *
          * <p>Only one timer can be registered for a given time.
          *
          * <p>Note: Because only PTFs taking set semantic tables support state, and timers are a

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionRestoreTests.java

@@ -40,6 +40,7 @@ public List<TableTestProgram> programs() {
                 ProcessTableFunctionTestPrograms.PROCESS_MAP_STATE_RESTORE,
                 ProcessTableFunctionTestPrograms.PROCESS_UPDATING_OUTPUT_UPSERT_RESTORE,
                 ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_RESTORE,
-                ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY_RESTORE);
+                ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY_RESTORE,
+                ProcessTableFunctionTestPrograms.PROCESS_LATE_EVENTS_RESTORE);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java

@@ -87,6 +87,7 @@ public List<TableTestProgram> programs() {
                 ProcessTableFunctionTestPrograms.PROCESS_NAMED_TIMERS,
                 ProcessTableFunctionTestPrograms.PROCESS_UNNAMED_TIMERS,
                 ProcessTableFunctionTestPrograms.PROCESS_LATE_EVENTS,
+                ProcessTableFunctionTestPrograms.PROCESS_ROW_LATE_EVENTS,
                 ProcessTableFunctionTestPrograms.PROCESS_SCALAR_ARGS_TIME,
                 ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY_TIME,
                 ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_ON_TIME,