Milesian111 commented on code in PR #29078:
URL: https://github.com/apache/flink/pull/29078#discussion_r3976470071
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -620,6 +621,72 @@ def top_n(
distinct = drop_duplicates
unique = drop_duplicates
+ # ======================== Filtering & Ordering ========================
+
+ @PublicEvolving()
+ def sort(
+ self,
+ by: Union[str, Expression, List[Union[str, Expression]]],
+ *,
+ descending: Union[bool, List[bool]] = False,
+ nulls_first: Union[bool, List[bool]] = None,
+ ) -> "DataFrame":
+ """
+ Sort rows globally by one or more columns or expressions.
+
+ This method builds a new DataFrame plan without executing a Flink job.
The ``by``
+ expressions must not already specify ``asc`` or ``desc``; use
``descending`` to control
+ their direction. When ``nulls_first`` is omitted, the Table API
default is used: NULLs
+ are ordered last for ascending keys and first for descending keys.
+
+ The result is globally sorted across all parallel partitions. For
unbounded tables, this
+ operation requires a time-attribute sort or a subsequent fetch
operation.
+
+ :param by: Column name or expression, or a list of them, used as sort
keys.
+ :param descending: Whether to sort in descending order, either for all
keys or once per
+ key.
+ :param nulls_first: Whether to place NULLs first, either for all keys
or once per key. When
+ omitted, the Table API default applies.
+ :return: A new sorted DataFrame.
+ :raises TypeError: If ``by``, ``descending`` or ``nulls_first`` has an
unsupported type.
+ :raises ValueError: If ``by`` is empty, option lengths do not match, a
column does not
+ exist, or an expression already specifies ``asc`` or ``desc``.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records(
+ ... [(2, "b"), (1, "a")], schema=["id", "name"]
+ ... )
+ >>> ascending = df.sort("id")
+ >>> mixed = df.sort(["id", "name"], descending=[False, True])
+
+ .. versionadded:: 2.4.0
+ """
+ order_keys = _normalize_order_by(by, "by")
+ if order_keys is None:
+ raise TypeError("by must be a string, an expression, or a list or
tuple of them")
+ columns = self._table.get_resolved_schema().get_column_names()
+ for key in order_keys:
+ if isinstance(key, str) and key not in columns:
+ raise ValueError(
+ "by column '%s' does not exist, available columns: %s" %
(key, columns)
+ )
+ if isinstance(key, Expression) and
_contains_ordering_expression(key):
+ raise ValueError(
+ "sort() expressions must not specify asc or desc; use
descending instead"
+ )
+
+ descending_values = _normalize_descending(descending, len(order_keys))
Review 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!
--
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]
