private Object deserializePrimitive(Object datum, Schema fileSchema, Schema recordSchema,
      PrimitiveTypeInfo columnType) throws AvroSerdeException {
    switch (columnType.getPrimitiveCategory()){
    case STRING:
      return datum.toString(); // To workaround AvroUTF8
      // This also gets us around the Enum issue since we just take the value
      // and convert it to a string. Yay!
    case BINARY:
      if (recordSchema.getType() == Type.FIXED){
        Fixed fixed = (Fixed) datum;
        return fixed.bytes();
      } else if (recordSchema.getType() == Type.BYTES){
        return AvroSerdeUtils.getBytesFromByteBuffer((ByteBuffer) datum);
      } else {
        throw new AvroSerdeException("Unexpected Avro schema for Binary TypeInfo: " + recordSchema.getType());
      }
    case DECIMAL:
      if (fileSchema == null) {
        throw new AvroSerdeException("File schema is missing for decimal field. Reader schema is " + columnType);
      }

      int scale = 0;
      try {
        scale = AvroSerdeUtils.getIntFromSchema(fileSchema, AvroSerDe.AVRO_PROP_SCALE);
      } catch(Exception ex) {
        throw new AvroSerdeException("Failed to obtain scale value from file schema: " + fileSchema, ex);
      }

      HiveDecimal dec = AvroSerdeUtils.getHiveDecimalFromByteBuffer((ByteBuffer) datum, scale);
      JavaHiveDecimalObjectInspector oi = (JavaHiveDecimalObjectInspector)
          PrimitiveObjectInspectorFactory.getPrimitiveJavaObjectInspector((DecimalTypeInfo)columnType);
      return oi.set(null, dec);
    case CHAR:
      if (fileSchema == null) {
        throw new AvroSerdeException("File schema is missing for char field. Reader schema is " + columnType);
      }

      int maxLength = 0;
      try {
        maxLength = AvroSerdeUtils.getIntFromSchema(fileSchema, AvroSerDe.AVRO_PROP_MAX_LENGTH);
      } catch (Exception ex) {
        throw new AvroSerdeException("Failed to obtain maxLength value for char field from file schema: " + fileSchema, ex);
      }

      String str = datum.toString();
      HiveChar hc = new HiveChar(str, maxLength);
      return hc;
    case VARCHAR:
      if (fileSchema == null) {
        throw new AvroSerdeException("File schema is missing for varchar field. Reader schema is " + columnType);
      }

      maxLength = 0;
      try {
        maxLength = AvroSerdeUtils.getIntFromSchema(fileSchema, AvroSerDe.AVRO_PROP_MAX_LENGTH);
      } catch (Exception ex) {
        throw new AvroSerdeException("Failed to obtain maxLength value for varchar field from file schema: " + fileSchema, ex);
      }

      str = datum.toString();
      HiveVarchar hvc = new HiveVarchar(str, maxLength);
      return hvc;
    case DATE: {
      if (recordSchema.getType() != Type.INT) {
        throw new AvroSerdeException("Unexpected Avro schema for Date TypeInfo: " + recordSchema.getType());
      }

      final boolean skipProlepticConversion;
      if (writerProleptic != null) {
        skipProlepticConversion = writerProleptic;
      } else {
        if (configuration != null) {
          skipProlepticConversion = HiveConf.getBoolVar(
              configuration, HiveConf.ConfVars.HIVE_AVRO_PROLEPTIC_GREGORIAN_DEFAULT);
        } else {
          skipProlepticConversion = HiveConf.ConfVars.HIVE_AVRO_PROLEPTIC_GREGORIAN_DEFAULT.defaultBoolVal;
        }
      }

      return Date.ofEpochMilli(DateWritableV2.daysToMillis(
          skipProlepticConversion ? (Integer) datum : CalendarUtils.convertDateToProleptic((Integer) datum)));
    }
    case TIMESTAMP: {
      if (recordSchema.getType() != Type.LONG) {
        throw new AvroSerdeException(
            "Unexpected Avro schema for Date TypeInfo: " + recordSchema.getType());
      }
      // If a time zone is found in file metadata (property name: writer.time.zone), convert the
      // timestamp to that (writer) time zone in order to emulate time zone agnostic behavior.
      // If not, then the file was written by an older version of hive, so we convert the timestamp
      // to the server's (reader) time zone for backwards compatibility reasons - unless the
      // session level configuration hive.avro.timestamp.skip.conversion is set to true, in which
      // case we assume it was written by a time zone agnostic writer, so we don't convert it.
      final boolean skipUTCConversion;
      if (configuration != null) {
        skipUTCConversion = HiveConf.getBoolVar(
            configuration, HiveConf.ConfVars.HIVE_AVRO_TIMESTAMP_SKIP_CONVERSION);
      } else {
        skipUTCConversion = HiveConf.ConfVars.HIVE_AVRO_TIMESTAMP_SKIP_CONVERSION.defaultBoolVal;
      }
      final boolean legacyConversion;
      if (writerZoneConversionLegacy != null) {
        legacyConversion = writerZoneConversionLegacy;
      } else if (writerTimezone != null) {
        legacyConversion = false;
      } else if (configuration != null) {
        legacyConversion =
            HiveConf.getBoolVar(configuration, HiveConf.ConfVars.HIVE_AVRO_TIMESTAMP_LEGACY_CONVERSION_ENABLED);
      } else {
        legacyConversion = HiveConf.ConfVars.HIVE_AVRO_TIMESTAMP_LEGACY_CONVERSION_ENABLED.defaultBoolVal;
      }
      ZoneId convertToTimeZone;
      if (writerTimezone != null) {
        convertToTimeZone = writerTimezone;
      } else if (skipUTCConversion) {
        convertToTimeZone = ZoneOffset.UTC;
      } else {
        convertToTimeZone = TimeZone.getDefault().toZoneId();
      }
      final boolean skipProlepticConversion;
      if (writerProleptic != null) {
        skipProlepticConversion = writerProleptic;
      } else {
        if (configuration != null) {
          skipProlepticConversion = HiveConf.getBoolVar(
              configuration, HiveConf.ConfVars.HIVE_AVRO_PROLEPTIC_GREGORIAN_DEFAULT);
        } else {
          skipProlepticConversion = HiveConf.ConfVars.HIVE_AVRO_PROLEPTIC_GREGORIAN_DEFAULT.defaultBoolVal;
        }
      }
      Timestamp timestamp = TimestampTZUtil.convertTimestampToZone(
          Timestamp.ofEpochMilli((Long) datum), ZoneOffset.UTC, convertToTimeZone, legacyConversion);
      if (!skipProlepticConversion) {
        timestamp = Timestamp.ofEpochMilli(
            CalendarUtils.convertTimeToProleptic(timestamp.toEpochMilli()));
      }
      return timestamp;
    }
    default:
      return datum;
    }
  }
Deserialize an Avro record, recursing into its component fields and
deserializing them as well.  Fields of the record are matched by name
against fields in the Hive row.

Because Avro has some data types that Hive does not, these are converted
during deserialization to types Hive will work with.

@param columnNames List of columns Hive is expecting from record.
@param columnTypes List of column types matched by index to names
@param writable Instance of GenericAvroWritable to deserialize
@param readerSchema Schema of the writable to deserialize
@return A list of objects suitable for Hive to work with further
@throws AvroSerdeException For any exception during deserialization
serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java