[SPARK-27457][SQL] modify bean encoder to support avro objects #24367

mazeboard · 2019-04-13T21:59:10Z

What changes were proposed in this pull request?

Currently we modified JavaTypeInference to be able to create encoders for Avro objects; we have now three solutions, the ones in the PR #24299, and #22878 and this PR (fewer code changes); which one is better?

How was this patch tested?

We added one test in ExpressionencoderSuite and used the following program to test it locally:

  implicit val avroExampleEncoder = Encoders.bean[AvroExample1](classOf[AvroExample1]).asInstanceOf[ExpressionEncoder[AvroExample1]]
  val input: AvroExample1 = AvroExample1.newBuilder()
    .setMyarray(List("Foo", "Bar").asJava)
    .setMyboolean(true)
    .setMybytes(java.nio.ByteBuffer.wrap("MyBytes".getBytes()))
    .setMydouble(2.5)
    .setMyfixed(new Magic("magic".getBytes))
    .setMyfloat(25.0F)
    .setMyint(100)
    .setMylong(10L)
    .setMystring("hello")
    .setMymap(Map(
      "foo" -> new java.lang.Integer(1),
      "bar" -> new java.lang.Integer(2)).asJava)
    .setMymoney(Money.newBuilder().setAmount(100.0F).setCurrency(Currency.EUR).build())
    .build()

  val row: InternalRow = avroExampleEncoder.toRow(input)

  val output: AvroExample1 = avroExampleEncoder.resolveAndBind().fromRow(row)

  val ds: Dataset[AvroExample1] = List(input).toDS()

  println(ds.schema)
  println(ds.collect().toList)

  ds.write.format("avro").save("example1")

  val fooDF = spark.read.format("avro").load("example1")

  val fooDS = fooDF.as[AvroExample1]

  println(fooDS.collect().toList)

…eBuffer

gengliangwang · 2019-04-19T12:54:28Z

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

+        method.getParameterCount == 1)
+      if (a.getName == b.getName ||
+        (a.getName.indexOf("get") == 0 && b.getName.indexOf("set") == 0 &&
+          a.getName.substring(3) == b.getName.substring(3))) &&


It is a bit hacky here, compared with #24299

Yes, I will try to simplify

We need to check that the getter and setter functions are either prefixed with get and set respectively or both functions are not prefixed

I did this modification in #24299

I will try to make the refactor the code if possible

gengliangwang · 2019-04-19T12:55:18Z

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

  encodeDecodeTest(Option("abc"), "option of string")
  encodeDecodeTest(Option.empty[String], "empty option of string")

