[FLINK-39256][table] Support ORDER BY clause in Process Table Functions

[twalthr]

What is the purpose of the change

This PR implements ORDER BY support for Process Table Functions (PTFs), enabling guaranteed ordered processing of rows within partitions based on time attributes and additional sort keys.

Brief change log

1. ORDER BY Clause Support

• PTFs can now specify ORDER BY in table arguments with set semantics
• First column must be a time attribute (TIMESTAMP/TIMESTAMP_LTZ) in ascending order
• Supports secondary sort columns with flexible ASC/DESC and NULLS FIRST/LAST options
• Example: SELECT * FROM my_ptf(input => TABLE t PARTITION BY k ORDER BY (ts ASC, priority DESC))

2. Enhanced Time Context API

• Added TimeContext#tableWatermark() - returns the watermark of the specific input table currently being processed
• Improved TimeContext#currentWatermark() documentation to clarify it returns the global minimum watermark across all inputs
• Critical for multi-input scenarios where each table has independent watermarks

3. Runtime Implementation

• InputSortBuffer: Timer-based sorting service that buffers rows by timestamp and emits in sorted order when watermarks advance
• MailboxPartialWatermarkProcessor: Manages partial watermark tracking per input table for multi-input PTFs
• Optimized timer usage from O(K × T) to O(K) where K = keys and T = unique timestamps
• Late events (arriving after watermark) are automatically dropped to maintain ordering guarantees

4. TableSemantics Enhancements

• Added orderByColumns() - returns ORDER BY column indices
• Added orderByDirections() - returns sort directions with null handling
• Added SortDirection enum (ASC_NULLS_FIRST, ASC_NULLS_LAST, DESC_NULLS_FIRST, DESC_NULLS_LAST)
• Enables runtime access to ordering metadata within PTF implementations

5. Planner Integration

• Extended RexTableArgCall to track ORDER BY information
• Updated type inference system to validate and propagate ordering constraints
• Enhanced serialization/deserialization for plan persistence

Verifying this change

This change added tests and can be verified as follows:

ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_TABLE_API
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_SINGLE_PARTITION_TABLE_API
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_RESTORE
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY_RESTORE

and various unit tests.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs

[flinkbot]

CI report:

• bb83b36 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[snuyanzin]

will it work with Event as record?
I wonder if we should apply it to examples as well

[snuyanzin]

do we need it to have Serializable?
aren't java's enums serializable by default?

[snuyanzin]

can we instead reuse Calcite's Direction https://github.com/apache/calcite/blob/0ce0c5ae2f7055f13551ebcde8e1aaa2ff17b469/core/src/main/java/org/apache/calcite/rel/RelFieldCollation.java#L54
?

[snuyanzin]

also I guess in case we know something more about data like if it is monotonic, then order by might be optimized (Calcite's RelFieldCollation already has some functionality for that)

[snuyanzin]

Suggested change

	public PartitionedTable orderBy(Expression... fields) {
	public PartitionedTable orderBy(Expression... expressions) {

in general it could be not only fields, right?

[snuyanzin]

We have some weird tests with expressions in ORDER BY like for instance

flink/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala

Lines 1035 to 1036 in d8010df

	"SELECT d, h, dense_rank() over (order by cast(d as tinyint), cast(h as smallint) desc)" +
	" FROM Table5",

are there any limitations for PTF case?

can there be order by by some PTF?
do we have any validation around this?

[snuyanzin]

Suggested change

	this(partitionExpressions, Collections.emptyList(), child);
	this(partitionExpressions, List.of(), child);

[snuyanzin]

Suggested change

	this(table, partitionKeys, Collections.emptyList());
	this(table, partitionKeys, List.of());

[snuyanzin]

Suggested change

	"test", mailboxExecutor, Collections.emptyList(), consumedWatermarks::add);
	"test", mailboxExecutor, List.of(), consumedWatermarks::add);

[snuyanzin]

Suggested change

	Arrays.asList(service1, service2, service3),
	List.of(service1, service2, service3),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),

[snuyanzin]

Suggested change

	Collections.singletonList(timerService),
	List.of(timerService),