dianfu commented on code in PR #29078:
URL: https://github.com/apache/flink/pull/29078#discussion_r3966809893
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -560,6 +561,82 @@ def drop_duplicates(
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_sort_flags(descending, len(order_keys),
"descending")
Review 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.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -560,6 +561,82 @@ def drop_duplicates(
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_sort_flags(descending, len(order_keys),
"descending")
+ nulls_values: List[Optional[bool]] = (
+ [None] * len(order_keys)
+ if nulls_first is None
+ else _normalize_sort_flags(nulls_first, len(order_keys),
"nulls_first")
+ )
+
+ order_expressions: List[Expression] = []
+ for key, is_descending, is_nulls_first in zip(
+ order_keys, descending_values, nulls_values
+ ):
+ expression = table_col(key) if isinstance(key, str) else key
+ if is_nulls_first is not None:
Review 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`
##########
flink-python/pyflink/dataframe/tests/test_dataframe.py:
##########
@@ -1903,6 +2000,62 @@ def _ordered_dataframe(self):
)
return pf.from_table(table.order_by(table.id))
+ def _unsorted_dataframe(self):
+ table = self.t_env.sql_query(
+ "SELECT * FROM (VALUES (3, 'C'), (1, 'A'), (2, 'B')) AS T(id,
name)"
+ )
+ return pf.from_table(table)
+
+ def _nullable_dataframe(self):
+ table = self.t_env.sql_query(
+ "SELECT * FROM (VALUES (CAST(NULL AS INT), 'NULL'), (2, 'B'), (1,
'A')) "
+ "AS T(id, name)"
+ )
+ return pf.from_table(table)
+
+ def test_sort_returns_rows_in_ascending_order(self):
+ self.assertEqual(
+ self._unsorted_dataframe().sort("id").collect(),
+ [Row(1, "A"), Row(2, "B"), Row(3, "C")],
+ )
+
+ def test_sort_supports_per_key_descending_order(self):
+ dataframe = pf.from_table(
+ self.t_env.sql_query(
+ "SELECT * FROM (VALUES (1, 10, 'A'), (1, 20, 'B'), (2, 5,
'C')) "
+ "AS T(group_id, score, name)"
+ )
+ )
+
+ self.assertEqual(
+ dataframe.sort(["group_id", "score"], descending=[False,
True]).collect(),
+ [Row(1, 20, "B"), Row(1, 10, "A"), Row(2, 5, "C")],
+ )
+
+ def test_sort_supports_explicit_null_ordering(self):
Review 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.
--
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]
