diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/logical/visitors/VariableUtilities.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/logical/visitors/VariableUtilities.java
index 08f985b..f3cf0a4 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/logical/visitors/VariableUtilities.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/logical/visitors/VariableUtilities.java
@@ -12,77 +12,86 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors;
-
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.apache.commons.lang3.mutable.Mutable;
-
-import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
-import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
-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.typing.ITypingContext;
-import edu.uci.ics.hyracks.algebricks.core.algebra.visitors.ILogicalOperatorVisitor;
-
-public class VariableUtilities {
-
-    public static void getUsedVariables(ILogicalOperator op, Collection<LogicalVariable> usedVariables)
-            throws AlgebricksException {
-        ILogicalOperatorVisitor<Void, Void> visitor = new UsedVariableVisitor(usedVariables);
-        op.accept(visitor, null);
-    }
-
-    public static void getProducedVariables(ILogicalOperator op, Collection<LogicalVariable> producedVariables)
-            throws AlgebricksException {
-        ILogicalOperatorVisitor<Void, Void> visitor = new ProducedVariableVisitor(producedVariables);
-        op.accept(visitor, null);
-    }
-
-    public static void getLiveVariables(ILogicalOperator op, Collection<LogicalVariable> schemaVariables)
-            throws AlgebricksException {
-        ILogicalOperatorVisitor<Void, Void> visitor = new SchemaVariableVisitor(schemaVariables);
-        op.accept(visitor, null);
-    }
-
-    public static void getUsedVariablesInDescendantsAndSelf(ILogicalOperator op, Collection<LogicalVariable> vars)
-            throws AlgebricksException {
-        // DFS traversal
-        VariableUtilities.getUsedVariables(op, vars);
-        for (Mutable<ILogicalOperator> c : op.getInputs()) {
-            getUsedVariablesInDescendantsAndSelf(c.getValue(), vars);
-        }
-    }
-
-    public static void substituteVariables(ILogicalOperator op, LogicalVariable v1, LogicalVariable v2,
-            ITypingContext ctx) throws AlgebricksException {
-        substituteVariables(op, v1, v2, true, ctx);
-    }
-    
-    public static void substituteVariablesInDescendantsAndSelf(ILogicalOperator op, LogicalVariable v1,
-            LogicalVariable v2, ITypingContext ctx) throws AlgebricksException {
-        for (Mutable<ILogicalOperator> childOp : op.getInputs()) {
-            substituteVariablesInDescendantsAndSelf(childOp.getValue(), v1, v2, ctx);
-        }
-        substituteVariables(op, v1, v2, true, ctx);
-    }
-    
-    public static void substituteVariables(ILogicalOperator op, LogicalVariable v1, LogicalVariable v2,
-            boolean goThroughNts, ITypingContext ctx) throws AlgebricksException {
-        ILogicalOperatorVisitor<Void, Pair<LogicalVariable, LogicalVariable>> visitor = new SubstituteVariableVisitor(
-                goThroughNts, ctx);
-        op.accept(visitor, new Pair<LogicalVariable, LogicalVariable>(v1, v2));
-    }
-    
-    public static <T> boolean varListEqualUnordered(List<T> var, List<T> varArg) {
-        Set<T> varSet = new HashSet<T>();
-        Set<T> varArgSet = new HashSet<T>();
-        varSet.addAll(var);
-        varArgSet.addAll(varArg);
-        return varSet.equals(varArgSet);
-    }
-
-}
+package edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+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.typing.ITypingContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.visitors.ILogicalOperatorVisitor;
+
+public class VariableUtilities {
+
+    public static void getUsedVariables(ILogicalOperator op, Collection<LogicalVariable> usedVariables)
+            throws AlgebricksException {
+        ILogicalOperatorVisitor<Void, Void> visitor = new UsedVariableVisitor(usedVariables);
+        op.accept(visitor, null);
+    }
+
+    public static void getProducedVariables(ILogicalOperator op, Collection<LogicalVariable> producedVariables)
+            throws AlgebricksException {
+        ILogicalOperatorVisitor<Void, Void> visitor = new ProducedVariableVisitor(producedVariables);
+        op.accept(visitor, null);
+    }
+
+    public static void getLiveVariables(ILogicalOperator op, Collection<LogicalVariable> schemaVariables)
+            throws AlgebricksException {
+        ILogicalOperatorVisitor<Void, Void> visitor = new SchemaVariableVisitor(schemaVariables);
+        op.accept(visitor, null);
+    }
+
+    public static void getUsedVariablesInDescendantsAndSelf(ILogicalOperator op, Collection<LogicalVariable> vars)
+            throws AlgebricksException {
+        // DFS traversal
+        VariableUtilities.getUsedVariables(op, vars);
+        for (Mutable<ILogicalOperator> c : op.getInputs()) {
+            getUsedVariablesInDescendantsAndSelf(c.getValue(), vars);
+        }
+    }
+
+    public static void getProducedVariablesInDescendantsAndSelf(ILogicalOperator op, Collection<LogicalVariable> vars)
+            throws AlgebricksException {
+        // DFS traversal
+        VariableUtilities.getProducedVariables(op, vars);
+        for (Mutable<ILogicalOperator> c : op.getInputs()) {
+            getProducedVariablesInDescendantsAndSelf(c.getValue(), vars);
+        }
+    }
+
+    public static void substituteVariables(ILogicalOperator op, LogicalVariable v1, LogicalVariable v2,
+            ITypingContext ctx) throws AlgebricksException {
+        substituteVariables(op, v1, v2, true, ctx);
+    }
+
+    public static void substituteVariablesInDescendantsAndSelf(ILogicalOperator op, LogicalVariable v1,
+            LogicalVariable v2, ITypingContext ctx) throws AlgebricksException {
+        for (Mutable<ILogicalOperator> childOp : op.getInputs()) {
+            substituteVariablesInDescendantsAndSelf(childOp.getValue(), v1, v2, ctx);
+        }
+        substituteVariables(op, v1, v2, true, ctx);
+    }
+
+    public static void substituteVariables(ILogicalOperator op, LogicalVariable v1, LogicalVariable v2,
+            boolean goThroughNts, ITypingContext ctx) throws AlgebricksException {
+        ILogicalOperatorVisitor<Void, Pair<LogicalVariable, LogicalVariable>> visitor = new SubstituteVariableVisitor(
+                goThroughNts, ctx);
+        op.accept(visitor, new Pair<LogicalVariable, LogicalVariable>(v1, v2));
+    }
+
+    public static <T> boolean varListEqualUnordered(List<T> var, List<T> varArg) {
+        Set<T> varSet = new HashSet<T>();
+        Set<T> varArgSet = new HashSet<T>();
+        varSet.addAll(var);
+        varArgSet.addAll(varArg);
+        return varSet.equals(varArgSet);
+    }
+
+}
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/util/OperatorPropertiesUtil.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/util/OperatorPropertiesUtil.java
index 0838291..6648252 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/util/OperatorPropertiesUtil.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/util/OperatorPropertiesUtil.java
@@ -21,6 +21,7 @@
 import org.apache.commons.lang3.mutable.Mutable;
 
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
@@ -99,6 +100,55 @@
         }
     }
 
