/**
 * 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.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.hadoop.hive.serde2.ByteStream.Output;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.io.WritableUtils;

import edu.uci.ics.hivesterix.serde.lazy.objectinspector.LazyObjectInspectorFactory;
import edu.uci.ics.hivesterix.serde.lazy.objectinspector.primitive.PrimitiveObjectInspectorFactory;

/**
 * LazyUtils.
 * 
 */
public final class LazyUtils {

	/**
	 * Convert the byte array to an int starting from the given offset. Refer to
	 * code by aeden on DZone Snippets:
	 * 
	 * @param b
	 *            the byte array
	 * @param offset
	 *            the array offset
	 * @return the integer
	 */
	public static int byteArrayToInt(byte[] b, int offset) {
		int value = 0;
		for (int i = 0; i < 4; i++) {
			int shift = (4 - 1 - i) * 8;
			value += (b[i + offset] & 0x000000FF) << shift;
		}
		return value;
	}

	/**
	 * Convert the byte array to a long starting from the given offset.
	 * 
	 * @param b
	 *            the byte array
	 * @param offset
	 *            the array offset
	 * @return the long
	 */
	public static long byteArrayToLong(byte[] b, int offset) {
		long value = 0;
		for (int i = 0; i < 8; i++) {
			int shift = (8 - 1 - i) * 8;
			value += ((long) (b[i + offset] & 0x00000000000000FF)) << shift;
		}
		return value;
	}

	/**
	 * Convert the byte array to a short starting from the given offset.
	 * 
	 * @param b
	 *            the byte array
	 * @param offset
	 *            the array offset
	 * @return the short
	 */
	public static short byteArrayToShort(byte[] b, int offset) {
		short value = 0;
		value += (b[offset] & 0x000000FF) << 8;
		value += (b[offset + 1] & 0x000000FF);
		return value;
	}

	/**
	 * Record is the unit that data is serialized in. A record includes two
	 * parts. The first part stores the size of the element and the second part
	 * stores the real element. size element record ->
	 * |----|-------------------------|
	 * 
	 * A RecordInfo stores two information of a record, the size of the "size"
	 * part which is the element offset and the size of the element part which
	 * is element size.
	 */
	public static class RecordInfo {
		public RecordInfo() {
			elementOffset = 0;
			elementSize = 0;
		}

		public byte elementOffset;
		public int elementSize;

		@Override
		public String toString() {
			return "(" + elementOffset + ", " + elementSize + ")";
		}
	}

	static VInt vInt = new LazyUtils.VInt();

	/**
	 * Check a particular field and set its size and offset in bytes based on
	 * the field type and the bytes arrays.
	 * 
	 * For void, boolean, byte, short, int, long, float and double, there is no
	 * offset and the size is fixed. For string, map, list, struct, the first
	 * four bytes are used to store the size. So the offset is 4 and the size is
	 * computed by concating the first four bytes together. The first four bytes
	 * are defined with respect to the offset in the bytes arrays.
	 * 
	 * @param objectInspector
	 *            object inspector of the field
	 * @param bytes
	 *            bytes arrays store the table row
	 * @param offset
	 *            offset of this field
	 * @param recordInfo
	 *            modify this byteinfo object and return it
	 */
	public static void checkObjectByteInfo(ObjectInspector objectInspector,
			byte[] bytes, int offset, RecordInfo recordInfo) {
		Category category = objectInspector.getCategory();
		switch (category) {
		case PRIMITIVE:
			PrimitiveCategory primitiveCategory = ((PrimitiveObjectInspector) objectInspector)
					.getPrimitiveCategory();
			switch (primitiveCategory) {
			case VOID:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = 0;
				break;
			case BOOLEAN:
			case BYTE:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = 1;
				break;
			case SHORT:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = 2;
				break;
			case FLOAT:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = 4;
				break;
			case DOUBLE:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = 8;
				break;
			case INT:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = WritableUtils
						.decodeVIntSize(bytes[offset]);
				break;
			case LONG:
				recordInfo.elementOffset = 0;
				recordInfo.elementSize = WritableUtils
						.decodeVIntSize(bytes[offset]);
				break;
			case STRING:
				// using vint instead of 4 bytes
				LazyUtils.readVInt(bytes, offset, vInt);
				recordInfo.elementOffset = vInt.length;
				recordInfo.elementSize = vInt.value;
				break;
			default: {
				throw new RuntimeException("Unrecognized primitive type: "
						+ primitiveCategory);
			}
			}
			break;
		case LIST:
		case MAP:
		case STRUCT:
			recordInfo.elementOffset = 4;
			recordInfo.elementSize = LazyUtils.byteArrayToInt(bytes, offset);
			break;
		default: {
			throw new RuntimeException("Unrecognized non-primitive type: "
					+ category);
		}
		}
	}

