Timm0 commented on code in PR #29022:
URL: https://github.com/apache/flink/pull/29022#discussion_r3871572302
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -836,6 +914,132 @@ def agg(self, *aggs: Expression, **named_aggs:
Expression) -> DataFrame:
# ======================== Internal Helpers ========================
+def _normalize_subset(subset: Union[str, List[str], None]) ->
Optional[List[str]]:
+ if subset is None:
+ return None
+ if isinstance(subset, str):
+ return [subset]
+ if isinstance(subset, (list, tuple)):
+ if not subset:
+ raise ValueError("subset must not be empty")
+ for name in subset:
+ if not isinstance(name, str):
+ raise TypeError("subset must be a string or a list of strings")
+ return list(subset)
+ raise TypeError("subset must be a string or a list of strings")
+
+
+def _normalize_order_by(
+ order_by: Union[str, Expression, List[Union[str, Expression]], None],
+) -> Optional[List[Union[str, Expression]]]:
+ if order_by is None:
+ return None
+
+ values = order_by if isinstance(order_by, (list, tuple)) else [order_by]
+ keys: List[Union[str, Expression]] = []
+ for value in values:
+ if isinstance(value, (str, Expression)):
+ keys.append(value)
+ else:
+ raise TypeError(
+ "order_by must be a string, an expression, or a list or tuple
of them"
+ )
+
+ if not keys:
+ raise ValueError("order_by must not be empty")
+
+ return keys
+
+
+def _normalize_nulls_first(
+ nulls_first: Union[bool, List[bool], None], order_len: int
+) -> Optional[List[bool]]:
+ if nulls_first is None:
+ return None
+
+ if isinstance(nulls_first, bool):
+ values = [nulls_first] * order_len
+ elif isinstance(nulls_first, (list, tuple)):
+ for value in nulls_first:
+ if not isinstance(value, bool):
+ raise TypeError("nulls_first must be a boolean or a list of
booleans")
+ values = list(nulls_first)
+ else:
+ raise TypeError("nulls_first must be a boolean or a list of booleans")
+
+ if len(values) != order_len:
+ raise ValueError("nulls_first must have the same length as order_by")
+
+ return values
+
+
+def _build_deduplication_query(
+ table: Table,
+ columns: List[str],
+ subset_keys: List[str],
+ order_keys: Optional[List[Union[str, Expression]]],
+ keep: str,
+ nulls: Optional[List[bool]],
+) -> Table:
+ direction = "DESC" if keep == "last" else "ASC"
+ taken = set(columns)
+
+ if order_keys is None:
+ # Arrival order (processing time); keep decides its direction.
+ order_terms = ["PROCTIME() " + direction]
+ else:
+ order_terms = []
+ for index, key in enumerate(order_keys):
+ if isinstance(key, str):
+ if key not in columns:
+ raise ValueError(
+ "order_by column '%s' does not exist, available
columns: %s"
+ % (key, columns)
+ )
+ name = key
+ else:
+ # An Expression cannot be rendered to SQL text, so materialize
it as a helper
+ # column and reference it by name.
+ name = _unique_name("__pf_order_%d" % index, taken)
+ taken.add(name)
+ table = table.add_columns(key.alias(name))
+ term = _quote_identifier(name) + " " + direction
+ if nulls is not None:
+ term += " NULLS FIRST" if nulls[index] else " NULLS LAST"
+ order_terms.append(term)
+
+ rank_column = _quote_identifier(_unique_name("__pf_row_number", taken))
+ source = _quote_identifier(str(table))
+ select_list = ", ".join(_quote_identifier(name) for name in columns)
+ partition_by = ", ".join(_quote_identifier(name) for name in subset_keys)
+ query = (
Review Comment:
Good catch, I'll create a ticket and work on it as a follow-up
--
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]
