twalthr commented on code in PR #27929:
URL: https://github.com/apache/flink/pull/27929#discussion_r3085890431
##########
docs/content/docs/dev/table/functions/ptfs.md:
##########
@@ -1015,6 +1030,157 @@ not needed anymore via `Context#clearAllTimers()` or
`TimeContext#clearTimer(Str
{{< top >}}
+Ordering
+--------
+
+A PTF that takes a table with set semantics can optionally specify an ORDER BY
clause in the
+function call to define the order in which rows are processed within each
partition. The ORDER BY
+clause guarantees that rows are delivered to the eval() method in the
specified order.
+
+The ORDER BY clause requires that the first column is a time attribute column
(i.e., a
+TIMESTAMP or TIMESTAMP_LTZ column with a watermark declaration). The first
ORDER BY column must
+be specified in ascending order. This ensures that rows are processed in
event-time order.
+Additional non-temporal columns can be specified as secondary sort keys to
define the ordering of
+rows with the same timestamp.
+
+{{< tabs "2137eeed-order-by-syntax" >}}
+{{< tab "SQL" >}}
+```sql
+SELECT * FROM my_ptf(
+ input_table => TABLE source_table
+ PARTITION BY user_id
+ ORDER BY (event_time ASC, priority DESC NULLS FIRST)
+)
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+env.from("source_table")
+ .partitionBy($("user_id"))
+ .orderBy($("event_time").asc(), $("priority").desc())
+ .process(MyPTF.class, descriptor("event_time").asArgument("on_time"))
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### Difference Between ORDER BY and on_time Argument
+
+While both ORDER BY and the `on_time` argument relate to time attributes, they
serve
+different purposes:
+
+- **on_time**: Declares which time attribute column powers the time context
(`TimeContext#time()`) and
+ output timestamp. It does NOT affect the processing order of rows.
+- **ORDER BY**: Physically buffers and sorts rows within each partition to
guarantee ordered delivery
+ to the eval() method. If both ORDER BY and `on_time` are specified for the
same table argument, they
+ must reference the same time attribute column.
+
+### Ordering Guarantees and Late Events
+
+When ORDER BY is specified on a time attribute column, the framework maintains
a sort buffer
+per partition and input table to reorder out-of-order events. The sort buffer
is flushed when the
+watermark for the given input table advances, at which point all buffered rows
with timestamps
+less than or equal to the watermark are delivered to the eval() method in
sorted order. Late
+events (arriving after the watermark) are dropped to maintain the ordering
guarantee.
+
+The following example demonstrates ordered processing with secondary sorting.
First, the function implementation:
+
+```java
+// Function that processes events in order and captures the ordering
+public static class OrderedProcessor extends ProcessTableFunction<List<Event>>
{
+ public static class Event {
Review Comment:
I converted it to `record`. But overall, it is also nice if example show
various ways of using the API. It doesn't need to be record.
--
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]
