Skip to content

[SPARK-39575][AVRO] add ByteBuffer#rewind after ByteBuffer#get in Avr… #36973

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

Closed
wzx140 wants to merge 3 commits into from
Closed

[SPARK-39575][AVRO] add ByteBuffer#rewind after ByteBuffer#get in Avr… #36973

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bd917ce
[SPARK-39575][AVRO] add ByteBuffer#rewind after ByteBuffer#get in Avr…
Jun 24, 2022
1cb156c
remove unnecessary nested type
Jun 24, 2022
83094a8
fix code style
Jun 25, 2022
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
2 changes: 2 additions & 0 deletions connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,8 @@ private[sql] class AvroDeserializer(
case b: ByteBuffer =>
val bytes = new Array[Byte](b.remaining)
b.get(bytes)
// Do not forget to reset the position
b.rewind()
Copy link
Member

@srowen srowen Jun 24, 2022

Choose a reason for hiding this comment

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

I think you're right but can you explain why this is needed?

Copy link
Contributor Author

@wzx140 wzx140 Jun 24, 2022

Choose a reason for hiding this comment

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

HeapBuffer.get(bytes) puts the data from POS to the end into bytes, and sets POS as the end. The next call will return empty bytes. You can take a look at added unit test. The second call of deserializer will return an InternalRow with empty binary column.

Copy link
Member

@srowen srowen Jun 24, 2022

Choose a reason for hiding this comment

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

Sounds good. I wonder why this never surfaced before? seems like it would mean any binary cols in Avro don't work?

Copy link
Contributor Author

@wzx140 wzx140 Jun 24, 2022

Choose a reason for hiding this comment

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

Maybe this is not common to call this twice to deserialize one avro data object

bytes
case b: Array[Byte] => b
case other =>
Expand Down
21 changes: 21 additions & 0 deletions ...ector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,4 +360,25 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite
None,
new OrderedFilters(Seq(Not(EqualTo("Age", 39))), sqlSchema))
}

test("AvroDeserializer with binary type") {
val jsonFormatSchema =
"""
|{
| "type": "record",
| "name": "record",
| "fields" : [
| {"name": "a", "type": "bytes"}
| ]
|}
""".stripMargin
val avroSchema = new Schema.Parser().parse(jsonFormatSchema)
val avroRecord = new GenericData.Record(avroSchema)
val bb = java.nio.ByteBuffer.wrap(Array[Byte](97, 48, 53))
avroRecord.put("a", bb)

val expected = InternalRow(Array[Byte](97, 48, 53))
checkDeserialization(avroSchema, avroRecord, Some(expected))
checkDeserialization(avroSchema, avroRecord, Some(expected))
}
}