diff --git a/hivesterix/hivesterix-runtime/pom.xml b/hivesterix/hivesterix-runtime/pom.xml
index e4f5416..6d10286 100644
--- a/hivesterix/hivesterix-runtime/pom.xml
+++ b/hivesterix/hivesterix-runtime/pom.xml
@@ -20,16 +20,14 @@
 	<parent>
                 <groupId>edu.uci.ics.hyracks</groupId>
                 <artifactId>hivesterix</artifactId>
-                <version>0.2.7-SNAPSHOT</version>
+                <version>0.2.10-SNAPSHOT</version>
         </parent>
 
 	<dependencies>
-        <dependency>
-			<groupId>javax.servlet</groupId>
-			<artifactId>servlet-api</artifactId>
-			<version>2.5</version>
-			<type>jar</type>
-			<scope>compile</scope>
+		<dependency>
+  			<groupId>sqlline</groupId>
+  			<artifactId>sqlline</artifactId>
+  			<version>1_0_2</version>
 		</dependency>
 		<dependency>
 			<groupId>junit</groupId>
@@ -38,285 +36,105 @@
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>args4j</groupId>
-			<artifactId>args4j</artifactId>
-			<version>2.0.12</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.json</groupId>
-			<artifactId>json</artifactId>
-			<version>20090211</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.eclipse.jetty</groupId>
-			<artifactId>jetty-server</artifactId>
-			<version>8.0.0.M1</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.eclipse.jetty</groupId>
-			<artifactId>jetty-servlet</artifactId>
-			<version>8.0.0.M1</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>jline</groupId>
-			<artifactId>jline</artifactId>
-			<version>0.9.94</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.datanucleus</groupId>
-			<artifactId>datanucleus-core</artifactId>
-			<version>2.0.3</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.datanucleus</groupId>
-			<artifactId>datanucleus-connectionpool</artifactId>
-			<version>2.0.3</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.datanucleus</groupId>
-			<artifactId>datanucleus-enhancer</artifactId>
-			<version>2.0.3</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.datanucleus</groupId>
-			<artifactId>datanucleus-rdbms</artifactId>
-			<version>2.0.3</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-dbcp</groupId>
-			<artifactId>commons-dbcp</artifactId>
-			<version>1.4</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-pool</groupId>
-			<artifactId>commons-pool</artifactId>
-			<version>1.5.4</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-collections</groupId>
-			<artifactId>commons-collections</artifactId>
-			<version>3.2.1</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-lang</groupId>
-			<artifactId>commons-lang</artifactId>
-			<version>2.4</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>javax</groupId>
-			<artifactId>jdo2-api</artifactId>
-			<version>2.3-ec</version>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>com.facebook</groupId>
-			<artifactId>libfb303</artifactId>
-			<version>0.5.0</version>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache.thrift</groupId>
-			<artifactId>libthrift</artifactId>
-			<version>0.5.0</version>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache.commons</groupId>
-			<artifactId>cli</artifactId>
-			<version>1.2</version>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache</groupId>
-			<artifactId>log4j</artifactId>
-			<version>1.2.15</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.antlr</groupId>
-			<artifactId>antlr-runtime</artifactId>
-			<version>3.0.1</version>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-cli</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-common</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-exec</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-hwi</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-jdbc</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-metastore</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-service</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-shims</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.hadoop.hive</groupId>
+			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-serde</artifactId>
-			<version>0.7.0</version>
+			<version>0.11.0</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.slf4j</groupId>
-			<artifactId>slf4j-api</artifactId>
-			<version>1.6.1</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-cli</groupId>
-			<artifactId>commons-cli</artifactId>
-			<version>1.2</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.slf4j</groupId>
-			<artifactId>slf4j-log4j12</artifactId>
-			<version>1.6.1</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>commons-logging</groupId>
-			<artifactId>commons-logging</artifactId>
-			<version>1.1.1</version>
-			<type>jar</type>
-			<classifier>api</classifier>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>com.google.guava</groupId>
-			<artifactId>guava</artifactId>
-			<version>r06</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.antlr</groupId>
-			<artifactId>stringtemplate</artifactId>
-			<version>3.2</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache.derby</groupId>
-			<artifactId>derby</artifactId>
-			<version>10.8.1.2</version>
-			<type>jar</type>
-			<scope>compile</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.apache.hbase</groupId>
-			<artifactId>hbase</artifactId>
-			<version>0.90.3</version>
-			<type>jar</type>
-			<scope>compile</scope>
+			<groupId>org.apache.hive</groupId>
+			<artifactId>hive-hbase-handler</artifactId>
+			<version>0.11.0</version>
 		</dependency>
 		<dependency>
 			<groupId>edu.uci.ics.hyracks</groupId>
 			<artifactId>algebricks-compiler</artifactId>
-			<version>0.2.7-SNAPSHOT</version>
+			<version>0.2.10-SNAPSHOT</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
 			<groupId>edu.uci.ics.hyracks</groupId>
 			<artifactId>hyracks-control-cc</artifactId>
-			<version>0.2.7-SNAPSHOT</version>
+			<version>0.2.10-SNAPSHOT</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
 			<groupId>edu.uci.ics.hyracks</groupId>
 			<artifactId>hyracks-control-nc</artifactId>
-			<version>0.2.7-SNAPSHOT</version>
+			<version>0.2.10-SNAPSHOT</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
 			<groupId>edu.uci.ics.hyracks</groupId>
 			<artifactId>hivesterix-serde</artifactId>
