/**
 * 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 edu.uci.ics.hivesterix.serde.lazy;

import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.serde2.lazy.ByteArrayRef;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;

import edu.uci.ics.hivesterix.serde.lazy.LazyUtils.RecordInfo;
import edu.uci.ics.hivesterix.serde.lazy.LazyUtils.VInt;
import edu.uci.ics.hivesterix.serde.lazy.objectinspector.LazyMapObjectInspector;

/**
 * LazyMap is serialized as follows: start A b c b c b c end bytes[] ->
 * |--------|---|---|---|---| ... |---|---|
 * Section A is the null-bytes. Suppose the map has N key-value pairs, then
 * there are (N*2+7)/8 bytes used as null-bytes. Each bit corresponds to a key
 * or a value and it indicates whether that key or value is null (0) or not null
 * (1).
 * After A, all the bytes are actual serialized data of the map, which are
 * key-value pairs. b represent the keys and c represent the values. Each of
 * them is again a LazyObject.
 */

@SuppressWarnings("rawtypes")
public class LazyMap extends LazyNonPrimitive<LazyMapObjectInspector> {

    private static Log LOG = LogFactory.getLog(LazyMap.class.getName());

    /**
     * Whether the data is already parsed or not.
     */
    boolean parsed;

    /**
     * The size of the map. Only valid when the data is parsed. -1 when the map
     * is NULL.
     */
    int mapSize = 0;

    /**
     * The beginning position and length of key[i] and value[i]. Only valid when
     * the data is parsed.
     */
    int[] keyStart;
    int[] keyLength;
    int[] valueStart;
    int[] valueLength;
    /**
     * Whether valueObjects[i]/keyObjects[i] is initialized or not.
     */
    boolean[] keyInited;
    boolean[] valueInited;

    /**
     * Whether valueObjects[i]/keyObjects[i] is null or not This could not be
     * inferred from the length of the object. In particular, a 0-length string
     * is not null.
     */
    boolean[] keyIsNull;
    boolean[] valueIsNull;

    /**
     * The keys are stored in an array of LazyPrimitives.
     */
    LazyPrimitive<?, ?>[] keyObjects;
    /**
     * The values are stored in an array of LazyObjects. value[index] will start
     * from KeyEnd[index] + 1, and ends before KeyStart[index+1] - 1.
     */
    LazyObject[] valueObjects;

    protected LazyMap(LazyMapObjectInspector oi) {
        super(oi);
    }

    /**
     * Set the row data for this LazyMap.
     * 
     * @see LazyObject#init(ByteArrayRef, int, int)
     */
    @Override
    public void init(byte[] bytes, int start, int length) {
        super.init(bytes, start, length);
        parsed = false;
    }

    /**
     * Adjust the size of arrays: keyStart, keyLength valueStart, valueLength
     * keyInited, keyIsNull valueInited, valueIsNull.
     */
    protected void adjustArraySize(int newSize) {
        if (keyStart == null || keyStart.length < newSize) {
            keyStart = new int[newSize];
            keyLength = new int[newSize];
            valueStart = new int[newSize];
            valueLength = new int[newSize];
            keyInited = new boolean[newSize];
            keyIsNull = new boolean[newSize];
            valueInited = new boolean[newSize];
            valueIsNull = new boolean[newSize];
            keyObjects = new LazyPrimitive<?, ?>[newSize];
            valueObjects = new LazyObject[newSize];
        }
    }

    boolean nullMapKey = false;
    VInt vInt = new LazyUtils.VInt();
    RecordInfo recordInfo = new LazyUtils.RecordInfo();

    /**
     * Parse the byte[] and fill keyStart, keyLength, keyIsNull valueStart,
     * valueLength and valueIsNull.
     */
    private void parse() {

        // get the VInt that represents the map size
        LazyUtils.readVInt(bytes, start, vInt);
        mapSize = vInt.value;
        if (0 == mapSize) {
            parsed = true;
            return;
        }

        // adjust arrays
        adjustArraySize(mapSize);

        // find out the null-bytes
        int mapByteStart = start + vInt.length;
        int nullByteCur = mapByteStart;
        int nullByteEnd = mapByteStart + (mapSize * 2 + 7) / 8;
        int lastElementByteEnd = nullByteEnd;

        // parsing the keys and values one by one
        for (int i = 0; i < mapSize; i++) {
            // parse a key
            keyIsNull[i] = true;
            if ((bytes[nullByteCur] & (1 << ((i * 2) % 8))) != 0) {
                keyIsNull[i] = false;
                LazyUtils.checkObjectByteInfo(((MapObjectInspector) oi).getMapKeyObjectInspector(), bytes,
                        lastElementByteEnd, recordInfo);
                keyStart[i] = lastElementByteEnd + recordInfo.elementOffset;
                keyLength[i] = recordInfo.elementSize;
                lastElementByteEnd = keyStart[i] + keyLength[i];
            } else if (!nullMapKey) {
                nullMapKey = true;
                LOG.warn("Null map key encountered! Ignoring similar problems.");
            }

            // parse a value
            valueIsNull[i] = true;
            if ((bytes[nullByteCur] & (1 << ((i * 2 + 1) % 8))) != 0) {
                valueIsNull[i] = false;
                LazyUtils.checkObjectByteInfo(((MapObjectInspector) oi).getMapValueObjectInspector(), bytes,
                        lastElementByteEnd, recordInfo);
                valueStart[i] = lastElementByteEnd + recordInfo.elementOffset;
                valueLength[i] = recordInfo.elementSize;
                lastElementByteEnd = valueStart[i] + valueLength[i];
            }

            // move onto the next null byte
            if (3 == (i % 4)) {
                nullByteCur++;
            }
        }

        Arrays.fill(keyInited, 0, mapSize, false);
        Arrays.fill(valueInited, 0, mapSize, false);
        parsed = true;
    }

