/*
 * 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.asterix.runtime.formats;

import java.io.DataOutput;
import java.util.Arrays;
import java.util.List;

import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.dataflow.data.nontagged.MissingWriterFactory;
import org.apache.asterix.dataflow.data.nontagged.NullWriterFactory;
import org.apache.asterix.formats.base.IDataFormat;
import org.apache.asterix.formats.nontagged.ADMPrinterFactoryProvider;
import org.apache.asterix.formats.nontagged.BinaryBooleanInspector;
import org.apache.asterix.formats.nontagged.BinaryComparatorFactoryProvider;
import org.apache.asterix.formats.nontagged.BinaryHashFunctionFactoryProvider;
import org.apache.asterix.formats.nontagged.BinaryHashFunctionFamilyProvider;
import org.apache.asterix.formats.nontagged.BinaryIntegerInspector;
import org.apache.asterix.formats.nontagged.CSVPrinterFactoryProvider;
import org.apache.asterix.formats.nontagged.CleanJSONPrinterFactoryProvider;
import org.apache.asterix.formats.nontagged.LosslessADMJSONPrinterFactoryProvider;
import org.apache.asterix.formats.nontagged.LosslessJSONPrinterFactoryProvider;
import org.apache.asterix.formats.nontagged.NormalizedKeyComputerFactoryProvider;
import org.apache.asterix.formats.nontagged.PredicateEvaluatorFactoryProvider;
import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
import org.apache.asterix.formats.nontagged.TypeTraitProvider;
import org.apache.asterix.om.base.ABoolean;
import org.apache.asterix.om.base.AInt32;
import org.apache.asterix.om.base.AMissing;
import org.apache.asterix.om.base.ANull;
import org.apache.asterix.om.base.AOrderedList;
import org.apache.asterix.om.base.AString;
import org.apache.asterix.om.base.IAObject;
import org.apache.asterix.om.constants.AsterixConstantValue;
import org.apache.asterix.om.functions.BuiltinFunctions;
import org.apache.asterix.om.functions.IFunctionDescriptor;
import org.apache.asterix.om.functions.IFunctionManager;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.runtime.base.UnnestingPositionWriterFactory;
import org.apache.asterix.runtime.evaluators.common.CreateMBREvalFactory;
import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.common.utils.Triple;
import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.IAlgebricksConstantValue;
import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionEvalSizeComputer;
import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableEvalSizeEnvironment;
import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
import org.apache.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
import org.apache.hyracks.algebricks.data.IBinaryBooleanInspectorFactory;
import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
import org.apache.hyracks.algebricks.data.IBinaryHashFunctionFactoryProvider;
import org.apache.hyracks.algebricks.data.IBinaryHashFunctionFamilyProvider;
import org.apache.hyracks.algebricks.data.IBinaryIntegerInspectorFactory;
import org.apache.hyracks.algebricks.data.INormalizedKeyComputerFactoryProvider;
import org.apache.hyracks.algebricks.data.IPrinterFactoryProvider;
import org.apache.hyracks.algebricks.data.ISerializerDeserializerProvider;
import org.apache.hyracks.algebricks.data.ITypeTraitProvider;
import org.apache.hyracks.algebricks.data.IUnnestingPositionWriterFactory;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
import org.apache.hyracks.algebricks.runtime.evaluators.ColumnAccessEvalFactory;
import org.apache.hyracks.algebricks.runtime.evaluators.ConstantEvalFactory;
import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
import org.apache.hyracks.api.dataflow.value.IPredicateEvaluatorFactoryProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.SourceLocation;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;

public class NonTaggedDataFormat implements IDataFormat {

    static final NonTaggedDataFormat INSTANCE = new NonTaggedDataFormat();

    private static LogicalVariable METADATA_DUMMY_VAR = new LogicalVariable(-1);

    public static final String NON_TAGGED_DATA_FORMAT = "org.apache.asterix.runtime.formats.NonTaggedDataFormat";

    private NonTaggedDataFormat() {
    }

    @Override
    public IBinaryBooleanInspectorFactory getBinaryBooleanInspectorFactory() {
        return BinaryBooleanInspector.FACTORY;
    }

    @Override
    public IBinaryComparatorFactoryProvider getBinaryComparatorFactoryProvider() {
        return BinaryComparatorFactoryProvider.INSTANCE;
    }

    @Override
    public IBinaryHashFunctionFactoryProvider getBinaryHashFunctionFactoryProvider() {
        return BinaryHashFunctionFactoryProvider.INSTANCE;
    }

    @Override
    public ISerializerDeserializerProvider getSerdeProvider() {
        return SerializerDeserializerProvider.INSTANCE; // done
    }

    @Override
    public ITypeTraitProvider getTypeTraitProvider() {
        return TypeTraitProvider.INSTANCE;
    }

    @SuppressWarnings("unchecked")
    @Override
    public IScalarEvaluatorFactory getFieldAccessEvaluatorFactory(IFunctionManager functionManager, ARecordType recType,
            List<String> fldName, int recordColumn, SourceLocation sourceLoc) throws AlgebricksException {
        IScalarEvaluatorFactory recordEvalFactory = new ColumnAccessEvalFactory(recordColumn);

        if (fldName.size() == 1) {
            String[] names = recType.getFieldNames();
            ArrayBackedValueStorage abvs = new ArrayBackedValueStorage();
            DataOutput dos = abvs.getDataOutput();

            String fieldName = fldName.get(0);
            for (int i = 0; i < names.length; i++) {
                if (names[i].equals(fieldName)) {
                    try {
                        AInt32 ai = new AInt32(i);
                        SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(ai.getType()).serialize(ai,
                                dos);
                    } catch (HyracksDataException e) {
                        throw new AlgebricksException(e);
                    }
                    IScalarEvaluatorFactory fldIndexEvalFactory =
                            new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(), abvs.getLength()));
                    IFunctionDescriptor fDesc =
                            functionManager.lookupFunction(BuiltinFunctions.FIELD_ACCESS_BY_INDEX, sourceLoc);
                    fDesc.setSourceLocation(sourceLoc);
                    fDesc.setImmutableStates(recType);
                    return fDesc.createEvaluatorFactory(
                            new IScalarEvaluatorFactory[] { recordEvalFactory, fldIndexEvalFactory });
                }
            }

            if (recType.isOpen()) {
                AString as = new AString(fieldName);
                try {
                    SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(as.getType()).serialize(as, dos);
                } catch (HyracksDataException e) {
                    throw new AlgebricksException(e);
                }
                IScalarEvaluatorFactory fldNameEvalFactory =
                        new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(), abvs.getLength()));
                IFunctionDescriptor fDesc =
                        functionManager.lookupFunction(BuiltinFunctions.FIELD_ACCESS_BY_NAME, sourceLoc);
                fDesc.setSourceLocation(sourceLoc);
                return fDesc.createEvaluatorFactory(
                        new IScalarEvaluatorFactory[] { recordEvalFactory, fldNameEvalFactory });
            }
        }

        if (fldName.size() > 1) {
            IFunctionDescriptor fDesc = functionManager.lookupFunction(BuiltinFunctions.FIELD_ACCESS_NESTED, sourceLoc);
            fDesc.setSourceLocation(sourceLoc);
            fDesc.setImmutableStates(recType, fldName);
            return fDesc.createEvaluatorFactory(new IScalarEvaluatorFactory[] { recordEvalFactory });
        }

        throw new AlgebricksException("Could not find field " + fldName + " in the schema.");
    }

    @SuppressWarnings("unchecked")
    @Override
    public IScalarEvaluatorFactory[] createMBRFactory(IFunctionManager functionManager, ARecordType recType,
            List<String> fldName, int recordColumn, int dimension, List<String> filterFieldName, boolean isPointMBR,
            SourceLocation sourceLoc) throws AlgebricksException {
        IScalarEvaluatorFactory evalFactory =
                getFieldAccessEvaluatorFactory(functionManager, recType, fldName, recordColumn, sourceLoc);
        int numOfFields = isPointMBR ? dimension : dimension * 2;
        IScalarEvaluatorFactory[] evalFactories =
                new IScalarEvaluatorFactory[numOfFields + (filterFieldName == null ? 0 : 1)];

        ArrayBackedValueStorage abvs1 = new ArrayBackedValueStorage();
        DataOutput dos1 = abvs1.getDataOutput();
        try {
            AInt32 ai = new AInt32(dimension);
            SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(ai.getType()).serialize(ai, dos1);
        } catch (HyracksDataException e) {
            throw new AlgebricksException(e);
        }
        IScalarEvaluatorFactory dimensionEvalFactory =
                new ConstantEvalFactory(Arrays.copyOf(abvs1.getByteArray(), abvs1.getLength()));

        for (int i = 0; i < numOfFields; i++) {
            ArrayBackedValueStorage abvs2 = new ArrayBackedValueStorage();
            DataOutput dos2 = abvs2.getDataOutput();
            try {
                AInt32 ai = new AInt32(i);
                SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(ai.getType()).serialize(ai, dos2);
            } catch (HyracksDataException e) {
                throw new AlgebricksException(e);
            }
            IScalarEvaluatorFactory coordinateEvalFactory =
                    new ConstantEvalFactory(Arrays.copyOf(abvs2.getByteArray(), abvs2.getLength()));

            evalFactories[i] = new CreateMBREvalFactory(evalFactory, dimensionEvalFactory, coordinateEvalFactory);
        }
        if (filterFieldName != null) {
            evalFactories[numOfFields] =
                    getFieldAccessEvaluatorFactory(functionManager, recType, filterFieldName, recordColumn, sourceLoc);
        }
        return evalFactories;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Triple<IScalarEvaluatorFactory, ScalarFunctionCallExpression, IAType> partitioningEvaluatorFactory(
            IFunctionManager functionManager, ARecordType recType, List<String> fldName, SourceLocation sourceLoc)
            throws AlgebricksException {
        String[] names = recType.getFieldNames();
        int n = names.length;
        if (fldName.size() > 1) {
            for (int i = 0; i < n; i++) {
                if (names[i].equals(fldName.get(0))) {
                    IScalarEvaluatorFactory recordEvalFactory =
                            new ColumnAccessEvalFactory(GlobalConfig.DEFAULT_INPUT_DATA_COLUMN);
                    ArrayBackedValueStorage abvs = new ArrayBackedValueStorage();
                    DataOutput dos = abvs.getDataOutput();
                    try {
                        AInt32 ai = new AInt32(i);
                        SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(ai.getType()).serialize(ai,
                                dos);
                    } catch (HyracksDataException e) {
                        throw new AlgebricksException(e);
                    }
                    IScalarEvaluatorFactory fldIndexEvalFactory =
                            new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(), abvs.getLength()));
                    IFunctionDescriptor fDesc =
                            functionManager.lookupFunction(BuiltinFunctions.FIELD_ACCESS_BY_INDEX, sourceLoc);
                    fDesc.setSourceLocation(sourceLoc);
                    fDesc.setImmutableStates(recType);
                    IScalarEvaluatorFactory evalFactory = fDesc.createEvaluatorFactory(
                            new IScalarEvaluatorFactory[] { recordEvalFactory, fldIndexEvalFactory });
                    IFunctionInfo finfoAccess =
                            BuiltinFunctions.getBuiltinFunctionInfo(BuiltinFunctions.FIELD_ACCESS_BY_INDEX);

                    ScalarFunctionCallExpression partitionFun = new ScalarFunctionCallExpression(finfoAccess,
                            new MutableObject<>(new VariableReferenceExpression(METADATA_DUMMY_VAR)),
                            new MutableObject<>(new ConstantExpression(new AsterixConstantValue(new AInt32(i)))));
                    partitionFun.setSourceLocation(sourceLoc);
                    return new Triple<>(evalFactory, partitionFun, recType.getFieldTypes()[i]);
                }
            }
        } else {
            IScalarEvaluatorFactory recordEvalFactory =
                    new ColumnAccessEvalFactory(GlobalConfig.DEFAULT_INPUT_DATA_COLUMN);
            ArrayBackedValueStorage abvs = new ArrayBackedValueStorage();
            DataOutput dos = abvs.getDataOutput();
            AOrderedList as = new AOrderedList(fldName);
            try {
                SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(as.getType()).serialize(as, dos);
            } catch (HyracksDataException e) {
                throw new AlgebricksException(e);
            }
            IFunctionDescriptor fDesc = functionManager.lookupFunction(BuiltinFunctions.FIELD_ACCESS_NESTED, sourceLoc);
            fDesc.setSourceLocation(sourceLoc);
            fDesc.setImmutableStates(recType, fldName);
            IScalarEvaluatorFactory evalFactory =
                    fDesc.createEvaluatorFactory(new IScalarEvaluatorFactory[] { recordEvalFactory });
            IFunctionInfo finfoAccess = BuiltinFunctions.getBuiltinFunctionInfo(BuiltinFunctions.FIELD_ACCESS_NESTED);

            ScalarFunctionCallExpression partitionFun = new ScalarFunctionCallExpression(finfoAccess,
                    new MutableObject<>(new VariableReferenceExpression(METADATA_DUMMY_VAR)),
                    new MutableObject<>(new ConstantExpression(new AsterixConstantValue(as))));
            partitionFun.setSourceLocation(sourceLoc);
            return new Triple<>(evalFactory, partitionFun, recType.getSubFieldType(fldName));
        }
        throw new AlgebricksException("Could not find field " + fldName + " in the schema.");
    }

    @Override
    public IPrinterFactoryProvider getADMPrinterFactoryProvider() {
        return ADMPrinterFactoryProvider.INSTANCE;
    }

    @Override
    public IPrinterFactoryProvider getLosslessJSONPrinterFactoryProvider() {
        return LosslessJSONPrinterFactoryProvider.INSTANCE;
    }

    @Override
    public IPrinterFactoryProvider getLosslessADMJSONPrinterFactoryProvider() {
        return LosslessADMJSONPrinterFactoryProvider.INSTANCE;
    }

    @Override
    public IPrinterFactoryProvider getCleanJSONPrinterFactoryProvider() {
        return CleanJSONPrinterFactoryProvider.INSTANCE;
    }

    @Override
    public IPrinterFactoryProvider getCSVPrinterFactoryProvider() {
        return CSVPrinterFactoryProvider.INSTANCE;
    }

    @SuppressWarnings("unchecked")
    @Override
    public IScalarEvaluatorFactory getConstantEvalFactory(IAlgebricksConstantValue value) throws AlgebricksException {
        IAObject obj;
        if (value.isMissing()) {
            obj = AMissing.MISSING;
        } else if (value.isNull()) {
            obj = ANull.NULL;
        } else if (value.isTrue()) {
            obj = ABoolean.TRUE;
        } else if (value.isFalse()) {
            obj = ABoolean.FALSE;
        } else {
            AsterixConstantValue acv = (AsterixConstantValue) value;
            obj = acv.getObject();
        }
        ArrayBackedValueStorage abvs = new ArrayBackedValueStorage();
        DataOutput dos = abvs.getDataOutput();
        try {
            SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(obj.getType()).serialize(obj, dos);
        } catch (HyracksDataException e) {
            throw new AlgebricksException(e);
        }
        return new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(), abvs.getLength()));
    }

    @Override
    public IBinaryIntegerInspectorFactory getBinaryIntegerInspectorFactory() {
        return BinaryIntegerInspector.FACTORY;
    }

    @Override
    public IMissingWriterFactory getMissingWriterFactory() {
        return MissingWriterFactory.INSTANCE;
    }

    @Override
    public IMissingWriterFactory getNullWriterFactory() {
        return NullWriterFactory.INSTANCE;
    }

    @Override
    public IUnnestingPositionWriterFactory getUnnestingPositionWriterFactory() {
        return UnnestingPositionWriterFactory.INSTANCE;
    }

    @Override
    public IExpressionEvalSizeComputer getExpressionEvalSizeComputer() {
        return new IExpressionEvalSizeComputer() {
            @Override
            public int getEvalSize(ILogicalExpression expr, IVariableEvalSizeEnvironment env)
                    throws AlgebricksException {
                switch (expr.getExpressionTag()) {
                    case CONSTANT: {
                        ConstantExpression c = (ConstantExpression) expr;
                        if (c == ConstantExpression.MISSING) {
                            return 1;
                        } else if (c == ConstantExpression.FALSE || c == ConstantExpression.TRUE) {
                            return 2;
                        } else {
                            AsterixConstantValue acv = (AsterixConstantValue) c.getValue();
                            IAObject o = acv.getObject();
                            switch (o.getType().getTypeTag()) {
                                case DOUBLE:
                                    return 9;
                                case FLOAT:
                                    return 5;
                                case BOOLEAN:
                                    return 2;
                                case MISSING:
                                    return 1;
                                case NULL:
                                    return 1;
                                case TINYINT:
                                    return 2;
                                case SMALLINT:
                                    return 3;
                                case INTEGER:
                                    return 5;
                                case BIGINT:
                                    return 9;
                                default:
                                    return -1;
                            }
                        }
                    }
                    case FUNCTION_CALL: {
                        AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) expr;
                        if (f.getFunctionIdentifier().equals(BuiltinFunctions.TID)) {
                            return 5;
                        } else {
                            // TODO
                            return -1;
                        }
                    }
                    default: {
                        // TODO
                        return -1;
                    }
                }
            }
        };
    }

    @Override
    public INormalizedKeyComputerFactoryProvider getNormalizedKeyComputerFactoryProvider() {
        return NormalizedKeyComputerFactoryProvider.INSTANCE;
    }

    @Override
    public IBinaryHashFunctionFamilyProvider getBinaryHashFunctionFamilyProvider() {
        return BinaryHashFunctionFamilyProvider.INSTANCE;
    }

    @Override
    public IPredicateEvaluatorFactoryProvider getPredicateEvaluatorFactoryProvider() {
        return PredicateEvaluatorFactoryProvider.INSTANCE;
    }
}
