/***********************************************************************
 * Copyright (c) 2013-2025 Commonwealth Computer Research, Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Apache License, Version 2.0
 * which accompanies this distribution and is available at
 * http://www.opensource.org/licenses/apache2.0.php.
 ***********************************************************************/

package org.locationtech.geomesa.features.avro
package serialization

import org.apache.avro.LogicalTypes.LogicalTypeFactory
import org.apache.avro.io.{Decoder, Encoder}
import org.apache.avro.{LogicalType, LogicalTypes, Schema, SchemaBuilder}
import org.geotools.api.feature.`type`.AttributeDescriptor
import org.locationtech.geomesa.utils.cache.SoftThreadLocal
import org.locationtech.geomesa.utils.geotools.ObjectType
import org.locationtech.geomesa.utils.geotools.ObjectType.ObjectType
import org.locationtech.geomesa.utils.text.{WKBUtils, WKTUtils}
import org.locationtech.jts.geom.Geometry

import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets
import java.util
import java.util.{Date, Locale, UUID}

/**
 * Trait for reading an avro field, corresponding to a simple feature attribute
 *
 * @tparam T type of attribute
 */
trait AvroField[T] {

  /**
   * Schema of the field
   *
   * @return
   */
  def schema: Schema

  /**
   * Read the field value
   *
   * @param in input
   * @return
   */
  def read(in: Decoder): T = {
    if (in.readIndex() == 1) {
      in.readNull()
      null.asInstanceOf[T]
    } else {
      readNonNull(in)
    }
  }

  /**
   * Skip over the field value - may be more efficient than reading and discarding it
   *
   * @param in input
   */
  def skip(in: Decoder): Unit = {
    if (in.readIndex() == 1) {
      in.readNull()
    } else {
      skipNonNull(in)
    }
  }

  /**
   * Write the field value
   *
   * @param out output
   * @param value value to write
   */
  def write(out: Encoder, value: T): Unit = {
    if (value == null) {
      out.writeIndex(1)
      out.writeNull()
    } else {
      out.writeIndex(0)
      writeNonNull(out, value)
    }
  }

  /**
   * Gets a version-specific instance of this field.
   * See org.locationtech.geomesa.features.avro.SerializationVersions
   *
   * @param version version
   * @return
   */
  def withVersion(version: Int): AvroField[T] = this

  protected def readNonNull(in: Decoder): T
  protected def skipNonNull(in: Decoder): Unit
  protected def writeNonNull(out: Encoder, value: T): Unit
}

object AvroField {

  private val buffers = new SoftThreadLocal[(ByteBuffer, Array[Byte])]

  Seq(WkbLogicalType, WktLogicalType, ListLogicalType, MapLogicalType)
      .foreach(lt => LogicalTypes.register(lt.getName, lt))

  /**
   * Get the field corresponding to the attribute descriptor
   *
   * @param descriptor descriptor
   * @return
   */
  def apply(descriptor: AttributeDescriptor): AvroField[AnyRef] = apply(ObjectType.selectType(descriptor))

  private def apply(types: Seq[ObjectType]): AvroField[AnyRef] = {
    val field = types.head match {
      case ObjectType.STRING   => StringField
      case ObjectType.INT      => IntField
      case ObjectType.LONG     => LongField
      case ObjectType.FLOAT    => FloatField
      case ObjectType.DOUBLE   => DoubleField
      case ObjectType.BOOLEAN  => BooleanField
      case ObjectType.DATE     => DateField
      case ObjectType.UUID     => UuidBinaryField
      case ObjectType.GEOMETRY => GeometryField
      case ObjectType.LIST     => new ListOpaqueField(types(1))
      case ObjectType.MAP      => new MapOpaqueField(types(1), types(2))
      case ObjectType.BYTES    => BytesField
      case t => throw new IllegalStateException(s"Unexpected descriptor type: $t")
    }
    field.asInstanceOf[AvroField[AnyRef]]
  }

  abstract class StandardField[T](val name: String) extends AvroField[T] {
    override def read(in: Decoder): T = readNonNull(in)
    override def skip(in: Decoder): Unit = skipNonNull(in)
    override def write(out: Encoder, value: T): Unit = writeNonNull(out, value)
  }

