/**
 * 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.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.serde.Constants;
import org.apache.hadoop.hive.serde2.ByteStream;
import org.apache.hadoop.hive.serde2.ByteStream.Output;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
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.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.BooleanObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.ByteObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.DoubleObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.FloatObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.IntObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.ShortObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

/**
 * The LazySerDe class combines the lazy property of LazySimpleSerDe class and
 * the binary property of BinarySortable class. Lazy means a field is not
 * deserialized until required. Binary means a field is serialized in binary
 * compact format.
 */
public class LazySerDe implements SerDe {

	public static final Log LOG = LogFactory.getLog(LazySerDe.class.getName());

	public LazySerDe() {
	}

	List<String> columnNames;
	List<TypeInfo> columnTypes;

	TypeInfo rowTypeInfo;
	ObjectInspector cachedObjectInspector;

	// The object for storing row data
	LazyColumnar cachedLazyStruct;

	/**
	 * Initialize the SerDe with configuration and table information.
	 */
	@Override
	public void initialize(Configuration conf, Properties tbl)
			throws SerDeException {
		// Get column names and types
		String columnNameProperty = tbl.getProperty(Constants.LIST_COLUMNS);
		String columnTypeProperty = tbl
				.getProperty(Constants.LIST_COLUMN_TYPES);
		if (columnNameProperty.length() == 0) {
			columnNames = new ArrayList<String>();
		} else {
			columnNames = Arrays.asList(columnNameProperty.split(","));
		}
		if (columnTypeProperty.length() == 0) {
			columnTypes = new ArrayList<TypeInfo>();
		} else {
			columnTypes = TypeInfoUtils
					.getTypeInfosFromTypeString(columnTypeProperty);
		}
		assert (columnNames.size() == columnTypes.size());
		// Create row related objects
		rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames,
				columnTypes);
		// Create the object inspector and the lazy binary struct object
		cachedObjectInspector = LazyUtils.getLazyObjectInspectorFromTypeInfo(
				rowTypeInfo, true);
		cachedLazyStruct = (LazyColumnar) LazyFactory
				.createLazyObject(cachedObjectInspector);
		// output debug info
		LOG.debug("LazySerDe initialized with: columnNames=" + columnNames
				+ " columnTypes=" + columnTypes);
	}

	/**
	 * Returns the ObjectInspector for the row.
	 */
	@Override
	public ObjectInspector getObjectInspector() throws SerDeException {
		return cachedObjectInspector;
	}

	/**
	 * Returns the Writable Class after serialization.
	 */
	@Override
	public Class<? extends Writable> getSerializedClass() {
		return BytesWritable.class;
	}

	// The wrapper for byte array
	ByteArrayRef byteArrayRef;

	/**
	 * Deserialize a table record to a Lazy struct.
	 */
	@SuppressWarnings("deprecation")
	@Override
	public Object deserialize(Writable field) throws SerDeException {
		if (byteArrayRef == null) {
			byteArrayRef = new ByteArrayRef();
		}
		if (field instanceof BytesWritable) {
			BytesWritable b = (BytesWritable) field;
			if (b.getSize() == 0) {
				return null;
			}
			// For backward-compatibility with hadoop 0.17
			byteArrayRef.setData(b.get());
			cachedLazyStruct.init(byteArrayRef.getData(), 0, b.getSize());
		} else if (field instanceof Text) {
			Text t = (Text) field;
			if (t.getLength() == 0) {
				return null;
			}
			byteArrayRef.setData(t.getBytes());
			cachedLazyStruct.init(byteArrayRef.getData(), 0, t.getLength());
		} else {
			throw new SerDeException(getClass().toString()
					+ ": expects either BytesWritable or Text object!");
		}
		return cachedLazyStruct;
	}

	/**
	 * The reusable output buffer and serialize byte buffer.
	 */
	BytesWritable serializeBytesWritable = new BytesWritable();
	ByteStream.Output serializeByteStream = new ByteStream.Output();

	/**
	 * Serialize an object to a byte buffer in a binary compact way.
	 */
	@Override
	public Writable serialize(Object obj, ObjectInspector objInspector)
			throws SerDeException {
		// make sure it is a struct record or not
		serializeByteStream.reset();

		if (objInspector.getCategory() != Category.STRUCT) {
			// serialize the primitive object
			serialize(serializeByteStream, obj, objInspector);
		} else {
			// serialize the row as a struct
			serializeStruct(serializeByteStream, obj,
					(StructObjectInspector) objInspector);
		}
		// return the serialized bytes
		serializeBytesWritable.set(serializeByteStream.getData(), 0,
				serializeByteStream.getCount());
		return serializeBytesWritable;
	}

	boolean nullMapKey = false;

	/**
	 * Serialize a struct object without writing the byte size. This function is
	 * shared by both row serialization and struct serialization.
	 * 
	 * @param byteStream
	 *            the byte stream storing the serialization data
	 * @param obj
	 *            the struct object to serialize
	 * @param objInspector
	 *            the struct object inspector
	 */
	private void serializeStruct(Output byteStream, Object obj,
			StructObjectInspector soi) {
		// do nothing for null struct
		if (null == obj) {
			return;
		}
		/*
		 * Interleave serializing one null byte and 8 struct fields in each
		 * round, in order to support data deserialization with different table
		 * schemas
		 */
		List<? extends StructField> fields = soi.getAllStructFieldRefs();
		int size = fields.size();
		int lasti = 0;
		byte nullByte = 0;
		for (int i = 0; i < size; i++) {
			// set bit to 1 if a field is not null
			if (null != soi.getStructFieldData(obj, fields.get(i))) {
				nullByte |= 1 << (i % 8);
			}
			// write the null byte every eight elements or
			// if this is the last element and serialize the
			// corresponding 8 struct fields at the same time
			if (7 == i % 8 || i == size - 1) {
				serializeByteStream.write(nullByte);
				for (int j = lasti; j <= i; j++) {
					serialize(serializeByteStream, soi.getStructFieldData(obj,
							fields.get(j)), fields.get(j)
							.getFieldObjectInspector());
				}
				lasti = i + 1;
				nullByte = 0;
			}
		}
	}

	/**
	 * A recursive function that serialize an object to a byte buffer based on
	 * its object inspector.
	 * 
	 * @param byteStream
	 *            the byte stream storing the serialization data
	 * @param obj
	 *            the object to serialize
	 * @param objInspector
	 *            the object inspector
	 */
	private void serialize(Output byteStream, Object obj,
			ObjectInspector objInspector) {

		// do nothing for null object
		if (null == obj) {
			return;
		}

		switch (objInspector.getCategory()) {
		case PRIMITIVE: {
			PrimitiveObjectInspector poi = (PrimitiveObjectInspector) objInspector;
			switch (poi.getPrimitiveCategory()) {
			case VOID: {
				return;
			}
			case BOOLEAN: {
				boolean v = ((BooleanObjectInspector) poi).get(obj);
				byteStream.write((byte) (v ? 1 : 0));
				return;
			}
			case BYTE: {
				ByteObjectInspector boi = (ByteObjectInspector) poi;
				byte v = boi.get(obj);
				byteStream.write(v);
				return;
			}
			case SHORT: {
				ShortObjectInspector spoi = (ShortObjectInspector) poi;
				short v = spoi.get(obj);
				byteStream.write((byte) (v >> 8));
				byteStream.write((byte) (v));
				return;
			}
			case INT: {
				IntObjectInspector ioi = (IntObjectInspector) poi;
				int v = ioi.get(obj);
				LazyUtils.writeVInt(byteStream, v);
				return;
			}
			case LONG: {
				LongObjectInspector loi = (LongObjectInspector) poi;
				long v = loi.get(obj);
				LazyUtils.writeVLong(byteStream, v);
				return;
			}
			case FLOAT: {
				FloatObjectInspector foi = (FloatObjectInspector) poi;
				int v = Float.floatToIntBits(foi.get(obj));
				byteStream.write((byte) (v >> 24));
				byteStream.write((byte) (v >> 16));
				byteStream.write((byte) (v >> 8));
				byteStream.write((byte) (v));
				return;
			}
			case DOUBLE: {
				DoubleObjectInspector doi = (DoubleObjectInspector) poi;
				long v = Double.doubleToLongBits(doi.get(obj));
				byteStream.write((byte) (v >> 56));
				byteStream.write((byte) (v >> 48));
				byteStream.write((byte) (v >> 40));
				byteStream.write((byte) (v >> 32));
				byteStream.write((byte) (v >> 24));
				byteStream.write((byte) (v >> 16));
				byteStream.write((byte) (v >> 8));
				byteStream.write((byte) (v));
				return;
			}
			case STRING: {
				StringObjectInspector soi = (StringObjectInspector) poi;
				Text t = soi.getPrimitiveWritableObject(obj);
				/* write byte size of the string which is a vint */
				int length = t.getLength();
				LazyUtils.writeVInt(byteStream, length);
				/* write string itself */
				byte[] data = t.getBytes();
				byteStream.write(data, 0, length);
				return;
			}
			default: {
				throw new RuntimeException("Unrecognized type: "
						+ poi.getPrimitiveCategory());
			}
			}
		}
		case LIST: {
			ListObjectInspector loi = (ListObjectInspector) objInspector;
			ObjectInspector eoi = loi.getListElementObjectInspector();

			// 1/ reserve spaces for the byte size of the list
			// which is a integer and takes four bytes
			int byteSizeStart = byteStream.getCount();
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			int listStart = byteStream.getCount();

			// 2/ write the size of the list as a VInt
			int size = loi.getListLength(obj);
			LazyUtils.writeVInt(byteStream, size);

			// 3/ write the null bytes
			byte nullByte = 0;
			for (int eid = 0; eid < size; eid++) {
				// set the bit to 1 if an element is not null
				if (null != loi.getListElement(obj, eid)) {
					nullByte |= 1 << (eid % 8);
				}
				// store the byte every eight elements or
				// if this is the last element
				if (7 == eid % 8 || eid == size - 1) {
					byteStream.write(nullByte);
					nullByte = 0;
				}
			}

			// 4/ write element by element from the list
			for (int eid = 0; eid < size; eid++) {
				serialize(byteStream, loi.getListElement(obj, eid), eoi);
			}

			// 5/ update the list byte size
			int listEnd = byteStream.getCount();
			int listSize = listEnd - listStart;
			byte[] bytes = byteStream.getData();
			bytes[byteSizeStart] = (byte) (listSize >> 24);
			bytes[byteSizeStart + 1] = (byte) (listSize >> 16);
			bytes[byteSizeStart + 2] = (byte) (listSize >> 8);
			bytes[byteSizeStart + 3] = (byte) (listSize);

			return;
		}
		case MAP: {
			MapObjectInspector moi = (MapObjectInspector) objInspector;
			ObjectInspector koi = moi.getMapKeyObjectInspector();
			ObjectInspector voi = moi.getMapValueObjectInspector();
			Map<?, ?> map = moi.getMap(obj);

			// 1/ reserve spaces for the byte size of the map
			// which is a integer and takes four bytes
			int byteSizeStart = byteStream.getCount();
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			int mapStart = byteStream.getCount();

			// 2/ write the size of the map which is a VInt
			int size = map.size();
			LazyUtils.writeVInt(byteStream, size);

			// 3/ write the null bytes
			int b = 0;
			byte nullByte = 0;
			for (Map.Entry<?, ?> entry : map.entrySet()) {
				// set the bit to 1 if a key is not null
				if (null != entry.getKey()) {
					nullByte |= 1 << (b % 8);
				} else if (!nullMapKey) {
					nullMapKey = true;
					LOG.warn("Null map key encountered! Ignoring similar problems.");
				}
				b++;
				// set the bit to 1 if a value is not null
				if (null != entry.getValue()) {
					nullByte |= 1 << (b % 8);
				}
				b++;
				// write the byte to stream every 4 key-value pairs
				// or if this is the last key-value pair
				if (0 == b % 8 || b == size * 2) {
					byteStream.write(nullByte);
					nullByte = 0;
				}
			}

			// 4/ write key-value pairs one by one
			for (Map.Entry<?, ?> entry : map.entrySet()) {
				serialize(byteStream, entry.getKey(), koi);
				serialize(byteStream, entry.getValue(), voi);
			}

			// 5/ update the byte size of the map
			int mapEnd = byteStream.getCount();
			int mapSize = mapEnd - mapStart;
			byte[] bytes = byteStream.getData();
			bytes[byteSizeStart] = (byte) (mapSize >> 24);
			bytes[byteSizeStart + 1] = (byte) (mapSize >> 16);
			bytes[byteSizeStart + 2] = (byte) (mapSize >> 8);
			bytes[byteSizeStart + 3] = (byte) (mapSize);

			return;
		}
		case STRUCT: {
			// 1/ reserve spaces for the byte size of the struct
			// which is a integer and takes four bytes
			int byteSizeStart = byteStream.getCount();
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			byteStream.write((byte) 0);
			int structStart = byteStream.getCount();

			// 2/ serialize the struct
			serializeStruct(byteStream, obj,
					(StructObjectInspector) objInspector);

			// 3/ update the byte size of the struct
			int structEnd = byteStream.getCount();
			int structSize = structEnd - structStart;
			byte[] bytes = byteStream.getData();
			bytes[byteSizeStart] = (byte) (structSize >> 24);
			bytes[byteSizeStart + 1] = (byte) (structSize >> 16);
			bytes[byteSizeStart + 2] = (byte) (structSize >> 8);
			bytes[byteSizeStart + 3] = (byte) (structSize);

			return;
		}
		default: {
			throw new RuntimeException("Unrecognized type: "
					+ objInspector.getCategory());
		}
		}
	}
}