+    /**
+     * Adds the free variables of the operator path from
+     * op to dest, where dest is a direct/indirect input operator of op in the query plan.
+     * 
+     * @param op
+     *            , the start operator.
+     * @param dest
+     *            , the destination operator (a direct/indirect input operator).
+     * @param freeVars
+     *            - The collection to which the free variables will be added.
+     */
+    public static void getFreeVariablesInPath(ILogicalOperator op, ILogicalOperator dest, Set<LogicalVariable> freeVars)
+            throws AlgebricksException {
+        Set<LogicalVariable> producedVars = new ListSet<LogicalVariable>();
+        VariableUtilities.getLiveVariables(op, freeVars);
+        collectUsedAndProducedVariablesInPath(op, dest, freeVars, producedVars);
+        freeVars.removeAll(producedVars);
+    }
+
+    /**
+     * @param op
+     *            , the start operator.
+     * @param dest
+     *            , the destination operator (a direct/indirect input operator).
+     * @param usedVars
+     *            , the collection of used variables.
+     * @param producedVars
+     *            , the collection of produced variables.
+     * @return if the current operator is on the path from the original start operator to the destination operator.
+     * @throws AlgebricksException
+     */
+    private static boolean collectUsedAndProducedVariablesInPath(ILogicalOperator op, ILogicalOperator dest,
+            Set<LogicalVariable> usedVars, Set<LogicalVariable> producedVars) throws AlgebricksException {
+        if (op == dest) {
+            return true;
+        }
+        boolean onPath = false;
+        for (Mutable<ILogicalOperator> childRef : op.getInputs()) {
+            if (collectUsedAndProducedVariablesInPath(childRef.getValue(), dest, usedVars, producedVars)) {
+                onPath = true;
+            }
+        }
+        if (onPath) {
+            VariableUtilities.getUsedVariables(op, usedVars);
+            VariableUtilities.getProducedVariables(op, producedVars);
+        }
+        return onPath;
+    }
+
     public static void getFreeVariablesInSubplans(AbstractOperatorWithNestedPlans op, Set<LogicalVariable> freeVars)
             throws AlgebricksException {
         for (ILogicalPlan p : op.getNestedPlans()) {
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java
new file mode 100644
index 0000000..1397956
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java
@@ -0,0 +1,199 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * This rule eliminates a subplan with the following pattern:
+ * -- SUBPLAN
+ * -- OP (where OP produces exactly one tuple)
+ * The live variables at OP will not be used after SUBPLAN.
+ * Note: This rule must be applied after
+ * the RemoveRedundantVariablesRule (to avoid the lineage analysis of variable cardinality).
+ * 
+ * @author yingyib
+ */
+public class EliminateSubplanWithInputCardinalityOneRule implements IAlgebraicRewriteRule {
+    /** The pointer to the topmost operator */
+    private Mutable<ILogicalOperator> rootRef;
+    /** Whether the rule has even been invoked */
+    private boolean invoked = false;
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        if (!invoked) {
+            rootRef = opRef;
+            invoked = true;
+        }
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getInputs().size() <= 0) {
+            return false;
+        }
+        boolean changed = false;
+        for (Mutable<ILogicalOperator> subplanRef : op.getInputs()) {
+            AbstractLogicalOperator op1 = (AbstractLogicalOperator) subplanRef.getValue();
+            if (op1.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+                continue;
+            }
+
+            SubplanOperator subplan = (SubplanOperator) op1;
+            Set<LogicalVariable> usedVarsUp = new ListSet<LogicalVariable>();
+            OperatorPropertiesUtil.getFreeVariablesInPath(rootRef.getValue(), subplan, usedVarsUp);
+            // TODO(buyingyi): figure out the rewriting for subplan operators with multiple subplans.
+            if (subplan.getNestedPlans().size() != 1) {
+                continue;
+            }
+
+            ILogicalOperator subplanInputOperator = subplan.getInputs().get(0).getValue();
+            Set<LogicalVariable> subplanInputVars = new ListSet<LogicalVariable>();
+            VariableUtilities.getLiveVariables(subplanInputOperator, subplanInputVars);
+            int subplanInputVarSize = subplanInputVars.size();
+            subplanInputVars.removeAll(usedVarsUp);
+            // Makes sure the free variables are only used in the subplan.
+            if (subplanInputVars.size() < subplanInputVarSize) {
+                continue;
+            }
+            Set<LogicalVariable> freeVars = new ListSet<LogicalVariable>();
+            OperatorPropertiesUtil.getFreeVariablesInSubplans(subplan, freeVars);
+            boolean cardinalityOne = isCardinalityOne(subplan.getInputs().get(0), freeVars);
+            if (cardinalityOne) {
+                /** If the cardinality of freeVars in the subplan is one, the subplan can be removed. */
+                ILogicalPlan plan = subplan.getNestedPlans().get(0);
+
+                List<Mutable<ILogicalOperator>> rootRefs = plan.getRoots();
+                // TODO(buyingyi): investigate the case of multi-root plans.
+                if (rootRefs.size() != 1) {
+                    continue;
+                }
+                Set<Mutable<ILogicalOperator>> ntsSet = new ListSet<Mutable<ILogicalOperator>>();
+                findNts(rootRefs.get(0), ntsSet);
+
+                /** Replaces nts with the input operator of the subplan. */
+                for (Mutable<ILogicalOperator> nts : ntsSet) {
+                    nts.setValue(subplanInputOperator);
+                }
+                subplanRef.setValue(rootRefs.get(0).getValue());
+                changed = true;
+            } else {
+                continue;
+            }
+        }
+        return changed;
+    }
+
+    /**
+     * Whether the cardinality of the input free variables are one.
+     * 
+     * @param opRef
+     *            the operator to be checked (including its input operators)
+     * @param freeVars
+     *            variables to be checked for produced operators
+     * @return true if every input variable has cardinality one; false otherwise.
+     * @throws AlgebricksException
+     */
+    private boolean isCardinalityOne(Mutable<ILogicalOperator> opRef, Set<LogicalVariable> freeVars)
+            throws AlgebricksException {
+        Set<LogicalVariable> varsWithCardinalityOne = new ListSet<LogicalVariable>();
+        Set<LogicalVariable> varsLiveAtUnnestAndJoin = new ListSet<LogicalVariable>();
+        isCardinalityOne(opRef, freeVars, varsWithCardinalityOne, varsLiveAtUnnestAndJoin);
+        varsWithCardinalityOne.removeAll(varsLiveAtUnnestAndJoin);
+        return varsWithCardinalityOne.equals(freeVars);
+    }
+
+    /**
+     * Recursively adding variables which has cardinality one and in int the input free variable set.
+     * 
+     * @param opRef
+     *            , the current operator reference.
+     * @param freeVars
+     *            , a set of variables.
+     * @param varsWithCardinalityOne
+     *            , variables in the free variable set with cardinality one at the time they are created.
+     * @param varsLiveAtUnnestAndJoin
+     *            , live variables at Unnest and Join. The cardinalities of those variables can become more than one
+     *            even if their cardinalities were one at the time those variables were created.
+     * @throws AlgebricksException
+     */
+    private void isCardinalityOne(Mutable<ILogicalOperator> opRef, Set<LogicalVariable> freeVars,
+            Set<LogicalVariable> varsWithCardinalityOne, Set<LogicalVariable> varsLiveAtUnnestAndJoin)
+            throws AlgebricksException {
+        AbstractLogicalOperator operator = (AbstractLogicalOperator) opRef.getValue();
+        List<LogicalVariable> producedVars = new ArrayList<LogicalVariable>();
+        VariableUtilities.getProducedVariables(operator, producedVars);
+        if (operator.getOperatorTag() == LogicalOperatorTag.UNNEST
+                || operator.getOperatorTag() == LogicalOperatorTag.INNERJOIN
+                || operator.getOperatorTag() == LogicalOperatorTag.LEFTOUTERJOIN) {
+            VariableUtilities.getLiveVariables(operator, varsLiveAtUnnestAndJoin);
+        }
+        if (operator.getOperatorTag() == LogicalOperatorTag.AGGREGATE) {
+            for (LogicalVariable producedVar : producedVars) {
+                if (freeVars.contains(producedVar)) {
+                    varsWithCardinalityOne.add(producedVar);
+                }
+            }
+        }
+        if (varsWithCardinalityOne.size() == freeVars.size()) {
+            return;
+        }
+        for (Mutable<ILogicalOperator> childRef : operator.getInputs()) {
+            isCardinalityOne(childRef, freeVars, varsWithCardinalityOne, varsLiveAtUnnestAndJoin);
+        }
+    }
+
+    /**
+     * Find the NestedTupleSource operator in the direct/undirect input operators of opRef.
+     * 
+     * @param opRef
+     *            , the current operator reference.
+     * @param ntsSet
+     *            , the set NestedTupleSource operator references.
+     */
+    private void findNts(Mutable<ILogicalOperator> opRef, Set<Mutable<ILogicalOperator>> ntsSet) {
+        int childSize = opRef.getValue().getInputs().size();
+        if (childSize == 0) {
+            AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+            if (op.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+                ntsSet.add(opRef);
+            }
+            return;
+        }
+        for (Mutable<ILogicalOperator> childRef : opRef.getValue().getInputs()) {
+            findNts(childRef, ntsSet);
+        }
+    }
+}
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroduceGroupByForSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroduceGroupByForSubplanRule.java
index 1c66317..ab7835d 100644
--- a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroduceGroupByForSubplanRule.java
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroduceGroupByForSubplanRule.java
@@ -27,6 +27,7 @@
 import org.apache.commons.lang3.mutable.MutableObject;
 
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
 import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
