Merge branch 'apache:master' into windowFuncFix

Author: xin-aurora

python/docs/source/reference/pyspark.sql/functions.rst

@@ -132,6 +132,7 @@ Mathematical Functions
     radians
     rand
     randn
+    random
     rint
     round
     sec
@@ -164,6 +165,7 @@ String Functions
     char
     char_length
     character_length
+    chr
     collate
     collation
     concat_ws
@@ -192,6 +194,7 @@ String Functions
     overlay
     position
     printf
+    quote
     randstr
     regexp_count
     regexp_extract
@@ -631,6 +634,7 @@ Misc Functions
     try_reflect
     typeof
     user
+    uuid
     version
 
 

python/pyspark/ml/connect/functions.py

@@ -61,7 +61,7 @@ def _test() -> None:
     from pyspark.testing import should_test_connect
 
     if not should_test_connect:
-        print(f"Skipping pyspark.ml.connect.functions doctests", file=sys.stderr)
+        print("Skipping pyspark.ml.connect.functions doctests", file=sys.stderr)
         sys.exit(0)
 
     import doctest

python/pyspark/sql/connect/tvf.py

@@ -130,7 +130,7 @@ def _test() -> None:
     from pyspark.testing import should_test_connect
 
     if not should_test_connect:
-        print(f"Skipping pyspark.sql.connect.tvf doctests", file=sys.stderr)
+        print("Skipping pyspark.sql.connect.tvf doctests", file=sys.stderr)
         sys.exit(0)
 
     import doctest

python/pyspark/sql/tests/connect/test_df_debug.py

@@ -17,49 +17,45 @@
 
 import unittest
 
-from pyspark.sql.tests.connect.test_connect_basic import SparkConnectSQLTestCase
-from pyspark.testing.connectutils import should_test_connect
+from pyspark.testing.connectutils import ReusedConnectTestCase
 from pyspark.testing.utils import have_graphviz, graphviz_requirement_message
 
-if should_test_connect:
-    from pyspark.sql.connect.dataframe import DataFrame
 
-
-class SparkConnectDataFrameDebug(SparkConnectSQLTestCase):
+class SparkConnectDataFrameDebug(ReusedConnectTestCase):
     def test_df_debug_basics(self):
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         x = df.collect()  # noqa: F841
         ei = df.executionInfo
 
         root, graph = ei.metrics.extract_graph()
         self.assertIn(root, graph, "The root must be rooted in the graph")
 
     def test_df_quey_execution_empty_before_execution(self):
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         ei = df.executionInfo
         self.assertIsNone(ei, "The query execution must be None before the action is executed")
 
     def test_df_query_execution_with_writes(self):
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         df.write.save("/tmp/test_df_query_execution_with_writes", format="json", mode="overwrite")
         ei = df.executionInfo
         self.assertIsNotNone(
             ei, "The query execution must be None after the write action is executed"
         )
 
     def test_query_execution_text_format(self):
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         df.collect()
         self.assertIn("HashAggregate", df.executionInfo.metrics.toText())
 
         # Different execution mode.
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         df.toPandas()
         self.assertIn("HashAggregate", df.executionInfo.metrics.toText())
 
     @unittest.skipIf(not have_graphviz, graphviz_requirement_message)
     def test_df_query_execution_metrics_to_dot(self):
-        df: DataFrame = self.connect.range(100).repartition(10).groupBy("id").count()
+        df = self.spark.range(100).repartition(10).groupBy("id").count()
         x = df.collect()  # noqa: F841
         ei = df.executionInfo
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -409,7 +409,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
             throw QueryCompilationErrors.windowSpecificationNotDefinedError(windowName)
 
           case e: Expression if e.checkInputDataTypes().isFailure =>
