[SPARK-24305][SQL][FOLLOWUP] Avoid serialization of private fields in collection expressions. #21352

mn-mikke · 2018-05-17T11:56:53Z

What changes were proposed in this pull request?

The PR tries to avoid serialization of private fields of already added collection functions and follows up on comments in SPARK-23922 and SPARK-23935

How was this patch tested?

Run tests from:

CollectionExpressionSuite.scala
DataFrameFunctionsSuite.scala

… new collection expressions.

mn-mikke · 2018-05-17T11:58:15Z

cc @ueshin @mgaido91

mgaido91 · 2018-05-17T12:12:41Z

in most of the cases these are lazy fields which are called only on the execution, so I don't think they are serialized. And in some cases (I am thinking to the ordering ones) with this patch we are executing the same computation for each row, which means a performance regression.

I am not sure honestly we do need this. @cloud-fan what do you think?

cloud-fan · 2018-05-17T12:36:19Z

adding @transient is no harm, I'm fine with it.

cloud-fan · 2018-05-17T12:37:02Z

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

@@ -870,6 +870,7 @@ case class ArrayMin(child: Expression) extends UnaryExpression with ImplicitCast

  override protected def nullSafeEval(input: Any): Any = {
    var min: Any = null
+    val ordering = TypeUtils.getInterpretedOrdering(dataType)


We can't do this, nullSafeEval is called for every input record

Ok, I will fix it.

…variable local.

mgaido91 · 2018-05-17T13:13:36Z

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

@@ -388,7 +388,8 @@ case class Reverse(child: Expression) extends UnaryExpression with ImplicitCastI

  override def dataType: DataType = child.dataType

-  lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType
+  @transient
+  private lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType


this is not serialized since it is lazy and it is used only in the eval methods. I think we don't need this change, so.

What about the arrayCodeGen method, isn't it executed before seriealization?

code generation is executed on each partition so I'd say no, it is not executed before serialization.

I feel the benefit of adding @transient is: we don't need to think about if this method is called on executor side only. We never serialize it.

@mgaido91 Thanks for clarification! I thought that code gets generated and compiled on the driver and sent to each executor.

sure, np. Anyway, if you all think that it is good to add it, then let's do it and ignore my comments related to @transient please.

mgaido91 · 2018-05-17T13:14:45Z

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

@@ -552,7 +553,8 @@ case class Slice(x: Expression, start: Expression, length: Expression)

  override def children: Seq[Expression] = Seq(x, start, length)

-  lazy val elementType: DataType = x.dataType.asInstanceOf[ArrayType].elementType
+  @transient
+  private lazy val elementType: DataType = x.dataType.asInstanceOf[ArrayType].elementType


What about genCodeForResult?

same as before :)

mgaido91 · 2018-05-17T13:15:43Z

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

@@ -837,6 +839,7 @@ case class ArrayMin(child: Expression) extends UnaryExpression with ImplicitCast

  override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)

+  @transient


Yeah, here it isn't necessary.

mgaido91 · 2018-05-17T13:16:21Z

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

@@ -902,6 +905,7 @@ case class ArrayMax(child: Expression) extends UnaryExpression with ImplicitCast

  override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)

+  @transient


mgaido91 · 2018-05-17T13:17:05Z

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

