Skip to content

[SPARK-58488][PYTHON] Fix type annotations for DataFrameWriter and DataStreamWriter partitionBy/clusterBy/bucketBy/sortBy #57693

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Spenserrrr wants to merge 6 commits into
base: master
Choose a base branch
from
Open

[SPARK-58488][PYTHON] Fix type annotations for DataFrameWriter and DataStreamWriter partitionBy/clusterBy/bucketBy/sortBy #57693

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
26f45bc
[SPARK-58488][PYTHON] Fix type annotations for DataFrameWriter and Da…
Spenserrrr Jul 31, 2026
627984b
[SPARK-58488][PYTHON] Fix type annotation for DataFrameNaFunctions.re…
Spenserrrr Jul 31, 2026
00d2069
[SPARK-58488][PYTHON] Remove unnecessary type: ignore in DataFrameWri…
Spenserrrr Jul 31, 2026
5161e6d
[SPARK-58488][PYTHON] Use Sequence[str] for DataFrameWriter bucketBy/…
Spenserrrr Jul 31, 2026
42e8bb3
[SPARK-58488][PYTHON] Apply ruff format to writer varargs changes
Spenserrrr Aug 1, 2026
b2b8a14
[SPARK-58488][PYTHON] Use Sequence[str] for save/saveAsTable partitio…
Spenserrrr Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion python/pyspark/sql/classic/dataframe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2053,6 +2053,7 @@ def replace(
def replace(
self,
to_replace: Dict["LiteralType", "OptionalPrimitiveType"],
*,
subset: Optional[List[str]] = ...,
) -> ParentDataFrame: ...

Expand All @@ -2064,7 +2065,7 @@ def replace(
subset: Optional[List[str]] = ...,
) -> ParentDataFrame: ...

def replace( # type: ignore[misc]
def replace(
self,
to_replace: Union[List["LiteralType"], Dict["LiteralType", "OptionalPrimitiveType"]],
value: Optional[
Expand Down
45 changes: 21 additions & 24 deletions python/pyspark/sql/connect/readwriter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
# limitations under the License.
#
from typing import Dict
from typing import Optional, Union, List, overload, Tuple, cast, Callable
from typing import Optional, Sequence, Union, List, overload, cast, Callable
from typing import TYPE_CHECKING

from pyspark.sql.connect.plan import (
Expand Down Expand Up @@ -52,7 +52,6 @@
__all__ = ["DataFrameReader", "DataFrameWriter"]

PathOrPaths = Union[str, List[str]]
TupleOrListOfString = Union[List[str], Tuple[str, ...]]


class OptionUtils:
Expand Down Expand Up @@ -620,11 +619,11 @@ def options(self, **options: "OptionalPrimitiveType") -> "DataFrameWriter":
def partitionBy(self, *cols: str) -> "DataFrameWriter": ...

@overload
def partitionBy(self, *cols: List[str]) -> "DataFrameWriter": ...
def partitionBy(self, __cols: Sequence[str]) -> "DataFrameWriter": ...

def partitionBy(self, *cols: Union[str, List[str]]) -> "DataFrameWriter":
if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
cols = cols[0] # type: ignore[assignment]
def partitionBy(self, *cols: Union[str, Sequence[str]]) -> "DataFrameWriter":
if len(cols) == 1 and not isinstance(cols[0], str) and isinstance(cols[0], Sequence):
cols = tuple(cols[0])

self._write.partitioning_cols = cast(List[str], cols)
return self
Expand All @@ -635,10 +634,10 @@ def partitionBy(self, *cols: Union[str, List[str]]) -> "DataFrameWriter":
def bucketBy(self, numBuckets: int, col: str, *cols: str) -> "DataFrameWriter": ...

@overload
def bucketBy(self, numBuckets: int, col: TupleOrListOfString) -> "DataFrameWriter": ...
def bucketBy(self, numBuckets: int, col: Sequence[str]) -> "DataFrameWriter": ...

def bucketBy(
self, numBuckets: int, col: Union[str, TupleOrListOfString], *cols: Optional[str]
self, numBuckets: int, col: Union[str, Sequence[str]], *cols: Optional[str]
) -> "DataFrameWriter":
if not isinstance(numBuckets, int):
raise PySparkTypeError(
Expand All @@ -650,7 +649,7 @@ def bucketBy(
},
)

if isinstance(col, (list, tuple)):
if not isinstance(col, str) and isinstance(col, Sequence):
if cols:
raise PySparkValueError(
errorClass="CANNOT_SET_TOGETHER",
Expand All @@ -659,7 +658,7 @@ def bucketBy(
},
)

col, cols = col[0], col[1:] # type: ignore[assignment]
col, cols = col[0], tuple(col[1:])

for c in cols:
if not isinstance(c, str):
Expand Down Expand Up @@ -691,12 +690,10 @@ def bucketBy(
def sortBy(self, col: str, *cols: str) -> "DataFrameWriter": ...

@overload
def sortBy(self, col: TupleOrListOfString) -> "DataFrameWriter": ...
def sortBy(self, col: Sequence[str]) -> "DataFrameWriter": ...

def sortBy(
self, col: Union[str, TupleOrListOfString], *cols: Optional[str]
) -> "DataFrameWriter":
if isinstance(col, (list, tuple)):
def sortBy(self, col: Union[str, Sequence[str]], *cols: Optional[str]) -> "DataFrameWriter":
if not isinstance(col, str) and isinstance(col, Sequence):
if cols:
raise PySparkValueError(
errorClass="CANNOT_SET_TOGETHER",
Expand All @@ -705,7 +702,7 @@ def sortBy(
},
)

col, cols = col[0], col[1:] # type: ignore[assignment]
col, cols = col[0], tuple(col[1:])

for c in cols:
if not isinstance(c, str):
Expand Down Expand Up @@ -736,11 +733,11 @@ def sortBy(
def clusterBy(self, *cols: str) -> "DataFrameWriter": ...

@overload
def clusterBy(self, *cols: List[str]) -> "DataFrameWriter": ...
def clusterBy(self, __cols: Sequence[str]) -> "DataFrameWriter": ...

def clusterBy(self, *cols: Union[str, List[str]]) -> "DataFrameWriter":
if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
cols = cols[0] # type: ignore[assignment]
def clusterBy(self, *cols: Union[str, Sequence[str]]) -> "DataFrameWriter":
if len(cols) == 1 and not isinstance(cols[0], str) and isinstance(cols[0], Sequence):
cols = tuple(cols[0])
assert len(cols) > 0, "clusterBy needs one or more clustering columns."
self._write.clustering_cols = cast(List[str], cols)
return self
Expand All @@ -752,7 +749,7 @@ def save(
path: Optional[str] = None,
format: Optional[str] = None,
mode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
**options: "OptionalPrimitiveType",
) -> None:
self.mode(mode).options(**options)
Expand Down Expand Up @@ -785,7 +782,7 @@ def saveAsTable(
name: str,
format: Optional[str] = None,
mode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
**options: "OptionalPrimitiveType",
) -> None:
self.mode(mode).options(**options)
Expand Down Expand Up @@ -830,7 +827,7 @@ def parquet(
self,
path: str,
mode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
compression: Optional[str] = None,
) -> None:
self.mode(mode)
Expand Down Expand Up @@ -933,7 +930,7 @@ def orc(
self,
path: str,
mode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
compression: Optional[str] = None,
) -> None:
self.mode(mode)
Expand Down
24 changes: 12 additions & 12 deletions python/pyspark/sql/connect/streaming/readwriter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import re
import sys
import pickle
from typing import cast, overload, Callable, Dict, List, Optional, TYPE_CHECKING, Union
from typing import cast, overload, Callable, Dict, List, Optional, Sequence, TYPE_CHECKING, Union

from pyspark.serializers import CloudPickleSerializer
from pyspark.sql.connect.plan import (
Expand Down Expand Up @@ -508,11 +508,11 @@ def options(self, **options: "OptionalPrimitiveType") -> "DataStreamWriter":
def partitionBy(self, *cols: str) -> "DataStreamWriter": ...

@overload
def partitionBy(self, __cols: List[str]) -> "DataStreamWriter": ...
def partitionBy(self, __cols: Sequence[str]) -> "DataStreamWriter": ...

def partitionBy(self, *cols: str) -> "DataStreamWriter": # type: ignore[misc]
if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
cols = cols[0]
def partitionBy(self, *cols: Union[str, Sequence[str]]) -> "DataStreamWriter":
if len(cols) == 1 and not isinstance(cols[0], str) and isinstance(cols[0], Sequence):
cols = tuple(cols[0])
# Clear any existing columns (if any).
while len(self._write_proto.partitioning_column_names) > 0:
self._write_proto.partitioning_column_names.pop()
Expand All @@ -525,11 +525,11 @@ def partitionBy(self, *cols: str) -> "DataStreamWriter": # type: ignore[misc]
def clusterBy(self, *cols: str) -> "DataStreamWriter": ...

@overload
def clusterBy(self, __cols: List[str]) -> "DataStreamWriter": ...
def clusterBy(self, __cols: Sequence[str]) -> "DataStreamWriter": ...

def clusterBy(self, *cols: str) -> "DataStreamWriter": # type: ignore[misc]
if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
cols = cols[0]
def clusterBy(self, *cols: Union[str, Sequence[str]]) -> "DataStreamWriter":
if len(cols) == 1 and not isinstance(cols[0], str) and isinstance(cols[0], Sequence):
cols = tuple(cols[0])
# Clear any existing columns (if any).
while len(self._write_proto.clustering_column_names) > 0:
self._write_proto.clustering_column_names.pop()
Expand Down Expand Up @@ -681,7 +681,7 @@ def _start_internal(
tableName: Optional[str] = None,
format: Optional[str] = None,
outputMode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
queryName: Optional[str] = None,
**options: "OptionalPrimitiveType",
) -> StreamingQuery:
Expand Down Expand Up @@ -727,7 +727,7 @@ def start(
path: Optional[str] = None,
format: Optional[str] = None,
outputMode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
queryName: Optional[str] = None,
**options: "OptionalPrimitiveType",
) -> "StreamingQuery":
Expand All @@ -748,7 +748,7 @@ def toTable(
tableName: str,
format: Optional[str] = None,
outputMode: Optional[str] = None,
partitionBy: Optional[Union[str, List[str]]] = None,
partitionBy: Optional[Union[str, Sequence[str]]] = None,
queryName: Optional[str] = None,
**options: "OptionalPrimitiveType",
) -> "StreamingQuery":
Expand Down
3 changes: 2 additions & 1 deletion python/pyspark/sql/dataframe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7173,6 +7173,7 @@ def replace(
def replace(
self,
to_replace: Dict["LiteralType", "OptionalPrimitiveType"],
*,
subset: Optional[List[str]] = ...,
) -> DataFrame: ...

Expand All @@ -7184,7 +7185,7 @@ def replace(
subset: Optional[List[str]] = ...,
) -> DataFrame: ...

@dispatch_df_method # type: ignore[misc]
@dispatch_df_method
def replace(
self,
to_replace: Union[List["LiteralType"], Dict["LiteralType", "OptionalPrimitiveType"]],
Expand Down
Loading