/*
 * 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.external.library.java.base;

import static org.apache.asterix.om.utils.RecordUtil.FULLY_OPEN_RECORD_TYPE;

import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.asterix.builders.RecordBuilder;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.dataflow.data.nontagged.serde.ARecordSerializerDeserializer;
import org.apache.asterix.dataflow.data.nontagged.serde.AStringSerializerDeserializer;
import org.apache.asterix.external.api.IJObject;
import org.apache.asterix.om.base.AMutableString;
import org.apache.asterix.om.base.ARecord;
import org.apache.asterix.om.base.IAObject;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.IAType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.util.LogRedactionUtil;

public final class JRecord extends JComplexObject<Map<String, Object>> {

    private static final AStringSerializerDeserializer aStringSerDer = AStringSerializerDeserializer.INSTANCE;
    private ARecordType recordType;
    private IJObject[] fields;
    private Map<String, IJObject> openFields;
    RecordBuilder recordBuilder = new RecordBuilder();
    ArrayBackedValueStorage fieldNameBuffer = new ArrayBackedValueStorage();
    ArrayBackedValueStorage fieldValueBuffer = new ArrayBackedValueStorage();
    AMutableString nameString = new AMutableString("");

    public JRecord() {
        this.recordType = FULLY_OPEN_RECORD_TYPE;
        this.fields = new IJObject[] {};
        this.openFields = new LinkedHashMap<>();
    }

    public JRecord(ARecordType recordType, IJObject[] fields) {
        this.recordType = recordType;
        this.fields = fields;
        this.openFields = new LinkedHashMap<>();
    }

    public JRecord(ARecordType recordType, IJObject[] fields, Map<String, IJObject> openFields) {
        this(recordType, fields);
        this.openFields = openFields;
    }

    public void addField(String fieldName, IJObject fieldValue) throws HyracksDataException {
        int pos = getFieldPosByName(fieldName);
        if (pos >= 0) {
            throw new RuntimeDataException(ErrorCode.LIBRARY_JAVA_JOBJECTS_FIELD_ALREADY_DEFINED, "closed");
        }
        if (openFields.get(fieldName) != null) {
            throw new RuntimeDataException(ErrorCode.LIBRARY_JAVA_JOBJECTS_FIELD_ALREADY_DEFINED, "open");
        }
        openFields.put(fieldName, fieldValue);
    }

    public IJObject getValueByName(String fieldName) throws HyracksDataException {
        // check closed part
        int fieldPos = getFieldPosByName(fieldName);
        if (fieldPos >= 0) {
            return fields[fieldPos];
        } else {
            // check open part
            IJObject fieldValue = openFields.get(fieldName);
            if (fieldValue == null) {
                throw new RuntimeDataException(ErrorCode.LIBRARY_JAVA_JOBJECTS_UNKNOWN_FIELD,
                        LogRedactionUtil.userData(fieldName));
            }
            return fieldValue;
        }
    }

    public void setValueAtPos(int pos, IJObject jObject) {
        fields[pos] = jObject;
    }

    public IAType getIAType() {
        return recordType;
    }

    public void setField(String fieldName, IJObject fieldValue) throws HyracksDataException {
        int pos = getFieldPosByName(fieldName);
        if (pos >= 0) {
            fields[pos] = fieldValue;
        } else {
            openFields.put(fieldName, fieldValue);
        }
    }

    private int getFieldPosByName(String fieldName) {
        int index = 0;
        String[] fieldNames = recordType.getFieldNames();
        for (String name : fieldNames) {
            if (name.equals(fieldName)) {
                return index;
            }
            index++;
        }
        return -1;
    }

    public ARecordType getRecordType() {
        return recordType;
    }

    @Override
    public void serialize(DataOutput output, boolean writeTypeTag) throws HyracksDataException {
        recordBuilder.reset(recordType);
        int index = 0;
        for (IJObject jObject : fields) {
            fieldValueBuffer.reset();
            jObject.serialize(fieldValueBuffer.getDataOutput(), writeTypeTag);
            recordBuilder.addField(index, fieldValueBuffer);
            index++;
        }

        try {
            if (openFields != null && !openFields.isEmpty()) {
                for (Map.Entry<String, IJObject> entry : openFields.entrySet()) {
                    fieldNameBuffer.reset();
                    fieldValueBuffer.reset();
                    nameString.setValue(entry.getKey());
                    fieldNameBuffer.getDataOutput().write(ATypeTag.STRING.serialize());
                    aStringSerDer.serialize(nameString, fieldNameBuffer.getDataOutput());
                    entry.getValue().serialize(fieldValueBuffer.getDataOutput(), writeTypeTag);
                    recordBuilder.addField(fieldNameBuffer, fieldValueBuffer);
                }
            }
        } catch (IOException ae) {
            throw HyracksDataException.create(ae);
        }
        recordBuilder.write(output, writeTypeTag);
    }

    @Override
    public IAObject getIAObject() {
        // As the open part can be changed any time, we cannot pre-allocate the arrays.
        int numberOfOpenFields = openFields.size();
        String[] openFieldNames = new String[numberOfOpenFields];
        IAType[] openFieldTypes = new IAType[numberOfOpenFields];
        IAObject[] openFieldValues = new IAObject[numberOfOpenFields];
        IAObject[] closedFieldValues = new IAObject[fields.length];
        int idx = 0;
        for (Map.Entry<String, IJObject> entry : openFields.entrySet()) {
            openFieldNames[idx] = entry.getKey();
            openFieldTypes[idx] = entry.getValue().getIAObject().getType();
            openFieldValues[idx] = entry.getValue().getIAObject();
        }
        for (int iter1 = 0; iter1 < fields.length; iter1++) {
            closedFieldValues[iter1] = fields[iter1].getIAObject();
        }
        ARecordType openPartRecType = new ARecordType(null, openFieldNames, openFieldTypes, true);
        ARecordType mergedRecordType = ARecordSerializerDeserializer.mergeRecordTypes(recordType, openPartRecType);
        IAObject[] mergedFields = ARecordSerializerDeserializer.mergeFields(closedFieldValues, openFieldValues);

        return new ARecord(mergedRecordType, mergedFields);
    }

    @Override
    public void setValueGeneric(Map<String, Object> o) throws HyracksDataException {
        reset();
        for (Map.Entry<String, Object> e : o.entrySet()) {
            IAType asxClass = JObject.convertType(e.getValue().getClass());
            IJObject obj = pool.allocate(asxClass);
            obj.setValueGeneric(e.getValue());
            openFields.put(e.getKey(), obj);
        }
    }

    @Override
    public Map<String, Object> getValueGeneric() {
        HashMap<String, Object> rec = new HashMap<>();
        String[] closedFieldNames = recordType.getFieldNames();
        int idx = 0;
        for (IJObject j : fields) {
            rec.put(closedFieldNames[idx], j.getValueGeneric());
            idx++;
        }
        openFields.entrySet().forEach(e -> rec.put(e.getKey(), e.getValue().getValueGeneric()));
        return rec;
    }

    @Override
    public void reset() throws HyracksDataException {
        if (openFields != null && !openFields.isEmpty()) {
            openFields.clear();
        }
        if (fields != null) {
            for (IJObject field : fields) {
                if (field != null) {
                    field.reset();
                }
            }
        }
    }

    public void reset(IJObject[] fields, Map<String, IJObject> openFields) throws HyracksDataException {
        this.reset();
        this.fields = fields;
        this.openFields = openFields;
    }

}