  /**
   * The serialization version
   */
  case object VersionField extends StandardField[Int]("__version__") {
    override val schema: Schema = SchemaBuilder.builder().intType()
    override protected def readNonNull(in: Decoder): Int = in.readInt()
    override protected def skipNonNull(in: Decoder): Unit = in.readInt()
    override protected def writeNonNull(out: Encoder, value: Int): Unit = out.writeInt(value)
  }

  /**
   * The feature id
   */
  case object FidField extends StandardField[String]("__fid__") {
    override val schema: Schema = SchemaBuilder.builder().stringType()
    override protected def readNonNull(in: Decoder): String = in.readString()
    override protected def skipNonNull(in: Decoder): Unit = in.readString()
    override protected def writeNonNull(out: Encoder, value: String): Unit = out.writeString(value)
  }

  /**
   * The feature user data
   */
  case object UserDataField extends StandardField[java.util.Map[AnyRef, AnyRef]]("__userdata__") {
    override val schema: Schema = UserDataSchema.schema
    override protected def readNonNull(in: Decoder): java.util.Map[AnyRef, AnyRef] =
      AvroUserDataSerialization.deserialize(in)
    override protected def skipNonNull(in: Decoder): Unit = AvroUserDataSerialization.deserialize(in)
    override protected def writeNonNull(out: Encoder, value: java.util.Map[AnyRef, AnyRef]): Unit =
      AvroUserDataSerialization.serialize(out, value)
    override def withVersion(version: Int): AvroField[util.Map[AnyRef, AnyRef]] =
      if (version < 5) { UserDataFieldV4 } else { this }
  }

  /**
   * Deprecated feature user data serialization
   */
  case object UserDataFieldV4 extends StandardField[java.util.Map[AnyRef, AnyRef]]("__userdata__") {
    // note that for versions < 4 the schema is incorrect and a bug
    override val schema: Schema = UserDataSchema.schema
    override protected def readNonNull(in: Decoder): java.util.Map[AnyRef, AnyRef] =
      AvroUserDataSerializationV4.deserialize(in)
    override protected def skipNonNull(in: Decoder): Unit = AvroUserDataSerializationV4.deserialize(in)
    // noinspection ScalaDeprecation
    override protected def writeNonNull(out: Encoder, value: java.util.Map[AnyRef, AnyRef]): Unit =
      AvroUserDataSerializationV4.serialize(out, value)
    override def withVersion(version: Int): AvroField[util.Map[AnyRef, AnyRef]] =
      if (version < 5) { this } else { UserDataField }
  }

  private object UserDataSchema {

    private val kvType =
      SchemaBuilder.builder().unionOf
          .nullType.and
          .stringType.and
          .intType.and
          .longType.and
          .floatType.and
          .doubleType.and
          .booleanType.and
          .bytesType
          .endUnion()

    private val itemType =
      SchemaBuilder.builder().record("userDataItem").fields()
          .name("key").`type`(kvType).noDefault()
          .name("value").`type`(kvType).noDefault()
          .endRecord()

    val schema: Schema = SchemaBuilder.builder().array().items(itemType)
  }

  /**
   * String type field
   */
  case object StringField extends AvroField[String] {
    override val schema: Schema = SchemaBuilder.nullable().stringType()
    override protected def readNonNull(in: Decoder): String = {
      // note: we don't use the string reading methods, as internal avro state can get corrupted and cause
      // exceptions in BinaryDecoder.scratchUtf8
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      val length = bb.remaining
      if (bytes.length < length) {
        bytes = Array.ofDim(length)
      }
      buffers.put((bb, bytes))
      bb.get(bytes, 0, length)
      new String(bytes, 0, length, StandardCharsets.UTF_8)
    }
    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()
    override protected def writeNonNull(out: Encoder, value: String): Unit = out.writeString(value)
  }

