[GitHub] [spark] HyukjinKwon commented on a change in pull request #35656: [SPARK-38107][SQL] Use error classes in the compilation errors of python/pandas UDFs

[GitBox <gi...@apache.org>]

HyukjinKwon commented on a change in pull request #35656:
URL: https://github.com/apache/spark/pull/35656#discussion_r814452730



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
##########
@@ -101,4 +102,91 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
     assert(e.message ===
       "The argument_index of string format cannot contain position 0$.")
   }
+
+  test("CANNOT_USE_MIXTURE") {
+    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_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_JOIN_CONDITION"))
+    assert(e.message ===
+      "Using PythonUDF in join condition of join type LeftOuter is not supported")
+  }
+
+  test("UNSUPPORTED_PANDAS_UDF_AGGREGATE_EXPRESSION") {
+    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_PANDAS_UDF_AGGREGATE_EXPRESSION"))
+    assert(e.message ===
+      "Pivot does not support Pandas UDF aggregate expressions.")
+  }
+
+   test("UNSUPPORTED_STREAMING_AGGREGATION") {

Review comment:
       Let's remove this and `groupAggPandasUDFUnsupportedByStreamingAggError` for now because the `AnalysisException` class is not user-facing. cc @MaxGekk @HeartSaVioR  to confirm




-- 
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