package edu.uci.ics.hivesterix.logical.expression;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.mutable.Mutable;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;

import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AggregateFunctionCallExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IExpressionTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IPartialAggregationTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;

public class HivePartialAggregationTypeComputer implements
		IPartialAggregationTypeComputer {

	public static IPartialAggregationTypeComputer INSTANCE = new HivePartialAggregationTypeComputer();

	@Override
	public Object getType(ILogicalExpression expr,
			IVariableTypeEnvironment env,
			IMetadataProvider<?, ?> metadataProvider)
			throws AlgebricksException {
		if (expr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
			IExpressionTypeComputer tc = HiveExpressionTypeComputer.INSTANCE;
			/**
			 * function expression
			 */
			AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expr;

			/**
			 * argument expressions, types, object inspectors
			 */
			List<Mutable<ILogicalExpression>> arguments = funcExpr
					.getArguments();
			List<TypeInfo> argumentTypes = new ArrayList<TypeInfo>();

			/**
			 * get types of argument
			 */
			for (Mutable<ILogicalExpression> argument : arguments) {
				TypeInfo type = (TypeInfo) tc.getType(argument.getValue(),
						metadataProvider, env);
				argumentTypes.add(type);
			}

			ObjectInspector[] childrenOIs = new ObjectInspector[argumentTypes
					.size()];

			/**
			 * get object inspector
			 */
			for (int i = 0; i < argumentTypes.size(); i++) {
				childrenOIs[i] = TypeInfoUtils
						.getStandardWritableObjectInspectorFromTypeInfo(argumentTypes
								.get(i));
			}

			/**
			 * type inference for scalar function
			 */
			if (funcExpr instanceof AggregateFunctionCallExpression) {
				/**
				 * hive aggregation info
				 */
				AggregationDesc aggregateDesc = (AggregationDesc) ((HiveFunctionInfo) funcExpr
						.getFunctionInfo()).getInfo();
				/**
				 * type inference for aggregation function
				 */
				GenericUDAFEvaluator result = aggregateDesc
						.getGenericUDAFEvaluator();

				ObjectInspector returnOI = null;
				try {
					returnOI = result.init(
							getPartialMode(aggregateDesc.getMode()),
							childrenOIs);
				} catch (HiveException e) {
					e.printStackTrace();
				}
				TypeInfo exprType = TypeInfoUtils
						.getTypeInfoFromObjectInspector(returnOI);
				return exprType;
			} else {
				throw new IllegalStateException("illegal expressions "
						+ expr.getClass().getName());
			}
		} else {
			throw new IllegalStateException("illegal expressions "
					+ expr.getClass().getName());
		}
	}

	private Mode getPartialMode(Mode mode) {
		Mode partialMode;
		if (mode == Mode.FINAL)
			partialMode = Mode.PARTIAL2;
		else if (mode == Mode.COMPLETE)
			partialMode = Mode.PARTIAL1;
		else
			partialMode = mode;
		return partialMode;
	}
}