From 00382581f9b679b33cd59422ac3060c36e0fea04 Mon Sep 17 00:00:00 2001 From: Max Gekk Date: Wed, 7 Dec 2022 08:34:58 +0300 Subject: [PATCH 1/2] Support parameterized SQL by PySpark sql --- python/pyspark/sql/session.py | 3 ++- python/pyspark/sql/sql_formatter.py | 8 ++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/python/pyspark/sql/session.py b/python/pyspark/sql/session.py index e96eef79b87ec..b80d2dbaaaf00 100644 --- a/python/pyspark/sql/session.py +++ b/python/pyspark/sql/session.py @@ -1376,7 +1376,8 @@ def sql(self, sqlQuery: str, **kwargs: Any) -> DataFrame: if len(kwargs) > 0: sqlQuery = formatter.format(sqlQuery, **kwargs) try: - return DataFrame(self._jsparkSession.sql(sqlQuery), self) + args = formatter.unused_args + return DataFrame(self._jsparkSession.sql(sqlQuery, args), self) finally: if len(kwargs) > 0: formatter.clear() diff --git a/python/pyspark/sql/sql_formatter.py b/python/pyspark/sql/sql_formatter.py index 5e79b9ff5ea98..c8a9c8e17437d 100644 --- a/python/pyspark/sql/sql_formatter.py +++ b/python/pyspark/sql/sql_formatter.py @@ -37,6 +37,7 @@ class SQLStringFormatter(string.Formatter): def __init__(self, session: "SparkSession") -> None: self._session: "SparkSession" = session self._temp_views: List[Tuple[DataFrame, str]] = [] + self.unused_args: Mapping[str, str] = {} def get_field(self, field_name: str, args: Sequence[Any], kwargs: Mapping[str, Any]) -> Any: obj, first = super(SQLStringFormatter, self).get_field(field_name, args, kwargs) @@ -78,7 +79,14 @@ def _convert_value(self, val: Any, field_name: str) -> Optional[str]: else: return val + def check_unused_args(self, used_args: Sequence[Any], args: Sequence[Any], kwargs: Mapping[str, Any]) -> None: + for k in kwargs: + if isinstance(kwargs[k], str) and not used_args.__contains__(k): + self.unused_args[k] = kwargs[k] + return super().check_unused_args(used_args, args, kwargs) + def clear(self) -> None: for _, n in self._temp_views: self._session.catalog.dropTempView(n) self._temp_views = [] + self.unused_args = [] From 368de7bc851b961c80022aae5875900310d5dd8a Mon Sep 17 00:00:00 2001 From: Max Gekk Date: Wed, 21 Dec 2022 21:36:05 +0300 Subject: [PATCH 2/2] Fix coding style of sql_formatter.py --- python/pyspark/sql/sql_formatter.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/python/pyspark/sql/sql_formatter.py b/python/pyspark/sql/sql_formatter.py index c8a9c8e17437d..9c669fa8091d8 100644 --- a/python/pyspark/sql/sql_formatter.py +++ b/python/pyspark/sql/sql_formatter.py @@ -79,7 +79,9 @@ def _convert_value(self, val: Any, field_name: str) -> Optional[str]: else: return val - def check_unused_args(self, used_args: Sequence[Any], args: Sequence[Any], kwargs: Mapping[str, Any]) -> None: + def check_unused_args( + self, used_args: Sequence[Any], args: Sequence[Any], kwargs: Mapping[str, Any] + ) -> None: for k in kwargs: if isinstance(kwargs[k], str) and not used_args.__contains__(k): self.unused_args[k] = kwargs[k]