You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2025/04/24 11:11:30 UTC
(spark) branch master updated: [SPARK-51898][PYTHON][CONNECT][TESTS] Refactor SparkConnectErrorTests to support connect-only mode
This is an automated email from the ASF dual-hosted git repository.
ruifengz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8da4904cfdb7 [SPARK-51898][PYTHON][CONNECT][TESTS] Refactor SparkConnectErrorTests to support connect-only mode
8da4904cfdb7 is described below
commit 8da4904cfdb78ec74402b3e81ad2ebfc2faaff0e
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Thu Apr 24 19:11:11 2025 +0800
[SPARK-51898][PYTHON][CONNECT][TESTS] Refactor SparkConnectErrorTests to support connect-only mode
### What changes were proposed in this pull request?
`SparkConnectSQLTestCase`, as a child class of `ReusedMixedTestCase`, supports compare classic and connect directly. via `self.spark` vs `spark.connect`.
So it cannot be used in connect-only mode, moreover, `SparkConnectSQLTestCase` has some expensive setup which is not necessary.
### Why are the changes needed?
1, to improve test coverage;
2, to avoid unnecessary setups.
### Does this PR introduce _any_ user-facing change?
no, test-only
### How was this patch tested?
CI
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #50693 from zhengruifeng/fix_test_connect_error.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
.../sql/tests/connect/test_connect_error.py | 71 +++++++++++++---------
1 file changed, 41 insertions(+), 30 deletions(-)
diff --git a/python/pyspark/sql/tests/connect/test_connect_error.py b/python/pyspark/sql/tests/connect/test_connect_error.py
index 01047741f674..3f540f83dcb7 100644
--- a/python/pyspark/sql/tests/connect/test_connect_error.py
+++ b/python/pyspark/sql/tests/connect/test_connect_error.py
@@ -20,44 +20,42 @@ import unittest
from pyspark.errors import PySparkAttributeError
from pyspark.errors.exceptions.base import SessionNotSameException
from pyspark.sql.types import Row
-from pyspark.testing.connectutils import should_test_connect
+from pyspark.sql import functions as F
from pyspark.errors import PySparkTypeError
-from pyspark.sql.tests.connect.test_connect_basic import SparkConnectSQLTestCase
+from pyspark.testing.connectutils import ReusedConnectTestCase
-if should_test_connect:
- from pyspark.sql.connect.session import SparkSession as RemoteSparkSession
- from pyspark.sql.connect import functions as CF
- from pyspark.sql.connect.column import Column
- from pyspark.errors.exceptions.connect import AnalysisException
-
-class SparkConnectErrorTests(SparkConnectSQLTestCase):
+class SparkConnectErrorTests(ReusedConnectTestCase):
def test_recursion_handling_for_plan_logging(self):
"""SPARK-45852 - Test that we can handle recursion in plan logging."""
- cdf = self.connect.range(1)
+ cdf = self.spark.range(1)
for x in range(400):
- cdf = cdf.withColumn(f"col_{x}", CF.lit(x))
+ cdf = cdf.withColumn(f"col_{x}", F.lit(x))
# Calling schema will trigger logging the message that will in turn trigger the message
# conversion into protobuf that will then trigger the recursion error.
self.assertIsNotNone(cdf.schema)
- result = self.connect._client._proto_to_string(cdf._plan.to_proto(self.connect._client))
+ result = self.spark._client._proto_to_string(cdf._plan.to_proto(self.spark._client))
self.assertIn("recursion", result)
def test_error_handling(self):
+ from pyspark.errors.exceptions.connect import AnalysisException
+
# SPARK-41533 Proper error handling for Spark Connect
- df = self.connect.range(10).select("id2")
+ df = self.spark.range(10).select("id2")
with self.assertRaises(AnalysisException):
df.collect()
def test_invalid_column(self):
+ from pyspark.errors.exceptions.connect import AnalysisException
+
# SPARK-41812: fail df1.select(df2.col)
data1 = [Row(a=1, b=2, c=3)]
- cdf1 = self.connect.createDataFrame(data1)
+ cdf1 = self.spark.createDataFrame(data1)
data2 = [Row(a=2, b=0)]
- cdf2 = self.connect.createDataFrame(data2)
+ cdf2 = self.spark.createDataFrame(data2)
with self.assertRaises(AnalysisException):
cdf1.select(cdf2.a).schema
@@ -81,11 +79,13 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
cdf1.select(cdf2.a).schema
def test_invalid_star(self):
+ from pyspark.errors.exceptions.connect import AnalysisException
+
data1 = [Row(a=1, b=2, c=3)]
- cdf1 = self.connect.createDataFrame(data1)
+ cdf1 = self.spark.createDataFrame(data1)
data2 = [Row(a=2, b=0)]
- cdf2 = self.connect.createDataFrame(data2)
+ cdf2 = self.spark.createDataFrame(data2)
# Can find the target plan node, but fail to resolve with it
with self.assertRaisesRegex(
@@ -101,7 +101,7 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
"CANNOT_RESOLVE_DATAFRAME_COLUMN",
):
# column 'a has been replaced
- cdf3 = cdf1.withColumn("a", CF.lit(0))
+ cdf3 = cdf1.withColumn("a", F.lit(0))
cdf3.select(cdf1["*"]).schema
# Can not find the target plan node by plan id
@@ -119,15 +119,24 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
cdf1.join(cdf1).select(cdf1["*"]).schema
def test_deduplicate_within_watermark_in_batch(self):
- df = self.connect.read.table(self.tbl_name)
- with self.assertRaisesRegex(
- AnalysisException,
- "dropDuplicatesWithinWatermark is not supported with batch DataFrames/DataSets",
- ):
- df.dropDuplicatesWithinWatermark().toPandas()
+ from pyspark.errors.exceptions.connect import AnalysisException
+
+ table_name = "tmp_table_for_test_deduplicate_within_watermark_in_batch"
+ with self.table(table_name):
+ self.spark.createDataFrame(
+ [Row(key=i, value=str(i)) for i in range(100)]
+ ).write.saveAsTable(table_name)
+
+ with self.assertRaisesRegex(
+ AnalysisException,
+ "dropDuplicatesWithinWatermark is not supported with batch DataFrames/DataSets",
+ ):
+ self.spark.read.table(table_name).dropDuplicatesWithinWatermark().toPandas()
def test_different_spark_session_join_or_union(self):
- df = self.connect.range(10).limit(3)
+ from pyspark.sql.connect.session import SparkSession as RemoteSparkSession
+
+ df = self.spark.range(10).limit(3)
spark2 = RemoteSparkSession(connection="sc://localhost")
df2 = spark2.range(10).limit(3)
@@ -158,7 +167,7 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
def test_unsupported_functions(self):
# SPARK-41225: Disable unsupported functions.
- df = self.connect.read.table(self.tbl_name)
+ df = self.spark.range(10)
with self.assertRaises(NotImplementedError):
df.toJSON()
with self.assertRaises(NotImplementedError):
@@ -167,7 +176,7 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
def test_unsupported_jvm_attribute(self):
# Unsupported jvm attributes for Spark session.
unsupported_attrs = ["_jsc", "_jconf", "_jvm", "_jsparkSession"]
- spark_session = self.connect
+ spark_session = self.spark
for attr in unsupported_attrs:
with self.assertRaises(PySparkAttributeError) as pe:
getattr(spark_session, attr)
@@ -180,7 +189,7 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
# Unsupported jvm attributes for DataFrame.
unsupported_attrs = ["_jseq", "_jdf", "_jmap", "_jcols"]
- cdf = self.connect.range(10)
+ cdf = self.spark.range(10)
for attr in unsupported_attrs:
with self.assertRaises(PySparkAttributeError) as pe:
getattr(cdf, attr)
@@ -212,12 +221,14 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
)
def test_column_cannot_be_constructed_from_string(self):
+ from pyspark.sql.connect.column import Column
+
with self.assertRaises(TypeError):
Column("col")
def test_select_none(self):
with self.assertRaises(PySparkTypeError) as e1:
- self.connect.range(1).select(None)
+ self.spark.range(1).select(None)
self.check_error(
exception=e1.exception,
@@ -228,7 +239,7 @@ class SparkConnectErrorTests(SparkConnectSQLTestCase):
def test_ym_interval_in_collect(self):
# YearMonthIntervalType is not supported in python side arrow conversion
with self.assertRaises(PySparkTypeError):
- self.connect.sql("SELECT INTERVAL '10-8' YEAR TO MONTH AS interval").first()
+ self.spark.sql("SELECT INTERVAL '10-8' YEAR TO MONTH AS interval").first()
if __name__ == "__main__":
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org