grundprinzip commented on code in PR #38768:
URL: https://github.com/apache/spark/pull/38768#discussion_r1030246542
##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -79,23 +67,23 @@ def agg(self, exprs: Optional[MeasuresType] = None) ->
"DataFrame":
)
return res
- def _map_cols_to_dict(self, fun: str, cols: List[Union[Column, str]]) ->
Dict[str, str]:
- return {x if isinstance(x, str) else x.name(): fun for x in cols}
+ def _map_cols_to_expression(self, fun: str, col: Union[Column, str]) ->
Sequence[Expression]:
+ return [ScalarFunctionExpression(fun, Column(col) if isinstance(col,
str) else col)]
- def min(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("min", list(cols))
+ def min(self, col: Union[Column, str]) -> "DataFrame":
Review Comment:
```suggestion
def min(self, col: Union[Expression, str]) -> "DataFrame":
```
##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -79,23 +67,23 @@ def agg(self, exprs: Optional[MeasuresType] = None) ->
"DataFrame":
)
return res
- def _map_cols_to_dict(self, fun: str, cols: List[Union[Column, str]]) ->
Dict[str, str]:
- return {x if isinstance(x, str) else x.name(): fun for x in cols}
+ def _map_cols_to_expression(self, fun: str, col: Union[Column, str]) ->
Sequence[Expression]:
+ return [ScalarFunctionExpression(fun, Column(col) if isinstance(col,
str) else col)]
- def min(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("min", list(cols))
+ def min(self, col: Union[Column, str]) -> "DataFrame":
+ expr = self._map_cols_to_expression("min", col)
return self.agg(expr)
- def max(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("max", list(cols))
+ def max(self, col: Union[Column, str]) -> "DataFrame":
+ expr = self._map_cols_to_expression("max", col)
return self.agg(expr)
- def sum(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("sum", list(cols))
+ def sum(self, col: Union[Column, str]) -> "DataFrame":
Review Comment:
```suggestion
def sum(self, col: Union[Expression, str]) -> "DataFrame":
```
##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -79,23 +67,23 @@ def agg(self, exprs: Optional[MeasuresType] = None) ->
"DataFrame":
)
return res
- def _map_cols_to_dict(self, fun: str, cols: List[Union[Column, str]]) ->
Dict[str, str]:
- return {x if isinstance(x, str) else x.name(): fun for x in cols}
+ def _map_cols_to_expression(self, fun: str, col: Union[Column, str]) ->
Sequence[Expression]:
+ return [ScalarFunctionExpression(fun, Column(col) if isinstance(col,
str) else col)]
- def min(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("min", list(cols))
+ def min(self, col: Union[Column, str]) -> "DataFrame":
+ expr = self._map_cols_to_expression("min", col)
return self.agg(expr)
- def max(self, *cols: Union[Column, str]) -> "DataFrame":
- expr = self._map_cols_to_dict("max", list(cols))
+ def max(self, col: Union[Column, str]) -> "DataFrame":
Review Comment:
```suggestion
def max(self, col: Union[Expression, str]) -> "DataFrame":
```
##########
python/pyspark/sql/connect/plan.py:
##########
@@ -558,29 +557,19 @@ def _repr_html_(self) -> str:
class Aggregate(LogicalPlan):
- MeasureType = Tuple["ExpressionOrString", str]
- MeasuresType = Sequence[MeasureType]
- OptMeasuresType = Optional[MeasuresType]
-
def __init__(
self,
child: Optional["LogicalPlan"],
grouping_cols: List[Column],
- measures: OptMeasuresType,
Review Comment:
There is no longer a way to call this with empty measures?
##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -164,8 +152,21 @@ def selectExpr(self, *expr: Union[str, List[str]]) ->
"DataFrame":
return DataFrame.withPlan(plan.Project(self._plan, *sql_expr),
session=self._session)
- def agg(self, exprs: Optional[GroupingFrame.MeasuresType]) -> "DataFrame":
- return self.groupBy().agg(exprs)
+ def agg(self, *exprs: Union[Expression, Dict[str, str]]) -> "DataFrame":
+ if not exprs:
+ raise ValueError("exprs should not be empty")
Review Comment:
```suggestion
raise ValueError("Argument 'exprs' must not be empty")
```
##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -79,23 +67,23 @@ def agg(self, exprs: Optional[MeasuresType] = None) ->
"DataFrame":
)
return res
- def _map_cols_to_dict(self, fun: str, cols: List[Union[Column, str]]) ->
Dict[str, str]:
- return {x if isinstance(x, str) else x.name(): fun for x in cols}
+ def _map_cols_to_expression(self, fun: str, col: Union[Column, str]) ->
Sequence[Expression]:
Review Comment:
```suggestion
def _map_cols_to_expression(self, fun: str, col: Union[Expression, str])
-> Sequence[Expression]:
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
