ConvertToLocalRelation should tolerate expression reusing output object

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericDatumWriter, GenericRecord, GenericRecordBuilder}
import org.apache.avro.io.EncoderFactory
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.avro.from_avro
import org.apache.spark.sql.functions.col

import java.io.ByteArrayOutputStream

object TestApp extends App {
  // Payload container
  case class TestEvent(payload: Array[Byte])
  // Deserialized Payload
  case class TestPayload(message: String)
  // Schema for Payload
  val simpleSchema =
    """
      |{
      |"type": "record",
      |"name" : "Payload",
      |"fields" : [ {"name" : "message", "type" : [ "string", "null" ] } ]
      |}
    """.stripMargin
  // Convert TestPayload into avro binary
  def generateSimpleSchemaBinary(record: TestPayload, avsc: String): Array[Byte] = {
    val schema = new Schema.Parser().parse(avsc)
    val out = new ByteArrayOutputStream()
    val writer = new GenericDatumWriter[GenericRecord](schema)
    val encoder = EncoderFactory.get().binaryEncoder(out, null)
    val rootRecord = new GenericRecordBuilder(schema).set("message", record.message).build()
    writer.write(rootRecord, encoder)
    encoder.flush()
    out.toByteArray
  }

  val spark: SparkSession = SparkSession.builder().master("local[*]").getOrCreate()
  import spark.implicits._
  List(
    TestPayload("one"),
    TestPayload("two"),
    TestPayload("three"),
    TestPayload("four")
  ).map(payload => TestEvent(generateSimpleSchemaBinary(payload, simpleSchema)))
    .toDS()
    .withColumn("deserializedPayload", from_avro(col("payload"), simpleSchema))
    .show(truncate = false)
}