/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.avro;

import org.apache.avro.Conversions;
import org.apache.avro.JsonProperties;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Array;
import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.specific.SpecificRecord;
import org.apache.avro.util.Utf8;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.SchemaIdentifier;
import org.apache.nifi.serialization.record.StandardSchemaIdentifier;
import org.apache.nifi.serialization.record.type.ArrayDataType;
import org.apache.nifi.serialization.record.type.ChoiceDataType;
import org.apache.nifi.serialization.record.type.DecimalDataType;
import org.apache.nifi.serialization.record.type.MapDataType;
import org.apache.nifi.serialization.record.type.RecordDataType;
import org.apache.nifi.serialization.record.util.DataTypeUtils;
import org.apache.nifi.serialization.record.util.IllegalTypeConversionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.sql.Blob;
import java.sql.Time;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

public class AvroTypeUtil {
    private static final Logger logger = LoggerFactory.getLogger(AvroTypeUtil.class);
    public static final String AVRO_SCHEMA_FORMAT = "avro";

    private static final String LOGICAL_TYPE_DATE = "date";
    private static final String LOGICAL_TYPE_TIME_MILLIS = "time-millis";
    private static final String LOGICAL_TYPE_TIME_MICROS = "time-micros";
    private static final String LOGICAL_TYPE_TIMESTAMP_MILLIS = "timestamp-millis";
    private static final String LOGICAL_TYPE_TIMESTAMP_MICROS = "timestamp-micros";
    private static final String LOGICAL_TYPE_DECIMAL = "decimal";


    public static Schema extractAvroSchema(final RecordSchema recordSchema) {
        if (recordSchema == null) {
            throw new IllegalArgumentException("RecordSchema cannot be null");
        }

        final Optional<String> schemaFormatOption = recordSchema.getSchemaFormat();
        if (!schemaFormatOption.isPresent()) {
            return buildAvroSchema(recordSchema);
        }

        final String schemaFormat = schemaFormatOption.get();
        if (!schemaFormat.equals(AVRO_SCHEMA_FORMAT)) {
            return buildAvroSchema(recordSchema);
        }

        final Optional<String> textOption = recordSchema.getSchemaText();
        if (!textOption.isPresent()) {
            return buildAvroSchema(recordSchema);
        }

        final String text = textOption.get();
        return new Schema.Parser().parse(text);
    }

    private static Schema buildAvroSchema(final RecordSchema recordSchema) {
        final List<Field> avroFields = new ArrayList<>(recordSchema.getFieldCount());
        for (final RecordField recordField : recordSchema.getFields()) {
            avroFields.add(buildAvroField(recordField, ""));
        }

        final Schema avroSchema = Schema.createRecord("nifiRecord", null, "org.apache.nifi", false, avroFields);
        return avroSchema;
    }

    private static Field buildAvroField(final RecordField recordField, String fieldNamePrefix) {
        final Schema schema = buildAvroSchema(recordField.getDataType(), recordField.getFieldName(), fieldNamePrefix, recordField.isNullable());

        final Field field;
        final String recordFieldName = recordField.getFieldName();
        if (isValidAvroFieldName(recordFieldName)) {
            field = new Field(recordField.getFieldName(), schema, null, recordField.getDefaultValue());
        } else {
            final String validName = createValidAvroFieldName(recordField.getFieldName());
            field = new Field(validName, schema, null, recordField.getDefaultValue());
            field.addAlias(recordField.getFieldName());
        }

        for (final String alias : recordField.getAliases()) {
            field.addAlias(alias);
        }

        return field;
    }

    private static boolean isValidAvroFieldName(final String fieldName) {
        // Avro field names must match the following criteria:
        // 1. Must be non-empty
        // 2. Must begin with a letter or an underscore
        // 3. Must consist only of letters, underscores, and numbers.
        if (fieldName.isEmpty()) {
            return false;
        }

        final char firstChar = fieldName.charAt(0);
        if (firstChar != '_' && !Character.isLetter(firstChar)) {
            return false;
        }

        for (int i=1; i < fieldName.length(); i++) {
            final char c = fieldName.charAt(i);
            if (c != '_' && !Character.isLetterOrDigit(c)) {
                return false;
            }
        }

        return true;
    }

    private static String createValidAvroFieldName(final String fieldName) {
        if (fieldName.isEmpty()) {
            return "UNNAMED_FIELD";
        }

        final StringBuilder sb = new StringBuilder();

        final char firstChar = fieldName.charAt(0);
        if (firstChar == '_' || Character.isLetter(firstChar)) {
            sb.append(firstChar);
        } else {
            sb.append("_");
        }

        for (int i=1; i < fieldName.length(); i++) {
            final char c = fieldName.charAt(i);
            if (c == '_' || Character.isLetterOrDigit(c)) {
                sb.append(c);
            } else {
                sb.append("_");
            }
        }

        return sb.toString();
    }