@@ -160,6 +161,16 @@
                 testForNull = innerUnnest.getVariable();
                 break;
             }
+            case RUNNINGAGGREGATE: {
+                ILogicalOperator inputToRunningAggregate = right.getInputs().get(0).getValue();
+                Set<LogicalVariable> producedVars = new ListSet<LogicalVariable>();
+                VariableUtilities.getProducedVariables(inputToRunningAggregate, producedVars);
+                if (!producedVars.isEmpty()) {
+                    // Select [ $y != null ]
+                    testForNull = producedVars.iterator().next();
+                }
+                break;
+            }
             case DATASOURCESCAN: {
                 DataSourceScanOperator innerScan = (DataSourceScanOperator) right;
                 // Select [ $y != null ]
@@ -184,7 +195,8 @@
         IFunctionInfo finfoNot = context.getMetadataProvider().lookupFunction(AlgebricksBuiltinFunctions.NOT);
         ScalarFunctionCallExpression nonNullTest = new ScalarFunctionCallExpression(finfoNot,
                 new MutableObject<ILogicalExpression>(isNullTest));
-        SelectOperator selectNonNull = new SelectOperator(new MutableObject<ILogicalExpression>(nonNullTest), false, null);
+        SelectOperator selectNonNull = new SelectOperator(new MutableObject<ILogicalExpression>(nonNullTest), false,
+                null);
         GroupByOperator g = new GroupByOperator();
         Mutable<ILogicalOperator> newSubplanRef = new MutableObject<ILogicalOperator>(subplan);
         NestedTupleSourceOperator nts = new NestedTupleSourceOperator(new MutableObject<ILogicalOperator>(g));
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java
new file mode 100644
index 0000000..4dfede0
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * This rule pushes a subplan on top of a group-by into the
+ * nested plan of the group-by.
+ * 
+ * @author yingyib
+ */
+public class PushSubplanIntoGroupByRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        ILogicalOperator parentOperator = opRef.getValue();
+        if (parentOperator.getInputs().size() <= 0) {
+            return false;
+        }
+        boolean changed = false;
+        for (Mutable<ILogicalOperator> ref : parentOperator.getInputs()) {
+            AbstractLogicalOperator op = (AbstractLogicalOperator) ref.getValue();
+            /** Only processes subplan operator. */
+            if (op.getOperatorTag() == LogicalOperatorTag.SUBPLAN) {
+                AbstractLogicalOperator child = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
+                /** Only processes the case a group-by operator is the input of the subplan operator. */
+                if (child.getOperatorTag() == LogicalOperatorTag.GROUP) {
+                    SubplanOperator subplan = (SubplanOperator) op;
+                    GroupByOperator gby = (GroupByOperator) child;
+                    List<ILogicalPlan> subplanNestedPlans = subplan.getNestedPlans();
+                    List<ILogicalPlan> gbyNestedPlans = gby.getNestedPlans();
+                    List<ILogicalPlan> subplanNestedPlansToRemove = new ArrayList<ILogicalPlan>();
+                    for (ILogicalPlan subplanNestedPlan : subplanNestedPlans) {
+                        List<Mutable<ILogicalOperator>> rootOpRefs = subplanNestedPlan.getRoots();
+                        List<Mutable<ILogicalOperator>> rootOpRefsToRemove = new ArrayList<Mutable<ILogicalOperator>>();
+                        for (Mutable<ILogicalOperator> rootOpRef : rootOpRefs) {
+                            /** Gets free variables in the root operator of a nested plan and its descent. */
+                            Set<LogicalVariable> freeVars = new ListSet<LogicalVariable>();
+                            VariableUtilities.getUsedVariablesInDescendantsAndSelf(rootOpRef.getValue(), freeVars);
+                            Set<LogicalVariable> producedVars = new ListSet<LogicalVariable>();
+                            VariableUtilities.getProducedVariablesInDescendantsAndSelf(rootOpRef.getValue(),
+                                    producedVars);
+                            freeVars.removeAll(producedVars);
+
+                            /**
+                             * Checks whether the above freeVars are all contained in live variables
+                             * of one nested plan inside the group-by operator.
+                             * If yes, then the subplan can be pushed into the nested plan of the group-by.
+                             */
+                            for (ILogicalPlan gbyNestedPlan : gbyNestedPlans) {
+                                List<Mutable<ILogicalOperator>> gbyRootOpRefs = gbyNestedPlan.getRoots();
+                                for (Mutable<ILogicalOperator> gbyRootOpRef : gbyRootOpRefs) {
+                                    Set<LogicalVariable> liveVars = new ListSet<LogicalVariable>();
+                                    VariableUtilities.getLiveVariables(gbyRootOpRef.getValue(), liveVars);
+                                    if (liveVars.containsAll(freeVars)) {
+                                        /** Does the actual push. */
+                                        Mutable<ILogicalOperator> ntsRef = downToNts(rootOpRef);
+                                        ntsRef.setValue(gbyRootOpRef.getValue());
+                                        gbyRootOpRef.setValue(rootOpRef.getValue());
+                                        rootOpRefsToRemove.add(rootOpRef);
+                                        changed = true;
+                                    }
+                                }
+                            }
+                        }
+                        rootOpRefs.removeAll(rootOpRefsToRemove);
+                        if (rootOpRefs.size() == 0) {
+                            subplanNestedPlansToRemove.add(subplanNestedPlan);
+                        }
+                    }
+                    subplanNestedPlans.removeAll(subplanNestedPlansToRemove);
+                    if (subplanNestedPlans.size() == 0) {
+                        ref.setValue(gby);
+                    }
+                }
+            }
+        }
+        return changed;
+    }
+
+    private Mutable<ILogicalOperator> downToNts(Mutable<ILogicalOperator> opRef) {
+        Mutable<ILogicalOperator> currentOpRef = opRef;
+        while (currentOpRef.getValue().getInputs().size() > 0) {
+            currentOpRef = currentOpRef.getValue().getInputs().get(0);
+        }
+        return currentOpRef;
+    }
+}
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
index 918be11..c8dc852 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
@@ -37,9 +37,9 @@
 public class NestedPlansRunningAggregatorFactory implements IAggregatorDescriptorFactory {
 
     private static final long serialVersionUID = 1L;
-    private AlgebricksPipeline[] subplans;
-    private int[] keyFieldIdx;
-    private int[] decorFieldIdx;
+    private final AlgebricksPipeline[] subplans;
+    private final int[] keyFieldIdx;
+    private final int[] decorFieldIdx;
 
     public NestedPlansRunningAggregatorFactory(AlgebricksPipeline[] subplans, int[] keyFieldIdx, int[] decorFieldIdx) {
         this.subplans = subplans;
@@ -93,7 +93,6 @@
                 for (int i = 0; i < pipelines.length; i++) {
                     outputWriter.setInputIdx(i);
                     pipelines[i].writeTuple(accessor.getBuffer(), tIndex);
-                    pipelines[i].forceFlush();
                 }
             }
 
@@ -103,13 +102,16 @@
                 for (int i = 0; i < pipelines.length; i++) {
                     outputWriter.setInputIdx(i);
                     pipelines[i].writeTuple(accessor.getBuffer(), tIndex);
-                    pipelines[i].forceFlush();
                 }
             }
 
             @Override
             public boolean outputFinalResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
                     AggregateState state) throws HyracksDataException {
+                for (int i = 0; i < pipelines.length; ++i) {
+                    outputWriter.setInputIdx(i);
+                    pipelines[i].close();
+                }
                 return false;
             }
 
