package edu.uci.ics.hivesterix.logical.plan.visitor;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.mutable.Mutable;
import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.hadoop.hive.ql.exec.UnionOperator;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;

import edu.uci.ics.hivesterix.logical.plan.visitor.base.DefaultVisitor;
import edu.uci.ics.hivesterix.logical.plan.visitor.base.Translator;
import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;

public class UnionVisitor extends DefaultVisitor {

    List<Mutable<ILogicalOperator>> parents = new ArrayList<Mutable<ILogicalOperator>>();

    @Override
    public Mutable<ILogicalOperator> visit(UnionOperator operator, Mutable<ILogicalOperator> AlgebricksParentOperator,
            Translator t) throws AlgebricksException {

        parents.add(AlgebricksParentOperator);
        if (operator.getParentOperators().size() > parents.size()) {
            return null;
        }

        List<LogicalVariable> leftVars = new ArrayList<LogicalVariable>();
        List<LogicalVariable> rightVars = new ArrayList<LogicalVariable>();

        VariableUtilities.getUsedVariables(parents.get(0).getValue(), leftVars);
        VariableUtilities.getUsedVariables(parents.get(1).getValue(), rightVars);

        List<Triple<LogicalVariable, LogicalVariable, LogicalVariable>> triples = new ArrayList<Triple<LogicalVariable, LogicalVariable, LogicalVariable>>();
        List<LogicalVariable> unionVars = new ArrayList<LogicalVariable>();

        for (int i = 0; i < leftVars.size(); i++) {
            LogicalVariable unionVar = t.getVariable(
                    leftVars.get(i).getId() + "union" + AlgebricksParentOperator.hashCode(),
                    TypeInfoFactory.unknownTypeInfo);
            unionVars.add(unionVar);
            Triple<LogicalVariable, LogicalVariable, LogicalVariable> triple = new Triple<LogicalVariable, LogicalVariable, LogicalVariable>(
                    leftVars.get(i), rightVars.get(i), unionVar);
            t.replaceVariable(leftVars.get(i), unionVar);
            t.replaceVariable(rightVars.get(i), unionVar);
            triples.add(triple);
        }
        ILogicalOperator currentOperator = new edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.UnionAllOperator(
                triples);
        for (Mutable<ILogicalOperator> parent : parents)
            currentOperator.getInputs().add(parent);

        t.rewriteOperatorOutputSchema(unionVars, operator);
        parents.clear();
        return new MutableObject<ILogicalOperator>(currentOperator);
    }

}