fullstack/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AbstractExpressionEvaluator.java

package edu.uci.ics.hivesterix.runtime.evaluator;

import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.io.BytesWritable;

import edu.uci.ics.hivesterix.serde.lazy.LazyFactory;
import edu.uci.ics.hivesterix.serde.lazy.LazyObject;
import edu.uci.ics.hivesterix.serde.lazy.LazySerDe;
import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyEvaluator;
import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;

public abstract class AbstractExpressionEvaluator implements ICopyEvaluator {

    private List<ICopyEvaluator> children;

    private ExprNodeEvaluator evaluator;

    private IDataOutputProvider out;

    private ObjectInspector inspector;

    /**
     * output object inspector
     */
    private ObjectInspector outputInspector;

    /**
     * cached row object
     */
    private LazyObject<? extends ObjectInspector> cachedRowObject;

    /**
     * serializer/derialzer for lazy object
     */
    private SerDe lazySer;

    /**
     * data output
     */
    DataOutput dataOutput;

    public AbstractExpressionEvaluator(ExprNodeEvaluator hiveEvaluator, ObjectInspector oi, IDataOutputProvider output)
            throws AlgebricksException {
        evaluator = hiveEvaluator;
        out = output;
        inspector = oi;
        dataOutput = out.getDataOutput();
    }

    protected ObjectInspector getRowInspector() {
        return null;
    }

    protected IDataOutputProvider getIDataOutputProvider() {
        return out;
    }

    protected ExprNodeEvaluator getHiveEvaluator() {
        return evaluator;
    }

    public ObjectInspector getObjectInspector() {
        return inspector;
    }

    @Override
    public void evaluate(IFrameTupleReference r) throws AlgebricksException {
        // initialize hive evaluator
        try {
            if (outputInspector == null)
                outputInspector = evaluator.initialize(inspector);
        } catch (Exception e) {
            e.printStackTrace();
            throw new AlgebricksException(e.getMessage());
        }

        readIntoCache(r);
        try {
            Object result = evaluator.evaluate(cachedRowObject);

            // if (result == null) {
            // result = evaluator.evaluate(cachedRowObject);
            //
            // // check if result is null
            //
            // String errorMsg = "serialize null object in  \n output " +
            // outputInspector.toString() + " \n input "
            // + inspector.toString() + "\n ";
            // errorMsg += "";
            // List<Object> columns = ((StructObjectInspector)
            // inspector).getStructFieldsDataAsList(cachedRowObject);
            // for (Object column : columns) {
            // errorMsg += column.toString() + " ";
            // }
            // errorMsg += "\n";
            // Log.info(errorMsg);
            // System.out.println(errorMsg);
            // // result = new BooleanWritable(true);
            // throw new IllegalStateException(errorMsg);
            // }

            serializeResult(result);
        } catch (HiveException e) {
            e.printStackTrace();
            throw new AlgebricksException(e.getMessage());
        } catch (IOException e) {
            e.printStackTrace();
            throw new AlgebricksException(e.getMessage());
        }
    }

    /**
     * serialize the result
     * 
     * @param result
     *            the evaluation result
     * @throws IOException
     * @throws AlgebricksException
     */
    private void serializeResult(Object result) throws IOException, AlgebricksException {
        if (lazySer == null)
            lazySer = new LazySerDe();

        try {
            BytesWritable outputWritable = (BytesWritable) lazySer.serialize(result, outputInspector);
            dataOutput.write(outputWritable.getBytes(), 0, outputWritable.getLength());
        } catch (SerDeException e) {
            throw new AlgebricksException(e);
        }
    }

    /**
     * bind the tuple reference to the cached row object
     * 
     * @param r
     */
    private void readIntoCache(IFrameTupleReference r) {
        if (cachedRowObject == null)
            cachedRowObject = (LazyObject<? extends ObjectInspector>) LazyFactory.createLazyObject(inspector);
        cachedRowObject.init(r);
    }

    /**
     * set a list of children of this evaluator
     * 
     * @param children
     */
    public void setChildren(List<ICopyEvaluator> children) {
        this.children = children;
    }

    public void addChild(ICopyEvaluator child) {
        if (children == null)
            children = new ArrayList<ICopyEvaluator>();
        children.add(child);
    }

}