    private static Schema buildAvroSchema(final DataType dataType, final String fieldName, String fieldNamePrefix, final boolean nullable) {
        final Schema schema;

        switch (dataType.getFieldType()) {
            case ARRAY:
                final ArrayDataType arrayDataType = (ArrayDataType) dataType;
                final DataType elementDataType = arrayDataType.getElementType();
                if (RecordFieldType.BYTE.equals(elementDataType.getFieldType())) {
                    schema = Schema.create(Type.BYTES);
                } else {
                    final Schema elementType = buildAvroSchema(elementDataType, fieldName, fieldNamePrefix, false);
                    schema = Schema.createArray(elementType);
                }
                break;
            case BIGINT:
                schema = Schema.create(Type.STRING);
                break;
            case BOOLEAN:
                schema = Schema.create(Type.BOOLEAN);
                break;
            case BYTE:
                schema = Schema.create(Type.INT);
                break;
            case CHAR:
                schema = Schema.create(Type.STRING);
                break;
            case CHOICE:
                final ChoiceDataType choiceDataType = (ChoiceDataType) dataType;
                final List<DataType> options = choiceDataType.getPossibleSubTypes();

                // We need to keep track of which types have been added to the union, because if we have
                // two elements in the UNION with the same type, it will fail - even if the logical type is
                // different. So if we have an int and a logical type date (which also has a 'concrete type' of int)
                // then an Exception will be thrown when we try to create the union. To avoid this, we just keep track
                // of the Types and avoid adding it in such a case.
                final List<Schema> unionTypes = new ArrayList<>(options.size());
                final Set<Type> typesAdded = new HashSet<>();

                int optionCounter = 1;
                for (final DataType option : options) {
                    final Schema optionSchema = buildAvroSchema(option, fieldName, fieldNamePrefix, false);
                    if (!typesAdded.contains(optionSchema.getType())) {
                        unionTypes.add(optionSchema);
                        typesAdded.add(optionSchema.getType());
                    } else if (Type.RECORD.equals(optionSchema.getType()) && !unionTypes.contains(optionSchema)) {
                        final Schema indexedOptionSchema = buildAvroSchema(option, fieldName + ++optionCounter, fieldNamePrefix, false);
                        unionTypes.add(indexedOptionSchema);
                    }
                }

                schema = Schema.createUnion(unionTypes);
                break;
            case DATE:
                schema = Schema.create(Type.INT);
                LogicalTypes.date().addToSchema(schema);
                break;
            case DOUBLE:
                schema = Schema.create(Type.DOUBLE);
                break;
            case FLOAT:
                schema = Schema.create(Type.FLOAT);
                break;
            case INT:
                schema = Schema.create(Type.INT);
                break;
            case LONG:
                schema = Schema.create(Type.LONG);
                break;
            case DECIMAL:
                final DecimalDataType decimalDataType = (DecimalDataType) dataType;
                schema = Schema.create(Type.BYTES);
                LogicalTypes.decimal(decimalDataType.getPrecision(), decimalDataType.getScale()).addToSchema(schema);
                break;
            case MAP:
                schema = Schema.createMap(buildAvroSchema(((MapDataType) dataType).getValueType(), fieldName, fieldNamePrefix, false));
                break;
            case RECORD:
                final RecordDataType recordDataType = (RecordDataType) dataType;
                final RecordSchema childSchema = recordDataType.getChildSchema();

                final List<Field> childFields = new ArrayList<>(childSchema.getFieldCount());
                for (final RecordField field : childSchema.getFields()) {
                    String childFieldNamePrefix = StringUtils.isBlank(fieldNamePrefix) ? fieldName + "_" : fieldNamePrefix + fieldName + "_";
                    childFields.add(buildAvroField(field, childFieldNamePrefix));
                }

                schema = Schema.createRecord(fieldNamePrefix + fieldName + "Type", null, "org.apache.nifi", false, childFields);
                break;
            case SHORT:
                schema = Schema.create(Type.INT);
                break;
            case STRING:
                schema = Schema.create(Type.STRING);
                break;
            case TIME:
                schema = Schema.create(Type.INT);
                LogicalTypes.timeMillis().addToSchema(schema);
                break;
            case TIMESTAMP:
                schema = Schema.create(Type.LONG);
                LogicalTypes.timestampMillis().addToSchema(schema);
                break;
            default:
                return null;
        }

        if (nullable) {
            return nullable(schema);
        } else {
            return schema;
        }
    }

    private static Schema nullable(final Schema schema) {
        if (schema.getType() == Type.UNION) {
            final List<Schema> unionTypes = new ArrayList<>(schema.getTypes());
            final Schema nullSchema = Schema.create(Type.NULL);
            if (unionTypes.contains(nullSchema)) {
                return schema;
            }

            unionTypes.add(nullSchema);
            return Schema.createUnion(unionTypes);
        }

        return Schema.createUnion(Schema.create(Type.NULL), schema);
    }

    /**
     * Returns a DataType for the given Avro Schema
     *
     * @param avroSchema the Avro Schema to convert
     * @return a Data Type that corresponds to the given Avro Schema
     */
    public static DataType determineDataType(final Schema avroSchema) {
        return determineDataType(avroSchema, new HashMap<>());
    }

