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/03/10 09:47:58 UTC
[GitHub] [spark] MaxGekk commented on a change in pull request #35656: [SPARK-38107][SQL] Use error classes in the compilation errors of python/pandas UDFs
MaxGekk commented on a change in pull request #35656:
URL: https://github.com/apache/spark/pull/35656#discussion_r823529710
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +101,69 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
assert(e.message ===
"The argument_index of string format cannot contain position 0$.")
}
+
+ test("CANNOT_USE_MIXTURE: Using aggregate function with grouped aggregate pandas UDF") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy("CustomerId")
+ .agg(pandasTestUDF(df("Quantity")), sum(df("Quantity"))).collect()
Review comment:
Please, fix indentation here. Should be 2 spaces.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +101,69 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
assert(e.message ===
"The argument_index of string format cannot contain position 0$.")
}
+
+ test("CANNOT_USE_MIXTURE: Using aggregate function with grouped aggregate pandas UDF") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy("CustomerId")
+ .agg(pandasTestUDF(df("Quantity")), sum(df("Quantity"))).collect()
+ }
+
+ assert(e.errorClass === Some("CANNOT_USE_MIXTURE"))
+ assert(e.message ===
+ "Cannot use a mixture of aggregate function and group aggregate pandas UDF")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using Python UDF with unsupported join condition") {
+ import IntegratedUDFTestUtils._
+
+ val df1 = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val df2 = Seq(
+ ("Bob", 17850),
+ ("Alice", 17850),
+ ("Tom", 17851)
+ ).toDF("CustomerName", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pythonTestUDF = TestPythonUDF(name = "python_udf")
+ df1.join(
+ df2, pythonTestUDF(df1("CustomerID") === df2("CustomerID")), "leftouter").collect()
Review comment:
Please, fix indentation.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +101,69 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
assert(e.message ===
"The argument_index of string format cannot contain position 0$.")
}
+
+ test("CANNOT_USE_MIXTURE: Using aggregate function with grouped aggregate pandas UDF") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy("CustomerId")
+ .agg(pandasTestUDF(df("Quantity")), sum(df("Quantity"))).collect()
+ }
+
+ assert(e.errorClass === Some("CANNOT_USE_MIXTURE"))
+ assert(e.message ===
+ "Cannot use a mixture of aggregate function and group aggregate pandas UDF")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using Python UDF with unsupported join condition") {
+ import IntegratedUDFTestUtils._
+
+ val df1 = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val df2 = Seq(
+ ("Bob", 17850),
+ ("Alice", 17850),
+ ("Tom", 17851)
+ ).toDF("CustomerName", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pythonTestUDF = TestPythonUDF(name = "python_udf")
+ df1.join(
+ df2, pythonTestUDF(df1("CustomerID") === df2("CustomerID")), "leftouter").collect()
+ }
+
+ assert(e.errorClass === Some("UNSUPPORTED_FEATURE"))
+ assert(e.message ===
+ "The feature is not supported: " +
+ "Using PythonUDF in join condition of join type LeftOuter is not supported")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using pandas UDF aggregate expression with pivot") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy(df("CustomerID")).pivot(df("CustomerID")).agg(pandasTestUDF(df("Quantity")))
Review comment:
Change indentation
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +101,69 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
assert(e.message ===
"The argument_index of string format cannot contain position 0$.")
}
+
+ test("CANNOT_USE_MIXTURE: Using aggregate function with grouped aggregate pandas UDF") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy("CustomerId")
+ .agg(pandasTestUDF(df("Quantity")), sum(df("Quantity"))).collect()
+ }
+
+ assert(e.errorClass === Some("CANNOT_USE_MIXTURE"))
+ assert(e.message ===
+ "Cannot use a mixture of aggregate function and group aggregate pandas UDF")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using Python UDF with unsupported join condition") {
+ import IntegratedUDFTestUtils._
+
+ val df1 = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val df2 = Seq(
+ ("Bob", 17850),
+ ("Alice", 17850),
+ ("Tom", 17851)
+ ).toDF("CustomerName", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pythonTestUDF = TestPythonUDF(name = "python_udf")
+ df1.join(
+ df2, pythonTestUDF(df1("CustomerID") === df2("CustomerID")), "leftouter").collect()
+ }
+
+ assert(e.errorClass === Some("UNSUPPORTED_FEATURE"))
Review comment:
Check sqlState too
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +101,69 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
assert(e.message ===
"The argument_index of string format cannot contain position 0$.")
}
+
+ test("CANNOT_USE_MIXTURE: Using aggregate function with grouped aggregate pandas UDF") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy("CustomerId")
+ .agg(pandasTestUDF(df("Quantity")), sum(df("Quantity"))).collect()
+ }
+
+ assert(e.errorClass === Some("CANNOT_USE_MIXTURE"))
+ assert(e.message ===
+ "Cannot use a mixture of aggregate function and group aggregate pandas UDF")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using Python UDF with unsupported join condition") {
+ import IntegratedUDFTestUtils._
+
+ val df1 = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+ val df2 = Seq(
+ ("Bob", 17850),
+ ("Alice", 17850),
+ ("Tom", 17851)
+ ).toDF("CustomerName", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pythonTestUDF = TestPythonUDF(name = "python_udf")
+ df1.join(
+ df2, pythonTestUDF(df1("CustomerID") === df2("CustomerID")), "leftouter").collect()
+ }
+
+ assert(e.errorClass === Some("UNSUPPORTED_FEATURE"))
+ assert(e.message ===
+ "The feature is not supported: " +
+ "Using PythonUDF in join condition of join type LeftOuter is not supported")
+ }
+
+ test("UNSUPPORTED_FEATURE: Using pandas UDF aggregate expression with pivot") {
+ import IntegratedUDFTestUtils._
+
+ val df = Seq(
+ (536361, "85123A", 2, 17850),
+ (536362, "85123B", 4, 17850),
+ (536363, "86123A", 6, 17851)
+ ).toDF("InvoiceNo", "StockCode", "Quantity", "CustomerID")
+
+ val e = intercept[AnalysisException] {
+ val pandasTestUDF = TestGroupedAggPandasUDF(name = "pandas_udf")
+ df.groupBy(df("CustomerID")).pivot(df("CustomerID")).agg(pandasTestUDF(df("Quantity")))
+ }
+
+ assert(e.errorClass === Some("UNSUPPORTED_FEATURE"))
Review comment:
check sqlState
--
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