[FLINK-40421][python] Add sorting APIs to DataFrame API - #29078
[FLINK-40421][python] Add sorting APIs to DataFrame API#29078Milesian111 wants to merge 1 commit into
Conversation
3c5cf58 to
f71e1d1
Compare
|
@flinkbot run azure |
dianfu
left a comment
dianfu
left a comment
There was a problem hiding this comment.
@Milesian111 Good work! Have left a few comments.
| "sort() expressions must not specify asc or desc; use descending instead" | ||
| ) | ||
|
|
||
| descending_values = _normalize_sort_flags(descending, len(order_keys), "descending") |
There was a problem hiding this comment.
PR #29105 has introduced a method _normalize_descending which is similar to _normalize_sort_flags. Could you rebase the PR and check if we could use that method? Besides, there is code conflict with master, should be caused by the above PR.
| order_keys, descending_values, nulls_values | ||
| ): | ||
| expression = table_col(key) if isinstance(key, str) else key | ||
| if is_nulls_first is not None: |
There was a problem hiding this comment.
Adding expression.is_null as the leading sort key prevents the streaming planner from recognizing a temporal sort.
For example, df.sort(\"ts\") on a rowtime column produces TemporalSort, but df.sort(\"ts\", nulls_first=True) becomes ORDER BY ts IS NULL DESC, ts ASC and fails with Sort on a non-time-attribute field is not supported.
Please represent null placement as part of the original key, e.g. ORDER BY ts ASC NULLS FIRST
| [Row(1, 20, "B"), Row(1, 10, "A"), Row(2, 5, "C")], | ||
| ) | ||
|
|
||
| def test_sort_supports_explicit_null_ordering(self): |
There was a problem hiding this comment.
This test calls collect() five times, launching five separate Flink jobs, while the default ascending and descending cases are already covered elsewhere. Please keep only the essential non-default combinations, for example descending=True, nulls_first=False and one expression-key case, to reduce the integration-test runtime.
133b640 to
b9af998
Compare
| self.assertIsNot(result, self.dataframe) | ||
| self.assertIsNot(result.to_table(), self.dataframe.to_table()) | ||
| self.assertEqual( | ||
| result.to_table()._j_table.getQueryOperation().getOrder().toString(), |
There was a problem hiding this comment.
These assertions still assume that sort() calls Table.order_by() and returns a query operation with getOrder(). After switching to sql_query(), the root operation does not expose that method, so four tests will fail with Method getOrder([]) does not exist.
We can update the tests to verify the generated SQL or observable behavior. For example, sort("id", nulls_first=True) should generate ORDER BY id ASC NULLS FIRST.
| "sort() expressions must not specify asc or desc; use descending instead" | ||
| ) | ||
|
|
||
| descending_values = _normalize_descending(descending, len(order_keys)) |
There was a problem hiding this comment.
These helpers hard-code order_by in their length-mismatch errors, but sort() exposes the parameter as by.
We can take the same way as _normalize_order_by by introducing parameter_name or use a generic term such as sort keys. The same issue applies to nulls_first.
There was a problem hiding this comment.
Thanks for the review and applying the remaining adjustments. Sorry for the delayed reply.
I planned to address all 5 suggestions before responding, but you’ve gone ahead, closed this PR and committed the last two fixes haha.
I thought I should reply first to sync my progress with reviewer next time.
Appreciate you taking care of these again!
Add lazy DataFrame.sort backed by Table.order_by with per-key direction and explicit null ordering. Reject ordered expressions and cover validation, planning, batch behavior, and mypy-compatible typing with focused tests. Generated-by: Codex GPT-5
b9af998 to
b927482
Compare
3 participants
What is the purpose of the change
This pull request adds sorting support to the PyFlink DataFrame API. It introduces a lazy
DataFrame.sort()transformation backed by the existing Table APIorder_by()operation, allowing users to sort by one or more columns or expressions without triggering execution.Brief change log
DataFrame.sort()with support for single-column and multi-column sorting.descendingoptions.nulls_firstoptions..ascor.desc.Verifying this change
This change added focused tests covering:
The focused sorting tests and complete DataFrame batch integration tests pass locally. The full DataFrame test module has one unrelated Windows-only failure involving pre-epoch timestamp conversion in
time.mktime.Does this pull request potentially affect one of the following parts:
Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Codex GPT-5