	/**
	 * A zero-compressed encoded long.
	 */
	public static class VLong {
		public VLong() {
			value = 0;
			length = 0;
		}

		public long value;
		public byte length;
	};

	/**
	 * Reads a zero-compressed encoded long from a byte array and returns it.
	 * 
	 * @param bytes
	 *            the byte array
	 * @param offset
	 *            offset of the array to read from
	 * @param vlong
	 *            storing the deserialized long and its size in byte
	 */
	public static void readVLong(byte[] bytes, int offset, VLong vlong) {
		byte firstByte = bytes[offset];
		vlong.length = (byte) WritableUtils.decodeVIntSize(firstByte);
		if (vlong.length == 1) {
			vlong.value = firstByte;
			return;
		}
		long i = 0;
		for (int idx = 0; idx < vlong.length - 1; idx++) {
			byte b = bytes[offset + 1 + idx];
			i = i << 8;
			i = i | (b & 0xFF);
		}
		vlong.value = (WritableUtils.isNegativeVInt(firstByte) ? (i ^ -1L) : i);
	}

	/**
	 * A zero-compressed encoded integer.
	 */
	public static class VInt implements Serializable {
		private static final long serialVersionUID = 1L;

		public VInt() {
			value = 0;
			length = 0;
		}

		public int value;
		public byte length;
	};

	/**
	 * Reads a zero-compressed encoded int from a byte array and returns it.
	 * 
	 * @param bytes
	 *            the byte array
	 * @param offset
	 *            offset of the array to read from
	 * @param vInt
	 *            storing the deserialized int and its size in byte
	 */
	public static void readVInt(byte[] bytes, int offset, VInt vInt) {
		byte firstByte = bytes[offset];
		vInt.length = (byte) WritableUtils.decodeVIntSize(firstByte);
		if (vInt.length == 1) {
			vInt.value = firstByte;
			return;
		}
		int i = 0;
		for (int idx = 0; idx < vInt.length - 1; idx++) {
			byte b = bytes[offset + 1 + idx];
			i = i << 8;
			i = i | (b & 0xFF);
		}
		vInt.value = (WritableUtils.isNegativeVInt(firstByte) ? (i ^ -1) : i);
	}

	/**
	 * Writes a zero-compressed encoded int to a byte array.
	 * 
	 * @param byteStream
	 *            the byte array/stream
	 * @param i
	 *            the int
	 */
	public static void writeVInt(Output byteStream, int i) {
		writeVLong(byteStream, i);
	}

	/**
	 * Write a zero-compressed encoded long to a byte array.
	 * 
	 * @param byteStream
	 *            the byte array/stream
	 * @param l
	 *            the long
	 */
	public static void writeVLong(Output byteStream, long l) {
		if (l >= -112 && l <= 127) {
			byteStream.write((byte) l);
			return;
		}

		int len = -112;
		if (l < 0) {
			l ^= -1L; // take one's complement'
			len = -120;
		}

		long tmp = l;
		while (tmp != 0) {
			tmp = tmp >> 8;
			len--;
		}

		byteStream.write((byte) len);

		len = (len < -120) ? -(len + 120) : -(len + 112);

		for (int idx = len; idx != 0; idx--) {
			int shiftbits = (idx - 1) * 8;
			long mask = 0xFFL << shiftbits;
			byteStream.write((byte) ((l & mask) >> shiftbits));
		}
	}

	static Map<TypeInfo, ObjectInspector> cachedLazyObjectInspector = new ConcurrentHashMap<TypeInfo, ObjectInspector>();