  /**
   * Int type field
   */
  case object IntField extends AvroField[java.lang.Integer] {
    override val schema: Schema = SchemaBuilder.nullable().intType()
    override protected def readNonNull(in: Decoder): java.lang.Integer = Int.box(in.readInt())
    override protected def skipNonNull(in: Decoder): Unit = in.readInt()
    override protected def writeNonNull(out: Encoder, value: java.lang.Integer): Unit = out.writeInt(value)
  }

  /**
   * Long type field
   */
  case object LongField extends AvroField[java.lang.Long] {
    override val schema: Schema = SchemaBuilder.nullable().longType()
    override protected def readNonNull(in: Decoder): java.lang.Long = Long.box(in.readLong())
    override protected def skipNonNull(in: Decoder): Unit = in.readLong()
    override protected def writeNonNull(out: Encoder, value: java.lang.Long): Unit = out.writeLong(value)
  }

  /**
   * Float type field
   */
  case object FloatField extends AvroField[java.lang.Float] {
    override val schema: Schema = SchemaBuilder.nullable().floatType()
    override protected def readNonNull(in: Decoder): java.lang.Float = Float.box(in.readFloat())
    override protected def skipNonNull(in: Decoder): Unit = in.readFloat()
    override protected def writeNonNull(out: Encoder, value: java.lang.Float): Unit = out.writeFloat(value)
  }

  /**
   * Double type field
   */
  case object DoubleField extends AvroField[java.lang.Double] {
    override val schema: Schema = SchemaBuilder.nullable().doubleType()
    override protected def readNonNull(in: Decoder): java.lang.Double = Double.box(in.readDouble())
    override protected def skipNonNull(in: Decoder): Unit = in.readDouble()
    override protected def writeNonNull(out: Encoder, value: java.lang.Double): Unit = out.writeDouble(value)
  }

  /**
   * Boolean type field
   */
  case object BooleanField extends AvroField[java.lang.Boolean] {
    override val schema: Schema = SchemaBuilder.nullable().booleanType()
    override protected def readNonNull(in: Decoder): java.lang.Boolean = Boolean.box(in.readBoolean())
    override protected def skipNonNull(in: Decoder): Unit = in.readBoolean()
    override protected def writeNonNull(out: Encoder, value: java.lang.Boolean): Unit = out.writeBoolean(value)
  }

  /**
   * Date type field, stored as millis since epoch
   */
  case object DateField extends AvroField[Date] {
    override val schema: Schema =
      SchemaBuilder.nullable().`type`(LogicalTypes.timestampMillis.addToSchema(SchemaBuilder.builder().longType()))
    override protected def readNonNull(in: Decoder): Date = new Date(in.readLong())
    override protected def skipNonNull(in: Decoder): Unit = in.readLong()
    override protected def writeNonNull(out: Encoder, value: Date): Unit = out.writeLong(value.getTime)
  }

  /**
   * UUID type field, stored as binary two longs
   */
  case object UuidBinaryField extends AvroField[UUID] {

    override val schema: Schema = SchemaBuilder.nullable().bytesType()

    def decode(buf: ByteBuffer): UUID = new UUID(buf.getLong, buf.getLong)

    override protected def readNonNull(in: Decoder): UUID = {
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      buffers.put((bb, bytes))
      decode(bb)
    }

    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()

    override protected def writeNonNull(out: Encoder, value: UUID): Unit = {
      val buf =
        ByteBuffer.allocate(16)
          .putLong(value.getMostSignificantBits)
          .putLong(value.getLeastSignificantBits)
          .flip.asInstanceOf[ByteBuffer]
      out.writeBytes(buf)
    }

    override def withVersion(version: Int): AvroField[UUID] =
      if (version == SerializationVersions.NativeCollectionVersion) { UuidRecordField } else { this }
  }

  /**
   * UUID type field, stored as a record with two longs
   */
  case object UuidRecordField extends AvroField[UUID] {

    override val schema: Schema =
      SchemaBuilder.nullable().record("uuid").fields()
        .name("msb").`type`().longType().noDefault()
        .name("lsb").`type`().longType().noDefault()
        .endRecord()

    override protected def readNonNull(in: Decoder): UUID = new UUID(in.readLong(), in.readLong())

    override protected def skipNonNull(in: Decoder): Unit = {
      in.readLong()
      in.readLong()
    }

    override protected def writeNonNull(out: Encoder, value: UUID): Unit = {
      out.writeLong(value.getMostSignificantBits)
      out.writeLong(value.getLeastSignificantBits)
    }

    override def withVersion(version: Int): AvroField[UUID] =
      if (version == SerializationVersions.NativeCollectionVersion) { this } else { UuidBinaryField }
  }