-            TypeCoercionValidation.failOnTypeCheckResult(e, operator)
+            TypeCoercionValidation.failOnTypeCheckResult(e, Some(operator))
 
           case c: Cast if !c.resolved =>
             throw SparkException.internalError(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionValidation.scala

@@ -27,14 +27,14 @@ import org.apache.spark.sql.types.DataType
 object TypeCoercionValidation extends QueryErrorsBase {
   private val DATA_TYPE_MISMATCH_ERROR = TreeNodeTag[Unit]("dataTypeMismatchError")
 
-  def failOnTypeCheckResult(e: Expression, operator: LogicalPlan): Nothing = {
+  def failOnTypeCheckResult(e: Expression, operator: Option[LogicalPlan] = None): Nothing = {
     e.checkInputDataTypes() match {
       case checkRes: TypeCheckResult.DataTypeMismatch =>
         e.setTagValue(DATA_TYPE_MISMATCH_ERROR, ())
         e.dataTypeMismatch(e, checkRes)
       case TypeCheckResult.TypeCheckFailure(message) =>
         e.setTagValue(DATA_TYPE_MISMATCH_ERROR, ())
-        val extraHint = TypeCoercionValidation.getHintForExpressionCoercion(operator)
+        val extraHint = operator.map(getHintForExpressionCoercion(_)).getOrElse("")
         e.failAnalysis(
           errorClass = "DATATYPE_MISMATCH.TYPE_CHECK_FAILURE_WITH_HINT",
           messageParameters = Map("sqlExpr" -> toSQLExpr(e), "msg" -> message, "hint" -> extraHint)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala

@@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.analysis.{
   withPosition,
   FunctionResolution,
   GetViewColumnByNameAndOrdinal,
+  TypeCoercionValidation,
   UnresolvedAlias,
   UnresolvedAttribute,
   UnresolvedFunction,
@@ -984,6 +985,10 @@ class ExpressionResolver(
   }
 
   private def validateResolvedExpressionGenerically(resolvedExpression: Expression): Unit = {
+    if (resolvedExpression.checkInputDataTypes().isFailure) {
+      TypeCoercionValidation.failOnTypeCheckResult(resolvedExpression)
+    }
+
     if (!resolvedExpression.resolved) {
       throwSinglePassFailedToResolveExpression(resolvedExpression)
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeExtractors.scala

@@ -20,10 +20,11 @@ package org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.QueryContext
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode}
 import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTRACT_VALUE, TreePattern}
 import org.apache.spark.sql.catalyst.util.{quoteIdentifier, ArrayData, GenericArrayData, MapData, TypeUtils}
-import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase, QueryExecutionErrors}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
@@ -90,7 +91,7 @@ object ExtractValue {
   }
 }
 
-trait ExtractValue extends Expression {
+trait ExtractValue extends Expression with QueryErrorsBase {
   override def nullIntolerant: Boolean = true
   final override val nodePatterns: Seq[TreePattern] = Seq(EXTRACT_VALUE)
   val child: Expression
@@ -314,6 +315,30 @@ case class GetArrayItem(
     })
   }
 
+  override def checkInputDataTypes(): TypeCheckResult = {
+    (left.dataType, right.dataType) match {
+      case (_: ArrayType, e2) if !e2.isInstanceOf[IntegralType] =>
+        DataTypeMismatch(
+          errorSubClass = "UNEXPECTED_INPUT_TYPE",
+          messageParameters = Map(
+            "paramIndex" -> ordinalNumber(1),
+            "requiredType" -> toSQLType(IntegralType),
+            "inputSql" -> toSQLExpr(right),
+            "inputType" -> toSQLType(right.dataType))
+        )
+      case (e1, _) if !e1.isInstanceOf[ArrayType] =>
+        DataTypeMismatch(
+          errorSubClass = "UNEXPECTED_INPUT_TYPE",
+          messageParameters = Map(
+            "paramIndex" -> ordinalNumber(0),
+            "requiredType" -> toSQLType(TypeCollection(ArrayType)),
+            "inputSql" -> toSQLExpr(left),
+            "inputType" -> toSQLType(left.dataType))
+        )
+      case _ => TypeCheckResult.TypeCheckSuccess
+    }
+  }
+
   override protected def withNewChildrenInternal(
       newLeft: Expression, newRight: Expression): GetArrayItem =
     copy(child = newLeft, ordinal = newRight)