You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/10/22 11:34:09 UTC
(spark) branch branch-4.3 updated: [SPARK-59909][4.3][PYTHON] Fix PySpark errors raised with undefined error conditions or mismatched message parameters
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 18416c7b42d8 [SPARK-59909][4.3][PYTHON] Fix PySpark errors raised with undefined error conditions or mismatched message parameters
18416c7b42d8 is described below
commit 18416c7b42d8b9a1e4647f688aa1a4519f1e1ad7
Author: Jubin Soni <ju...@gmail.com>
AuthorDate: Thu Oct 22 11:33:54 2026 +0000
[SPARK-59909][4.3][PYTHON] Fix PySpark errors raised with undefined error conditions or mismatched message parameters
### What changes were proposed in this pull request?
Backport of #59170 to `branch-4.3`.
Several PySpark call sites raise a `PySparkException` whose error condition is not defined in `error-conditions.json`, or whose `messageParameters` do not match the template. Building the exception fails, so users get an internal error (`ValueError: Cannot find main error class ...` or `AssertionError: Undefined error message parameter ...`) instead of the intended message. This PR fixes those call sites and adds a test that validates every literal error class raised in PySpark sources.
The change is the same as the merged commit, except for two test-only adjustments needed on this branch:
- `PythonDataSourceWorkerUtilsTests` does not exist on `branch-4.3`, so the new `test_deserialize_unsupported_filter` is added in a minimal class of that name, without the unrelated profiler test from master.
- `DataTypeTests` does not mix in `PySparkErrorTestUtils` on this branch, so `test_variant_builder_size_limit` asserts `getCondition()` and `getMessageParameters()` directly instead of using `check_error`.
### Why are the changes needed?
All eight affected call sites exist on `branch-4.3`, so users of 4.3.x get internal errors instead of the intended messages. For example:
```python
>>> spark.readStream.changes(123)
ValueError: Cannot find main error class 'NOT_STR'
```
### Does this PR introduce _any_ user-facing change?
Yes, the same as #59170: the affected APIs raise their intended error messages instead of internal errors. The variant size error is defined as `VARIANT_SIZE_LIMIT`, matching the JVM condition name.
### How was this patch tested?
- The new `ErrorsTest.test_error_classes_used_in_source_are_valid` passes against the `branch-4.3` sources, so no other broken raise sites exist on this branch.
- `test_errors.py`, `DataTypeTests` and `PythonDataSourceWorkerUtilsTests` pass locally (22 tests).
- The JVM-backed tests (`test_readwriter.py`, `test_python_datasource.py`) run in CI.
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes - Claude Code was used as a pair-programming assistant. All changes were reviewed and verified by the author. Generated-by: Claude Opus 5.5
Closes #59458 from jubins/j-spark-59909-backport-4.3.
Authored-by: Jubin Soni <ju...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
python/pyspark/errors/error-conditions.json | 19 ++++++-
python/pyspark/errors/tests/test_errors.py | 65 ++++++++++++++++++++++
python/pyspark/sql/connect/dataframe.py | 2 +-
python/pyspark/sql/streaming/readwriter.py | 8 ++-
python/pyspark/sql/tests/test_python_datasource.py | 21 ++++++-
python/pyspark/sql/tests/test_readwriter.py | 11 +++-
python/pyspark/sql/tests/test_types.py | 10 ++++
python/pyspark/sql/variant_utils.py | 15 ++++-
8 files changed, 141 insertions(+), 10 deletions(-)
diff --git a/python/pyspark/errors/error-conditions.json b/python/pyspark/errors/error-conditions.json
index 07ce4314763a..756d55f1ae22 100644
--- a/python/pyspark/errors/error-conditions.json
+++ b/python/pyspark/errors/error-conditions.json
@@ -194,7 +194,7 @@
},
"DATA_SOURCE_EXTRANEOUS_FILTERS": {
"message": [
- "<type>.pushFilters() returned filters that are not part of the input. Make sure that each returned filter is one of the input filters by reference."
+ "<type>.pushFilters() returned filters that are not part of the input. Make sure that each returned filter is one of the input filters by reference. Input filters: <input>. Extraneous filters: <extraneous>."
]
},
"DATA_SOURCE_INVALID_RETURN_TYPE": {
@@ -430,6 +430,11 @@
"Invalid number of dataframes in group <dataframes_in_group>."
]
},
+ "INVALID_OPERATION_UUID_ID": {
+ "message": [
+ "Parameter value <arg_name> must be a valid UUID format: <origin>"
+ ]
+ },
"INVALID_PANDAS_UDF": {
"message": [
"Invalid function: <detail>"
@@ -1119,6 +1124,12 @@
"<data_type> is not supported in conversion to Arrow."
]
},
+ "UNSUPPORTED_FILTER": {
+ "message": [
+ "Unsupported filter <name>."
+ ],
+ "sqlState": "0A000"
+ },
"UNSUPPORTED_JOIN_TYPE": {
"message": [
"Unsupported join type: '<typ>'. Supported join types include: <supported>."
@@ -1224,6 +1235,12 @@
"Value for `<arg_name>` must be between <lower_bound> and <upper_bound> (inclusive), got <actual>"
]
},
+ "VARIANT_SIZE_LIMIT": {
+ "message": [
+ "Cannot build variant bigger than <sizeLimit> bytes."
+ ],
+ "sqlState": "22023"
+ },
"WKB_PARSE_ERROR" : {
"message" : [
"Error parsing WKB: <parseError> at position <pos>"
diff --git a/python/pyspark/errors/tests/test_errors.py b/python/pyspark/errors/tests/test_errors.py
index e477f4a99b0e..3e8b17577806 100644
--- a/python/pyspark/errors/tests/test_errors.py
+++ b/python/pyspark/errors/tests/test_errors.py
@@ -15,9 +15,12 @@
# limitations under the License.
#
+import ast
import json
+import os
import unittest
+import pyspark
from pyspark.errors import PySparkRuntimeError, PySparkValueError
from pyspark.errors.error_classes import ERROR_CLASSES_JSON
from pyspark.errors.utils import ErrorClassesReader
@@ -49,6 +52,68 @@ class ErrorsTest(unittest.TestCase):
json.loads(ERROR_CLASSES_JSON, object_pairs_hook=detect_duplication)
+ def test_error_classes_used_in_source_are_valid(self):
+ # PySparkException builds its message from error-conditions.json whenever no message is
+ # passed. For every such call with a literal errorClass, whatever the exception class is
+ # named, the condition must be defined and the literal messageParameters keys must match
+ # the placeholders of the message template. Otherwise building the exception fails and
+ # the intended error message is lost.
+
+ # Known issues tracked separately. Remove the entry when the linked issue is fixed.
+ known_issues = {
+ # SPARK-57965
+ (
+ os.path.join("pipelines", "spark_connect_graph_element_registry.py"),
+ "UNSUPPORTED_PIPELINES_DATASET_TYPE",
+ ),
+ }
+
+ error_reader = ErrorClassesReader()
+ root = os.path.dirname(pyspark.__file__)
+ failures = []
+ for dirpath, _, filenames in os.walk(root):
+ if "tests" in os.path.relpath(dirpath, root).split(os.sep):
+ continue
+ for filename in filenames:
+ if not filename.endswith(".py"):
+ continue
+ path = os.path.join(dirpath, filename)
+ relpath = os.path.relpath(path, root)
+ with open(path, encoding="utf-8") as f:
+ tree = ast.parse(f.read(), filename=path)
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call):
+ continue
+ kwargs = {kw.arg: kw.value for kw in node.keywords}
+ # A message passed positionally or as message= is used as is, and the
+ # condition may then be a JVM one that is not in error-conditions.json.
+ message = kwargs.get("message", ast.Constant(value=None))
+ if node.args or not (
+ isinstance(message, ast.Constant) and message.value is None
+ ):
+ continue
+ error_class = kwargs.get("errorClass")
+ if not (
+ isinstance(error_class, ast.Constant) and isinstance(error_class.value, str)
+ ):
+ continue
+ if (relpath, error_class.value) in known_issues:
+ continue
+ params = kwargs.get("messageParameters", ast.Dict(keys=[], values=[]))
+ try:
+ if isinstance(params, ast.Dict) and all(
+ isinstance(k, ast.Constant) for k in params.keys
+ ):
+ keys = [k.value for k in params.keys]
+ error_reader.get_error_message(error_class.value, {k: "" for k in keys})
+ else:
+ # Parameters are not a literal dict, so only check the condition.
+ error_reader.get_message_template(error_class.value)
+ except (AssertionError, ValueError) as e:
+ failures.append(f"{relpath}:{node.lineno}: {e}")
+
+ self.assertEqual(failures, [], "\n".join(failures))
+
def test_invalid_error_class(self):
with self.assertRaisesRegex(ValueError, "Cannot find main error class"):
PySparkValueError(errorClass="invalid", messageParameters={})
diff --git a/python/pyspark/sql/connect/dataframe.py b/python/pyspark/sql/connect/dataframe.py
index b6362b607b9b..85850192c03f 100644
--- a/python/pyspark/sql/connect/dataframe.py
+++ b/python/pyspark/sql/connect/dataframe.py
@@ -137,7 +137,7 @@ class DataFrame(ParentDataFrame):
if self._session is None:
raise PySparkRuntimeError(
errorClass="NO_ACTIVE_SESSION",
- messageParameters={"operator": "__init__"},
+ messageParameters={},
)
# Check whether _repr_html is supported or not, we use it to avoid calling RPC twice
diff --git a/python/pyspark/sql/streaming/readwriter.py b/python/pyspark/sql/streaming/readwriter.py
index 6b7faa622207..28ae1a9d22d0 100644
--- a/python/pyspark/sql/streaming/readwriter.py
+++ b/python/pyspark/sql/streaming/readwriter.py
@@ -987,8 +987,12 @@ class DataStreamReader(OptionUtils):
return self._df(self._jreader.changes(tableName))
else:
raise PySparkTypeError(
- errorClass="NOT_STR",
- messageParameters={"arg_name": "tableName", "arg_type": type(tableName).__name__},
+ errorClass="NOT_EXPECTED_TYPE",
+ messageParameters={
+ "arg_name": "tableName",
+ "expected_type": "str",
+ "arg_type": type(tableName).__name__,
+ },
)
diff --git a/python/pyspark/sql/tests/test_python_datasource.py b/python/pyspark/sql/tests/test_python_datasource.py
index 675c4b19fad9..249b76c9602f 100644
--- a/python/pyspark/sql/tests/test_python_datasource.py
+++ b/python/pyspark/sql/tests/test_python_datasource.py
@@ -25,8 +25,9 @@ from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import Callable, Iterable, List, Union, Iterator, Tuple
+from unittest import mock
-from pyspark.errors import AnalysisException, PythonException
+from pyspark.errors import AnalysisException, PySparkNotImplementedError, PythonException
from pyspark.memory_profiler_ext import has_memory_profiler
from pyspark.sql.datasource import (
CaseInsensitiveDict,
@@ -66,6 +67,19 @@ from pyspark.testing.utils import (
from pyspark.util import is_remote_only
+class PythonDataSourceWorkerUtilsTests(unittest.TestCase):
+ @unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
+ def test_deserialize_unsupported_filter(self):
+ with mock.patch.dict(os.environ, {"SPARK_PYTHON_RUNTIME": "PYTHON_WORKER"}):
+ from pyspark.sql.worker.data_source_pushdown_filters import deserializeFilter
+
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ deserializeFilter({"name": "UnknownFilter", "columnPath": ["x"], "isNegated": False})
+
+ self.assertEqual(pe.exception.getCondition(), "UNSUPPORTED_FILTER")
+ self.assertEqual(pe.exception.getMessageParameters(), {"name": "UnknownFilter"})
+
+
@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
class BasePythonDataSourceTestsMixin:
spark: SparkSession
@@ -364,7 +378,10 @@ class BasePythonDataSourceTestsMixin:
with self.sql_conf({"spark.sql.python.filterPushdown.enabled": True}):
self.spark.dataSource.register(TestDataSource)
- with self.assertRaisesRegex(Exception, "DATA_SOURCE_EXTRANEOUS_FILTERS"):
+ with self.assertRaisesRegex(
+ Exception,
+ "DATA_SOURCE_EXTRANEOUS_FILTERS.*returned filters that are not part of the input",
+ ):
self.spark.read.format("test").load().filter("x = 1").show()
def test_filter_pushdown_error(self):
diff --git a/python/pyspark/sql/tests/test_readwriter.py b/python/pyspark/sql/tests/test_readwriter.py
index 5dd5b1ebff63..412a364c928f 100644
--- a/python/pyspark/sql/tests/test_readwriter.py
+++ b/python/pyspark/sql/tests/test_readwriter.py
@@ -447,7 +447,16 @@ class ReadwriterV2TestsMixin:
class ReadwriterTests(ReadwriterTestsMixin, ReusedSQLTestCase):
- pass
+ # Classic only: Spark Connect's DataStreamReader.changes does not validate the argument type.
+ def test_streaming_changes_rejects_non_str_table_name(self):
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.readStream.changes(123)
+
+ self.check_error(
+ exception=pe.exception,
+ errorClass="NOT_EXPECTED_TYPE",
+ messageParameters={"arg_name": "tableName", "expected_type": "str", "arg_type": "int"},
+ )
class ReadwriterV2Tests(ReadwriterV2TestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/test_types.py b/python/pyspark/sql/tests/test_types.py
index c1e0ee6f096a..85c6f35b54be 100644
--- a/python/pyspark/sql/tests/test_types.py
+++ b/python/pyspark/sql/tests/test_types.py
@@ -3046,6 +3046,16 @@ class DataTypeTests(unittest.TestCase):
self.assertEqual(list(r), [1, 2, 3])
self.assertEqual([v for v in r], [1, 2, 3])
+ def test_variant_builder_size_limit(self):
+ from pyspark.sql.variant_utils import VariantBuilder
+
+ for json_str in ['"%s"' % ("a" * 100), '{"%s": 1}' % ("k" * 100)]:
+ with self.assertRaises(PySparkValueError) as pe:
+ VariantBuilder(size_limit=64).build(json_str)
+
+ self.assertEqual(pe.exception.getCondition(), "VARIANT_SIZE_LIMIT")
+ self.assertEqual(pe.exception.getMessageParameters(), {"sizeLimit": "64"})
+
class DataTypeVerificationTests(unittest.TestCase, PySparkErrorTestUtils):
def test_verify_type_exception_msg(self):
diff --git a/python/pyspark/sql/variant_utils.py b/python/pyspark/sql/variant_utils.py
index 80b7efeaca10..0b044e630b84 100644
--- a/python/pyspark/sql/variant_utils.py
+++ b/python/pyspark/sql/variant_utils.py
@@ -563,14 +563,20 @@ class VariantBuilder:
# calculation in case of pathological data.
max_size = max(dictionary_string_size, num_keys)
if max_size > self.size_limit:
- raise PySparkValueError(errorClass="VARIANT_SIZE_LIMIT_EXCEEDED", messageParameters={})
+ raise PySparkValueError(
+ errorClass="VARIANT_SIZE_LIMIT",
+ messageParameters={"sizeLimit": str(self.size_limit)},
+ )
offset_size = self._get_integer_size(max_size)
offset_start = 1 + offset_size
string_start = offset_start + (num_keys + 1) * offset_size
metadata_size = string_start + dictionary_string_size
if metadata_size > self.size_limit:
- raise PySparkValueError(errorClass="VARIANT_SIZE_LIMIT_EXCEEDED", messageParameters={})
+ raise PySparkValueError(
+ errorClass="VARIANT_SIZE_LIMIT",
+ messageParameters={"sizeLimit": str(self.size_limit)},
+ )
metadata = bytearray()
header_byte = VariantUtils.VERSION | ((offset_size - 1) << 6)
@@ -631,7 +637,10 @@ class VariantBuilder:
def _check_capacity(self, additional: int) -> None:
required = len(self.value) + additional
if required > self.size_limit:
- raise PySparkValueError(errorClass="VARIANT_SIZE_LIMIT_EXCEEDED", messageParameters={})
+ raise PySparkValueError(
+ errorClass="VARIANT_SIZE_LIMIT",
+ messageParameters={"sizeLimit": str(self.size_limit)},
+ )
def _primitive_header(self, type: int) -> bytes:
return bytes([(type << 2) | VariantUtils.PRIMITIVE])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org