[SPARK-31450][SQL] Make ExpressionEncoder thread-safe #28223

hvanhovell · 2020-04-15T09:04:00Z

What changes were proposed in this pull request?

This PR moves the ExpressionEncoder.toRow and ExpressionEncoder.fromRow functions into their own function objects(ExpressionEncoder.Serializer & ExpressionEncoder.Deserializer). This effectively makes the ExpressionEncoder stateless, thread-safe and (more) reusable. The function objects are not thread safe, however they are documented as such and should be used in a more limited scope (making it easier to reason about thread safety).

Why are the changes needed?

ExpressionEncoders are not thread-safe. We had various (nasty) bugs because of this.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

hvanhovell · 2020-04-15T09:08:40Z

Reviewers please first take a look at the ExpressionEncoder class and specifically the naming of the function objects.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

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

...talyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ObjectExpressionsSuite.scala

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/ArrayDataIndexedSeqSuite.scala

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

hvanhovell · 2020-04-15T09:48:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

+   * Function that deserializes an [[InternalRow]] into an object of type `T`. Instances of this
+   * class are not meant to be thread-safe.
+   */
+  abstract class Deserializer[T] extends (InternalRow => T)


Implementation note. I opted to go with abstract classes so we can get monomorphic call sites in many cases.

SparkQA · 2020-04-15T10:14:54Z

Test build #121320 has finished for PR 28223 at commit cc8400c.

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

SparkQA · 2020-04-15T10:31:03Z

Test build #121319 has finished for PR 28223 at commit a7f948c.

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

SparkQA · 2020-04-15T10:50:09Z

Test build #121313 has finished for PR 28223 at commit e6c3391.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
abstract class Deserializer[T] extends (InternalRow => T)
abstract class Serializer[T] extends (T => InternalRow)

SparkQA · 2020-04-15T11:09:31Z

Test build #121316 has finished for PR 28223 at commit b7699a2.

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

SparkQA · 2020-04-15T18:35:31Z

Test build #121327 has finished for PR 28223 at commit 2c402f9.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
abstract class Deserializer[T] extends (InternalRow => T) with Serializable
abstract class Serializer[T] extends (T => InternalRow) with Serializable

hvanhovell · 2020-04-15T18:42:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

-  } catch {
-    case e: Exception =>
-      throw new RuntimeException(s"Error while encoding: $e\n" +
+  def createSerializer(): Serializer[T] = new Serializer[T] {


This currently relies on use serializing the enclosing encoder as well. We technically don't need the entire encoder but only a couple fields. I could move this class into the companion object and just use the fields I need.

yea that would be better

I'm also thinking about if we just need to pass the original (de)serializer expressions and do the optimization inside Serializer and Deserializer lazily.

There is some cost to the optimization. So I would like to do it only once.

dongjoon-hyun · 2020-04-15T18:58:09Z

Thank you, @hvanhovell . Do we need a benchmark about this change?

hvanhovell · 2020-04-15T19:09:15Z

@dongjoon-hyun it should be a bit of a lateral move performance wise. The expensive bit is generating the code and compiling it, and we are definitely not avoiding that. It might be a bit quicker because it does not excessively copy the expression encoder (which is not for free because of the work which is done in the constructor). What would you like to see benchmarked?

dongjoon-hyun · 2020-04-15T20:15:41Z

Got it. Thank you for confirmation, @hvanhovell . I was just wondering. Your comment is enough for me.

viirya

Looks like in a good shape already.

viirya · 2020-04-15T22:39:14Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

+    } catch {
+      case e: Exception =>
+        throw new RuntimeException(s"Error while decoding: $e\n" +
+          s"${deserializer.simpleString(SQLConf.get.maxToStringFields)}", e)


As pointed already, some fields like deserializer are in enclosing encoder, and so currently looks like we will serialize entire encoder? Actually we did serialize entire encoder currently but yea it is better we can get rid of unnecessary.

cloud-fan · 2020-04-16T04:08:32Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

+
+    private def initialize(): Unit = {
+      inputRow = new GenericInternalRow(1)
+      extractProjection = GenerateUnsafeProjection.generate(optimizedSerializer)


why can't this two be lazy val? performance concerns?

Yeah, lazy vals are not free and the extractProjection is on the hot path.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoderSuite.scala

HyukjinKwon · 2020-04-16T04:23:52Z

cc also @JoshRosen and @zsxwing from #26076

viirya · 2020-04-16T05:03:51Z

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

- * == Implementation ==
- *  - Encoders are not required to be thread-safe and thus they do not need to use locks to guard
- *    against concurrent access if they reuse internal buffers to improve performance.
- *


Removing this means Encoders must be thread-safe? Do we need explicit comment for that?

I think it should be thread-safe.

srowen · 2020-04-16T14:23:54Z

I feel like @JoshRosen looked at this too a while ago

SparkQA · 2020-04-16T15:17:11Z

Test build #121361 has finished for PR 28223 at commit 5bcff74.

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

SparkQA · 2020-04-16T17:31:09Z

Test build #121364 has finished for PR 28223 at commit bef4d8d.

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

SparkQA · 2020-04-17T01:02:26Z

Test build #121376 has finished for PR 28223 at commit ced396f.

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

dongjoon-hyun · 2020-04-17T01:47:36Z

Thank you, @hvanhovell and all.
Merged to master/3.0.

### What changes were proposed in this pull request? This PR moves the `ExpressionEncoder.toRow` and `ExpressionEncoder.fromRow` functions into their own function objects(`ExpressionEncoder.Serializer` & `ExpressionEncoder.Deserializer`). This effectively makes the `ExpressionEncoder` stateless, thread-safe and (more) reusable. The function objects are not thread safe, however they are documented as such and should be used in a more limited scope (making it easier to reason about thread safety). ### Why are the changes needed? ExpressionEncoders are not thread-safe. We had various (nasty) bugs because of this. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Existing tests. Closes #28223 from hvanhovell/SPARK-31450. Authored-by: herman <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit fab4ca5) Signed-off-by: Dongjoon Hyun <[email protected]>

HeartSaVioR · 2020-04-17T03:19:50Z

Late LGTM.

Make ExpressionEncoder thread-safe

e6c3391

probot-autolabeler bot added ML SQL STRUCTURED STREAMING labels Apr 15, 2020

hvanhovell requested review from cloud-fan, viirya and zsxwing April 15, 2020 09:04