    public static DataType determineDataType(final Schema avroSchema, Map<String, DataType> knownRecordTypes) {

        if (knownRecordTypes == null) {
            throw new IllegalArgumentException("'knownRecordTypes' cannot be null.");
        }

        final Type avroType = avroSchema.getType();

        final LogicalType logicalType = avroSchema.getLogicalType();
        if (logicalType != null) {
            final String logicalTypeName = logicalType.getName();
            switch (logicalTypeName) {
                case LOGICAL_TYPE_DATE:
                    return RecordFieldType.DATE.getDataType();
                case LOGICAL_TYPE_TIME_MILLIS:
                case LOGICAL_TYPE_TIME_MICROS:
                    return RecordFieldType.TIME.getDataType();
                case LOGICAL_TYPE_TIMESTAMP_MILLIS:
                case LOGICAL_TYPE_TIMESTAMP_MICROS:
                    return RecordFieldType.TIMESTAMP.getDataType();
                case LOGICAL_TYPE_DECIMAL:
                    final LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType;
                    return RecordFieldType.DECIMAL.getDecimalDataType(decimal.getPrecision(), decimal.getScale());
            }
        }

        switch (avroType) {
            case ARRAY:
                final DataType elementType = determineDataType(avroSchema.getElementType(), knownRecordTypes);
                final boolean elementsNullable = isNullable(avroSchema.getElementType());
                return RecordFieldType.ARRAY.getArrayDataType(elementType, elementsNullable);
            case BYTES:
            case FIXED:
                return RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType());
            case BOOLEAN:
                return RecordFieldType.BOOLEAN.getDataType();
            case DOUBLE:
                return RecordFieldType.DOUBLE.getDataType();
            case ENUM:
                return RecordFieldType.ENUM.getEnumDataType(avroSchema.getEnumSymbols());
            case STRING:
                return RecordFieldType.STRING.getDataType();
            case FLOAT:
                return RecordFieldType.FLOAT.getDataType();
            case INT:
                return RecordFieldType.INT.getDataType();
            case LONG:
                return RecordFieldType.LONG.getDataType();
            case RECORD: {
                String schemaFullName = avroSchema.getNamespace() + "." + avroSchema.getName();

                if (knownRecordTypes.containsKey(schemaFullName)) {
                    return knownRecordTypes.get(schemaFullName);
                } else {
                    SimpleRecordSchema recordSchema = new SimpleRecordSchema(SchemaIdentifier.EMPTY);
                    recordSchema.setSchemaName(avroSchema.getName());
                    recordSchema.setSchemaNamespace(avroSchema.getNamespace());
                    DataType recordSchemaType = RecordFieldType.RECORD.getRecordDataType(recordSchema);
                    knownRecordTypes.put(schemaFullName, recordSchemaType);

                    final List<Field> avroFields = avroSchema.getFields();
                    final List<RecordField> recordFields = new ArrayList<>(avroFields.size());

                    for (final Field field : avroFields) {
                        final String fieldName = field.name();
                        final Schema fieldSchema = field.schema();
                        final DataType fieldType = determineDataType(fieldSchema, knownRecordTypes);
                        final boolean nullable = isNullable(fieldSchema);
                        addFieldToList(recordFields, field, fieldName, fieldSchema, fieldType, nullable);
                    }

                    recordSchema.setFields(recordFields);
                    return recordSchemaType;
                }
            }
            case NULL:
                return RecordFieldType.STRING.getDataType();
            case MAP:
                final Schema valueSchema = avroSchema.getValueType();
                final DataType valueType = determineDataType(valueSchema, knownRecordTypes);
                final boolean valuesNullable = isNullable(valueSchema);
                return RecordFieldType.MAP.getMapDataType(valueType, valuesNullable);
            case UNION: {
                final List<Schema> nonNullSubSchemas = getNonNullSubSchemas(avroSchema);

                if (nonNullSubSchemas.size() == 1) {
                    return determineDataType(nonNullSubSchemas.get(0), knownRecordTypes);
                }

                final List<DataType> possibleChildTypes = new ArrayList<>(nonNullSubSchemas.size());
                for (final Schema subSchema : nonNullSubSchemas) {
                    final DataType childDataType = determineDataType(subSchema, knownRecordTypes);
                    possibleChildTypes.add(childDataType);
                }

                return RecordFieldType.CHOICE.getChoiceDataType(possibleChildTypes);
            }
        }

