Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-26403][SQL] Support pivoting using array column for pivot(column) API #23349

Closed
wants to merge 5 commits into from
Closed

[SPARK-26403][SQL] Support pivoting using array column for pivot(column) API #23349

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
741ee69
Support pivoting using array column for `pivot(column)` API
HyukjinKwon Dec 19, 2018
825d14c
Alternative take
HyukjinKwon Dec 30, 2018
392c87f
Remove literal test
HyukjinKwon Dec 30, 2018
7f8c3e9
Revert "Alternative take"
HyukjinKwon Jan 2, 2019
d91ade6
Revert "Remove literal test"
HyukjinKwon Jan 2, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ object Literal {
case t: Timestamp => Literal(DateTimeUtils.fromJavaTimestamp(t), TimestampType)
case d: Date => Literal(DateTimeUtils.fromJavaDate(d), DateType)
case a: Array[Byte] => Literal(a, BinaryType)
case a: collection.mutable.WrappedArray[_] => apply(a.array)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if you look at the doc of Row.get, the official type for array type is scala.collection.Seq

In practice, we should never match collection.mutable.WrappedArray explicitly. Users usually don't create this type explicitly.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we can easily support Seq with minimised change here - let me take another look for this. That's why I targeted to support an array column in pivot specifically. Maybe I can do the type check within pivot before we call Literal.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea, I'm not sure if I know the way for Seq here. I can make the changes inside of pivot to be more conservative, @cloud-fan. Let me know.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isn't it simply apply(seq.toArray)?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nope, that's not complied..

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

[error] /.../spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala:70: No ClassTag available for _
[error]     case a: Seq[_] => apply(a.toArray)
[error]                               ^
[error] one error found

We can't use type tag here as well since what we need is Any:

spark/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala

Line 422 in 86cc907

def pivot(pivotColumn: Column, values: Seq[Any]): RelationalGroupedDataset = {

I know there is some subtlety. That's why I tried to explain in details ..

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we provide the class tag manually? like

apply(seq.toArray(classTag[Any]))

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That wouldn't work

scala> Array(1,2,3).asInstanceOf[Array[Any]]
java.lang.ClassCastException: [I cannot be cast to [Ljava.lang.Object;
  ... 29 elided

scala> classOf[Array[Any]].getComponentType()
res1: Class[_] = class java.lang.Object

The cases below fail to compile.

apply(a.toArray[classTag[Any]])
apply(a.toArray[scala.reflect.classTag[Any]])

The case below causes runtime failure.

case a: Seq[Any] => apply(a.toArray)
Unsupported component type class java.lang.Object in arrays;
org.apache.spark.sql.AnalysisException: Unsupported component type class java.lang.Object in arrays;
	at org.apache.spark.sql.catalyst.expressions.Literal$.componentTypeToDataType(literals.scala:119)
	at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:72)
	at org.apache.spark.sql.RelationalGroupedDataset.$anonfun$pivot$2(RelationalGroupedDataset.scala:427)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237)
	at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
	at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:39)

case a: Array[_] =>
val elementType = componentTypeToDataType(a.getClass.getComponentType())
val dataType = ArrayType(elementType)
Expand Down
2 changes: 2 additions & 0 deletions ...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,8 @@ class LiteralExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
checkArrayLiteral(Array("a", "b", "c"))
checkArrayLiteral(Array(1.0, 4.0))
checkArrayLiteral(Array(CalendarInterval.MICROS_PER_DAY, CalendarInterval.MICROS_PER_HOUR))
val arr = collection.mutable.WrappedArray.make(Array(1.0, 4.0))
checkEvaluation(Literal(arr), toCatalyst(arr))
}

test("seq") {
Expand Down
11 changes: 11 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,4 +333,15 @@ class DataFramePivotSuite extends QueryTest with SharedSQLContext {
}
assert(exception.getMessage.contains("Unsupported literal type"))
}

test("SPARK-26403: pivoting by array column") {
val df = Seq(
(2, Seq.empty[String]),
(2, Seq("a", "x")),
(3, Seq.empty[String]),
(3, Seq("a", "x"))).toDF("x", "s")
val expected = Seq((3, 1, 1), (2, 1, 1)).toDF
val actual = df.groupBy("x").pivot("s").count()
checkAnswer(actual, expected)
}
}