  /**
   * Base trait for Geometry type fields
   */
  trait GeomField extends AvroField[Geometry] {

    override protected def readNonNull(in: Decoder): Geometry = {
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      val length = bb.remaining
      if (bytes.length < length) {
        bytes = Array.ofDim(length)
      }
      buffers.put((bb, bytes))
      bb.get(bytes, 0, length)

      parse(bytes, length)
    }

    protected def parse(bytes: Array[Byte], length: Int): Geometry

    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()
  }

  /**
   * Geometry type field, stored as WKB
   */
  case object GeometryField extends GeomField {

    override val schema: Schema =
      SchemaBuilder.nullable().`type`(WkbLogicalType.addToSchema(SchemaBuilder.builder().bytesType()))

    // note: WKBReader ignores any bytes after the geom
    override protected def parse(bytes: Array[Byte], length: Int): Geometry = WKBUtils.read(bytes)

    override protected def writeNonNull(out: Encoder, value: Geometry): Unit =
      out.writeBytes(ByteBuffer.wrap(WKBUtils.write(value)))

    override def withVersion(version: Int): AvroField[Geometry] =
      if (version == 1) { GeometryFieldV1 } else { this }
  }

  /**
   * Deprecated geometry type field, stored as WKT
   */
  case object GeometryFieldV1 extends GeomField {

    override val schema: Schema =
      SchemaBuilder.nullable().`type`(WktLogicalType.addToSchema(SchemaBuilder.builder().bytesType()))

    override protected def parse(bytes: Array[Byte], length: Int): Geometry =
      WKTUtils.read(new String(bytes, 0, length, StandardCharsets.UTF_8))

    override protected def writeNonNull(out: Encoder, value: Geometry): Unit =
      out.writeBytes(ByteBuffer.wrap(WKTUtils.write(value).getBytes(StandardCharsets.UTF_8)))

    override def withVersion(version: Int): AvroField[Geometry] =
      if (version == 1) { this } else { GeometryField }
  }

  /**
   * Byte array type field
   */
  case object BytesField extends AvroField[Array[Byte]] {

    override val schema: Schema = SchemaBuilder.nullable().bytesType()

    override protected def readNonNull(in: Decoder): Array[Byte] = {
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      buffers.put((bb, bytes))
      val value = Array.ofDim[Byte](bb.remaining)
      bb.get(value)
      value
    }

    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()

    override protected def writeNonNull(out: Encoder, value: Array[Byte]): Unit =
      out.writeBytes(ByteBuffer.wrap(value))
  }

  /**
   * List-type field, stored as binary using a custom serialization format
   *
   * @param items type of list items
   */
  class ListOpaqueField(items: ObjectType) extends AvroField[java.util.List[AnyRef]] {

    private val binding = items match {
      case ObjectType.BYTES => "byte[]"
      case t => t.toString.toLowerCase(Locale.US)
    }

    override val schema: Schema =
      SchemaBuilder.nullable().`type`(ListLogicalType.addToSchema(SchemaBuilder.builder().bytesType()))

    override protected def readNonNull(in: Decoder): java.util.List[AnyRef] = {
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      buffers.put((bb, bytes))
      CollectionSerialization.decodeList(bb)
    }

    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()

    override protected def writeNonNull(out: Encoder, value: java.util.List[AnyRef]): Unit =
      out.writeBytes(CollectionSerialization.encodeList(value.asInstanceOf[java.util.List[_]], binding))

    override def withVersion(version: Int): AvroField[util.List[AnyRef]] =
      if (version == SerializationVersions.NativeCollectionVersion) { new ListNativeField(items) } else { this }
  }

  /**
   * List-type field, stored as a native avro array
   *
   * @param items type of list items
   */
  class ListNativeField(items: ObjectType) extends AvroField[java.util.List[AnyRef]] {

