refine code to provide better atomicity

Author: Eric5553

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

@@ -263,18 +263,22 @@ abstract class Expression extends TreeNode[Expression] {
 
   protected def flatArguments: Iterator[Any] = stringArgs.flatMap {
     case t: Iterable[_] => t
-    case e: Expression => e.argumentString :: Nil
     case single => single :: Nil
   }
 
+  protected def flatArgumentStrings: Iterator[String] = flatArguments.map {
+    case e: Expression => e.argumentString
+    case arg: Any => arg.toString
+  }
+
   // Marks this as final, Expression.verboseString should never be called, and thus shouldn't be
   // overridden by concrete classes.
   final override def verboseString(maxFields: Int): String = simpleString(maxFields)
 
   override def simpleString(maxFields: Int): String = toString
 
   override def toString: String = prettyName + truncatedString(
-    flatArguments.toSeq, "(", ", ", ")", SQLConf.get.maxToStringFields)
+    flatArgumentStrings.toSeq, "(", ", ", ")", SQLConf.get.maxToStringFields)
 
   def argumentString: String = toString
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala

@@ -235,7 +235,7 @@ abstract class AggregateFunction extends Expression {
   /** String representation used in explain plans. */
   def toAggString(isDistinct: Boolean): String = {
     val start = if (isDistinct) "(distinct " else "("
-    prettyName + flatArguments.mkString(start, ", ", ")")
+    prettyName + flatArgumentStrings.mkString(start, ", ", ")")
   }
 }
 

sql/core/src/test/resources/sql-tests/results/explain.sql.out

@@ -91,7 +91,7 @@ Results [2]: [key#x, max#x]
 
 (6) Exchange 
 Input [2]: [key#x, max#x]
-Arguments: hashpartitioning(key#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, 4), true, [id=#x]
 
 (7) HashAggregate [codegen id : 2]
 Input [2]: [key#x, max#x]
@@ -158,7 +158,7 @@ Results [2]: [key#x, max#x]
 
 (6) Exchange 
 Input [2]: [key#x, max#x]
-Arguments: hashpartitioning(key#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, 4), true, [id=#x]
 
 (7) HashAggregate [codegen id : 2]
 Input [2]: [key#x, max#x]
@@ -246,7 +246,7 @@ Results [2]: [key#x, val#x]
 
 (11) Exchange 
 Input [2]: [key#x, val#x]
-Arguments: hashpartitioning(key#x, val#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, val, 4), true, [id=#x]
 
 (12) HashAggregate [codegen id : 4]
 Input [2]: [key#x, val#x]
@@ -832,7 +832,7 @@ Results [2]: [key#x, max#x]
 
 (6) Exchange 
 Input [2]: [key#x, max#x]
-Arguments: hashpartitioning(key#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, 4), true, [id=#x]
 
 (7) HashAggregate [codegen id : 4]
 Input [2]: [key#x, max#x]
@@ -917,7 +917,7 @@ Input [2]: [key#x, val#x]
 (3) HashAggregate 
 Input [2]: [key#x, val#x]
 Keys: []
-Functions [3]: [partial_count(val#x), partial_sum(cast(key#x as bigint)), partial_count(key#x) FILTER (WHERE (val#x > 1))]
+Functions [3]: [partial_count(val), partial_sum(cast(key#x as bigint)), partial_count(key) FILTER (WHERE (val#x > 1))]
 Aggregate Attributes [3]: [count#xL, sum#xL, count#xL]
 Results [3]: [count#xL, sum#xL, count#xL]
 
@@ -928,9 +928,9 @@ Arguments: SinglePartition, true, [id=#x]
 (5) HashAggregate [codegen id : 2]
 Input [3]: [count#xL, sum#xL, count#xL]
 Keys: []
-Functions [3]: [count(val#x), sum(cast(key#x as bigint)), count(key#x)]
-Aggregate Attributes [3]: [count(val#x)#xL, sum(cast(key#x as bigint))#xL, count(key#x)#xL]
-Results [2]: [(count(val#x)#xL + sum(cast(key#x as bigint))#xL) AS TOTAL#xL, count(key#x)#xL AS count(key) FILTER (WHERE (val > 1))#xL]
+Functions [3]: [count(val), sum(cast(key#x as bigint)), count(key)]
+Aggregate Attributes [3]: [count(val)#xL, sum(cast(key#x as bigint))#xL, count(key)#xL]
+Results [2]: [(count(val)#xL + sum(cast(key#x as bigint))#xL) AS TOTAL#xL, count(key)#xL AS count(key) FILTER (WHERE (val > 1))#xL]
 
 
 -- !query
@@ -967,7 +967,7 @@ Results [2]: [key#x, buf#x]
 
 (4) Exchange 
 Input [2]: [key#x, buf#x]
-Arguments: hashpartitioning(key#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, 4), true, [id=#x]
 
 (5) ObjectHashAggregate 
 Input [2]: [key#x, buf#x]
@@ -1017,7 +1017,7 @@ Results [2]: [key#x, min#x]
 
 (5) Exchange 
 Input [2]: [key#x, min#x]
-Arguments: hashpartitioning(key#x, 4), true, [id=#x]
+Arguments: hashpartitioning(key, 4), true, [id=#x]
 
 (6) Sort [codegen id : 2]
 Input [2]: [key#x, min#x]

sql/core/src/test/resources/sql-tests/results/pivot.sql.out

@@ -475,7 +475,7 @@ PIVOT (
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid pivot column 'named_struct(course, course#x, m, m#x)'. Pivot columns must be comparable.;
+Invalid pivot column 'named_struct(course, course, m, m)'. Pivot columns must be comparable.;
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/udf/udf-pivot.sql.out

@@ -441,7 +441,7 @@ PIVOT (
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid pivot column 'named_struct(course, course#x, m, m#x)'. Pivot columns must be comparable.;
+Invalid pivot column 'named_struct(course, course, m, m)'. Pivot columns must be comparable.;
 
 
 -- !query