	/**
	 * Returns the lazy binary object inspector that can be used to inspect an
	 * lazy binary object of that typeInfo
	 * 
	 * For primitive types, we use the standard writable object inspector.
	 */
	public static ObjectInspector getLazyObjectInspectorFromTypeInfo(
			TypeInfo typeInfo, boolean topLevel) {
		if (typeInfo == null)
			throw new IllegalStateException("illegal type null ");
		ObjectInspector result = cachedLazyObjectInspector.get(typeInfo);
		if (result == null) {
			switch (typeInfo.getCategory()) {
			case PRIMITIVE: {
				result = PrimitiveObjectInspectorFactory
						.getPrimitiveLazyObjectInspector(((PrimitiveTypeInfo) typeInfo)
								.getPrimitiveCategory());
				break;
			}
			case LIST: {
				ObjectInspector elementObjectInspector = getLazyObjectInspectorFromTypeInfo(
						((ListTypeInfo) typeInfo).getListElementTypeInfo(),
						false);
				result = LazyObjectInspectorFactory
						.getLazyListObjectInspector(elementObjectInspector);
				break;
			}
			case MAP: {
				MapTypeInfo mapTypeInfo = (MapTypeInfo) typeInfo;
				ObjectInspector keyObjectInspector = getLazyObjectInspectorFromTypeInfo(
						mapTypeInfo.getMapKeyTypeInfo(), false);
				ObjectInspector valueObjectInspector = getLazyObjectInspectorFromTypeInfo(
						mapTypeInfo.getMapValueTypeInfo(), false);
				result = LazyObjectInspectorFactory.getLazyMapObjectInspector(
						keyObjectInspector, valueObjectInspector);
				break;
			}
			case STRUCT: {
				StructTypeInfo structTypeInfo = (StructTypeInfo) typeInfo;
				List<String> fieldNames = structTypeInfo
						.getAllStructFieldNames();
				List<TypeInfo> fieldTypeInfos = structTypeInfo
						.getAllStructFieldTypeInfos();
				List<ObjectInspector> fieldObjectInspectors = new ArrayList<ObjectInspector>(
						fieldTypeInfos.size());

				for (int i = 0; i < fieldTypeInfos.size(); i++) {
					fieldObjectInspectors
							.add(getLazyObjectInspectorFromTypeInfo(
									fieldTypeInfos.get(i), false));
				}

				// if it is top level then create columnar
				if (topLevel)
					result = LazyObjectInspectorFactory
							.getLazyColumnarObjectInspector(fieldNames,
									fieldObjectInspectors);
				// if it is not top level then create struct
				else
					result = LazyObjectInspectorFactory
							.getLazyStructObjectInspector(fieldNames,
									fieldObjectInspectors);

				break;
			}
			default: {
				result = null;
			}
			}
			cachedLazyObjectInspector.put(typeInfo, result);
		}
		return result;
	}

	/**
	 * get top-level lazy object inspector
	 * 
	 * @param fieldNames
	 * @param fieldTypeInfos
	 * @return
	 */
	public static ObjectInspector getLazyObjectInspector(
			List<String> fieldNames, List<TypeInfo> fieldTypeInfos) {
		List<ObjectInspector> fieldObjectInspectors = new ArrayList<ObjectInspector>(
				fieldTypeInfos.size());
		for (int i = 0; i < fieldTypeInfos.size(); i++) {
			fieldObjectInspectors.add(getLazyObjectInspectorFromTypeInfo(
					fieldTypeInfos.get(i), false));
		}

		return LazyObjectInspectorFactory.getLazyColumnarObjectInspector(
				fieldNames, fieldObjectInspectors);
	}

	private LazyUtils() {
		// prevent instantiation
	}

	/**
	 * Returns -1 if the first byte sequence is lexicographically less than the
	 * second; returns +1 if the second byte sequence is lexicographically less
	 * than the first; otherwise return 0.
	 */
	public static int compare(byte[] b1, int start1, int length1, byte[] b2,
			int start2, int length2) {

		int min = Math.min(length1, length2);

		for (int i = 0; i < min; i++) {
			if (b1[start1 + i] == b2[start2 + i]) {
				continue;
			}
			if (b1[start1 + i] < b2[start2 + i]) {
				return -1;
			} else {
				return 1;
			}
		}

		if (length1 < length2) {
			return -1;
		}
		if (length1 > length2) {
			return 1;
		}
		return 0;
	}

	public static int hashBytes(byte[] data, int start, int len) {
		int hash = 1;
		for (int i = start; i < len; i++) {
			hash = (31 * hash) + data[i];
		}
		return hash;
	}

	/**
	 * Writes a zero-compressed encoded int to a byte array.
	 * 
	 * @param byteStream
	 *            the byte array/stream
	 * @param i
	 *            the int
	 */
	public static void writeVInt(DataOutput byteStream, int i)
			throws IOException {
		writeVLong(byteStream, i);
	}

	/**
	 * Write a zero-compressed encoded long to a byte array.
	 * 
	 * @param byteStream
	 *            the byte array/stream
	 * @param l
	 *            the long
	 */
	public static void writeVLong(DataOutput byteStream, long l)
			throws IOException {
		if (l >= -112 && l <= 127) {
			byteStream.write((byte) l);
			return;
		}

		int len = -112;
		if (l < 0) {
			l ^= -1L; // take one's complement'
			len = -120;
		}

		long tmp = l;
		while (tmp != 0) {
			tmp = tmp >> 8;
			len--;
		}

		byteStream.write((byte) len);

		len = (len < -120) ? -(len + 120) : -(len + 112);

		for (int idx = len; idx != 0; idx--) {
			int shiftbits = (idx - 1) * 8;
			long mask = 0xFFL << shiftbits;
			byteStream.write((byte) ((l & mask) >> shiftbits));
		}
	}
}