    import scala.collection.JavaConverters._

    private val field = AvroField.apply(Seq(items))

    override val schema: Schema = SchemaBuilder.nullable().array().items(field.schema)

    override protected def readNonNull(in: Decoder): java.util.List[AnyRef] = {
      var i = in.readArrayStart()
      val result = new java.util.ArrayList[AnyRef](i.toInt)
      while (i != 0) {
        var j = 0
        while (j < i) {
          result.add(field.read(in))
          j += 1
        }
        i = in.arrayNext()
      }
      result
    }

    override protected def skipNonNull(in: Decoder): Unit = {
      var i = in.skipArray()
      while (i != 0) {
        var j = 0
        while (j < i) {
          field.skip(in)
          j += 1
        }
        i = in.arrayNext()
      }
    }

    override protected def writeNonNull(out: Encoder, value: java.util.List[AnyRef]): Unit = {
      out.writeArrayStart()
      out.setItemCount(value.size)
      value.asScala.foreach { v =>
        out.startItem()
        field.write(out, v)
      }
      out.writeArrayEnd()
    }

    override def withVersion(version: Int): AvroField[java.util.List[AnyRef]] =
      if (version == SerializationVersions.NativeCollectionVersion) { this } else { new ListOpaqueField(items) }
  }

  /**
   * Map type field, stored as binary using a custom serialization format
   *
   * @param keys key type
   * @param values value type
   */
  class MapOpaqueField(keys: ObjectType, values: ObjectType) extends AvroField[java.util.Map[AnyRef, AnyRef]] {

    private val keyBinding = keys match {
      case ObjectType.BYTES => "byte[]"
      case t => t.toString.toLowerCase(Locale.US)
    }

    private val valueBinding = values match {
      case ObjectType.BYTES => "byte[]"
      case t => t.toString.toLowerCase(Locale.US)
    }

    override val schema: Schema =
      SchemaBuilder.nullable().`type`(MapLogicalType.addToSchema(SchemaBuilder.builder().bytesType()))

    override protected def readNonNull(in: Decoder): java.util.Map[AnyRef, AnyRef] = {
      var (bb, bytes) = buffers.getOrElseUpdate((ByteBuffer.allocate(16), Array.empty))
      bb = in.readBytes(bb)
      buffers.put((bb, bytes))
      CollectionSerialization.decodeMap(bb)
    }

    override protected def skipNonNull(in: Decoder): Unit = in.skipBytes()

    override protected def writeNonNull(out: Encoder, value: java.util.Map[AnyRef, AnyRef]): Unit =
      out.writeBytes(CollectionSerialization.encodeMap(value, keyBinding, valueBinding))

    override def withVersion(version: Int): AvroField[java.util.Map[AnyRef, AnyRef]] = {
      if (version == SerializationVersions.NativeCollectionVersion) {
        if (keys == ObjectType.STRING) {
          new MapNativeField(values)
        } else {
          new MapNativeRecordField(keys, values)
        }
      } else {
        this
      }
    }
  }

  /**
   * Map type field, stored as native avro using an array of key-value records
   *
   * @param keys key type
   * @param values value type
   */
  class MapNativeRecordField(keys: ObjectType, values: ObjectType)
      extends AvroField[java.util.Map[AnyRef, AnyRef]] {

    import scala.collection.JavaConverters._

    private val keyField = AvroField.apply(Seq(keys))
    private val valueField = AvroField.apply(Seq(values))

    override val schema: Schema =
      SchemaBuilder.builder().array().items().record("entry").fields()
        .name("key").`type`(keyField.schema).noDefault()
        .name("value").`type`(valueField.schema).noDefault()
        .endRecord()

    override protected def readNonNull(in: Decoder): java.util.Map[AnyRef, AnyRef] = {
      var i = in.readArrayStart()
      val result = new java.util.HashMap[AnyRef, AnyRef](i.toInt)
      while (i != 0) {
        var j = 0
        while (j < i) {
          result.put(keyField.read(in), valueField.read(in))
          j += 1
        }
        i = in.arrayNext()
      }
      result
    }

    override protected def skipNonNull(in: Decoder): Unit = {
      var i = in.skipArray()
      while (i != 0) {
        var j = 0
        while (j < i) {
          keyField.skip(in)
          valueField.skip(in)
          j += 1
        }
        i = in.arrayNext()
      }
    }

    override protected def writeNonNull(out: Encoder, value: java.util.Map[AnyRef, AnyRef]): Unit = {
      out.writeArrayStart()
      out.setItemCount(value.size)
      value.asScala.foreach { v =>
        out.startItem()
        keyField.write(out, v)
        valueField.write(out, v)
      }
      out.writeArrayEnd()
    }

    override def withVersion(version: Int): AvroField[java.util.Map[AnyRef, AnyRef]] = {
      if (version == SerializationVersions.NativeCollectionVersion) { this } else {
        new MapOpaqueField(keys, values)
      }
    }
  }