-			<version>0.2.7-SNAPSHOT</version>
+			<version>0.2.10-SNAPSHOT</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>
 			<groupId>edu.uci.ics.hyracks</groupId>
 			<artifactId>hivesterix-common</artifactId>
-			<version>0.2.7-SNAPSHOT</version>
+			<version>0.2.10-SNAPSHOT</version>
 			<type>jar</type>
 			<scope>compile</scope>
 		</dependency>
@@ -384,5 +202,19 @@
 			<id>hyracks-public-release</id>
 			<url>http://obelix.ics.uci.edu/nexus/content/repositories/hyracks-public-releases</url>
 		</repository>
+		<repository>
+			<releases>
+				<enabled>true</enabled>
+				<updatePolicy>always</updatePolicy>
+				<checksumPolicy>warn</checksumPolicy>
+			</releases>
+			<snapshots>
+				<enabled>true</enabled>
+				<updatePolicy>always</updatePolicy>
+				<checksumPolicy>fail</checksumPolicy>
+			</snapshots>
+			<id>plugins-release</id>
+			<url>http://repo.springsource.org/plugins-release</url>
+		</repository>
 	</repositories>
 </project>
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AbstractExpressionEvaluator.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AbstractExpressionEvaluator.java
index a5177c9..dd4fbe7 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AbstractExpressionEvaluator.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AbstractExpressionEvaluator.java
@@ -34,6 +34,7 @@
 import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