        return null;
    }

    private static List<Schema> getNonNullSubSchemas(final Schema avroSchema) {
        final List<Schema> unionFieldSchemas = avroSchema.getTypes();
        if (unionFieldSchemas == null) {
            return Collections.emptyList();
        }

        final List<Schema> nonNullTypes = new ArrayList<>(unionFieldSchemas.size());
        for (final Schema fieldSchema : unionFieldSchemas) {
            if (fieldSchema.getType() != Type.NULL) {
                nonNullTypes.add(fieldSchema);
            }
        }

        return nonNullTypes;
    }

    public static RecordSchema createSchema(final Schema avroSchema) {
        return createSchema(avroSchema, true);
    }

    public static RecordSchema createSchema(final Schema avroSchema, final boolean includeText) {
        if (avroSchema == null) {
            throw new IllegalArgumentException("Avro Schema cannot be null");
        }

        SchemaIdentifier identifier = new StandardSchemaIdentifier.Builder().name(avroSchema.getName()).build();
        return createSchema(avroSchema, includeText ? avroSchema.toString() : null, identifier);
    }

    /**
     * Converts an Avro Schema to a RecordSchema
     *
     * @param avroSchema the Avro Schema to convert
     * @param schemaText the textual representation of the schema
     * @param schemaId the identifier of the schema
     * @return the Corresponding Record Schema
     */
    public static RecordSchema createSchema(final Schema avroSchema, final String schemaText, final SchemaIdentifier schemaId) {
        if (avroSchema == null) {
            throw new IllegalArgumentException("Avro Schema cannot be null");
        }

        final String schemaFullName = avroSchema.getNamespace() + "." + avroSchema.getName();
        final SimpleRecordSchema recordSchema = schemaText == null ? new SimpleRecordSchema(schemaId) : new SimpleRecordSchema(schemaText, AVRO_SCHEMA_FORMAT, schemaId);
        recordSchema.setSchemaName(avroSchema.getName());
        recordSchema.setSchemaNamespace(avroSchema.getNamespace());
        final DataType recordSchemaType = RecordFieldType.RECORD.getRecordDataType(recordSchema);
        final Map<String, DataType> knownRecords = new HashMap<>();
        knownRecords.put(schemaFullName, recordSchemaType);

        final List<RecordField> recordFields = new ArrayList<>(avroSchema.getFields().size());
        for (final Field field : avroSchema.getFields()) {
            final String fieldName = field.name();
            final Schema fieldSchema = field.schema();
            final DataType dataType = AvroTypeUtil.determineDataType(fieldSchema, knownRecords);
            final boolean nullable = isNullable(fieldSchema);
            addFieldToList(recordFields, field, fieldName, fieldSchema, dataType, nullable);
        }

        recordSchema.setFields(recordFields);
        return recordSchema;
    }

    public static boolean isNullable(final Schema schema) {
        final Type schemaType = schema.getType();
        if (schemaType == Type.UNION) {
            for (final Schema unionSchema : schema.getTypes()) {
                if (isNullable(unionSchema)) {
                    return true;
                }
            }
        }

        return schemaType == Type.NULL;
    }

    public static Object[] convertByteArray(final byte[] bytes) {
        final Object[] array = new Object[bytes.length];
        for (int i = 0; i < bytes.length; i++) {
            array[i] = Byte.valueOf(bytes[i]);
        }
        return array;
    }

    public static ByteBuffer convertByteArray(final Object[] bytes) {
        final ByteBuffer bb = ByteBuffer.allocate(bytes.length);
        for (final Object o : bytes) {
            if (o instanceof Byte) {
                bb.put(((Byte) o).byteValue());
            } else {
                throw new IllegalTypeConversionException("Cannot convert value " + bytes + " of type " + bytes.getClass() + " to ByteBuffer");
            }
        }
        bb.flip();
        return bb;
    }

    /**
     * Method that attempts to map a record field into a provided schema
     * @param avroSchema - Schema to map into
     * @param recordField - The field of the record to be mapped
     * @return Pair with the LHS being the field name and RHS being the mapped field from the schema
     */
    protected static Pair<String, Field> lookupField(final Schema avroSchema, final RecordField recordField) {
        String fieldName = recordField.getFieldName();

        // Attempt to locate the field as is in a true 1:1 mapping with the same name
        Field field = avroSchema.getField(fieldName);
        if (field == null) {
            // No straight mapping was found, so check the aliases to see if it can be mapped
            for (final String alias : recordField.getAliases()) {
                field = avroSchema.getField(alias);
                if (field != null) {
                    fieldName = alias;
                    break;
                }
            }
        }

        if (field == null) {
            for (final Field childField : avroSchema.getFields()) {
                final Set<String> aliases = childField.aliases();
                if (aliases.isEmpty()) {
                    continue;
                }

                if (aliases.contains(fieldName)) {
                    field = childField;
                    break;
                }

                for (final String alias : recordField.getAliases()) {
                    if (aliases.contains(alias)) {
                        field = childField;
                        fieldName = alias;
                        break;
                    }
                }
            }
        }

        return new ImmutablePair<>(fieldName, field);
    }

    public static GenericRecord createAvroRecord(final Record record, final Schema avroSchema) throws IOException {
        return createAvroRecord(record, avroSchema, StandardCharsets.UTF_8);
    }

    public static GenericRecord createAvroRecord(final Record record, final Schema avroSchema, final Charset charset) throws IOException {
        final GenericRecord rec = new GenericData.Record(avroSchema);
        final RecordSchema recordSchema = record.getSchema();

        final Map<String, Object> recordValues = record.toMap();
        for (final Map.Entry<String, Object> entry : recordValues.entrySet()) {
            final Object rawValue = entry.getValue();
            if (rawValue == null) {
                continue;
            }

            final String rawFieldName = entry.getKey();
            final Optional<RecordField> optionalRecordField = recordSchema.getField(rawFieldName);
            if (!optionalRecordField.isPresent()) {
                continue;
            }

            final RecordField recordField = optionalRecordField.get();

            final Field field;
            final Field avroField = avroSchema.getField(rawFieldName);
            if (avroField == null) {
                final Pair<String, Field> fieldPair = lookupField(avroSchema, recordField);
                field = fieldPair.getRight();

                if (field == null) {
                    continue;
                }
            } else {
                field = avroField;
            }

            final String fieldName = field.name();
            final Object converted = convertToAvroObject(rawValue, field.schema(), fieldName, charset);
            rec.put(fieldName, converted);
        }

        return rec;
    }

    /**
     * Convert a raw value to an Avro object to serialize in Avro type system, using the provided character set when necessary.
     * The counter-part method which reads an Avro object back to a raw value is {@link #normalizeValue(Object, Schema, String)}.
     */
    public static Object convertToAvroObject(final Object rawValue, final Schema fieldSchema) {
        return convertToAvroObject(rawValue, fieldSchema, StandardCharsets.UTF_8);
    }

    /**
     * Convert a raw value to an Avro object to serialize in Avro type system, using the provided character set when necessary.
     * The counter-part method which reads an Avro object back to a raw value is {@link #normalizeValue(Object, Schema, String)}.
     */
    public static Object convertToAvroObject(final Object rawValue, final Schema fieldSchema, final Charset charset) {
        return convertToAvroObject(rawValue, fieldSchema, fieldSchema.getName(), charset);
    }

    /**
     * Adds fields to <tt>recordFields</tt> list.
     * @param recordFields - record fields are added to this list.
     * @param field - the field
     * @param fieldName - field name
     * @param fieldSchema - field schema
     * @param dataType -  data type
     * @param nullable - is nullable?
     */
    private static void addFieldToList(final List<RecordField> recordFields, final Field field, final String fieldName,
            final Schema fieldSchema, final DataType dataType, final boolean nullable) {
        if (field.defaultVal() == JsonProperties.NULL_VALUE) {
            recordFields.add(new RecordField(fieldName, dataType, field.aliases(), nullable));
        } else {
            Object defaultValue = field.defaultVal();
            if (defaultValue != null && fieldSchema.getType() == Schema.Type.ARRAY && !DataTypeUtils.isArrayTypeCompatible(defaultValue, ((ArrayDataType) dataType).getElementType())) {
                defaultValue = defaultValue instanceof List ? ((List<?>) defaultValue).toArray() : new Object[0];
            }
            recordFields.add(new RecordField(fieldName, dataType, defaultValue, field.aliases(), nullable));
        }
    }

    private static Long getLongFromTimestamp(final Object rawValue, final Schema fieldSchema, final String fieldName) {
        final String format = AvroTypeUtil.determineDataType(fieldSchema).getFormat();
        Timestamp t = DataTypeUtils.toTimestamp(rawValue, () -> DataTypeUtils.getDateFormat(format), fieldName);
        return t.getTime();
    }

    @SuppressWarnings("unchecked")
    private static Object convertToAvroObject(final Object rawValue, final Schema fieldSchema, final String fieldName, final Charset charset) {
        if (rawValue == null) {
            return null;
        }

        switch (fieldSchema.getType()) {
            case INT: {
                final LogicalType logicalType = fieldSchema.getLogicalType();
                if (logicalType == null) {
                    return DataTypeUtils.toInteger(rawValue, fieldName);
                }

                if (LOGICAL_TYPE_DATE.equals(logicalType.getName())) {
                    final String format = AvroTypeUtil.determineDataType(fieldSchema).getFormat();
                    final java.sql.Date date = DataTypeUtils.toDate(rawValue, () -> DataTypeUtils.getDateFormat(format), fieldName);
                    return (int) ChronoUnit.DAYS.between(Instant.EPOCH, Instant.ofEpochMilli(date.getTime()));
                } else if (LOGICAL_TYPE_TIME_MILLIS.equals(logicalType.getName())) {
                    final String format = AvroTypeUtil.determineDataType(fieldSchema).getFormat();
                    final Time time = DataTypeUtils.toTime(rawValue, () -> DataTypeUtils.getDateFormat(format), fieldName);
                    final Date date = new Date(time.getTime());
                    final Duration duration = Duration.between(date.toInstant().truncatedTo(ChronoUnit.DAYS), date.toInstant());
                    final long millisSinceMidnight = duration.toMillis();
                    return (int) millisSinceMidnight;
                }

                return DataTypeUtils.toInteger(rawValue, fieldName);
            }
            case LONG: {
                final LogicalType logicalType = fieldSchema.getLogicalType();
                if (logicalType == null) {
                    return DataTypeUtils.toLong(rawValue, fieldName);
                }

                if (LOGICAL_TYPE_TIME_MICROS.equals(logicalType.getName())) {
                    final long longValue = getLongFromTimestamp(rawValue, fieldSchema, fieldName);
                    final Date date = new Date(longValue);
                    final Duration duration = Duration.between(date.toInstant().truncatedTo(ChronoUnit.DAYS), date.toInstant());
                    return duration.toMillis() * 1000L;
                } else if (LOGICAL_TYPE_TIMESTAMP_MILLIS.equals(logicalType.getName())) {
                    return getLongFromTimestamp(rawValue, fieldSchema, fieldName);
                } else if (LOGICAL_TYPE_TIMESTAMP_MICROS.equals(logicalType.getName())) {
                    return getLongFromTimestamp(rawValue, fieldSchema, fieldName) * 1000L;
                }

                return DataTypeUtils.toLong(rawValue, fieldName);
            }
            case BYTES:
            case FIXED:
                final LogicalType logicalType = fieldSchema.getLogicalType();
                if (logicalType != null && LOGICAL_TYPE_DECIMAL.equals(logicalType.getName())) {
                    final LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) logicalType;
                    final BigDecimal rawDecimal;
                    if (rawValue instanceof BigDecimal) {
                        rawDecimal = (BigDecimal) rawValue;

                    } else if (rawValue instanceof Double) {
                        rawDecimal = BigDecimal.valueOf((Double) rawValue);

                    } else if (rawValue instanceof String) {
                        rawDecimal = new BigDecimal((String) rawValue);

                    } else if (rawValue instanceof Integer) {
                        rawDecimal = new BigDecimal((Integer) rawValue);

                    } else if (rawValue instanceof Long) {
                        rawDecimal = new BigDecimal((Long) rawValue);

                    } else {
                        throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a logical decimal");
                    }
                    // If the desired scale is different than this value's coerce scale.
                    final int desiredScale = decimalType.getScale();
                    final BigDecimal decimal = rawDecimal.scale() == desiredScale
                            ? rawDecimal : rawDecimal.setScale(desiredScale, BigDecimal.ROUND_HALF_UP);
                    return fieldSchema.getType() == Type.BYTES
                        ? new Conversions.DecimalConversion().toBytes(decimal, fieldSchema, logicalType) //return GenericByte
                        : new Conversions.DecimalConversion().toFixed(decimal, fieldSchema, logicalType); //return GenericFixed
                }
                if (rawValue instanceof byte[]) {
                    return ByteBuffer.wrap((byte[]) rawValue);
                }
                if (rawValue instanceof String) {
                    return ByteBuffer.wrap(((String) rawValue).getBytes(charset));
                }
                if (rawValue instanceof Object[]) {
                    if (fieldSchema.getType() == Type.FIXED && "INT96".equals(fieldSchema.getName())) {
                        Object[] rawObjects = (Object[]) rawValue;
                        byte[] rawBytes = new byte[rawObjects.length];
                        for (int elementIndex = 0; elementIndex < rawObjects.length; elementIndex++) {
                            rawBytes[elementIndex] = (Byte) rawObjects[elementIndex];
                        }

                        return new GenericData.Fixed(fieldSchema, rawBytes);
                    } else {
                        return AvroTypeUtil.convertByteArray((Object[]) rawValue);
                    }
                }
                try {
                    if (rawValue instanceof Blob) {
                        Blob blob = (Blob) rawValue;
                        return ByteBuffer.wrap(IOUtils.toByteArray(blob.getBinaryStream()));
                    } else {
                        throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a ByteBuffer");
                    }
                } catch (IllegalTypeConversionException itce) {
                    throw itce;
                } catch (Exception e) {
                    throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a ByteBuffer", e);
                }
            case MAP:
                if (rawValue instanceof Record) {
                    final Record recordValue = (Record) rawValue;
                    final Map<String, Object> map = new HashMap<>();
                    for (final RecordField recordField : recordValue.getSchema().getFields()) {
                        final Object v = recordValue.getValue(recordField);
                        if (v != null) {
                            map.put(recordField.getFieldName(), convertToAvroObject(v, fieldSchema.getValueType(), fieldName + "[" + recordField.getFieldName() + "]", charset));
                        }
                    }

                    return map;
                } else if (rawValue instanceof Map) {
                    final Map<String, Object> objectMap = (Map<String, Object>) rawValue;
                    final Map<String, Object> map = new HashMap<>(objectMap.size());
                    for (final String s : objectMap.keySet()) {
                        final Object converted = convertToAvroObject(objectMap.get(s), fieldSchema.getValueType(), fieldName + "[" + s + "]", charset);
                        map.put(s, converted);
                    }
                    return map;
                } else {
                    throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a Map");
                }
            case RECORD:
                final GenericData.Record avroRecord = new GenericData.Record(fieldSchema);

                final Set<Map.Entry<String, Object>> entries;
                if (rawValue instanceof Map) {
                    final Map<String, Object> map = (Map<String, Object>) rawValue;
                    entries = map.entrySet();
                } else if (rawValue instanceof Record) {
                    entries = new HashSet<>();
                    final Record record = (Record) rawValue;
                    record.getSchema().getFields().forEach(field -> entries.add(new AbstractMap.SimpleEntry<>(field.getFieldName(), record.getValue(field))));
                } else {
                    throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a Record");
                }
                for (final Map.Entry<String, Object> e : entries) {
                    final Object recordFieldValue = e.getValue();
                    final String recordFieldName = e.getKey();

                    final Field field = fieldSchema.getField(recordFieldName);
                    if (field == null) {
                        continue;
                    }

                    final Object converted = convertToAvroObject(recordFieldValue, field.schema(), fieldName + "/" + recordFieldName, charset);
                    avroRecord.put(recordFieldName, converted);
                }
                return avroRecord;
            case UNION:
                return convertUnionFieldValue(rawValue, fieldSchema, schema -> convertToAvroObject(rawValue, schema, fieldName, charset), fieldName);
            case ARRAY:
                final Object[] objectArray;
                if (rawValue instanceof List) {
                    objectArray = ((List) rawValue).toArray();
                } else if (rawValue instanceof Object[]) {
                    objectArray = (Object[]) rawValue;
                } else {
                    throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to an Array");
                }
                final List<Object> list = new ArrayList<>(objectArray.length);
                int i = 0;
                for (final Object o : objectArray) {
                    final Object converted = convertToAvroObject(o, fieldSchema.getElementType(), fieldName + "[" + i + "]", charset);
                    list.add(converted);
                    i++;
                }
                return list;
            case BOOLEAN:
                return DataTypeUtils.toBoolean(rawValue, fieldName);
            case DOUBLE:
                return DataTypeUtils.toDouble(rawValue, fieldName);
            case FLOAT:
                return DataTypeUtils.toFloat(rawValue, fieldName);
            case NULL:
                return null;
            case ENUM:
                List<String> enums = fieldSchema.getEnumSymbols();
                if(enums != null && enums.contains(rawValue)) {
                    return new GenericData.EnumSymbol(fieldSchema, rawValue);
                } else {
                    throw new IllegalTypeConversionException(rawValue + " is not a possible value of the ENUM" + enums + ".");
                }
            case STRING:
                if (rawValue instanceof String) {
                    return rawValue;
                }

                return DataTypeUtils.toString(rawValue, (String) null, charset);
        }

        return rawValue;
    }

    public static Map<String, Object> convertAvroRecordToMap(final GenericRecord avroRecord, final RecordSchema recordSchema) {
        return convertAvroRecordToMap(avroRecord, recordSchema, StandardCharsets.UTF_8);
    }

    public static Map<String, Object> convertAvroRecordToMap(final GenericRecord avroRecord, final RecordSchema recordSchema, final Charset charset) {
        final Map<String, Object> values = new HashMap<>(recordSchema.getFieldCount());

        for (final RecordField recordField : recordSchema.getFields()) {

            Object value = avroRecord.get(recordField.getFieldName());
            if (value == null) {
                for (final String alias : recordField.getAliases()) {
                    value = avroRecord.get(alias);
                    if (value != null) {
                        break;
                    }
                }
            }

            final String fieldName = recordField.getFieldName();
            try {
            final Field avroField = avroRecord.getSchema().getField(fieldName);
            if (avroField == null) {
                values.put(fieldName, null);
                continue;
            }

            final Schema fieldSchema = avroField.schema();
            final Object rawValue = normalizeValue(value, fieldSchema, fieldName);

            final DataType desiredType = recordField.getDataType();
            final Object coercedValue = DataTypeUtils.convertType(rawValue, desiredType, fieldName, charset);

            values.put(fieldName, coercedValue);
            } catch (Exception ex) {
                logger.debug("fail to convert field " + fieldName, ex );
                throw ex;
            }
        }

        return values;
    }

    /**
     * Convert value of a nullable union field.
     * @param originalValue original value
     * @param fieldSchema the union field schema
     * @param conversion the conversion function which takes a non-null field schema within the union field and returns a converted value
     * @return a converted value
     */
    private static Object convertUnionFieldValue(final Object originalValue, final Schema fieldSchema, final Function<Schema, Object> conversion, final String fieldName) {
        boolean foundNonNull = false;

        // It is an extremely common case to have a UNION type because a field can be NULL or some other type. In this situation,
        // we will have two possible types, and one of them will be null. When this happens, we can be much more efficient by simply
        // determining the non-null type and converting to that.
        final List<Schema> schemaTypes = fieldSchema.getTypes();
        if (schemaTypes.size() == 2) {
            final Schema firstSchema = schemaTypes.get(0);
            final Schema secondSchema = schemaTypes.get(1);

            if (firstSchema.getType() == Type.NULL) {
                return conversion.apply(secondSchema);
            }
            if (secondSchema.getType() == Type.NULL) {
                return conversion.apply(firstSchema);
            }
        }

        final Optional<Schema> mostSuitableType = DataTypeUtils.findMostSuitableType(
                originalValue,
                getNonNullSubSchemas(fieldSchema),
                AvroTypeUtil::determineDataType
        );
        if (mostSuitableType.isPresent()) {
            return conversion.apply(mostSuitableType.get());
        }

        for (final Schema subSchema : fieldSchema.getTypes()) {
            if (subSchema.getType() == Type.NULL) {
                continue;
            }

            foundNonNull = true;
            final DataType desiredDataType = AvroTypeUtil.determineDataType(subSchema);
            try {
                final Object convertedValue = conversion.apply(subSchema);

                if (isCompatibleDataType(convertedValue, desiredDataType)) {
                    return convertedValue;
                }

                // For logical types those store with different type (e.g. BigDecimal as ByteBuffer), check compatibility using the original rawValue
                if (subSchema.getLogicalType() != null && DataTypeUtils.isCompatibleDataType(originalValue, desiredDataType)) {
                    return convertedValue;
                }
            } catch (Exception e) {
                // If failed with one of possible types, continue with the next available option.
                if (logger.isDebugEnabled()) {
                    logger.debug("Cannot convert value {} to type {}", originalValue, desiredDataType, e);
                }
            }
        }

        if (foundNonNull) {
            throw new IllegalTypeConversionException("Cannot convert value " + originalValue + " of type " + originalValue.getClass()
                + " because no compatible types exist in the UNION for field " + fieldName);
        }

        return null;
    }

    private static boolean isCompatibleDataType(final Object value, final DataType dataType) {
        if (value == null) {
            return false;
        }

        switch (dataType.getFieldType()) {
            case RECORD:
                if (value instanceof GenericRecord || value instanceof SpecificRecord) {
                    return true;
                }
                break;
            case STRING:
                if (value instanceof Utf8) {
                    return true;
                }
                break;
            case ARRAY:
                if (value instanceof Array || value instanceof List || value instanceof ByteBuffer) {
                    return true;
                }
                break;
            case MAP:
                if (value instanceof Map) {
                    return true;
                }
        }

        return DataTypeUtils.isCompatibleDataType(value, dataType);
    }


    /**
     * Convert an Avro object to a normal Java objects for further processing.
     * The counter-part method which convert a raw value to an Avro object is {@link #convertToAvroObject(Object, Schema, String, Charset)}
     */
    private static Object normalizeValue(final Object value, final Schema avroSchema, final String fieldName) {
        if (value == null) {
            return null;
        }

        switch (avroSchema.getType()) {
            case INT: {
                final LogicalType logicalType = avroSchema.getLogicalType();
                if (logicalType == null) {
                    return value;
                }

                final String logicalName = logicalType.getName();
                if (LOGICAL_TYPE_DATE.equals(logicalName)) {
                    // date logical name means that the value is number of days since Jan 1, 1970
                    return new java.sql.Date(TimeUnit.DAYS.toMillis((int) value));
                } else if (LOGICAL_TYPE_TIME_MILLIS.equals(logicalName)) {
                    // time-millis logical name means that the value is number of milliseconds since midnight.
                    return new java.sql.Time((int) value);
                }

                break;
            }
            case LONG: {
                final LogicalType logicalType = avroSchema.getLogicalType();
                if (logicalType == null) {
                    return value;
                }

                final String logicalName = logicalType.getName();
                if (LOGICAL_TYPE_TIME_MICROS.equals(logicalName)) {
                    return new java.sql.Time(TimeUnit.MICROSECONDS.toMillis((long) value));
                } else if (LOGICAL_TYPE_TIMESTAMP_MILLIS.equals(logicalName)) {
                    return new java.sql.Timestamp((long) value);
                } else if (LOGICAL_TYPE_TIMESTAMP_MICROS.equals(logicalName)) {
                    return new java.sql.Timestamp(TimeUnit.MICROSECONDS.toMillis((long) value));
                }
                break;
            }
            case UNION:
                if (value instanceof GenericData.Record) {
                    final GenericData.Record avroRecord = (GenericData.Record) value;
                    return normalizeValue(value, avroRecord.getSchema(), fieldName);
                }
                return convertUnionFieldValue(value, avroSchema, schema -> normalizeValue(value, schema, fieldName), fieldName);
            case RECORD:
                final GenericData.Record record = (GenericData.Record) value;
                final Schema recordSchema = record.getSchema();
                final List<Field> recordFields = recordSchema.getFields();
                final Map<String, Object> values = new HashMap<>(recordFields.size());
                for (final Field field : recordFields) {
                    final Object avroFieldValue = record.get(field.name());
                    final Object fieldValue = normalizeValue(avroFieldValue, field.schema(), fieldName + "/" + field.name());
                    values.put(field.name(), fieldValue);
                }
                final RecordSchema childSchema = AvroTypeUtil.createSchema(recordSchema, false);
                return new MapRecord(childSchema, values);
            case BYTES:
                final ByteBuffer bb = (ByteBuffer) value;
                final LogicalType logicalType = avroSchema.getLogicalType();
                if (logicalType != null && LOGICAL_TYPE_DECIMAL.equals(logicalType.getName())) {
                    return new Conversions.DecimalConversion().fromBytes(bb, avroSchema, logicalType);
                }
                return AvroTypeUtil.convertByteArray(bb.array());
            case FIXED:
                final GenericFixed fixed = (GenericFixed) value;
                final LogicalType fixedLogicalType = avroSchema.getLogicalType();
                if (fixedLogicalType != null && LOGICAL_TYPE_DECIMAL.equals(fixedLogicalType.getName())) {
                    final ByteBuffer fixedByteBuffer = ByteBuffer.wrap(fixed.bytes());
                    return new Conversions.DecimalConversion().fromBytes(fixedByteBuffer, avroSchema, fixedLogicalType);
                }
                return AvroTypeUtil.convertByteArray(fixed.bytes());
            case ENUM:
                return value.toString();
            case NULL:
                return null;
            case STRING:
                return value.toString();
            case ARRAY:
                if (value instanceof List) {
                    final List<?> list = (List<?>) value;
                    final Object[] valueArray = new Object[list.size()];
                    for (int i = 0; i < list.size(); i++) {
                        final Schema elementSchema = avroSchema.getElementType();
                        valueArray[i] = normalizeValue(list.get(i), elementSchema, fieldName + "[" + i + "]");
                    }
                    return valueArray;
                } else {
                    final GenericData.Array<?> array = (GenericData.Array<?>) value;
                    final Object[] valueArray = new Object[array.size()];
                    for (int i = 0; i < array.size(); i++) {
                        final Schema elementSchema = avroSchema.getElementType();
                        valueArray[i] = normalizeValue(array.get(i), elementSchema, fieldName + "[" + i + "]");
                    }
                    return valueArray;
                }
            case MAP:
                final Map<?, ?> avroMap = (Map<?, ?>) value;
                final Map<String, Object> map = new HashMap<>(avroMap.size());
                for (final Map.Entry<?, ?> entry : avroMap.entrySet()) {
                    Object obj = entry.getValue();
                    if (obj instanceof Utf8 || obj instanceof CharSequence) {
                        obj = obj.toString();
                    }

                    final String key = entry.getKey().toString();
                    obj = normalizeValue(obj, avroSchema.getValueType(), fieldName + "[" + key + "]");

                    map.put(key, obj);
                }

                return map;
        }

        return value;
    }

}