  /**
   * Map type field, stored as a native avro map (which requires strings for keys)
   *
   * @param values value type
   */
  class MapNativeField(values: ObjectType) extends AvroField[java.util.Map[AnyRef, AnyRef]]{

    import scala.collection.JavaConverters._

    private val field = AvroField.apply(Seq(values))

    override val schema: Schema = SchemaBuilder.nullable().map().values(field.schema)

    override protected def readNonNull(in: Decoder): java.util.Map[AnyRef, AnyRef] = {
      var i = in.readMapStart()
      val result = new java.util.HashMap[AnyRef, AnyRef](i.toInt)
      while (i != 0) {
        var j = 0
        while (j < i) {
          val key = in.readString
          result.put(key, field.read(in))
          j += 1
        }
        i = in.mapNext()
      }
      result
    }

    override protected def skipNonNull(in: Decoder): Unit = {
      var i = in.skipMap()
      while (i != 0) {
        var j = 0
        while (j < i) {
          in.skipString()
          field.skip(in)
          j += 1
        }
        i = in.skipMap()
      }
    }

    override protected def writeNonNull(out: Encoder, value: java.util.Map[AnyRef, AnyRef]): Unit = {
      out.writeMapStart()
      out.setItemCount(value.size)
      value.asScala.foreach { case (k, v) =>
        out.startItem()
        out.writeString(k.asInstanceOf[String])
        field.write(out, v)
      }
      out.writeMapEnd()
    }

    override def withVersion(version: Int): AvroField[java.util.Map[AnyRef, AnyRef]] = {
      if (version == SerializationVersions.NativeCollectionVersion) { this } else {
        new MapOpaqueField(ObjectType.STRING, values)
      }
    }
  }

  object WkbLogicalType extends LogicalType("wkb") with LogicalTypeFactory {

    override def fromSchema(schema: Schema): LogicalType = this

    override def validate(schema: Schema): Unit = {
      super.validate(schema)
      if (schema.getType != Schema.Type.BYTES) {
        throw new IllegalArgumentException("WKB geometries can only be used with an underlying binary type")
      }
    }
  }

  object WktLogicalType extends LogicalType("wkt") with LogicalTypeFactory {

    override def fromSchema(schema: Schema): LogicalType = this

    override def validate(schema: Schema): Unit = {
      super.validate(schema)
      if (schema.getType != Schema.Type.BYTES && schema.getType != Schema.Type.STRING) {
        throw new IllegalArgumentException("WKT geometries can only be used with an underlying binary or string type")
      }
    }
  }

  object ListLogicalType extends LogicalType("list") with LogicalTypeFactory {

    override def fromSchema(schema: Schema): LogicalType = this

    override def validate(schema: Schema): Unit = {
      super.validate(schema)
      if (schema.getType != Schema.Type.BYTES) {
        throw new IllegalArgumentException("List can only be used with an underlying binary type")
      }
    }
  }

  object MapLogicalType extends LogicalType("map") with LogicalTypeFactory {

    override def fromSchema(schema: Schema): LogicalType = this

    override def validate(schema: Schema): Unit = {
      super.validate(schema)
      if (schema.getType != Schema.Type.BYTES) {
        throw new IllegalArgumentException("Map can only be used with an underlying binary type")
      }
    }
  }
}
