twalthr opened a new pull request, #27929:
URL: https://github.com/apache/flink/pull/27929
## 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
--
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]
