[SPARK-32585][SQL] Support scala enumeration in ScalaReflection #29403

ulysses-you · 2020-08-11T06:17:13Z

What changes were proposed in this pull request?

Add code in ScalaReflection to support scala enumeration and make enumeration type as string type in Spark.

Why are the changes needed?

We support java enum but failed with scala enum, it's better to keep the same behavior.

Here is a example.

package test

object TestEnum extends Enumeration {
  type TestEnum = Value
  val E1, E2, E3 = Value
}
import TestEnum._
case class TestClass(i: Int,  e: TestEnum) {
}

import test._
Seq(TestClass(1, TestEnum.E1)).toDS

Before this PR

Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for test.TestEnum.TestEnum
- field (class: "scala.Enumeration.Value", name: "e")
- root class: "test.TestClass"
  at org.apache.spark.sql.catalyst.ScalaReflection$.$anonfun$serializerFor$1(ScalaReflection.scala:567)
  at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:69)
  at org.apache.spark.sql.catalyst.ScalaReflection.cleanUpReflectionObjects(ScalaReflection.scala:882)
  at org.apache.spark.sql.catalyst.ScalaReflection.cleanUpReflectionObjects$(ScalaReflection.scala:881)

After this PR
org.apache.spark.sql.Dataset[test.TestClass] = [i: int, e: string]

Does this PR introduce any user-facing change?

Yes, user can make case class which include scala enumeration field as dataset.

How was this patch tested?

Add test.

SparkQA · 2020-08-11T07:05:02Z

Test build #127308 has finished for PR 29403 at commit 556b709.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-08-11T12:15:54Z

Test build #127319 has finished for PR 29403 at commit 0f98bef.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-08-12T05:33:52Z

Test build #127352 has finished for PR 29403 at commit 462b0f7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

srowen

@rednaxelafx would you have any thoughts? you might know of any potential issues in use of Scala enums

ulysses-you · 2020-08-31T00:26:55Z

I don't know if there exists any issue or reason why we not support it.

@maropu @cloud-fan @dongjoon-hyun do you have any thought ?

marmbrus

This seems generally useful to me. I would use it and there are blog posts that talk about working around this. A few comments on nullability.

marmbrus · 2020-09-25T23:33:36Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

            StructField(fieldName, dataType, nullable)
          }), nullable = true)
+      case t if isSubtype(t, localTypeOf[Enumeration#Value]) =>
+        Schema(StringType, nullable = false)


Why is nullable = false, I believe this should be reserved for primitive types that cannot be null.

Yeah, it should be true.

marmbrus · 2020-09-25T23:34:26Z

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala

+  test("SPARK-32585: Support scala enumeration in ScalaReflection") {
+    checkDataset(
+      Seq(FooClassWithEnum(1, FooEnum.E1), FooClassWithEnum(2, FooEnum.E2)).toDS(),
+      Seq(FooClassWithEnum(1, FooEnum.E1), FooClassWithEnum(2, FooEnum.E2)): _*


Related to the above comment, I would add a test case where the enum value is null.

Added the null test.

marmbrus · 2020-09-25T23:35:48Z

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ScalaReflectionSuite.scala

    assert(deserializerFor[FooWithAnnotation].dataType == ObjectType(classOf[FooWithAnnotation]))
  }
+
+  test("SPARK-32585: Support scala enumeration in ScalaReflection") {


I would consider putting a test case in ExpressionEncoderSuite as well as I think that checks various combinations of evaluation.

Thanks, not realized this test.

marmbrus · 2020-09-26T01:00:52Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

+        // we can call example.Foo.withName to deserialize string to enumeration.
+        val className = t.asInstanceOf[TypeRef].pre.typeSymbol.asClass.fullName
+        // this check is for spark-shell which give a default package name like '$line1.$read$$iw'
+        if (className.startsWith("$")) {


I don't actually understand this limitation?

It's actually a jline issue that default package is something like $line32.$read.$iw.$iw.$iw.$iw.$iw. We can't find Enum class using reflect. I choose to forbid it, but please tell me if there exists a better way to go.

marmbrus · 2020-09-26T01:01:38Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

+            s"Enumeration class required package name, but found $className")
+        }
+
+        val clazz = Utils.classForName(className)


An alternative:

val parent = t.asInstanceOf[TypeRef].pre.typeSymbol.asClass val cls = scala.reflect.runtime.universe .runtimeMirror(getClass.getClassLoader) .runtimeClass(parent)

Does this pure scala reflection fix the issues with the REPL?

Not fixed, exists the same issue.

Hmm, what is the error? Regardless, I'm not sure this limitation should hold up merging the basic support for enums.

Let me check this again.

After deep check some code, I found the magic.

we can get FooEnum class with default package using mirror. runtimeClass.

we failed in StaticInvoke which will re-reflect the class if class name end with $.

The StaticInvoke related code and pr #20753:

val objectName = staticObject.getName.stripSuffix("$") val cls = if (staticObject.getName == objectName) { staticObject } else { Utils.classForName(objectName) }

I cann't find more comment about this code, seems we can use staticObject directly instead of re-reflect it, is it ?
cc @kiszk

SparkQA · 2020-09-27T13:00:04Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33762/

SparkQA · 2020-09-27T13:16:34Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33762/

SparkQA · 2020-09-27T16:46:50Z

Test build #129147 has finished for PR 29403 at commit 68c163a.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
encodeDecodeTest(FooClassWithEnum(1, FooEnum.E1), \"case class with Int and scala Enum\")
encodeDecodeTest(FooEnum.E1, \"case class with Int and scala Enum\")

SparkQA · 2020-09-28T00:43:03Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33773/

SparkQA · 2020-09-28T01:02:21Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33773/

SparkQA · 2020-09-28T04:28:59Z

Test build #129158 has finished for PR 29403 at commit ed81055.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

marmbrus · 2020-09-28T15:35:34Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

+          ObjectType(getClassFromType(t)),
+          "withName",
+          createDeserializerForString(path, false) :: Nil,
+          returnNullable = false)


Should also be true?

Also, if I'm correct, it would be good to include a test case for this case. What happens if you actually do operations on the null value stored as an enum? Is the nullability of the resulting schema correct. Do operations like .isNotNull work correctly?

I think we need returnNullable = false since Enumeration.withName will never return null.

Yeah, you are correct.

What happens if you actually do operations on the null value stored as an enum

We will get an exception, and the behavior is same as Java enum what we supported.

case other if other.isEnum => createSerializerForString( Invoke(inputObject, "name", ObjectType(classOf[String]), returnNullable = false))

marmbrus · 2020-09-28T15:36:18Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

+            inputObject,
+            "toString",
+            ObjectType(classOf[java.lang.String]),
+            returnNullable = false))


Same as withName, toString also return a non-null value.

Ah, I see. I agree.

marmbrus · 2020-09-29T02:27:01Z

cc @tdas modulo the nullability concerns above, this LGTM.

tdas · 2020-09-29T02:58:31Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

            StructField(fieldName, dataType, nullable)
          }), nullable = true)