    /**
     * Get the value object with the index without checking parsed.
     * 
     * @param index
     *            The index into the array starting from 0
     */
    private LazyObject uncheckedGetValue(int index) {
        if (valueIsNull[index]) {
            return null;
        }
        if (!valueInited[index]) {
            valueInited[index] = true;
            if (valueObjects[index] == null) {
                valueObjects[index] = LazyFactory.createLazyObject(((MapObjectInspector) oi)
                        .getMapValueObjectInspector());
            }
            valueObjects[index].init(bytes, valueStart[index], valueLength[index]);
        }
        return valueObjects[index];
    }

    /**
     * Get the value in the map for the key.
     * If there are multiple matches (which is possible in the serialized
     * format), only the first one is returned.
     * The most efficient way to get the value for the key is to serialize the
     * key and then try to find it in the array. We do linear search because in
     * most cases, user only wants to get one or two values out of the map, and
     * the cost of building up a HashMap is substantially higher.
     * 
     * @param key
     *            The key object that we are looking for.
     * @return The corresponding value object, or NULL if not found
     */
    public Object getMapValueElement(Object key) {
        if (!parsed) {
            parse();
        }
        // search for the key
        for (int i = 0; i < mapSize; i++) {
            LazyPrimitive<?, ?> lazyKeyI = uncheckedGetKey(i);
            if (lazyKeyI == null) {
                continue;
            }
            // getWritableObject() will convert LazyPrimitive to actual
            // primitive
            // writable objects.
            Object keyI = lazyKeyI.getWritableObject();
            if (keyI == null) {
                continue;
            }
            if (keyI.equals(key)) {
                // Got a match, return the value
                LazyObject v = uncheckedGetValue(i);
                return v == null ? v : v.getObject();
            }
        }
        return null;
    }

    /**
     * Get the key object with the index without checking parsed.
     * 
     * @param index
     *            The index into the array starting from 0
     */
    private LazyPrimitive<?, ?> uncheckedGetKey(int index) {
        if (keyIsNull[index]) {
            return null;
        }
        if (!keyInited[index]) {
            keyInited[index] = true;
            if (keyObjects[index] == null) {
                // Keys are always primitive
                keyObjects[index] = LazyFactory
                        .createLazyPrimitiveClass((PrimitiveObjectInspector) ((MapObjectInspector) oi)
                                .getMapKeyObjectInspector());
            }
            keyObjects[index].init(bytes, keyStart[index], keyLength[index]);
        }
        return keyObjects[index];
    }

    /**
     * cachedMap is reused for different calls to getMap(). But each LazyMap has
     * a separate cachedMap so we won't overwrite the data by accident.
     */
    LinkedHashMap<Object, Object> cachedMap;

    /**
     * Return the map object representing this LazyMap. Note that the keyObjects
     * will be Writable primitive objects.
     * 
     * @return the map object
     */
    public Map<Object, Object> getMap() {
        if (!parsed) {
            parse();
        }
        if (cachedMap == null) {
            // Use LinkedHashMap to provide deterministic order
            cachedMap = new LinkedHashMap<Object, Object>();
        } else {
            cachedMap.clear();
        }

        // go through each element of the map
        for (int i = 0; i < mapSize; i++) {
            LazyPrimitive<?, ?> lazyKey = uncheckedGetKey(i);
            if (lazyKey == null) {
                continue;
            }
            Object key = lazyKey.getObject();
            // do not overwrite if there are duplicate keys
            if (key != null && !cachedMap.containsKey(key)) {
                LazyObject lazyValue = uncheckedGetValue(i);
                Object value = (lazyValue == null ? null : lazyValue.getObject());
                cachedMap.put(key, value);
            }
        }
        return cachedMap;
    }

    /**
     * Get the size of the map represented by this LazyMap.
     * 
     * @return The size of the map
     */
    public int getMapSize() {
        if (!parsed) {
            parse();
        }
        return mapSize;
    }
}