@@ -1147,6 +1151,7 @@ case class Concat(children: Seq[Expression]) extends Expression {

  override def dataType: DataType = children.map(_.dataType).headOption.getOrElse(StringType)

+  @transient


for all this related to code generation, please refer to #21352 (comment)

mgaido91 · 2018-05-17T13:20:31Z

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

@@ -1126,9 +1130,9 @@ case class ElementAt(left: Expression, right: Expression) extends GetMapValueUti
  """)
 case class Concat(children: Seq[Expression]) extends Expression {

-  private val MAX_ARRAY_LENGTH: Int = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
+  private def maxArrayLength: Int = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH


since ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH is a static value, what about using directly it in the methods instead of defining a quite useless def?

mgaido91 · 2018-05-17T13:20:59Z

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

@@ -1324,15 +1329,17 @@ case class Concat(children: Seq[Expression]) extends Expression {
  since = "2.4.0")
 case class Flatten(child: Expression) extends UnaryExpression {

-  private val MAX_ARRAY_LENGTH = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
+  private def maxArrayLength: Int = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH


mgaido91 · 2018-05-17T13:22:00Z

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

@@ -1483,7 +1490,7 @@ case class Flatten(child: Expression) extends UnaryExpression {
 case class ArrayRepeat(left: Expression, right: Expression)
  extends BinaryExpression with ExpectsInputTypes {

-  private val MAX_ARRAY_LENGTH = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
+  private def maxArrayLength: Int = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH


mgaido91 · 2018-05-17T13:23:22Z

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

  private lazy val childDataType: ArrayType = child.dataType.asInstanceOf[ArrayType]

  override def nullable: Boolean = child.nullable || childDataType.containsNull

  override def dataType: DataType = childDataType.elementType

-  lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType
+  @transient


…_LENGTH directly

SparkQA · 2018-05-17T15:42:40Z

Test build #90730 has finished for PR 21352 at commit ded67f5.

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

SparkQA · 2018-05-17T16:39:36Z

Test build #90733 has finished for PR 21352 at commit e96962e.

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

SparkQA · 2018-05-17T17:45:03Z

Test build #90740 has finished for PR 21352 at commit f6368b5.

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

SparkQA · 2018-05-17T21:37:01Z

Test build #90747 has finished for PR 21352 at commit 2862d3e.

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

ueshin · 2018-05-18T02:39:14Z

Jenkins, retest this please.

SparkQA · 2018-05-18T05:44:47Z

Test build #90767 has finished for PR 21352 at commit 2862d3e.

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

mn-mikke · 2018-05-18T07:25:00Z

retest this please

SparkQA · 2018-05-18T11:11:10Z

Test build #90781 has finished for PR 21352 at commit 2862d3e.

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

HyukjinKwon · 2018-07-16T02:34:48Z

ok to test

SparkQA · 2018-07-16T04:38:04Z

Test build #93062 has finished for PR 21352 at commit 2862d3e.

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

cloud-fan · 2018-07-16T06:19:54Z

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

@@ -1324,15 +1329,15 @@ case class Concat(children: Seq[Expression]) extends Expression {
  since = "2.4.0")
 case class Flatten(child: Expression) extends UnaryExpression {

-  private val MAX_ARRAY_LENGTH = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
-
+  @transient
  private lazy val childDataType: ArrayType = child.dataType.asInstanceOf[ArrayType]


we should only use lazy val if it can save some heavy computation. For things like this, I think def is good enough

…ering more expressions.

cloud-fan · 2018-07-16T15:21:00Z

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

-    case ArrayType(
+  @transient private lazy val dataTypeDetails: Option[(MapType, Boolean, Boolean)] = {
+    child.dataType match {
+      case ArrayType(
      StructType(Array(


the indentation seems wrong here.

cloud-fan · 2018-07-16T15:21:48Z

LGTM

mgaido91 · 2018-07-16T15:26:07Z

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

@@ -3226,7 +3218,7 @@ case class ArrayDistinct(child: Expression)

  override def dataType: DataType = child.dataType

-  @transient lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType
+  private def elementType: DataType = dataType.asInstanceOf[ArrayType].elementType


nit: as this is used in the eval method, with this PR we are re-evaluating this code for each row. Despite probably it is not a big issue, I'd rather not introduce perf regression. WDYT @cloud-fan ?

Yea, makes sense.

+1. If it's used in eval, let's use lazy val, as it will be called for every input. Are there more places like this?

yes, https://github.com/apache/spark/pull/21352/files#diff-9853dcf5ce3d2ac1e94d473197ff5768R183 for instance

@mgaido91 Thanks for your point!

SparkQA · 2018-07-16T19:16:56Z

Test build #93115 has finished for PR 21352 at commit 62c55ad.

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

SparkQA · 2018-07-16T19:41:27Z

Test build #93116 has finished for PR 21352 at commit 294ac69.

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

…d (Concat, MapConcat)

mgaido91 · 2018-07-17T09:57:50Z

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

-  @transient lazy val numberOfArrays: Int = children.length
+  override def nullable: Boolean = children.exists(_.nullable)
+
+  @transient private lazy val arrayElementTypes = {


nit: I think we don't need the braces

mgaido91 · 2018-07-17T09:59:34Z

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

@@ -360,7 +356,7 @@ case class MapEntries(child: Expression) extends UnaryExpression with ExpectsInp

  override def inputTypes: Seq[AbstractDataType] = Seq(MapType)

-  lazy val childDataType: MapType = child.dataType.asInstanceOf[MapType]
+  private def childDataType: MapType = child.dataType.asInstanceOf[MapType]


this should be a lazy val

I missed that one. Thanks!

mgaido91 · 2018-07-17T10:59:44Z

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

-        StructField(_, valueType, valueNullable, _))),
-      containsNull) => Some((MapType(keyType, valueType, valueNullable), keyNullable, containsNull))
-    case _ => None
+  @transient private lazy val dataTypeDetails: Option[(MapType, Boolean, Boolean)] = {


this is an unneeded change, isn't it?

Here I wanted to be consistent in terms of formatting. (@transient to be on the same line as private lazy val dataTypeDetails) After the change, two lines were exceeding 100 characters.

I see, but this seems an unneeded change to me and I think there are other places where we use this syntax, so I see no reason to change it

kiszk · 2018-07-17T13:10:57Z

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

+    child.dataType match {
+      case ArrayType(
+        StructType(Array(
+          StructField(_, kt, kn, _),


nit: Is there any reason to change variable names? It would be good to minimize differences for review and ease of understanding.

the motivation is described here. I will revert this piece of code shortly.

…tAt expression

SparkQA · 2018-07-17T13:39:22Z

Test build #93165 has finished for PR 21352 at commit 872ef99.

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

mgaido91 · 2018-07-17T14:40:52Z

LGTM

cloud-fan · 2018-07-17T15:07:43Z

thanks, merging to master!

SparkQA · 2018-07-17T17:34:12Z

Test build #93175 has finished for PR 21352 at commit 922d2f0.

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

[SPARK-24305][SQL][FOLLOWUP] Avoid serialization of private fields in…

ded67f5

… new collection expressions.

cloud-fan reviewed May 17, 2018

View reviewed changes

[SPARK-24305][SQL][FOLLOWUP] Reverting change of making the ordering …

e96962e

…variable local.

mgaido91 reviewed May 17, 2018

View reviewed changes

[SPARK-24305][SQL][FOLLOWUP] Using ByteArrayMethods.MAX_ROUNDED_ARRAY…

f6368b5

…_LENGTH directly

[SPARK-24305][SQL][FOLLOWUP] Fixing failing tests.

2862d3e

cloud-fan reviewed Jul 16, 2018

View reviewed changes

Merge remote-tracking branch 'spark/master' into SPARK-24305

a4d1e7f

mn-mikke mentioned this pull request Jul 16, 2018

[SPARK-23931][SQL] Adds arrays_zip function to sparksql #21045

Closed

[SPARK-24305][SQL][FOLLOWUP] Using def wherever it's possible and cov…

62c55ad

…ering more expressions.

cloud-fan reviewed Jul 16, 2018

View reviewed changes

mgaido91 reviewed Jul 16, 2018

View reviewed changes

[SPARK-24305][SQL][FOLLOWUP] Fixing indentation

294ac69

[SPARK-24305][SQL][FOLLOWUP] Making fields used in eval lazy-evaluated

94b86a2

mn-mikke added 2 commits July 17, 2018 11:44

Merge remote-tracking branch 'spark/master' into SPARK-24305

a0abc25

[SPARK-24305][SQL][FOLLOWUP] Making fields used in eval lazy-evaluate…

872ef99

…d (Concat, MapConcat)

mgaido91 reviewed Jul 17, 2018

View reviewed changes

[SPARK-24305][SQL][FOLLOWUP] Addressing review comments

fd3a945

kiszk reviewed Jul 17, 2018

View reviewed changes

[SPARK-24305][SQL][FOLLOWUP] A small optimization of Slice and Elemen…

922d2f0

…tAt expression

asfgit closed this in 4cf1bec Jul 17, 2018

		@@ -837,6 +839,7 @@ case class ArrayMin(child: Expression) extends UnaryExpression with ImplicitCast

		override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)

		@transient

		@@ -902,6 +905,7 @@ case class ArrayMax(child: Expression) extends UnaryExpression with ImplicitCast

		override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)

		@transient

		@@ -1147,6 +1151,7 @@ case class Concat(children: Seq[Expression]) extends Expression {

		override def dataType: DataType = children.map(_.dataType).headOption.getOrElse(StringType)

		@transient

[SPARK-24305][SQL][FOLLOWUP] Avoid serialization of private fields in collection expressions. #21352

[SPARK-24305][SQL][FOLLOWUP] Avoid serialization of private fields in collection expressions. #21352

Conversation

mn-mikke commented May 17, 2018

What changes were proposed in this pull request?

How was this patch tested?

mn-mikke commented May 17, 2018

mgaido91 commented May 17, 2018

cloud-fan commented May 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mgaido91 May 17, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 17, 2018

SparkQA commented May 17, 2018

SparkQA commented May 17, 2018

SparkQA commented May 17, 2018

ueshin commented May 18, 2018

SparkQA commented May 18, 2018

mn-mikke commented May 18, 2018

SparkQA commented May 18, 2018

HyukjinKwon commented Jul 16, 2018

SparkQA commented Jul 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 16, 2018

SparkQA commented Jul 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 17, 2018

mgaido91 commented Jul 17, 2018

cloud-fan commented Jul 17, 2018

SparkQA commented Jul 17, 2018

mgaido91 May 17, 2018 •

edited

Loading