+      case t if isSubtype(t, localTypeOf[Enumeration#Value]) =>
+        Schema(StringType, nullable = true)


So we are returning the schema generated from enum as nullable, but the serializer expression is not nullable (because it does not produce null). Why is that?

There are two side of nullable.

For schema, nullable means the column can be null value.

For returnNullable which used in Invoke and StaticInvoke means the method promise will never produce a null value.

Some code get from Invoke

def invoke( obj: Any, method: Method, arguments: Seq[Expression], input: InternalRow, dataType: DataType): Any = { val args = arguments.map(e => e.eval(input).asInstanceOf[Object]) if (needNullCheck && args.exists(_ == null)) { // return null if one of arguments is null null } else { val ret = method.invoke(obj, args: _*) val boxedClass = ScalaReflection.typeBoxedJavaMapping.get(dataType) if (boxedClass.isDefined) { boxedClass.get.cast(ret) } else { ret } } }

So effectively, in the serialized form, we are allowing nulls to be present in the column mapping to the enum field. But if there is indeed a row with a null in that column and we attempt to deserialize that rows, then it will cause a runtime failure...isnt it?

If this understanding is correct, then serializing will never produce null, and even if there is a null, serializing will fail. Then why keep this column nullable = true? I am not necessarily opposed to it, I am just trying to understand the rationale. cc @marmbrus

But if there is indeed a row with a null in that column and we attempt to deserialize that rows, then it will cause a runtime failure...isnt it

Actually we will get null back. It's a trick that if input value is null the serializing will return null.

SparkQA · 2020-09-29T07:05:01Z

Test build #129216 has finished for PR 29403 at commit 2d675ca.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-29T07:31:12Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33831/

SparkQA · 2020-09-29T07:52:01Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/33831/

ulysses-you · 2020-09-29T10:00:40Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

+        // the fullName of tpe is example.Foo.Foo, but we need example.Foo so that
+        // we can call example.Foo.withName to deserialize string to enumeration.
+        val parent = t.asInstanceOf[TypeRef].pre.typeSymbol.asClass
+        val cls = mirror.runtimeClass(parent)


follow scala reflection.

tdas · 2020-10-01T19:49:02Z

LGTM. Merging it to master.

ulysses-you · 2020-10-06T13:17:19Z

thanks for merging and review !

init

556b709

probot-autolabeler bot added the SQL label Aug 11, 2020

fix ut

0f98bef

fix

462b0f7

srowen reviewed Aug 12, 2020

View reviewed changes

marmbrus suggested changes Sep 25, 2020

View reviewed changes

marmbrus reviewed Sep 26, 2020

View reviewed changes

ulysses-you added 2 commits September 27, 2020 17:35

null

623a781

ExpressionEncoderSuite

68c163a

typo

ed81055

marmbrus reviewed Sep 28, 2020

View reviewed changes

tdas reviewed Sep 29, 2020

View reviewed changes

mirror

2d675ca

ulysses-you commented Sep 29, 2020

View reviewed changes

tdas approved these changes Oct 1, 2020

View reviewed changes

asfgit closed this in e62d247 Oct 1, 2020

ulysses-you deleted the SPARK-32585 branch March 10, 2021 06:07

[SPARK-32585][SQL] Support scala enumeration in ScalaReflection #29403

[SPARK-32585][SQL] Support scala enumeration in ScalaReflection #29403

Uh oh!

Conversation

ulysses-you commented Aug 11, 2020

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Aug 11, 2020

Uh oh!

SparkQA commented Aug 11, 2020

Uh oh!

SparkQA commented Aug 12, 2020

Uh oh!

srowen left a comment

Choose a reason for hiding this comment

Uh oh!

ulysses-you commented Aug 31, 2020

Uh oh!

marmbrus left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 27, 2020

Uh oh!

SparkQA commented Sep 27, 2020

Uh oh!

SparkQA commented Sep 27, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!