public Object deserialize(List<String> columnNames, List<TypeInfo> columnTypes,
                            Writable writable, Schema readerSchema) throws AvroSerdeException {
    if(!(writable instanceof AvroGenericRecordWritable)) {
      throw new AvroSerdeException("Expecting a AvroGenericRecordWritable");
    }

    if(row == null || row.size() != columnNames.size()) {
      row = new ArrayList<Object>(columnNames.size());
    } else {
      row.clear();
    }

    AvroGenericRecordWritable recordWritable = (AvroGenericRecordWritable) writable;
    GenericRecord r = recordWritable.getRecord();
    Schema fileSchema = recordWritable.getFileSchema();
    writerTimezone = recordWritable.getWriterTimezone();
    writerProleptic = recordWritable.getWriterProleptic();
    writerZoneConversionLegacy = recordWritable.getWriterZoneConversionLegacy(); 

    UID recordReaderId = recordWritable.getRecordReaderID();
    //If the record reader (from which the record is originated) is already seen and valid,
    //no need to re-encode the record.
    if(!noEncodingNeeded.contains(recordReaderId)) {
      SchemaReEncoder reEncoder = null;
      //Check if the record record is already encoded once. If it does
      //reuse the encoder.
      if(reEncoderCache.containsKey(recordReaderId)) {
        reEncoder = reEncoderCache.get(recordReaderId); //Reuse the re-encoder
      } else if (!r.getSchema().equals(readerSchema)) { //Evolved schema?
        //Create and store new encoder in the map for re-use
        reEncoder = new SchemaReEncoder(r.getSchema(), readerSchema);
        reEncoderCache.put(recordReaderId, reEncoder);
      } else{
        LOG.debug("Adding new valid RRID :" +  recordReaderId);
        noEncodingNeeded.add(recordReaderId);
      }
      if(reEncoder != null) {
        if (!warnedOnce) {
          LOG.warn("Received different schemas.  Have to re-encode: " +
              r.getSchema().toString(false) + "\nSIZE" + reEncoderCache + " ID " + recordReaderId);
          warnedOnce = true;
        }
        r = reEncoder.reencode(r);
      }
    }

    workerBase(row, fileSchema, columnNames, columnTypes, r);
    return row;
  }
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