+@SuppressWarnings("deprecation")
 public abstract class AbstractExpressionEvaluator implements ICopyEvaluator {
 
     private List<ICopyEvaluator> children;
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregationFunctionEvaluator.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregationFunctionEvaluator.java
index d061b23..87d2221 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregationFunctionEvaluator.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregationFunctionEvaluator.java
@@ -36,6 +36,7 @@
 import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyAggregateFunction;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
+@SuppressWarnings("deprecation")
 public class AggregationFunctionEvaluator implements ICopyAggregateFunction {
 
     /**
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregatuibFunctionSerializableEvaluator.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregatuibFunctionSerializableEvaluator.java
index f4b77b8..3f1cc27 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregatuibFunctionSerializableEvaluator.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/AggregatuibFunctionSerializableEvaluator.java
@@ -35,6 +35,7 @@
 import edu.uci.ics.hyracks.algebricks.runtime.base.ICopySerializableAggregateFunction;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
+@SuppressWarnings("deprecation")
 public class AggregatuibFunctionSerializableEvaluator implements ICopySerializableAggregateFunction {
 
     /**
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/UDTFFunctionEvaluator.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/UDTFFunctionEvaluator.java
index d91b806..b511d87 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/UDTFFunctionEvaluator.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/evaluator/UDTFFunctionEvaluator.java
@@ -35,6 +35,7 @@
 import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyUnnestingFunction;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
+@SuppressWarnings("deprecation")
 public class UDTFFunctionEvaluator implements ICopyUnnestingFunction, Collector {
 
     /**
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionFactory.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionFactory.java
index 09f0cb6..d65dc24 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionFactory.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionFactory.java
@@ -12,370 +12,383 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package edu.uci.ics.hivesterix.runtime.factory.evaluator;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
-import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluatorFactory;
-import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
-import org.apache.hadoop.hive.ql.exec.Utilities;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.plan.AggregationDesc;
-import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
-import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
-import org.apache.hadoop.hive.serde2.SerDe;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
-import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
-
-import edu.uci.ics.hivesterix.logical.expression.ExpressionTranslator;
-import edu.uci.ics.hivesterix.runtime.evaluator.AggregationFunctionEvaluator;
-import edu.uci.ics.hivesterix.runtime.jobgen.Schema;
-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.core.algebra.expressions.AggregateFunctionCallExpression;
-import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
-import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyAggregateFunction;
-import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyAggregateFunctionFactory;
-import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
-
-public class AggregationFunctionFactory implements ICopyAggregateFunctionFactory {
-
-    private static final long serialVersionUID = 1L;
-
-    /**
-     * list of parameters' serialization
-     */
-    private List<String> parametersSerialization = new ArrayList<String>();
-
-    /**
-     * the name of the udf
-     */
-    private String genericUDAFName;
-
-    /**
-     * aggregation mode
-     */
-    private GenericUDAFEvaluator.Mode mode;
-
-    /**
-     * list of type info
-     */
-    private List<TypeInfo> types = new ArrayList<TypeInfo>();
-
-    /**
-     * distinct or not
-     */
-    private boolean distinct;
-
-    /**
-     * the schema of incoming rows
-     */
-    private Schema rowSchema;
-
-    /**
-     * list of parameters
-     */
-    private transient List<ExprNodeDesc> parametersOrigin;
-
-    /**
-     * row inspector
-     */
-    private transient ObjectInspector rowInspector = null;
-
-    /**
-     * output object inspector
-     */
-    private transient ObjectInspector outputInspector = null;
-
-    /**
-     * output object inspector
-     */
-    private transient ObjectInspector outputInspectorPartial = null;
-
-    /**
-     * parameter inspectors
-     */
-    private transient ObjectInspector[] parameterInspectors = null;
-
-    /**
-     * expression desc
-     */
-    private transient HashMap<Long, List<ExprNodeDesc>> parameterExprs = new HashMap<Long, List<ExprNodeDesc>>();
-
-    /**
-     * evaluators
-     */
-    private transient HashMap<Long, ExprNodeEvaluator[]> evaluators = new HashMap<Long, ExprNodeEvaluator[]>();
-
-    /**
-     * cached parameter objects
-     */
-    private transient HashMap<Long, Object[]> cachedParameters = new HashMap<Long, Object[]>();
-
-    /**
-     * cached row object: one per thread
-     */
-    private transient HashMap<Long, LazyObject<? extends ObjectInspector>> cachedRowObjects = new HashMap<Long, LazyObject<? extends ObjectInspector>>();
-
-    /**
-     * we only use lazy serde to do serialization
-     */
-    private transient HashMap<Long, SerDe> serDe = new HashMap<Long, SerDe>();
-
-    /**
-     * udaf evaluators
-     */
-    private transient HashMap<Long, GenericUDAFEvaluator> udafsPartial = new HashMap<Long, GenericUDAFEvaluator>();
-
-    /**
-     * udaf evaluators
-     */
-    private transient HashMap<Long, GenericUDAFEvaluator> udafsComplete = new HashMap<Long, GenericUDAFEvaluator>();
-
-    /**
-     * aggregation function desc
-     */
-    private transient AggregationDesc aggregator;
-
-    /**
-     * @param aggregator
-     *            Algebricks function call expression
-     * @param oi
-     *            schema
-     */
-    public AggregationFunctionFactory(AggregateFunctionCallExpression expression, Schema oi,
-            IVariableTypeEnvironment env) throws AlgebricksException {
-
-        try {
-            aggregator = (AggregationDesc) ExpressionTranslator.getHiveExpression(expression, env);
-        } catch (Exception e) {
-            e.printStackTrace();
-            throw new AlgebricksException(e.getMessage());
-        }
-        init(aggregator.getParameters(), aggregator.getGenericUDAFName(), aggregator.getMode(),
-                aggregator.getDistinct(), oi);
-    }
-
-    /**
-     * constructor of aggregation function factory
-     * 
-     * @param inputs
-     * @param name
-     * @param udafMode
-     * @param distinct
-     * @param oi
-     */
-    private void init(List<ExprNodeDesc> inputs, String name, GenericUDAFEvaluator.Mode udafMode, boolean distinct,
-            Schema oi) {
-        parametersOrigin = inputs;
-        genericUDAFName = name;
-        mode = udafMode;
-        this.distinct = distinct;
-        rowSchema = oi;
-
-        for (ExprNodeDesc input : inputs) {
-            TypeInfo type = input.getTypeInfo();
-            if (type instanceof StructTypeInfo) {
-                types.add(TypeInfoFactory.doubleTypeInfo);
-            } else
-                types.add(type);
-
-            String s = Utilities.serializeExpression(input);
-            parametersSerialization.add(s);
-        }
-    }
-
-    @Override
-    public synchronized ICopyAggregateFunction createAggregateFunction(IDataOutputProvider provider)
-            throws AlgebricksException {
-        if (parametersOrigin == null) {
-            Configuration config = new Configuration();
-            config.setClassLoader(this.getClass().getClassLoader());
-            /**
-             * in case of class.forname(...) call in hive code
-             */
-            Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
-
-            parametersOrigin = new ArrayList<ExprNodeDesc>();
-            for (String serialization : parametersSerialization) {
-                parametersOrigin.add(Utilities.deserializeExpression(serialization, config));
-            }
-        }
-
-        /**
-         * exprs
-         */
-        if (parameterExprs == null)
-            parameterExprs = new HashMap<Long, List<ExprNodeDesc>>();
-
-        /**
-         * evaluators
-         */
-        if (evaluators == null)
-            evaluators = new HashMap<Long, ExprNodeEvaluator[]>();
-
-        /**
-         * cached parameter objects
-         */
-        if (cachedParameters == null)
-            cachedParameters = new HashMap<Long, Object[]>();
-
-        /**
-         * cached row object: one per thread
-         */
-        if (cachedRowObjects == null)
-            cachedRowObjects = new HashMap<Long, LazyObject<? extends ObjectInspector>>();
-
-        /**
-         * we only use lazy serde to do serialization
-         */
-        if (serDe == null)
-            serDe = new HashMap<Long, SerDe>();
-
-        /**
-         * UDAF functions
-         */
-        if (udafsComplete == null)
-            udafsComplete = new HashMap<Long, GenericUDAFEvaluator>();
-
-        /**
-         * UDAF functions
-         */
-        if (udafsPartial == null)
-            udafsPartial = new HashMap<Long, GenericUDAFEvaluator>();
-
-        if (parameterInspectors == null)
-            parameterInspectors = new ObjectInspector[parametersOrigin.size()];
-
-        if (rowInspector == null)
-            rowInspector = rowSchema.toObjectInspector();
-
-        // get current thread id
-        long threadId = Thread.currentThread().getId();
-
-        /**
-         * expressions, expressions are thread local
-         */
-        List<ExprNodeDesc> parameters = parameterExprs.get(threadId);
-        if (parameters == null) {
-            parameters = new ArrayList<ExprNodeDesc>();
-            for (ExprNodeDesc parameter : parametersOrigin)
-                parameters.add(parameter.clone());
-            parameterExprs.put(threadId, parameters);
-        }
-
-        /**
-         * cached parameter objects
-         */
-        Object[] cachedParas = cachedParameters.get(threadId);
-        if (cachedParas == null) {
-            cachedParas = new Object[parameters.size()];
-            cachedParameters.put(threadId, cachedParas);
-        }
-
-        /**
-         * cached row object: one per thread
-         */
-        LazyObject<? extends ObjectInspector> cachedRowObject = cachedRowObjects.get(threadId);
-        if (cachedRowObject == null) {
-            cachedRowObject = LazyFactory.createLazyObject(rowInspector);
-            cachedRowObjects.put(threadId, cachedRowObject);
-        }
-
-        /**
-         * we only use lazy serde to do serialization
-         */
-        SerDe lazySer = serDe.get(threadId);
-        if (lazySer == null) {
-            lazySer = new LazySerDe();
-            serDe.put(threadId, lazySer);
-        }
-
-        /**
-         * evaluators
-         */
-        ExprNodeEvaluator[] evals = evaluators.get(threadId);
-        if (evals == null) {
-            evals = new ExprNodeEvaluator[parameters.size()];
-            evaluators.put(threadId, evals);
-        }
-
-        GenericUDAFEvaluator udafPartial;
-        GenericUDAFEvaluator udafComplete;
-
-        // initialize object inspectors
-        try {
-            /**
-             * evaluators, udf, object inpsectors are shared in one thread
-             */
-            for (int i = 0; i < evals.length; i++) {
-                if (evals[i] == null) {
-                    evals[i] = ExprNodeEvaluatorFactory.get(parameters.get(i));
-                    if (parameterInspectors[i] == null) {
-                        parameterInspectors[i] = evals[i].initialize(rowInspector);
-                    } else {
-                        evals[i].initialize(rowInspector);
-                    }
-                }
-            }
-
-            udafComplete = udafsComplete.get(threadId);
-            if (udafComplete == null) {
-                try {
-                    udafComplete = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, types, distinct, false);
-                } catch (HiveException e) {
-                    throw new AlgebricksException(e);
-                }
-                udafsComplete.put(threadId, udafComplete);
-                udafComplete.init(mode, parameterInspectors);
-            }
-
-            // multiple stage group by, determined by the mode parameter
-            if (outputInspector == null)
-                outputInspector = udafComplete.init(mode, parameterInspectors);
-
-            // initial partial gby udaf
-            GenericUDAFEvaluator.Mode partialMode;
-            // adjust mode for external groupby
-            if (mode == GenericUDAFEvaluator.Mode.COMPLETE)
-                partialMode = GenericUDAFEvaluator.Mode.PARTIAL1;
-            else if (mode == GenericUDAFEvaluator.Mode.FINAL)
-                partialMode = GenericUDAFEvaluator.Mode.PARTIAL2;
-            else
-                partialMode = mode;
-            udafPartial = udafsPartial.get(threadId);
-            if (udafPartial == null) {
-                try {
-                    udafPartial = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, types, distinct, false);
-                } catch (HiveException e) {
-                    throw new AlgebricksException(e);
-                }
-                udafPartial.init(partialMode, parameterInspectors);
-                udafsPartial.put(threadId, udafPartial);
-            }
-
-            // multiple stage group by, determined by the mode parameter
-            if (outputInspectorPartial == null)
-                outputInspectorPartial = udafPartial.init(partialMode, parameterInspectors);
-        } catch (Exception e) {
-            e.printStackTrace();
-            throw new AlgebricksException(e);
-        }
-
-        return new AggregationFunctionEvaluator(parameters, types, genericUDAFName, mode, distinct, rowInspector,
-                provider.getDataOutput(), evals, parameterInspectors, cachedParas, lazySer, cachedRowObject,
-                udafPartial, udafComplete, outputInspector, outputInspectorPartial);
-    }
-
-    public String toString() {
-        return "aggregation function expression evaluator factory: " + this.genericUDAFName;
-    }
-}
+package edu.uci.ics.hivesterix.runtime.factory.evaluator;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
+import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluatorFactory;
+import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.AggregationDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
+
+import edu.uci.ics.hivesterix.logical.expression.ExpressionTranslator;
+import edu.uci.ics.hivesterix.runtime.evaluator.AggregationFunctionEvaluator;
+import edu.uci.ics.hivesterix.runtime.jobgen.Schema;
+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.hivesterix.serde.lazy.LazyUtils;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AggregateFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyAggregateFunction;
+import edu.uci.ics.hyracks.algebricks.runtime.base.ICopyAggregateFunctionFactory;
+import edu.uci.ics.hyracks.data.std.api.IDataOutputProvider;
+
+@SuppressWarnings("deprecation")
+public class AggregationFunctionFactory implements ICopyAggregateFunctionFactory {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * list of parameters' serialization
+     */
+    private List<String> parametersSerialization = new ArrayList<String>();
+
+    /**
+     * the name of the udf
+     */
+    private String genericUDAFName;
+
+    /**
+     * aggregation mode
+     */
+    private GenericUDAFEvaluator.Mode mode;
+
+    /**
+     * list of type info
+     */
+    private List<TypeInfo> types = new ArrayList<TypeInfo>();
+
+    /**
+     * distinct or not
+     */
+    private boolean distinct;
+
+    /**
+     * the schema of incoming rows
+     */
+    private Schema rowSchema;
+
+    /**
+     * list of parameters
+     */
+    private transient List<ExprNodeDesc> parametersOrigin;
+
+    /**
+     * row inspector
+     */
+    private transient ObjectInspector rowInspector = null;
+
+    /**
+     * output object inspector
+     */
+    private transient ObjectInspector outputInspector = null;
+
+    /**
+     * output object inspector
+     */
+    private transient ObjectInspector outputInspectorPartial = null;
+
+    /**
+     * parameter inspectors
+     */
+    private transient ObjectInspector[] parameterInspectors = null;
+
+    /**
+     * expression desc
+     */
+    private transient HashMap<Long, List<ExprNodeDesc>> parameterExprs = new HashMap<Long, List<ExprNodeDesc>>();
+
+    /**
+     * evaluators
+     */
+    private transient HashMap<Long, ExprNodeEvaluator[]> evaluators = new HashMap<Long, ExprNodeEvaluator[]>();
+
+    /**
+     * cached parameter objects
+     */
+    private transient HashMap<Long, Object[]> cachedParameters = new HashMap<Long, Object[]>();
+
+    /**
+     * cached row object: one per thread
+     */
+    private transient HashMap<Long, LazyObject<? extends ObjectInspector>> cachedRowObjects = new HashMap<Long, LazyObject<? extends ObjectInspector>>();
+
+    /**
+     * we only use lazy serde to do serialization
+     */
+    private transient HashMap<Long, SerDe> serDe = new HashMap<Long, SerDe>();
+
+    /**
+     * udaf evaluators
+     */
+    private transient HashMap<Long, GenericUDAFEvaluator> udafsPartial = new HashMap<Long, GenericUDAFEvaluator>();
+
+    /**
+     * udaf evaluators
+     */
+    private transient HashMap<Long, GenericUDAFEvaluator> udafsComplete = new HashMap<Long, GenericUDAFEvaluator>();
+
+    /**
+     * aggregation function desc
+     */
+    private transient AggregationDesc aggregator;
+
+    /**
+     * @param aggregator
+     *            Algebricks function call expression
+     * @param oi
+     *            schema
+     */
+    public AggregationFunctionFactory(AggregateFunctionCallExpression expression, Schema oi,
+            IVariableTypeEnvironment env) throws AlgebricksException {
+
+        try {
+            aggregator = (AggregationDesc) ExpressionTranslator.getHiveExpression(expression, env);
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new AlgebricksException(e.getMessage());
+        }
+        init(aggregator.getParameters(), aggregator.getGenericUDAFName(), aggregator.getMode(),
+                aggregator.getDistinct(), oi);
+    }
+
+    /**
+     * constructor of aggregation function factory
+     * 
+     * @param inputs
+     * @param name
+     * @param udafMode
+     * @param distinct
+     * @param oi
+     */
+    private void init(List<ExprNodeDesc> inputs, String name, GenericUDAFEvaluator.Mode udafMode, boolean distinct,
+            Schema oi) {
+        parametersOrigin = inputs;
+        genericUDAFName = name;
+        mode = udafMode;
+        this.distinct = distinct;
+        rowSchema = oi;
+
+        for (ExprNodeDesc input : inputs) {
+            TypeInfo type = input.getTypeInfo();
+            if (type instanceof StructTypeInfo) {
+                types.add(TypeInfoFactory.doubleTypeInfo);
+            } else {
+                types.add(type);
+            }
+
+            String s = Utilities.serializeExpression(input);
+            parametersSerialization.add(s);
+        }
+    }
+
+    @Override
+    public synchronized ICopyAggregateFunction createAggregateFunction(IDataOutputProvider provider)
+            throws AlgebricksException {
+        /**
+         * list of object inspectors correlated to types
+         */
+        List<ObjectInspector> oiListForTypes = new ArrayList<ObjectInspector>();
+        for (TypeInfo type : types) {
+            oiListForTypes.add(LazyUtils.getLazyObjectInspectorFromTypeInfo(type, false));
+        }
+
+        if (parametersOrigin == null) {
+            Configuration config = new Configuration();
+            config.setClassLoader(this.getClass().getClassLoader());
+            /**
+             * in case of class.forname(...) call in hive code
+             */
+            Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
+
+            parametersOrigin = new ArrayList<ExprNodeDesc>();
+            for (String serialization : parametersSerialization) {
+                parametersOrigin.add(Utilities.deserializeExpression(serialization, config));
+            }
+        }
+
+        /**
+         * exprs
+         */
+        if (parameterExprs == null)
+            parameterExprs = new HashMap<Long, List<ExprNodeDesc>>();
+
+        /**
+         * evaluators
+         */
+        if (evaluators == null)
+            evaluators = new HashMap<Long, ExprNodeEvaluator[]>();
+
+        /**
+         * cached parameter objects
+         */
+        if (cachedParameters == null)
+            cachedParameters = new HashMap<Long, Object[]>();
+
+        /**
+         * cached row object: one per thread
+         */
+        if (cachedRowObjects == null)
+            cachedRowObjects = new HashMap<Long, LazyObject<? extends ObjectInspector>>();
+
+        /**
+         * we only use lazy serde to do serialization
+         */
+        if (serDe == null)
+            serDe = new HashMap<Long, SerDe>();
+
+        /**
+         * UDAF functions
+         */
+        if (udafsComplete == null)
+            udafsComplete = new HashMap<Long, GenericUDAFEvaluator>();
+
+        /**
+         * UDAF functions
+         */
+        if (udafsPartial == null)
+            udafsPartial = new HashMap<Long, GenericUDAFEvaluator>();
+
+        if (parameterInspectors == null)
+            parameterInspectors = new ObjectInspector[parametersOrigin.size()];
+
+        if (rowInspector == null)
+            rowInspector = rowSchema.toObjectInspector();
+
+        // get current thread id
+        long threadId = Thread.currentThread().getId();
+
+        /**
+         * expressions, expressions are thread local
+         */
+        List<ExprNodeDesc> parameters = parameterExprs.get(threadId);
+        if (parameters == null) {
+            parameters = new ArrayList<ExprNodeDesc>();
+            for (ExprNodeDesc parameter : parametersOrigin)
+                parameters.add(parameter.clone());
+            parameterExprs.put(threadId, parameters);
+        }
+
+        /**
+         * cached parameter objects
+         */
+        Object[] cachedParas = cachedParameters.get(threadId);
+        if (cachedParas == null) {
+            cachedParas = new Object[parameters.size()];
+            cachedParameters.put(threadId, cachedParas);
+        }
+
+        /**
+         * cached row object: one per thread
+         */
+        LazyObject<? extends ObjectInspector> cachedRowObject = cachedRowObjects.get(threadId);
+        if (cachedRowObject == null) {
+            cachedRowObject = LazyFactory.createLazyObject(rowInspector);
+            cachedRowObjects.put(threadId, cachedRowObject);
+        }
+
+        /**
+         * we only use lazy serde to do serialization
+         */
+        SerDe lazySer = serDe.get(threadId);
+        if (lazySer == null) {
+            lazySer = new LazySerDe();
+            serDe.put(threadId, lazySer);
+        }
+
+        /**
+         * evaluators
+         */
+        ExprNodeEvaluator[] evals = evaluators.get(threadId);
+        if (evals == null) {
+            evals = new ExprNodeEvaluator[parameters.size()];
+            evaluators.put(threadId, evals);
+        }
+
+        GenericUDAFEvaluator udafPartial;
+        GenericUDAFEvaluator udafComplete;
+
+        // initialize object inspectors
+        try {
+            /**
+             * evaluators, udf, object inpsectors are shared in one thread
+             */
+            for (int i = 0; i < evals.length; i++) {
+                if (evals[i] == null) {
+                    evals[i] = ExprNodeEvaluatorFactory.get(parameters.get(i));
+                    if (parameterInspectors[i] == null) {
+                        parameterInspectors[i] = evals[i].initialize(rowInspector);
+                    } else {
+                        evals[i].initialize(rowInspector);
+                    }
+                }
+            }
+
+            udafComplete = udafsComplete.get(threadId);
+            if (udafComplete == null) {
+                try {
+                    udafComplete = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, oiListForTypes, distinct,
+                            false);
+                } catch (HiveException e) {
+                    throw new AlgebricksException(e);
+                }
+                udafsComplete.put(threadId, udafComplete);
+                udafComplete.init(mode, parameterInspectors);
+            }
+
+            // multiple stage group by, determined by the mode parameter
+            if (outputInspector == null)
+                outputInspector = udafComplete.init(mode, parameterInspectors);
+
+            // initial partial gby udaf
+            GenericUDAFEvaluator.Mode partialMode;
+            // adjust mode for external groupby
+            if (mode == GenericUDAFEvaluator.Mode.COMPLETE)
+                partialMode = GenericUDAFEvaluator.Mode.PARTIAL1;
+            else if (mode == GenericUDAFEvaluator.Mode.FINAL)
+                partialMode = GenericUDAFEvaluator.Mode.PARTIAL2;
+            else
+                partialMode = mode;
+            udafPartial = udafsPartial.get(threadId);
+            if (udafPartial == null) {
+                try {
+                    udafPartial = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, oiListForTypes, distinct,
+                            false);
+                } catch (HiveException e) {
+                    throw new AlgebricksException(e);
+                }
+                udafPartial.init(partialMode, parameterInspectors);
+                udafsPartial.put(threadId, udafPartial);
+            }
+
+            // multiple stage group by, determined by the mode parameter
+            if (outputInspectorPartial == null)
+                outputInspectorPartial = udafPartial.init(partialMode, parameterInspectors);
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new AlgebricksException(e);
+        }
+
+        return new AggregationFunctionEvaluator(parameters, types, genericUDAFName, mode, distinct, rowInspector,
+                provider.getDataOutput(), evals, parameterInspectors, cachedParas, lazySer, cachedRowObject,
+                udafPartial, udafComplete, outputInspector, outputInspectorPartial);
+    }
+
+    public String toString() {
+        return "aggregation function expression evaluator factory: " + this.genericUDAFName;
+    }
+}
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionSerializableFactory.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionSerializableFactory.java
index 71d11c0..c1ee814 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionSerializableFactory.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/factory/evaluator/AggregationFunctionSerializableFactory.java
@@ -39,12 +39,14 @@
 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.hivesterix.serde.lazy.LazyUtils;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AggregateFunctionCallExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import edu.uci.ics.hyracks.algebricks.runtime.base.ICopySerializableAggregateFunction;
 import edu.uci.ics.hyracks.algebricks.runtime.base.ICopySerializableAggregateFunctionFactory;
 