@@ -131,14 +133,7 @@
 
             @Override
             public void close() {
-                for (int i = 0; i < pipelines.length; ++i) {
-                    try {
-                        outputWriter.setInputIdx(i);
-                        pipelines[i].close();
-                    } catch (HyracksDataException e) {
-                        throw new IllegalStateException(e);
-                    }
-                }
+
             }
         };
     }
@@ -165,15 +160,15 @@
 
     private static class RunningAggregatorOutput implements IFrameWriter {
 
-        private FrameTupleAccessor[] tAccess;
-        private RecordDescriptor[] inputRecDesc;
+        private final FrameTupleAccessor[] tAccess;
+        private final RecordDescriptor[] inputRecDesc;
         private int inputIdx;
-        private ArrayTupleBuilder tb;
-        private ArrayTupleBuilder gbyTb;
-        private AlgebricksPipeline[] subplans;
-        private IFrameWriter outputWriter;
-        private ByteBuffer outputFrame;
-        private FrameTupleAppender outputAppender;
+        private final ArrayTupleBuilder tb;
+        private final ArrayTupleBuilder gbyTb;
+        private final AlgebricksPipeline[] subplans;
+        private final IFrameWriter outputWriter;
+        private final ByteBuffer outputFrame;
+        private final FrameTupleAppender outputAppender;
 
         public RunningAggregatorOutput(IHyracksTaskContext ctx, AlgebricksPipeline[] subplans, int numKeys,
                 int numDecors, IFrameWriter outputWriter) throws HyracksDataException {
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
index 7ecb288..2e1171c 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
@@ -27,7 +27,6 @@
 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.comm.io.FrameTupleAppender;
-import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
 import edu.uci.ics.hyracks.dataflow.std.group.preclustered.PreclusteredGroupWriter;
 
@@ -76,9 +75,7 @@
 
                 @Override
                 public void open() throws HyracksDataException {
-                    IAggregatorDescriptor aggregator = aggregatorFactory.createAggregator(ctx, inRecordDesc,
-                            outRecordDesc, groupFields, groupFields, writer);
-                    pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregator, inRecordDesc,
+                    pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregatorFactory, inRecordDesc,
                             outRecordDesc, writer);
                     pgw.open();
                 }
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
index 600d641..154f2d1 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
@@ -28,15 +28,14 @@
 import edu.uci.ics.hyracks.data.std.api.IPointable;
 import edu.uci.ics.hyracks.data.std.primitive.VoidPointable;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
-import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
 
 public class RunningAggregateRuntimeFactory extends AbstractOneInputOneOutputRuntimeFactory {
 
     private static final long serialVersionUID = 1L;
 
-    private int[] outColumns;
-    private IRunningAggregateEvaluatorFactory[] runningAggregates;
+    private final int[] outColumns;
+    private final IRunningAggregateEvaluatorFactory[] runningAggregates;
 
     /**
      * @param outColumns
@@ -83,17 +82,13 @@
         }
 
         return new AbstractOneInputOneOutputOneFramePushRuntime() {
-            private IPointable p = VoidPointable.FACTORY.createPointable();
-            private IRunningAggregateEvaluator[] raggs = new IRunningAggregateEvaluator[runningAggregates.length];
-            private ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(projectionList.length);
+            private final IPointable p = VoidPointable.FACTORY.createPointable();
+            private final IRunningAggregateEvaluator[] raggs = new IRunningAggregateEvaluator[runningAggregates.length];
+            private final ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(projectionList.length);
             private boolean first = true;
 
             @Override
             public void open() throws HyracksDataException {
-                if (!first) {
-                    FrameUtils.flushFrame(frame, writer);
-                    appender.reset(frame, true);
-                }
                 initAccessAppendRef(ctx);
                 if (first) {
                     first = false;
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java b/hyracks/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java
new file mode 100644
index 0000000..2de4256
--- /dev/null
+++ b/hyracks/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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.hyracks.dataflow.common.comm.io;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
+import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
+
+/**
+ * This class wraps the calls of FrameTupleAppender and
+ * allows user to not worry about flushing full frames.
+ * TODO(yingyib): cleanup existing usage of FrameTupleAppender.
+ * 
+ * @author yingyib
+ */
+public class FrameTupleAppenderWrapper {
+    private final FrameTupleAppender frameTupleAppender;
+    private final ByteBuffer outputFrame;
+    private final IFrameWriter outputWriter;
+
+    public FrameTupleAppenderWrapper(FrameTupleAppender frameTupleAppender, ByteBuffer outputFrame,
+            IFrameWriter outputWriter) {
+        this.frameTupleAppender = frameTupleAppender;
+        this.outputFrame = outputFrame;
+        this.outputWriter = outputWriter;
+    }
+
+    public void open() throws HyracksDataException {
+        outputWriter.open();
+    }
+
+    public void flush() throws HyracksDataException {
+        if (frameTupleAppender.getTupleCount() > 0) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+        }
+    }
+
+    public void close() throws HyracksDataException {
+        outputWriter.close();
+    }
+
+    public void fail() throws HyracksDataException {
+        outputWriter.fail();
+    }
+
+    public void reset(ByteBuffer buffer, boolean clear) {
+        frameTupleAppender.reset(buffer, clear);
+    }
+
+    public void appendSkipEmptyField(int[] fieldSlots, byte[] bytes, int offset, int length)
+            throws HyracksDataException {
+        if (!frameTupleAppender.append(fieldSlots, bytes, offset, length)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.appendSkipEmptyField(fieldSlots, bytes, offset, length)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void append(byte[] bytes, int offset, int length) throws HyracksDataException {
+        if (!frameTupleAppender.append(bytes, offset, length)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.append(bytes, offset, length)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void append(IFrameTupleAccessor tupleAccessor, int tStartOffset, int tEndOffset) throws HyracksDataException {
+        if (!frameTupleAppender.append(tupleAccessor, tStartOffset, tEndOffset)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.append(tupleAccessor, tStartOffset, tEndOffset)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void append(IFrameTupleAccessor tupleAccessor, int tIndex) throws HyracksDataException {
+        if (!frameTupleAppender.append(tupleAccessor, tIndex)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.append(tupleAccessor, tIndex)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void appendConcat(IFrameTupleAccessor accessor0, int tIndex0, IFrameTupleAccessor accessor1, int tIndex1)
+            throws HyracksDataException {
+        if (!frameTupleAppender.appendConcat(accessor0, tIndex0, accessor1, tIndex1)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.appendConcat(accessor0, tIndex0, accessor1, tIndex1)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void appendConcat(IFrameTupleAccessor accessor0, int tIndex0, int[] fieldSlots1, byte[] bytes1, int offset1,
+            int dataLen1) throws HyracksDataException {
+        if (!frameTupleAppender.appendConcat(accessor0, tIndex0, fieldSlots1, bytes1, offset1, dataLen1)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.appendConcat(accessor0, tIndex0, fieldSlots1, bytes1, offset1, dataLen1)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+    public void appendProjection(IFrameTupleAccessor accessor, int tIndex, int[] fields) throws HyracksDataException {
+        if (!frameTupleAppender.appendProjection(accessor, tIndex, fields)) {
+            FrameUtils.flushFrame(outputFrame, outputWriter);
+            frameTupleAppender.reset(outputFrame, true);
+            if (!frameTupleAppender.appendProjection(accessor, tIndex, fields)) {
+                throw new HyracksDataException("The output cannot be fit into a frame.");
+            }
+        }
+    }
+
+}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
index 71af928..9ce70c1 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
@@ -24,11 +24,9 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
-import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
 
-class PreclusteredGroupOperatorNodePushable extends
-        AbstractUnaryInputUnaryOutputOperatorNodePushable {
+class PreclusteredGroupOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
     private final IHyracksTaskContext ctx;
     private final int[] groupFields;
     private final IBinaryComparatorFactory[] comparatorFactories;
@@ -54,15 +52,13 @@
         for (int i = 0; i < comparatorFactories.length; ++i) {
             comparators[i] = comparatorFactories[i].createBinaryComparator();
         }
-        final IAggregatorDescriptor aggregator = aggregatorFactory.createAggregator(ctx, inRecordDescriptor,
-                outRecordDescriptor, groupFields, groupFields, writer);
         final ByteBuffer copyFrame = ctx.allocateFrame();
         final FrameTupleAccessor copyFrameAccessor = new FrameTupleAccessor(ctx.getFrameSize(), inRecordDescriptor);
         copyFrameAccessor.reset(copyFrame);
         ByteBuffer outFrame = ctx.allocateFrame();
         final FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
         appender.reset(outFrame, true);
-        pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregator, inRecordDescriptor,
+        pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregatorFactory, inRecordDescriptor,
                 outRecordDescriptor, writer);
         pgw.open();
     }
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
index 114463f..b67e236 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
@@ -24,22 +24,22 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppenderWrapper;
 import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
 import edu.uci.ics.hyracks.dataflow.std.group.AggregateState;
 import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
+import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
 
 public class PreclusteredGroupWriter implements IFrameWriter {
     private final int[] groupFields;
     private final IBinaryComparator[] comparators;
     private final IAggregatorDescriptor aggregator;
     private final AggregateState aggregateState;
-    private final IFrameWriter writer;
     private final ByteBuffer copyFrame;
     private final FrameTupleAccessor inFrameAccessor;
     private final FrameTupleAccessor copyFrameAccessor;
 
-    private final ByteBuffer outFrame;
-    private final FrameTupleAppender appender;
+    private final FrameTupleAppenderWrapper appenderWrapper;
     private final ArrayTupleBuilder tupleBuilder;
     private boolean outputPartial = false;
 
@@ -48,35 +48,36 @@
     private boolean isFailed = false;
 
     public PreclusteredGroupWriter(IHyracksTaskContext ctx, int[] groupFields, IBinaryComparator[] comparators,
-            IAggregatorDescriptor aggregator, RecordDescriptor inRecordDesc, RecordDescriptor outRecordDesc,
-            IFrameWriter writer, boolean outputPartial) throws HyracksDataException {
-        this(ctx, groupFields, comparators, aggregator, inRecordDesc, outRecordDesc, writer);
+            IAggregatorDescriptorFactory aggregatorFactory, RecordDescriptor inRecordDesc,
+            RecordDescriptor outRecordDesc, IFrameWriter writer, boolean outputPartial) throws HyracksDataException {
+        this(ctx, groupFields, comparators, aggregatorFactory, inRecordDesc, outRecordDesc, writer);
         this.outputPartial = outputPartial;
     }
 
     public PreclusteredGroupWriter(IHyracksTaskContext ctx, int[] groupFields, IBinaryComparator[] comparators,
-            IAggregatorDescriptor aggregator, RecordDescriptor inRecordDesc, RecordDescriptor outRecordDesc,
-            IFrameWriter writer) throws HyracksDataException {
+            IAggregatorDescriptorFactory aggregatorFactory, RecordDescriptor inRecordDesc,
+            RecordDescriptor outRecordDesc, IFrameWriter writer) throws HyracksDataException {
         this.groupFields = groupFields;
         this.comparators = comparators;
-        this.aggregator = aggregator;
+        this.aggregator = aggregatorFactory.createAggregator(ctx, inRecordDesc, outRecordDesc, groupFields,
+                groupFields, writer);
         this.aggregateState = aggregator.createAggregateStates();
-        this.writer = writer;
         copyFrame = ctx.allocateFrame();
         inFrameAccessor = new FrameTupleAccessor(ctx.getFrameSize(), inRecordDesc);
         copyFrameAccessor = new FrameTupleAccessor(ctx.getFrameSize(), inRecordDesc);
         copyFrameAccessor.reset(copyFrame);
 
-        outFrame = ctx.allocateFrame();
-        appender = new FrameTupleAppender(ctx.getFrameSize());
+        ByteBuffer outFrame = ctx.allocateFrame();
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
         appender.reset(outFrame, true);
+        appenderWrapper = new FrameTupleAppenderWrapper(appender, outFrame, writer);
 
         tupleBuilder = new ArrayTupleBuilder(outRecordDesc.getFields().length);
     }
 
     @Override
     public void open() throws HyracksDataException {
-        writer.open();
+        appenderWrapper.open();
         first = true;
     }
 
@@ -133,15 +134,9 @@
                 lastTupleIndex, aggregateState) : aggregator.outputFinalResult(tupleBuilder, lastTupleAccessor,
                 lastTupleIndex, aggregateState);
 
-        if (hasOutput
-                && !appender.appendSkipEmptyField(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
-                        tupleBuilder.getSize())) {
-            FrameUtils.flushFrame(outFrame, writer);
-            appender.reset(outFrame, true);
-            if (!appender.appendSkipEmptyField(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
-                    tupleBuilder.getSize())) {
-                throw new HyracksDataException("The output cannot be fit into a frame.");
-            }
+        if (hasOutput) {
+            appenderWrapper.appendSkipEmptyField(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
+                    tupleBuilder.getSize());
         }
 
     }
@@ -163,19 +158,17 @@
     @Override
     public void fail() throws HyracksDataException {
         isFailed = true;
-        writer.fail();
+        appenderWrapper.fail();
     }
 
     @Override
     public void close() throws HyracksDataException {
         if (!isFailed && !first) {
             writeOutput(copyFrameAccessor, copyFrameAccessor.getTupleCount() - 1);
-            if (appender.getTupleCount() > 0) {
-                FrameUtils.flushFrame(outFrame, writer);
-            }
+            appenderWrapper.flush();
         }
         aggregator.close();
         aggregateState.close();
-        writer.close();
+        appenderWrapper.close();
     }
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunGenerator.java
index 2a28dea..e695828 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunGenerator.java
@@ -28,7 +28,6 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.common.io.RunFileWriter;
-import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
 import edu.uci.ics.hyracks.dataflow.std.group.preclustered.PreclusteredGroupWriter;
 import edu.uci.ics.hyracks.dataflow.std.sort.Algorithm;
@@ -111,9 +110,7 @@
         for (int i = 0; i < comparators.length; i++) {
             comparators[i] = comparatorFactories[i].createBinaryComparator();
         }
-        IAggregatorDescriptor aggregator = aggregatorFactory.createAggregator(ctx, inRecordDesc, outRecordDesc,
-                groupFields, groupFields, writer);
-        PreclusteredGroupWriter pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregator,
+        PreclusteredGroupWriter pgw = new PreclusteredGroupWriter(ctx, groupFields, comparators, aggregatorFactory,
                 this.inRecordDesc, this.outRecordDesc, writer, true);
         pgw.open();
 
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunMerger.java b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunMerger.java
index 1f9b358..2a580d3 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunMerger.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/sort/ExternalSortGroupByRunMerger.java
@@ -31,7 +31,6 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
 import edu.uci.ics.hyracks.dataflow.common.io.RunFileReader;
 import edu.uci.ics.hyracks.dataflow.common.io.RunFileWriter;
-import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
 import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
 import edu.uci.ics.hyracks.dataflow.std.group.preclustered.PreclusteredGroupWriter;
 import edu.uci.ics.hyracks.dataflow.std.sort.IFrameSorter;
@@ -55,8 +54,8 @@
     private ByteBuffer outFrame;
     private FrameTupleAppender outFrameAppender;
 
-    private IFrameSorter frameSorter; // Used in External sort, no replacement
-                                      // selection
+    private final IFrameSorter frameSorter; // Used in External sort, no replacement
+    // selection
 
     private final int[] groupFields;
     private final INormalizedKeyComputer firstKeyNkc;
@@ -67,7 +66,7 @@
 
     private final int[] mergeSortFields;
     private final int[] mergeGroupFields;
-    private IBinaryComparator[] groupByComparators;
+    private final IBinaryComparator[] groupByComparators;
 
     // Constructor for external sort, no replacement selection
     public ExternalSortGroupByRunMerger(IHyracksTaskContext ctx, IFrameSorter frameSorter, List<IFrameReader> runs,
@@ -115,10 +114,8 @@
 
     public void process() throws HyracksDataException {
         IAggregatorDescriptorFactory aggregatorFactory = localSide ? partialAggregatorFactory : mergeAggregatorFactory;
-        IAggregatorDescriptor aggregator = aggregatorFactory.createAggregator(ctx, partialAggRecordDesc, outRecordDesc,
-                groupFields, groupFields, writer);
-        PreclusteredGroupWriter pgw = new PreclusteredGroupWriter(ctx, groupFields, groupByComparators, aggregator,
-                inputRecordDesc, outRecordDesc, writer, false);
+        PreclusteredGroupWriter pgw = new PreclusteredGroupWriter(ctx, groupFields, groupByComparators,
+                aggregatorFactory, inputRecordDesc, outRecordDesc, writer, false);
         try {
             if (runs.size() <= 0) {
                 pgw.open();
@@ -149,9 +146,7 @@
                         IFrameWriter mergeResultWriter = new RunFileWriter(newRun, ctx.getIOManager());
 
                         aggregatorFactory = localSide ? mergeAggregatorFactory : partialAggregatorFactory;
-                        aggregator = aggregatorFactory.createAggregator(ctx, partialAggRecordDesc,
-                                partialAggRecordDesc, mergeGroupFields, mergeGroupFields, mergeResultWriter);
-                        pgw = new PreclusteredGroupWriter(ctx, mergeGroupFields, groupByComparators, aggregator,
+                        pgw = new PreclusteredGroupWriter(ctx, mergeGroupFields, groupByComparators, aggregatorFactory,
                                 partialAggRecordDesc, partialAggRecordDesc, mergeResultWriter, true);
                         pgw.open();
 
@@ -166,10 +161,8 @@
                     }
                 }
                 if (!runs.isEmpty()) {
-                    aggregator = mergeAggregatorFactory.createAggregator(ctx, partialAggRecordDesc, outRecordDesc,
-                            mergeGroupFields, mergeGroupFields, writer);
-                    pgw = new PreclusteredGroupWriter(ctx, mergeGroupFields, groupByComparators, aggregator,
-                            partialAggRecordDesc, outRecordDesc, writer, false);
+                    pgw = new PreclusteredGroupWriter(ctx, mergeGroupFields, groupByComparators,
+                            mergeAggregatorFactory, partialAggRecordDesc, outRecordDesc, writer, false);
                     pgw.open();
                     IFrameReader[] runCursors = new RunFileReader[runs.size()];
                     for (int i = 0; i < runCursors.length; i++) {