You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/11/23 10:13:15 UTC
[GitHub] [spark] grundprinzip commented on a diff in pull request #38768: [SPARK-41230][CONNECT][PYTHON] Remove `str` from Aggregate expression type
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org