+  encodeDecodeTest(


I tried and the test case failed:

Can't compare maps! org.apache.avro.AvroRuntimeException: Can't compare maps! at org.apache.avro.generic.GenericData.compare(GenericData.java:984) at org.apache.avro.specific.SpecificData.compare(SpecificData.java:333) at org.apache.avro.generic.GenericData.compare(GenericData.java:961) at org.apache.avro.specific.SpecificData.compare(SpecificData.java:333) at org.apache.avro.generic.GenericData.compare(GenericData.java:946) at org.apache.avro.specific.SpecificRecordBase.compareTo(SpecificRecordBase.java:81) at org.apache.avro.specific.SpecificRecordBase.compareTo(SpecificRecordBase.java:30) at org.apache.spark.sql.catalyst.encoders.ExpressionEncoderSuite.$anonfun$encodeDecodeTest$1(ExpressionEncoderSuite.scala:442)

The issue here is how the input and the convertedBack objects are compared

if we replace the check, in line ExpressionEncoderSuite.scala:442,

left.asInstanceOf[Comparable[Any]].compareTo(right) == 0
By
left.asInstanceOf[Comparable[Any]].equals(right) == 0

the test for Avro encoder passes, but unfortunately other tests fail

Equality of objects is tricky.

The GenericData compare function

/** Comparison implementation. When equals is true, only checks for equality, * not for order. */ @SuppressWarnings(value="unchecked") protected int compare(Object o1, Object o2, Schema s, boolean equals) {

fails to compare Maps when the parameter equals is false

I propose the following

replace ExpressionEncoderSuite:434:444 lines

val isCorrect = (input, convertedBack) match { case (b1: Array[Byte], b2: Array[Byte]) => Arrays.equals(b1, b2) case (b1: Array[Int], b2: Array[Int]) => Arrays.equals(b1, b2) case (b1: Array[Array[_]], b2: Array[Array[_]]) => Arrays.deepEquals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (b1: Array[_], b2: Array[_]) => Arrays.equals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (left: Comparable[_], right: Comparable[_]) => left.asInstanceOf[Comparable[Any]].compareTo(right) == 0 case _ => input == convertedBack }

by

val convertedBackRow = encoder.toRow(convertedBack) val isCorrect = row == convertedBackRow

With the proposed modification all the tests in ExpressionEncoderSuite passes

~~I think this makes sense, the check should not depend on how the foreign objects implement~~ equality (see next comment)

we must compare input and convertedBack ; the equality row==convertedBackRow does not guarantee that the encoder is correct; indeed a buggy encoder may create a row with missing fields such that row==convertedBackRow is true while input==convertedBack is false.

For the avro encoder test the test input == convertedBack is true when the encoder is correct, thus we propose the following modification:

val isCorrect = input == convertedBack || ((input, convertedBack) match { case (b1: Array[Byte], b2: Array[Byte]) => Arrays.equals(b1, b2) case (b1: Array[Int], b2: Array[Int]) => Arrays.equals(b1, b2) case (b1: Array[Array[_]], b2: Array[Array[_]]) => Arrays.deepEquals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (b1: Array[_], b2: Array[_]) => Arrays.equals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (left: Comparable[_], right: Comparable[_]) => left.asInstanceOf[Comparable[Any]].compareTo(right) == 0 case _ => false })

With this modification all the tests pass

@SuppressWarnings

The issue here is how the input and the convertedBack objects are compared if we replace the check, in line ExpressionEncoderSuite.scala:442, left.asInstanceOf[Comparable[Any]].compareTo(right) == 0 By left.asInstanceOf[Comparable[Any]].equals(right) == 0 the test for Avro encoder passes, but unfortunately other tests fail Equality of objects is tricky. The GenericData compare function /** Comparison implementation. When equals is true, only checks for equality, * not for order. */ @SuppressWarnings(value="unchecked") protected int compare(Object o1, Object o2, Schema s, boolean equals) { fails to compare Maps when the parameter equals is false I propose the following replace ExpressionEncoderSuite:434:444 lines val isCorrect = (input, convertedBack) match { case (b1: Array[Byte], b2: Array[Byte]) => Arrays.equals(b1, b2) case (b1: Array[Int], b2: Array[Int]) => Arrays.equals(b1, b2) case (b1: Array[Array[_]], b2: Array[Array[_]]) => Arrays.deepEquals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (b1: Array[_], b2: Array[_]) => Arrays.equals(b1.asInstanceOf[Array[AnyRef]], b2.asInstanceOf[Array[AnyRef]]) case (left: Comparable[_], right: Comparable[_]) => left.asInstanceOf[Comparable[Any]].compareTo(right) == 0 case _ => input == convertedBack } by val convertedBackRow = encoder.toRow(convertedBack) val isCorrect = row == convertedBackRow With the proposed modification all the tests in ExpressionEncoderSuite passes

ravetdavid · 2019-05-17T14:41:15Z

👍 it's been a long time since we wanted to use dataset [T <: SpecificRecord]

mazeboard · 2019-05-18T08:53:14Z

The following example shows why we prefer the solution that modifies ScalaReflection (#24299)

In the example below, If we use bean encoder we must declare a tuple encoder as shown:

    implicit val barcodeEncoder = Encoders.bean[Barcode](classOf[Barcode]).asInstanceOf[ExpressionEncoder[Barcode]]
    implicit val tuplEncoder = Encoders.tuple[String, Money, CrpAddDesc](
      Encoders.STRING,
      Encoders.bean[Money](classOf[Money]).asInstanceOf[ExpressionEncoder[Money]],
      Encoders.bean[CrpAddDesc](classOf[CrpAddDesc]).asInstanceOf[ExpressionEncoder[CrpAddDesc]])
...
...
    val ds: Dataset[Barcode] = List(barcode).toDS()
    val x: Dataset[(String, Money, CrpAddDesc)] = ds.map(a => {
      (
        a.getBarcode,
        a.getPrdTaxVal,
        a.getCrpAttributes.get(0).getCrpAddDesc.get(0))
    })

if we do not declare the tuple encoder we get an error No Encoder found for common.lib.v1.Money

But with the ScalaReflection (#24299) modification we just need to declare an encoder for Barcode:

    implicit val barcodeEncoder = ExpressionEncoder[Barcode]()
      ...
      ...
    val ds: Dataset[Barcode] = List(barcode).toDS()
    val x: Dataset[(String, Money, CrpAddDesc)] = ds.map(a => {
      (
        a.getBarcode,
        a.getPrdTaxVal,
        a.getCrpAttributes.get(0).getCrpAddDesc.get(0))
    })

AmplabJenkins · 2019-09-16T18:14:24Z

Can one of the admins verify this patch?

github-actions · 2019-12-31T00:06:41Z

We're closing this PR because it hasn't been updated in a while.
This isn't a judgement on the merit of the PR in any way. It's just
a way of keeping the PR queue manageable.

If you'd like to revive this PR, please reopen it!

mazeboard added 5 commits April 13, 2019 15:30

modify resolution of bean properties and add encoder for java.nio.Byt…

6684afa

…eBuffer

return getter/setter methods

3a6fd95

use getterMethod.getGenericReturnType

0155276

add test for avro encoder using Encoders.bean

da6d507

add test files, and some fixes

5da63c8

mazeboard mentioned this pull request Apr 13, 2019

[SPARK-27388][SQL] encoder for objects defined by properties (ie. Avro) #24299

Closed

mazeboard added 5 commits April 14, 2019 15:14

add inferDataType for java.nio.ByteBuffer

5ad0ef8

simplify getObjectProperties

16dc3f1

fix style

46adf1a

fix propertyNames comparison

94ce8dd

refactor getObjectProperties

5834b3e

gengliangwang reviewed Apr 19, 2019

View reviewed changes

mazeboard added 3 commits April 19, 2019 16:24

refactor code of getObjectProperties

f0b4e3a

equality test first

3482242

use startsWith

2574e3d

dongjoon-hyun added the SQL label Jun 14, 2019

github-actions bot added the Stale label Dec 31, 2019

github-actions bot closed this Jan 1, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-27457][SQL] modify bean encoder to support avro objects #24367

[SPARK-27457][SQL] modify bean encoder to support avro objects #24367

Uh oh!

mazeboard commented Apr 13, 2019 •

edited

Loading

Uh oh!

gengliangwang Apr 19, 2019

Uh oh!

mazeboard Apr 19, 2019

Uh oh!

gengliangwang Apr 19, 2019

Uh oh!

mazeboard Apr 19, 2019 •

edited

Loading

Uh oh!

mazeboard Apr 19, 2019 •

edited

Loading

Uh oh!

ravetdavid commented May 17, 2019

Uh oh!

mazeboard commented May 18, 2019 •

edited

Loading

Uh oh!

AmplabJenkins commented Sep 16, 2019

Uh oh!

github-actions bot commented Dec 31, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

[SPARK-27457][SQL] modify bean encoder to support avro objects #24367

[SPARK-27457][SQL] modify bean encoder to support avro objects #24367

Uh oh!

Conversation

mazeboard commented Apr 13, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

gengliangwang Apr 19, 2019

Choose a reason for hiding this comment

Uh oh!

mazeboard Apr 19, 2019

Choose a reason for hiding this comment

Uh oh!

gengliangwang Apr 19, 2019

Choose a reason for hiding this comment

Uh oh!

mazeboard Apr 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mazeboard Apr 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ravetdavid commented May 17, 2019

Uh oh!

mazeboard commented May 18, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AmplabJenkins commented Sep 16, 2019

Uh oh!

github-actions bot commented Dec 31, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

mazeboard commented Apr 13, 2019 •

edited

Loading

mazeboard Apr 19, 2019 •

edited

Loading

mazeboard Apr 19, 2019 •

edited

Loading

mazeboard commented May 18, 2019 •

edited

Loading