+@SuppressWarnings("deprecation")
 public class AggregationFunctionSerializableFactory implements ICopySerializableAggregateFunctionFactory {
 
     private static final long serialVersionUID = 1L;
@@ -190,10 +192,19 @@
             String s = Utilities.serializeExpression(input);
             parametersSerialization.add(s);
         }
+
     }
 
     @Override
     public synchronized ICopySerializableAggregateFunction createAggregateFunction() throws AlgebricksException {
+        /**
+         * list of object inspectors correlated to types
+         */
+        List<ObjectInspector> oiListForTypes = new ArrayList<ObjectInspector>();
+        for (TypeInfo type : types) {
+            oiListForTypes.add(LazyUtils.getLazyObjectInspectorFromTypeInfo(type, false));
+        }
+
         if (parametersOrigin == null) {
             Configuration config = new Configuration();
             config.setClassLoader(this.getClass().getClassLoader());
@@ -328,7 +339,8 @@
             udafComplete = udafsComplete.get(threadId);
             if (udafComplete == null) {
                 try {
-                    udafComplete = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, types, distinct, false);
+                    udafComplete = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, oiListForTypes, distinct,
+                            false);
                 } catch (HiveException e) {
                     throw new AlgebricksException(e);
                 }
@@ -352,7 +364,8 @@
             udafPartial = udafsPartial.get(threadId);
             if (udafPartial == null) {
                 try {
-                    udafPartial = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, types, distinct, false);
+                    udafPartial = FunctionRegistry.getGenericUDAFEvaluator(genericUDAFName, oiListForTypes, distinct,
+                            false);
                 } catch (HiveException e) {
                     throw new AlgebricksException(e);
                 }
diff --git a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/operator/filewrite/HiveFileWritePushRuntime.java b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/operator/filewrite/HiveFileWritePushRuntime.java
index e89a4c4..4bbb21f 100644
--- a/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/operator/filewrite/HiveFileWritePushRuntime.java
+++ b/hivesterix/hivesterix-runtime/src/main/java/edu/uci/ics/hivesterix/runtime/operator/filewrite/HiveFileWritePushRuntime.java
@@ -12,156 +12,176 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package edu.uci.ics.hivesterix.runtime.operator.filewrite;
-
-import java.nio.ByteBuffer;
-
-import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
-import org.apache.hadoop.hive.ql.exec.OperatorFactory;
-import org.apache.hadoop.hive.ql.exec.RowSchema;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
-import org.apache.hadoop.mapred.JobConf;
-
-import edu.uci.ics.hivesterix.runtime.jobgen.Schema;
-import edu.uci.ics.hivesterix.serde.lazy.LazyColumnar;
-import edu.uci.ics.hivesterix.serde.lazy.objectinspector.LazyColumnarObjectInspector;
-import edu.uci.ics.hyracks.algebricks.runtime.base.IPushRuntime;
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
-
-@SuppressWarnings("deprecation")
-public class HiveFileWritePushRuntime implements IPushRuntime {
-
-	/**
-	 * frame tuple accessor to access byte buffer
-	 */
-	private final FrameTupleAccessor accessor;
-
-	/**
-	 * input object inspector
-	 */
-	private final ObjectInspector inputInspector;
-
-	/**
-	 * cachedInput
-	 */
-	private final LazyColumnar cachedInput;
-
-	/**
-	 * File sink operator of Hive
-	 */
-	private final FileSinkDesc fileSink;
-
-	/**
-	 * job configuration, which contain name node and other configuration
-	 * information
-	 */
-	private JobConf conf;
-
-	/**
-	 * input object inspector
-	 */
-	private final Schema inputSchema;
-
-	/**
-	 * a copy of hive schema representation
-	 */
-	private RowSchema rowSchema;
-
-	/**
-	 * the Hive file sink operator
-	 */
-	private FileSinkOperator fsOp;
-
-	/**
-	 * cached tuple object reference
-	 */
-	private FrameTupleReference tuple = new FrameTupleReference();
-
-	/**
-	 * @param spec
-	 * @param fsProvider
-	 */
-	public HiveFileWritePushRuntime(IHyracksTaskContext context,
-			RecordDescriptor inputRecordDesc, JobConf job, FileSinkDesc fs,
-			RowSchema schema, Schema oi) {
-		fileSink = fs;
-		fileSink.setGatherStats(false);
-
-		rowSchema = schema;
-		conf = job;
-		inputSchema = oi;
-
-		accessor = new FrameTupleAccessor(context.getFrameSize(),
-				inputRecordDesc);
-		inputInspector = inputSchema.toObjectInspector();
-		cachedInput = new LazyColumnar(
-				(LazyColumnarObjectInspector) inputInspector);
-	}
-
-	@Override
-	public void open() throws HyracksDataException {
-		fsOp = (FileSinkOperator) OperatorFactory.get(fileSink, rowSchema);
-		fsOp.setChildOperators(null);
-		fsOp.setParentOperators(null);
-		conf.setClassLoader(this.getClass().getClassLoader());
-
-		ObjectInspector[] inspectors = new ObjectInspector[1];
-		inspectors[0] = inputInspector;
-		try {
-			fsOp.initialize(conf, inspectors);
-			fsOp.setExecContext(null);
-		} catch (Exception e) {
-			e.printStackTrace();
-		}
-	}
-
-	@Override
-	public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
-		accessor.reset(buffer);
-		int n = accessor.getTupleCount();
-		try {
-			for (int i = 0; i < n; ++i) {
-				tuple.reset(accessor, i);
-				cachedInput.init(tuple);
-				fsOp.process(cachedInput, 0);
-			}
-		} catch (HiveException e) {
-			throw new HyracksDataException(e);
-		}
-	}
-
-	@Override
-	public void close() throws HyracksDataException {
-		try {
-			Thread.currentThread().setContextClassLoader(
-					this.getClass().getClassLoader());
-			fsOp.closeOp(false);
-		} catch (HiveException e) {
-			throw new HyracksDataException(e);
-		}
-	}
-
-	@Override
-	public void setFrameWriter(int index, IFrameWriter writer,
-			RecordDescriptor recordDesc) {
-		throw new IllegalStateException();
-	}
-
-	@Override
-	public void setInputRecordDescriptor(int index,
-			RecordDescriptor recordDescriptor) {
-	}
-
-	@Override
-	public void fail() throws HyracksDataException {
-
-	}
-
-}
+package edu.uci.ics.hivesterix.runtime.operator.filewrite;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.logging.Logger;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.exec.OperatorFactory;
+import org.apache.hadoop.hive.ql.exec.RowSchema;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.shims.ShimLoader;
+import org.apache.hadoop.mapred.JobConf;
+
+import edu.uci.ics.hivesterix.runtime.jobgen.Schema;
+import edu.uci.ics.hivesterix.serde.lazy.LazyColumnar;
+import edu.uci.ics.hivesterix.serde.lazy.objectinspector.LazyColumnarObjectInspector;
+import edu.uci.ics.hyracks.algebricks.runtime.base.IPushRuntime;
+import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+
+@SuppressWarnings("deprecation")
+public class HiveFileWritePushRuntime implements IPushRuntime {
+    private final static Logger LOGGER = Logger.getLogger(HiveFileWritePushRuntime.class.getName());
+
+    /**
+     * frame tuple accessor to access byte buffer
+     */
+    private final FrameTupleAccessor accessor;
+
+    /**
+     * input object inspector
+     */
+    private final ObjectInspector inputInspector;
+
+    /**
+     * cachedInput
+     */
+    private final LazyColumnar cachedInput;
+
+    /**
+     * File sink operator of Hive
+     */
+    private final FileSinkDesc fileSink;
+
+    /**
+     * job configuration, which contain name node and other configuration
+     * information
+     */
+    private JobConf conf;
+
+    /**
+     * input object inspector
+     */
+    private final Schema inputSchema;
+
+    /**
+     * a copy of hive schema representation
+     */
+    private RowSchema rowSchema;
+
+    /**
+     * the Hive file sink operator
+     */
+    private FileSinkOperator fsOp;
+
+    /**
+     * cached tuple object reference
+     */
+    private FrameTupleReference tuple = new FrameTupleReference();
+
+    /**
+     * @param spec
+     * @param fsProvider
+     */
+    public HiveFileWritePushRuntime(IHyracksTaskContext context, RecordDescriptor inputRecordDesc, JobConf job,
+            FileSinkDesc fs, RowSchema schema, Schema oi) {
+        fileSink = fs;
+        fileSink.setGatherStats(false);
+
+        rowSchema = schema;
+        conf = job;
+        inputSchema = oi;
+
+        accessor = new FrameTupleAccessor(context.getFrameSize(), inputRecordDesc);
+        inputInspector = inputSchema.toObjectInspector();
+        cachedInput = new LazyColumnar((LazyColumnarObjectInspector) inputInspector);
+    }
+
+    @Override
+    public void open() throws HyracksDataException {
+        fsOp = (FileSinkOperator) OperatorFactory.get(fileSink, rowSchema);
+        fsOp.setChildOperators(null);
+        fsOp.setParentOperators(null);
+        conf.setClassLoader(this.getClass().getClassLoader());
+
+        ObjectInspector[] inspectors = new ObjectInspector[1];
+        inspectors[0] = inputInspector;
+        try {
+            fsOp.initialize(conf, inspectors);
+            fsOp.setExecContext(null);
+            createTempDir();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+        accessor.reset(buffer);
+        int n = accessor.getTupleCount();
+        try {
+            for (int i = 0; i < n; ++i) {
+                tuple.reset(accessor, i);
+                cachedInput.init(tuple);
+                fsOp.process(cachedInput, 0);
+            }
+        } catch (HiveException e) {
+            throw new HyracksDataException(e);
+        }
+    }
+
+    @Override
+    public void close() throws HyracksDataException {
+        try {
+            Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
+            fsOp.closeOp(false);
+        } catch (HiveException e) {
+            throw new HyracksDataException(e);
+        }
+    }
+
+    @Override
+    public void setFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
+        throw new IllegalStateException();
+    }
+
+    @Override
+    public void setInputRecordDescriptor(int index, RecordDescriptor recordDescriptor) {
+    }
+
+    @Override
+    public void fail() throws HyracksDataException {
+
+    }
+
+    private void createTempDir() throws IOException {
+        FileSinkDesc fdesc = fsOp.getConf();
+        String tempDir = fdesc.getDirName();
+        if (tempDir != null) {
+            Path tempPath = Utilities.toTempPath(new Path(tempDir));
+            FileSystem fs = tempPath.getFileSystem(conf);
+            if (!fs.exists(tempPath)) {
+                try {
+                    fs.mkdirs(tempPath);
+                    ShimLoader.getHadoopShims().fileSystemDeleteOnExit(fs, tempPath);
+                } catch (IOException e) {
+                    //if the dir already exists, that should be fine; so log a warning msg
+                    LOGGER.warning("create tmp result directory fails.");
+                }
+            }
+        }
+    }
+
+}