diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/CountAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/CountAggregatorFactory.java
index e036470..a99cba9 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/CountAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/CountAggregatorFactory.java
@@ -50,44 +50,48 @@
         };
     }
 
-	@Override
-	public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
-		return new ISpillableFieldValueResultingAggregator() {
-			private int count;
+    @Override
+    public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
+        return new ISpillableFieldValueResultingAggregator() {
+            private int count;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeInt(count);
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeInt(count);
+                } catch (IOException e) {
+                    throw new HyracksDataException(e);
+                }
+            }
 
-			@Override
-			public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex)
-					throws HyracksDataException {
-				count = IntegerSerializerDeserializer.getInt(accessor.getBuffer().array(), accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2 + accessor.getFieldStartOffset(tIndex, fIndex));
+            @Override
+            public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                count = IntegerSerializerDeserializer.getInt(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
 
-			}
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				count++;
-			}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                count++;
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
-				count = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                count = 0;
+            }
 
-			@Override
-			public void accumulatePartialResult(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				count += IntegerSerializerDeserializer.getInt(accessor.getBuffer().array(), accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2 + accessor.getFieldStartOffset(tIndex, fIndex));
-			}
-		};
-	}
+            @Override
+            public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                count += IntegerSerializerDeserializer.getInt(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
+            }
+        };
+    }
 }
\ No newline at end of file
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/FloatSumAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/FloatSumAggregatorFactory.java
index 0b02378..7e6bc8c 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/FloatSumAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/FloatSumAggregatorFactory.java
@@ -25,123 +25,111 @@
  * SUM aggregator on float type data.
  * 
  * @author jarodwen
- * 
  */
-public class FloatSumAggregatorFactory implements
-		IFieldValueResultingAggregatorFactory {
+public class FloatSumAggregatorFactory implements IFieldValueResultingAggregatorFactory {
 
-	/**
+    /**
 	 * 
 	 */
-	private static final long serialVersionUID = 1L;
-	private int sumField;
+    private static final long serialVersionUID = 1L;
+    private int sumField;
 
-	public FloatSumAggregatorFactory(int field) {
-		this.sumField = field;
-	}
+    public FloatSumAggregatorFactory(int field) {
+        this.sumField = field;
+    }
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
-	 * IFieldValueResultingAggregatorFactory
-	 * #createFieldValueResultingAggregator()
-	 */
-	@Override
-	public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
-		return new IFieldValueResultingAggregator() {
+    /*
+     * (non-Javadoc)
+     * 
+     * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
+     * IFieldValueResultingAggregatorFactory
+     * #createFieldValueResultingAggregator()
+     */
+    @Override
+    public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
+        return new IFieldValueResultingAggregator() {
 
-			private float sum;
+            private float sum;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeFloat(sum);
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeFloat(sum);
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				sum = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                sum = 0;
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
-				sum += FloatSerializerDeserializer.getFloat(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
+                sum += FloatSerializerDeserializer.getFloat(accessor.getBuffer().array(), tupleOffset + 2 * fieldCount
+                        + fieldStart);
+            }
+        };
+    }
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
-	 * IFieldValueResultingAggregatorFactory
-	 * #createSpillableFieldValueResultingAggregator()
-	 */
-	@Override
-	public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
-		return new ISpillableFieldValueResultingAggregator() {
+    /*
+     * (non-Javadoc)
+     * 
+     * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
+     * IFieldValueResultingAggregatorFactory
+     * #createSpillableFieldValueResultingAggregator()
+     */
+    @Override
+    public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
+        return new ISpillableFieldValueResultingAggregator() {
 
-			private float sum;
+            private float sum;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeFloat(sum);
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeFloat(sum);
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
 
-			@Override
-			public void initFromPartial(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				sum = FloatSerializerDeserializer.getFloat(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
-			}
+            @Override
+            public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                sum = FloatSerializerDeserializer.getFloat(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				sum = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                sum = 0;
+            }
 
-			@Override
-			public void accumulatePartialResult(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				sum += FloatSerializerDeserializer.getFloat(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
-			}
+            @Override
+            public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                sum += FloatSerializerDeserializer.getFloat(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
-				sum += FloatSerializerDeserializer.getFloat(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
+                sum += FloatSerializerDeserializer.getFloat(accessor.getBuffer().array(), tupleOffset + 2 * fieldCount
+                        + fieldStart);
+            }
+        };
+    }
 
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/IFieldValueResultingAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/IFieldValueResultingAggregatorFactory.java
index b3ce8dd..1b7da7f 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/IFieldValueResultingAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/IFieldValueResultingAggregatorFactory.java
@@ -18,6 +18,6 @@
 
 public interface IFieldValueResultingAggregatorFactory extends Serializable {
     public IFieldValueResultingAggregator createFieldValueResultingAggregator();
-    
+
     public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator();
 }
\ No newline at end of file
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/ISpillableFieldValueResultingAggregator.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/ISpillableFieldValueResultingAggregator.java
index 95d4ac9..d52f458 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/ISpillableFieldValueResultingAggregator.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/ISpillableFieldValueResultingAggregator.java
@@ -20,32 +20,29 @@
 /**
  * An extended version of the {@link IFieldValueResultingAggregator} supporting
  * external aggregation.
- * 
  */
-public interface ISpillableFieldValueResultingAggregator extends
-		IFieldValueResultingAggregator {
+public interface ISpillableFieldValueResultingAggregator extends IFieldValueResultingAggregator {
 
-	/**
-	 * Called once per aggregator before calling accumulate for the first time.
-	 * 
-	 * @param accessor
-	 *            - Accessor to the data tuple.
-	 * @param tIndex
-	 *            - Index of the tuple in the accessor.
-	 * @throws HyracksDataException
-	 */
-	public void initFromPartial(IFrameTupleAccessor accessor, int tIndex,
-			int fIndex) throws HyracksDataException;
+    /**
+     * Called once per aggregator before calling accumulate for the first time.
+     * 
+     * @param accessor
+     *            - Accessor to the data tuple.
+     * @param tIndex
+     *            - Index of the tuple in the accessor.
+     * @throws HyracksDataException
+     */
+    public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex) throws HyracksDataException;
 
-	/**
-	 * Aggregate another partial result.
-	 * 
-	 * @param accessor
-	 * @param tIndex
-	 * @param fIndex
-	 * @throws HyracksDataException
-	 */
-	public void accumulatePartialResult(IFrameTupleAccessor accessor,
-			int tIndex, int fIndex) throws HyracksDataException;
+    /**
+     * Aggregate another partial result.
+     * 
+     * @param accessor
+     * @param tIndex
+     * @param fIndex
+     * @throws HyracksDataException
+     */
+    public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+            throws HyracksDataException;
 
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MinMaxAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MinMaxAggregatorFactory.java
index 95740b2..0c3b0d5 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MinMaxAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MinMaxAggregatorFactory.java
@@ -23,127 +23,115 @@
 
 /**
  * Min/Max aggregator factory
- *
  */
-public class MinMaxAggregatorFactory implements
-		IFieldValueResultingAggregatorFactory {
+public class MinMaxAggregatorFactory implements IFieldValueResultingAggregatorFactory {
 
-	/**
+    /**
 	 * 
 	 */
-	private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 1L;
 
-	/**
-	 * indicate the type of the value: true: max false: min
-	 */
-	private boolean type;
+    /**
+     * indicate the type of the value: true: max false: min
+     */
+    private boolean type;
 
-	/**
-	 * The field to be aggregated.
-	 */
-	private int field;
+    /**
+     * The field to be aggregated.
+     */
+    private int field;
 
-	public MinMaxAggregatorFactory(boolean type, int field) {
-		this.type = type;
-		this.field = field;
-	}
+    public MinMaxAggregatorFactory(boolean type, int field) {
+        this.type = type;
+        this.field = field;
+    }
 
-	@Override
-	public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
-		return new IFieldValueResultingAggregator() {
+    @Override
+    public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
+        return new IFieldValueResultingAggregator() {
 
-			private float minmax;
+            private float minmax;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeFloat(minmax);
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeFloat(minmax);
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				minmax = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                minmax = 0;
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, field);
-				float nval = FloatSerializerDeserializer.getFloat(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-				if ((type ? (nval > minmax) : (nval < minmax))) {
-					minmax = nval;
-				}
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, field);
+                float nval = FloatSerializerDeserializer.getFloat(accessor.getBuffer().array(), tupleOffset + 2
+                        * fieldCount + fieldStart);
+                if ((type ? (nval > minmax) : (nval < minmax))) {
+                    minmax = nval;
+                }
+            }
+        };
+    }
 
-	@Override
-	public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
-		return new ISpillableFieldValueResultingAggregator() {
+    @Override
+    public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
+        return new ISpillableFieldValueResultingAggregator() {
 
-			private float minmax;
+            private float minmax;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeFloat(minmax);
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeFloat(minmax);
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
 
-			@Override
-			public void initFromPartial(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				minmax = FloatSerializerDeserializer.getFloat(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
+            @Override
+            public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                minmax = FloatSerializerDeserializer.getFloat(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
 
-			}
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				minmax = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                minmax = 0;
+            }
 
-			@Override
-			public void accumulatePartialResult(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				minmax = FloatSerializerDeserializer.getFloat(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
+            @Override
+            public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                minmax = FloatSerializerDeserializer.getFloat(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
 
-			}
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, field);
-				float nval = FloatSerializerDeserializer.getFloat(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-				if ((type ? (nval > minmax) : (nval < minmax))) {
-					minmax = nval;
-				}
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, field);
+                float nval = FloatSerializerDeserializer.getFloat(accessor.getBuffer().array(), tupleOffset + 2
+                        * fieldCount + fieldStart);
+                if ((type ? (nval > minmax) : (nval < minmax))) {
+                    minmax = nval;
+                }
+            }
+        };
+    }
 
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MultiAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MultiAggregatorFactory.java
index b237d63..957c453 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MultiAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/MultiAggregatorFactory.java
@@ -27,170 +27,149 @@
 import edu.uci.ics.hyracks.dataflow.std.group.ISpillableAccumulatingAggregator;
 
 public class MultiAggregatorFactory implements IAccumulatingAggregatorFactory {
-	private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 1L;
 
-	private IFieldValueResultingAggregatorFactory[] aFactories;
+    private IFieldValueResultingAggregatorFactory[] aFactories;
 
-	public MultiAggregatorFactory(
-			IFieldValueResultingAggregatorFactory[] aFactories) {
-		this.aFactories = aFactories;
-	}
+    public MultiAggregatorFactory(IFieldValueResultingAggregatorFactory[] aFactories) {
+        this.aFactories = aFactories;
+    }
 
-	@Override
-	public ISpillableAccumulatingAggregator createSpillableAggregator(
-			IHyracksStageletContext ctx, RecordDescriptor inRecordDesc,
-			final RecordDescriptor outRecordDescriptor) {
-		final ISpillableFieldValueResultingAggregator aggregators[] = new ISpillableFieldValueResultingAggregator[aFactories.length];
-		for (int i = 0; i < aFactories.length; ++i) {
-			aggregators[i] = aFactories[i]
-					.createSpillableFieldValueResultingAggregator();
-		}
-		final ArrayTupleBuilder tb = new ArrayTupleBuilder(
-				outRecordDescriptor.getFields().length);
-		return new ISpillableAccumulatingAggregator() {
-			private boolean pending;
+    @Override
+    public ISpillableAccumulatingAggregator createSpillableAggregator(IHyracksStageletContext ctx,
+            RecordDescriptor inRecordDesc, final RecordDescriptor outRecordDescriptor) {
+        final ISpillableFieldValueResultingAggregator aggregators[] = new ISpillableFieldValueResultingAggregator[aFactories.length];
+        for (int i = 0; i < aFactories.length; ++i) {
+            aggregators[i] = aFactories[i].createSpillableFieldValueResultingAggregator();
+        }
+        final ArrayTupleBuilder tb = new ArrayTupleBuilder(outRecordDescriptor.getFields().length);
+        return new ISpillableAccumulatingAggregator() {
+            private boolean pending;
 
-			@Override
-			public boolean output(FrameTupleAppender appender,
-					IFrameTupleAccessor accessor, int tIndex,
-					int[] keyFieldIndexes) throws HyracksDataException {
-				if (!pending) {
-					tb.reset();
-					for (int i = 0; i < keyFieldIndexes.length; ++i) {
-						tb.addField(accessor, tIndex, keyFieldIndexes[i]);
-					}
-					DataOutput dos = tb.getDataOutput();
-					for (int i = 0; i < aggregators.length; ++i) {
-						aggregators[i].output(dos);
-						tb.addFieldEndOffset();
-					}
-				}
-				if (!appender.append(tb.getFieldEndOffsets(),
-						tb.getByteArray(), 0, tb.getSize())) {
-					pending = true;
-					return false;
-				}
-				return true;
-			}
+            @Override
+            public boolean output(FrameTupleAppender appender, IFrameTupleAccessor accessor, int tIndex,
+                    int[] keyFieldIndexes) throws HyracksDataException {
+                if (!pending) {
+                    tb.reset();
+                    for (int i = 0; i < keyFieldIndexes.length; ++i) {
+                        tb.addField(accessor, tIndex, keyFieldIndexes[i]);
+                    }
+                    DataOutput dos = tb.getDataOutput();
+                    for (int i = 0; i < aggregators.length; ++i) {
+                        aggregators[i].output(dos);
+                        tb.addFieldEndOffset();
+                    }
+                }
+                if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+                    pending = true;
+                    return false;
+                }
+                return true;
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				tb.reset();
-				for (int i = 0; i < aggregators.length; ++i) {
-					aggregators[i].init(accessor, tIndex);
-				}
-				pending = false;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                tb.reset();
+                for (int i = 0; i < aggregators.length; ++i) {
+                    aggregators[i].init(accessor, tIndex);
+                }
+                pending = false;
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				for (int i = 0; i < aggregators.length; ++i) {
-					aggregators[i].accumulate(accessor, tIndex);
-				}
-			}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                for (int i = 0; i < aggregators.length; ++i) {
+                    aggregators[i].accumulate(accessor, tIndex);
+                }
+            }
 
-			@Override
-			public void initFromPartial(IFrameTupleAccessor accessor,
-					int tIndex, int[] keyFieldIndexes)
-					throws HyracksDataException {
-				tb.reset();
-				for (int i = 0; i < aggregators.length; i++) {
-					aggregators[i].initFromPartial(accessor, tIndex,
-							keyFieldIndexes.length + i);
-				}
-				pending = false;
-			}
+            @Override
+            public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int[] keyFieldIndexes)
+                    throws HyracksDataException {
+                tb.reset();
+                for (int i = 0; i < aggregators.length; i++) {
+                    aggregators[i].initFromPartial(accessor, tIndex, keyFieldIndexes.length + i);
+                }
+                pending = false;
+            }
 
-			@Override
-			public void accumulatePartialResult(IFrameTupleAccessor accessor,
-					int tIndex, int[] keyFieldIndexes)
-					throws HyracksDataException {
-				for (int i = 0; i < aggregators.length; i++) {
-					aggregators[i].accumulatePartialResult(accessor, tIndex,
-							keyFieldIndexes.length + i);
-				}
+            @Override
+            public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int[] keyFieldIndexes)
+                    throws HyracksDataException {
+                for (int i = 0; i < aggregators.length; i++) {
+                    aggregators[i].accumulatePartialResult(accessor, tIndex, keyFieldIndexes.length + i);
+                }
 
-			}
+            }
 
-			@Override
-			public boolean output(FrameTupleAppender appender,
-					ArrayTupleBuilder tbder) throws HyracksDataException {
-				if (!pending) {
-					// TODO Here to be fixed:
-					DataOutput dos = tbder.getDataOutput();
-					for (int i = 0; i < aggregators.length; ++i) {
-						aggregators[i].output(dos);
-						tbder.addFieldEndOffset();
-					}
-				}
-				if (!appender.append(tbder.getFieldEndOffsets(),
-						tbder.getByteArray(), 0, tbder.getSize())) {
-					pending = true;
-					return false;
-				}
-				return true;
-			}
-		};
-	}
+            @Override
+            public boolean output(FrameTupleAppender appender, ArrayTupleBuilder tbder) throws HyracksDataException {
+                if (!pending) {
+                    // TODO Here to be fixed:
+                    DataOutput dos = tbder.getDataOutput();
+                    for (int i = 0; i < aggregators.length; ++i) {
+                        aggregators[i].output(dos);
+                        tbder.addFieldEndOffset();
+                    }
+                }
+                if (!appender.append(tbder.getFieldEndOffsets(), tbder.getByteArray(), 0, tbder.getSize())) {
+                    pending = true;
+                    return false;
+                }
+                return true;
+            }
+        };
+    }
 
-	@Override
-	public IAccumulatingAggregator createAggregator(
-			IHyracksStageletContext ctx, RecordDescriptor inRecordDesc,
-			RecordDescriptor outRecordDescriptor) throws HyracksDataException {
-		final IFieldValueResultingAggregator aggregators[] = new IFieldValueResultingAggregator[aFactories.length];
-		for (int i = 0; i < aFactories.length; ++i) {
-			aggregators[i] = aFactories[i]
-					.createFieldValueResultingAggregator();
-		}
-		final ArrayTupleBuilder tb = new ArrayTupleBuilder(
-				outRecordDescriptor.getFields().length);
-		return new IAccumulatingAggregator() {
+    @Override
+    public IAccumulatingAggregator createAggregator(IHyracksStageletContext ctx, RecordDescriptor inRecordDesc,
+            RecordDescriptor outRecordDescriptor) throws HyracksDataException {
+        final IFieldValueResultingAggregator aggregators[] = new IFieldValueResultingAggregator[aFactories.length];
+        for (int i = 0; i < aFactories.length; ++i) {
+            aggregators[i] = aFactories[i].createFieldValueResultingAggregator();
+        }
+        final ArrayTupleBuilder tb = new ArrayTupleBuilder(outRecordDescriptor.getFields().length);
+        return new IAccumulatingAggregator() {
 
-			private boolean pending;
+            private boolean pending;
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				tb.reset();
-				for (int i = 0; i < aggregators.length; ++i) {
-					aggregators[i].init(accessor, tIndex);
-				}
-				pending = false;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                tb.reset();
+                for (int i = 0; i < aggregators.length; ++i) {
+                    aggregators[i].init(accessor, tIndex);
+                }
+                pending = false;
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				for (int i = 0; i < aggregators.length; ++i) {
-					aggregators[i].accumulate(accessor, tIndex);
-				}
-			}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                for (int i = 0; i < aggregators.length; ++i) {
+                    aggregators[i].accumulate(accessor, tIndex);
+                }
+            }
 
-			@Override
-			public boolean output(FrameTupleAppender appender,
-					IFrameTupleAccessor accessor, int tIndex,
-					int[] keyFieldIndexes) throws HyracksDataException {
-				if (!pending) {
-					tb.reset();
-					for (int i = 0; i < keyFieldIndexes.length; ++i) {
-						tb.addField(accessor, tIndex, keyFieldIndexes[i]);
-					}
-					DataOutput dos = tb.getDataOutput();
-					for (int i = 0; i < aggregators.length; ++i) {
-						aggregators[i].output(dos);
-						tb.addFieldEndOffset();
-					}
-				}
-				if (!appender.append(tb.getFieldEndOffsets(),
-						tb.getByteArray(), 0, tb.getSize())) {
-					pending = true;
-					return false;
-				}
-				return true;
-			}
+            @Override
+            public boolean output(FrameTupleAppender appender, IFrameTupleAccessor accessor, int tIndex,
+                    int[] keyFieldIndexes) throws HyracksDataException {
+                if (!pending) {
+                    tb.reset();
+                    for (int i = 0; i < keyFieldIndexes.length; ++i) {
+                        tb.addField(accessor, tIndex, keyFieldIndexes[i]);
+                    }
+                    DataOutput dos = tb.getDataOutput();
+                    for (int i = 0; i < aggregators.length; ++i) {
+                        aggregators[i].output(dos);
+                        tb.addFieldEndOffset();
+                    }
+                }
+                if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+                    pending = true;
+                    return false;
+                }
+                return true;
+            }
 
-		};
-	}
+        };
+    }
 }
\ No newline at end of file
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/SumAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/SumAggregatorFactory.java
index cfb5293..64335f1 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/SumAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/aggregators/SumAggregatorFactory.java
@@ -24,123 +24,111 @@
 /**
  * SUM aggregator factory (for integer only; another SUM aggregator for floats
  * is available at {@link FloatSumAggregatorFactory})
- * 
  */
-public class SumAggregatorFactory implements
-		IFieldValueResultingAggregatorFactory {
+public class SumAggregatorFactory implements IFieldValueResultingAggregatorFactory {
 
-	private int sumField;
+    private int sumField;
 
-	public SumAggregatorFactory(int field) {
-		sumField = field;
-	}
+    public SumAggregatorFactory(int field) {
+        sumField = field;
+    }
 
-	/**
+    /**
 	 * 
 	 */
-	private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 1L;
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
-	 * IFieldValueResultingAggregatorFactory
-	 * #createFieldValueResultingAggregator()
-	 */
-	@Override
-	public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
-		return new IFieldValueResultingAggregator() {
-			private int sum;
+    /*
+     * (non-Javadoc)
+     * 
+     * @see edu.uci.ics.hyracks.dataflow.std.aggregators.
+     * IFieldValueResultingAggregatorFactory
+     * #createFieldValueResultingAggregator()
+     */
+    @Override
+    public IFieldValueResultingAggregator createFieldValueResultingAggregator() {
+        return new IFieldValueResultingAggregator() {
+            private int sum;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeInt(sum);
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeInt(sum);
+                } catch (IOException e) {
+                    throw new HyracksDataException(e);
+                }
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				sum++;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                sum++;
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
-				sum += IntegerSerializerDeserializer.getInt(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
+                sum += IntegerSerializerDeserializer.getInt(accessor.getBuffer().array(), tupleOffset + 2 * fieldCount
+                        + fieldStart);
+            }
+        };
+    }
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see edu.uci.ics.hyracks.dataflow.std.aggregators.spillable.
-	 * ISpillableFieldValueResultingAggregatorFactory
-	 * #createFieldValueResultingAggregator()
-	 */
-	@Override
-	public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
-		return new ISpillableFieldValueResultingAggregator() {
+    /*
+     * (non-Javadoc)
+     * 
+     * @see edu.uci.ics.hyracks.dataflow.std.aggregators.spillable.
+     * ISpillableFieldValueResultingAggregatorFactory
+     * #createFieldValueResultingAggregator()
+     */
+    @Override
+    public ISpillableFieldValueResultingAggregator createSpillableFieldValueResultingAggregator() {
+        return new ISpillableFieldValueResultingAggregator() {
 
-			private int sum;
+            private int sum;
 
-			@Override
-			public void output(DataOutput resultAcceptor)
-					throws HyracksDataException {
-				try {
-					resultAcceptor.writeInt(sum);
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
+            @Override
+            public void output(DataOutput resultAcceptor) throws HyracksDataException {
+                try {
+                    resultAcceptor.writeInt(sum);
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
 
-			@Override
-			public void initFromPartial(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				sum = IntegerSerializerDeserializer.getInt(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
-			}
+            @Override
+            public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                sum = IntegerSerializerDeserializer.getInt(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
+            }
 
-			@Override
-			public void init(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				sum = 0;
-			}
+            @Override
+            public void init(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                sum = 0;
+            }
 
-			@Override
-			public void accumulatePartialResult(IFrameTupleAccessor accessor,
-					int tIndex, int fIndex) throws HyracksDataException {
-				sum += IntegerSerializerDeserializer.getInt(
-						accessor.getBuffer().array(),
-						accessor.getTupleStartOffset(tIndex)
-								+ accessor.getFieldCount() * 2
-								+ accessor.getFieldStartOffset(tIndex, fIndex));
-			}
+            @Override
+            public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int fIndex)
+                    throws HyracksDataException {
+                sum += IntegerSerializerDeserializer.getInt(
+                        accessor.getBuffer().array(),
+                        accessor.getTupleStartOffset(tIndex) + accessor.getFieldCount() * 2
+                                + accessor.getFieldStartOffset(tIndex, fIndex));
+            }
 
-			@Override
-			public void accumulate(IFrameTupleAccessor accessor, int tIndex)
-					throws HyracksDataException {
-				int tupleOffset = accessor.getTupleStartOffset(tIndex);
-				int fieldCount = accessor.getFieldCount();
-				int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
-				sum += IntegerSerializerDeserializer.getInt(accessor
-						.getBuffer().array(), tupleOffset + 2 * fieldCount
-						+ fieldStart);
-			}
-		};
-	}
+            @Override
+            public void accumulate(IFrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+                int tupleOffset = accessor.getTupleStartOffset(tIndex);
+                int fieldCount = accessor.getFieldCount();
+                int fieldStart = accessor.getFieldStartOffset(tIndex, sumField);
+                sum += IntegerSerializerDeserializer.getInt(accessor.getBuffer().array(), tupleOffset + 2 * fieldCount
+                        + fieldStart);
+            }
+        };
+    }
 
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/file/PlainFileWriterOperatorDescriptor.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/file/PlainFileWriterOperatorDescriptor.java
index 1bad6a7..530207d 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/file/PlainFileWriterOperatorDescriptor.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/file/PlainFileWriterOperatorDescriptor.java
@@ -33,111 +33,104 @@
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
 
 /**
- * File writer to output plain text. 
- *
+ * File writer to output plain text.
  */
-public class PlainFileWriterOperatorDescriptor extends
-		AbstractSingleActivityOperatorDescriptor {
+public class PlainFileWriterOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
 
-	/**
-	 * 
-	 */
-	private static final long serialVersionUID = 1L;
-	
-	private IFileSplitProvider fileSplitProvider;
-	
-	private String delim;
+    /**
+     * 
+     */
+    private static final long serialVersionUID = 1L;
 
-	/**
-	 * @param spec
-	 * @param inputArity
-	 * @param outputArity
-	 */
-	public PlainFileWriterOperatorDescriptor(JobSpecification spec, IFileSplitProvider fileSplitProvider, String delim) {
-		super(spec, 1, 0);
-		this.fileSplitProvider = fileSplitProvider;
-		this.delim = delim;
-	}
+    private IFileSplitProvider fileSplitProvider;
 
-	/* (non-Javadoc)
-	 * @see edu.uci.ics.hyracks.api.dataflow.IActivityNode#createPushRuntime(edu.uci.ics.hyracks.api.context.IHyracksContext, edu.uci.ics.hyracks.api.job.IOperatorEnvironment, edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider, int, int)
-	 */
-	@Override
-	public IOperatorNodePushable createPushRuntime(IHyracksStageletContext ctx,
-			IOperatorEnvironment env,
-			IRecordDescriptorProvider recordDescProvider, final int partition,
-			int nPartitions) throws HyracksDataException {
-		// Output files
-		final FileSplit[] splits = fileSplitProvider.getFileSplits();
-		// Frame accessor
-		final FrameTupleAccessor frameTupleAccessor = new FrameTupleAccessor(ctx.getFrameSize(), recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0));
-		// Record descriptor
-		final RecordDescriptor recordDescriptor = recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0);
-		return new AbstractUnaryInputSinkOperatorNodePushable() {
-			private BufferedWriter out;
-			
-			private ByteBufferInputStream bbis;
-			
-			private DataInputStream di;
-			
-			@Override
-			public void open() throws HyracksDataException {
-				try {
+    private String delim;
+
+    /**
+     * @param spec
+     * @param inputArity
+     * @param outputArity
+     */
+    public PlainFileWriterOperatorDescriptor(JobSpecification spec, IFileSplitProvider fileSplitProvider, String delim) {
+        super(spec, 1, 0);
+        this.fileSplitProvider = fileSplitProvider;
+        this.delim = delim;
+    }
+
+    /* (non-Javadoc)
+     * @see edu.uci.ics.hyracks.api.dataflow.IActivityNode#createPushRuntime(edu.uci.ics.hyracks.api.context.IHyracksContext, edu.uci.ics.hyracks.api.job.IOperatorEnvironment, edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider, int, int)
+     */
+    @Override
+    public IOperatorNodePushable createPushRuntime(IHyracksStageletContext ctx, IOperatorEnvironment env,
+            IRecordDescriptorProvider recordDescProvider, final int partition, int nPartitions)
+    throws HyracksDataException {
+        // Output files
+        final FileSplit[] splits = fileSplitProvider.getFileSplits();
+        // Frame accessor
+        final FrameTupleAccessor frameTupleAccessor = new FrameTupleAccessor(ctx.getFrameSize(),
+                recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0));
+        // Record descriptor
+        final RecordDescriptor recordDescriptor = recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0);
+        return new AbstractUnaryInputSinkOperatorNodePushable() {
+            private BufferedWriter out;
+
+            private ByteBufferInputStream bbis;
+
+            private DataInputStream di;
+
+            @Override
+            public void open() throws HyracksDataException {
+                try {
                     out = new BufferedWriter(new FileWriter(splits[partition].getLocalFile().getFile()));
                     bbis = new ByteBufferInputStream();
                     di = new DataInputStream(bbis);
                 } catch (Exception e) {
                     throw new HyracksDataException(e);
                 }
-			}
-			
-			@Override
-			public void nextFrame(ByteBuffer buffer)
-					throws HyracksDataException {
-				try {
-					frameTupleAccessor.reset(buffer);
-					for (int tIndex = 0; tIndex < frameTupleAccessor
-							.getTupleCount(); tIndex++) {
-						int start = frameTupleAccessor
-								.getTupleStartOffset(tIndex)
-								+ frameTupleAccessor.getFieldSlotsLength();
-						bbis.setByteBuffer(buffer, start);
-						Object[] record = new Object[recordDescriptor
-								.getFields().length];
-						for (int i = 0; i < record.length; ++i) {
-							Object instance = recordDescriptor.getFields()[i]
-									.deserialize(di);
-							if (i == 0) {
-								out.write(String.valueOf(instance));
-							} else {
-								out.write(delim + String.valueOf(instance));
-							}
-						}
-						out.write("\n");
-					}
-				} catch (IOException ex) {
-					throw new HyracksDataException(ex);
-				}
-			}
-			
-			@Override
-			public void flush() throws HyracksDataException {
-				try {
-					out.flush();
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
-			
-			@Override
-			public void close() throws HyracksDataException {
-				try {
-					out.close();
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
-		};
-	}
+            }
+
+            @Override
+            public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+                try {
+                    frameTupleAccessor.reset(buffer);
+                    for (int tIndex = 0; tIndex < frameTupleAccessor.getTupleCount(); tIndex++) {
+                        int start = frameTupleAccessor.getTupleStartOffset(tIndex)
+                        + frameTupleAccessor.getFieldSlotsLength();
+                        bbis.setByteBuffer(buffer, start);
+                        Object[] record = new Object[recordDescriptor.getFields().length];
+                        for (int i = 0; i < record.length; ++i) {
+                            Object instance = recordDescriptor.getFields()[i].deserialize(di);
+                            if (i == 0) {
+                                out.write(String.valueOf(instance));
+                            } else {
+                                out.write(delim + String.valueOf(instance));
+                            }
+                        }
+                        out.write("\n");
+                    }
+                } catch (IOException ex) {
+                    throw new HyracksDataException(ex);
+                }
+            }
+
+            @Override
+            public void flush() throws HyracksDataException {
+                try {
+                    out.flush();
+                } catch (IOException e) {
+                    throw new HyracksDataException(e);
+                }
+            }
+
+            @Override
+            public void close() throws HyracksDataException {
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    throw new HyracksDataException(e);
+                }
+            }
+        };
+    }
 
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ExternalHashGroupOperatorDescriptor.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ExternalHashGroupOperatorDescriptor.java
index af78b7d..b13b087 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ExternalHashGroupOperatorDescriptor.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ExternalHashGroupOperatorDescriptor.java
@@ -52,717 +52,641 @@
 
 /**
  * This is an implementation of the external hash group operator.
- * 
  * The motivation of this operator is that when tuples are processed in
  * parallel, distinguished aggregating keys partitioned on one node may exceed
  * the main memory, so aggregation results should be output onto the disk to
  * make space for aggregating more input tuples.
- * 
- * 
  */
-public class ExternalHashGroupOperatorDescriptor extends
-		AbstractOperatorDescriptor {
+public class ExternalHashGroupOperatorDescriptor extends AbstractOperatorDescriptor {
 
-	/**
-	 * The input frame identifier (in the job environment)
-	 */
-	private static final String GROUPTABLES = "gtables";
+    /**
+     * The input frame identifier (in the job environment)
+     */
+    private static final String GROUPTABLES = "gtables";
 
-	/**
-	 * The runs files identifier (in the job environment)
-	 */
-	private static final String RUNS = "runs";
+    /**
+     * The runs files identifier (in the job environment)
+     */
+    private static final String RUNS = "runs";
 
-	/**
-	 * The fields used for grouping (grouping keys).
-	 */
-	private final int[] keyFields;
+    /**
+     * The fields used for grouping (grouping keys).
+     */
+    private final int[] keyFields;
 
-	/**
-	 * The comparator for checking the grouping conditions, corresponding to the
-	 * {@link #keyFields}.
-	 */
-	private final IBinaryComparatorFactory[] comparatorFactories;
+    /**
+     * The comparator for checking the grouping conditions, corresponding to the {@link #keyFields}.
+     */
+    private final IBinaryComparatorFactory[] comparatorFactories;
 
-	/**
-	 * The aggregator factory for the aggregating field, corresponding to the
-	 * {@link #aggregateFields}.
-	 */
-	private IAccumulatingAggregatorFactory aggregatorFactory;
+    /**
+     * The aggregator factory for the aggregating field, corresponding to the {@link #aggregateFields}.
+     */
+    private IAccumulatingAggregatorFactory aggregatorFactory;
 
-	/**
-	 * The maximum number of frames in the main memory.
-	 */
-	private final int framesLimit;
+    /**
+     * The maximum number of frames in the main memory.
+     */
+    private final int framesLimit;
 
-	/**
-	 * Indicate whether the final output will be sorted or not.
-	 */
-	private final boolean sortOutput;
+    /**
+     * Indicate whether the final output will be sorted or not.
+     */
+    private final boolean sortOutput;
 
-	/**
-	 * Partition computer factory
-	 */
-	private final ITuplePartitionComputerFactory tpcf;
+    /**
+     * Partition computer factory
+     */
+    private final ITuplePartitionComputerFactory tpcf;
 
-	/**
-	 * The size of the in-memory table, which should be specified now by the
-	 * creator of this operator descriptor.
-	 */
-	private final int tableSize;
+    /**
+     * The size of the in-memory table, which should be specified now by the
+     * creator of this operator descriptor.
+     */
+    private final int tableSize;
 
-	/**
-	 * XXX Logger for debug information
-	 */
-	private static Logger LOGGER = Logger
-			.getLogger(ExternalHashGroupOperatorDescriptor.class.getName());
+    /**
+     * XXX Logger for debug information
+     */
+    private static Logger LOGGER = Logger.getLogger(ExternalHashGroupOperatorDescriptor.class.getName());
 
-	/**
-	 * Constructor of the external hash group operator descriptor.
-	 * 
-	 * @param spec
-	 * @param keyFields
-	 *            The fields as keys of grouping.
-	 * @param framesLimit
-	 *            The maximum number of frames to be used in memory.
-	 * @param sortOutput
-	 *            Whether the output should be sorted or not. Note that if the
-	 *            input data is large enough for external grouping, the output
-	 *            will be sorted surely. The only case that when the output is
-	 *            not sorted is when the size of the input data can be grouped
-	 *            in memory and this parameter is false.
-	 * @param tpcf
-	 *            The partitioner.
-	 * @param comparatorFactories
-	 *            The comparators.
-	 * @param aggregatorFactory
-	 *            The aggregators.
-	 * @param recordDescriptor
-	 *            The record descriptor for the input data.
-	 * @param tableSize
-	 *            The maximum size of the in memory table usable to this
-	 *            operator.
-	 */
-	public ExternalHashGroupOperatorDescriptor(JobSpecification spec,
-			int[] keyFields, int framesLimit, boolean sortOutput,
-			ITuplePartitionComputerFactory tpcf,
-			IBinaryComparatorFactory[] comparatorFactories,
-			IAccumulatingAggregatorFactory aggregatorFactory,
-			RecordDescriptor recordDescriptor, int tableSize) {
-		super(spec, 1, 1);
-		this.framesLimit = framesLimit;
-		if (framesLimit <= 1) {
-			// Minimum of 2 frames: 1 for input records, and 1 for output
-			// aggregation results.
-			throw new IllegalStateException();
-		}
-		this.aggregatorFactory = aggregatorFactory;
-		this.keyFields = keyFields;
-		this.comparatorFactories = comparatorFactories;
+    /**
+     * Constructor of the external hash group operator descriptor.
+     * 
+     * @param spec
+     * @param keyFields
+     *            The fields as keys of grouping.
+     * @param framesLimit
+     *            The maximum number of frames to be used in memory.
+     * @param sortOutput
+     *            Whether the output should be sorted or not. Note that if the
+     *            input data is large enough for external grouping, the output
+     *            will be sorted surely. The only case that when the output is
+     *            not sorted is when the size of the input data can be grouped
+     *            in memory and this parameter is false.
+     * @param tpcf
+     *            The partitioner.
+     * @param comparatorFactories
+     *            The comparators.
+     * @param aggregatorFactory
+     *            The aggregators.
+     * @param recordDescriptor
+     *            The record descriptor for the input data.
+     * @param tableSize
+     *            The maximum size of the in memory table usable to this
+     *            operator.
+     */
+    public ExternalHashGroupOperatorDescriptor(JobSpecification spec, int[] keyFields, int framesLimit,
+            boolean sortOutput, ITuplePartitionComputerFactory tpcf, IBinaryComparatorFactory[] comparatorFactories,
+            IAccumulatingAggregatorFactory aggregatorFactory, RecordDescriptor recordDescriptor, int tableSize) {
+        super(spec, 1, 1);
+        this.framesLimit = framesLimit;
+        if (framesLimit <= 1) {
+            // Minimum of 2 frames: 1 for input records, and 1 for output
+            // aggregation results.
+            throw new IllegalStateException();
+        }
+        this.aggregatorFactory = aggregatorFactory;
+        this.keyFields = keyFields;
+        this.comparatorFactories = comparatorFactories;
 
-		this.sortOutput = sortOutput;
+        this.sortOutput = sortOutput;
 
-		this.tpcf = tpcf;
+        this.tpcf = tpcf;
 
-		this.tableSize = tableSize;
+        this.tableSize = tableSize;
 
-		// Set the record descriptor. Note that since this operator is a unary
-		// operator,
-		// only the first record descritpor is used here.
-		recordDescriptors[0] = recordDescriptor;
-	}
+        // Set the record descriptor. Note that since this operator is a unary
+        // operator,
+        // only the first record descritpor is used here.
+        recordDescriptors[0] = recordDescriptor;
+    }
 
-	/**
-	 * 
-	 */
-	private static final long serialVersionUID = 1L;
+    /**
+     * 
+     */
+    private static final long serialVersionUID = 1L;
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see
-	 * edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor#contributeTaskGraph
-	 * (edu.uci.ics.hyracks.api.dataflow.IActivityGraphBuilder)
-	 */
-	@Override
-	public void contributeTaskGraph(IActivityGraphBuilder builder) {
-		PartialAggregateActivity partialAggAct = new PartialAggregateActivity();
-		MergeActivity mergeAct = new MergeActivity();
+    /*
+     * (non-Javadoc)
+     * 
+     * @see
+     * edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor#contributeTaskGraph
+     * (edu.uci.ics.hyracks.api.dataflow.IActivityGraphBuilder)
+     */
+    @Override
+    public void contributeTaskGraph(IActivityGraphBuilder builder) {
+        PartialAggregateActivity partialAggAct = new PartialAggregateActivity();
+        MergeActivity mergeAct = new MergeActivity();
 
-		builder.addTask(partialAggAct);
-		builder.addSourceEdge(0, partialAggAct, 0);
+        builder.addTask(partialAggAct);
+        builder.addSourceEdge(0, partialAggAct, 0);
 
-		builder.addTask(mergeAct);
-		builder.addTargetEdge(0, mergeAct, 0);
+        builder.addTask(mergeAct);
+        builder.addTargetEdge(0, mergeAct, 0);
 
-		// FIXME Block or not?
-		builder.addBlockingEdge(partialAggAct, mergeAct);
+        // FIXME Block or not?
+        builder.addBlockingEdge(partialAggAct, mergeAct);
 
-	}
+    }
 
-	private class PartialAggregateActivity extends AbstractActivityNode {
+    private class PartialAggregateActivity extends AbstractActivityNode {
 
-		/**
-		 * 
-		 */
-		private static final long serialVersionUID = 1L;
+        /**
+         * 
+         */
+        private static final long serialVersionUID = 1L;
 
-		@Override
-		public IOperatorNodePushable createPushRuntime(
-				final IHyracksStageletContext ctx,
-				final IOperatorEnvironment env,
-				final IRecordDescriptorProvider recordDescProvider,
-				int partition, int nPartitions) {
-			// Create the in-memory hash table
-			final SpillableGroupingHashTable gTable = new SpillableGroupingHashTable(
-					ctx, keyFields, comparatorFactories, tpcf,
-					aggregatorFactory,
-					recordDescProvider.getInputRecordDescriptor(
-							getOperatorId(), 0), recordDescriptors[0],
-					// Always take one frame for the input records
-					framesLimit - 1, tableSize);
-			// Create the tuple accessor
-			final FrameTupleAccessor accessor = new FrameTupleAccessor(
-					ctx.getFrameSize(),
-					recordDescProvider.getInputRecordDescriptor(
-							getOperatorId(), 0));
-			// Create the partial aggregate activity node
-			IOperatorNodePushable op = new AbstractUnaryInputSinkOperatorNodePushable() {
+        @Override
+        public IOperatorNodePushable createPushRuntime(final IHyracksStageletContext ctx,
+                final IOperatorEnvironment env, final IRecordDescriptorProvider recordDescProvider, int partition,
+                int nPartitions) {
+            // Create the in-memory hash table
+            final SpillableGroupingHashTable gTable = new SpillableGroupingHashTable(ctx, keyFields,
+                    comparatorFactories, tpcf, aggregatorFactory, recordDescProvider.getInputRecordDescriptor(
+                            getOperatorId(), 0), recordDescriptors[0],
+                            // Always take one frame for the input records
+                            framesLimit - 1, tableSize);
+            // Create the tuple accessor
+            final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
+                    recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0));
+            // Create the partial aggregate activity node
+            IOperatorNodePushable op = new AbstractUnaryInputSinkOperatorNodePushable() {
 
-				/**
-				 * Run files
-				 */
-				private LinkedList<RunFileReader> runs;
+                /**
+                 * Run files
+                 */
+                private LinkedList<RunFileReader> runs;
 
-				@Override
-				public void close() throws HyracksDataException {
-					if (gTable.getFrameCount() >= 0) {
-						if (runs.size() <= 0) {
-							// All in memory
-							env.set(GROUPTABLES, gTable);
-						} else {
-							// flush the memory into the run file.
-							flushFramesToRun();
-						}
-					}
-					env.set(RUNS, runs);
-				}
+                @Override
+                public void close() throws HyracksDataException {
+                    if (gTable.getFrameCount() >= 0) {
+                        if (runs.size() <= 0) {
+                            // All in memory
+                            env.set(GROUPTABLES, gTable);
+                        } else {
+                            // flush the memory into the run file.
+                            flushFramesToRun();
+                        }
+                    }
+                    env.set(RUNS, runs);
+                }
 
-				@Override
-				public void flush() throws HyracksDataException {
+                @Override
+                public void flush() throws HyracksDataException {
 
-				}
+                }
 
-				/**
-				 * Process the next input buffer.
-				 * 
-				 * The actual insertion is processed in {@link #gTable}. It will
-				 * check whether it is possible to contain the data into the
-				 * main memory or not. If not, it will indicate the operator to
-				 * flush the content of the table into a run file.
-				 */
-				@Override
-				public void nextFrame(ByteBuffer buffer)
-						throws HyracksDataException {
-					accessor.reset(buffer);
-					int tupleCount = accessor.getTupleCount();
-					for (int i = 0; i < tupleCount; i++) {
-						// If the group table is too large, flush the table into
-						// a run file.
-						if (!gTable.insert(accessor, i)) {
-							flushFramesToRun();
-							if (!gTable.insert(accessor, i))
-								throw new HyracksDataException(
-										"Failed to insert a new buffer into the aggregate operator!");
-						}
-					}
+                /**
+                 * Process the next input buffer.
+                 * The actual insertion is processed in {@link #gTable}. It will
+                 * check whether it is possible to contain the data into the
+                 * main memory or not. If not, it will indicate the operator to
+                 * flush the content of the table into a run file.
+                 */
+                @Override
+                public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+                    accessor.reset(buffer);
+                    int tupleCount = accessor.getTupleCount();
+                    for (int i = 0; i < tupleCount; i++) {
+                        // If the group table is too large, flush the table into
+                        // a run file.
+                        if (!gTable.insert(accessor, i)) {
+                            flushFramesToRun();
+                            if (!gTable.insert(accessor, i))
+                                throw new HyracksDataException(
+                                "Failed to insert a new buffer into the aggregate operator!");
+                        }
+                    }
 
-				}
+                }
 
-				@Override
-				public void open() throws HyracksDataException {
-					runs = new LinkedList<RunFileReader>();
-					gTable.reset();
-				}
+                @Override
+                public void open() throws HyracksDataException {
+                    runs = new LinkedList<RunFileReader>();
+                    gTable.reset();
+                }
 
-				/**
-				 * Flush the content of the group table into a run file.
-				 * 
-				 * During the flushing, the hash table will be sorted as first.
-				 * After that, a run file handler is initialized and the hash
-				 * table is flushed into the run file.
-				 * 
-				 * @throws HyracksDataException
-				 */
-				private void flushFramesToRun() throws HyracksDataException {
-					// Sort the contents of the hash table.
-					gTable.sortFrames();
-					FileReference runFile;
-					try {
-						runFile = ctx.getJobletContext().createWorkspaceFile(
-								ExternalHashGroupOperatorDescriptor.class
-										.getSimpleName());
-					} catch (IOException e) {
-						throw new HyracksDataException(e);
-					}
-					RunFileWriter writer = new RunFileWriter(runFile,
-							ctx.getIOManager());
-					writer.open();
-					try {
-						gTable.flushFrames(writer, true);
-					} catch (Exception ex) {
-						throw new HyracksDataException(ex);
-					} finally {
-						writer.close();
-					}
-					gTable.reset();
-					runs.add(((RunFileWriter) writer).createReader());
-					LOGGER.warning("Created run file: "
-							+ runFile.getFile().getAbsolutePath());
-				}
+                /**
+                 * Flush the content of the group table into a run file.
+                 * During the flushing, the hash table will be sorted as first.
+                 * After that, a run file handler is initialized and the hash
+                 * table is flushed into the run file.
+                 * 
+                 * @throws HyracksDataException
+                 */
+                private void flushFramesToRun() throws HyracksDataException {
+                    // Sort the contents of the hash table.
+                    gTable.sortFrames();
+                    FileReference runFile;
+                    try {
+                        runFile = ctx.getJobletContext().createWorkspaceFile(
+                                ExternalHashGroupOperatorDescriptor.class.getSimpleName());
+                    } catch (IOException e) {
+                        throw new HyracksDataException(e);
+                    }
+                    RunFileWriter writer = new RunFileWriter(runFile, ctx.getIOManager());
+                    writer.open();
+                    try {
+                        gTable.flushFrames(writer, true);
+                    } catch (Exception ex) {
+                        throw new HyracksDataException(ex);
+                    } finally {
+                        writer.close();
+                    }
+                    gTable.reset();
+                    runs.add(((RunFileWriter) writer).createReader());
+                    LOGGER.warning("Created run file: " + runFile.getFile().getAbsolutePath());
+                }
 
-			};
+            };
 
-			return op;
-		}
+            return op;
+        }
 
-		@Override
-		public IOperatorDescriptor getOwner() {
-			return ExternalHashGroupOperatorDescriptor.this;
-		}
+        @Override
+        public IOperatorDescriptor getOwner() {
+            return ExternalHashGroupOperatorDescriptor.this;
+        }
 
-	}
+    }
 
-	private class MergeActivity extends AbstractActivityNode {
+    private class MergeActivity extends AbstractActivityNode {
 
-		/**
-		 * 
-		 */
-		private static final long serialVersionUID = 1L;
+        /**
+         * 
+         */
+        private static final long serialVersionUID = 1L;
 
-		@Override
-		public IOperatorNodePushable createPushRuntime(
-				final IHyracksStageletContext ctx,
-				final IOperatorEnvironment env,
-				IRecordDescriptorProvider recordDescProvider, int partition,
-				int nPartitions) {
-			final IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
-			for (int i = 0; i < comparatorFactories.length; ++i) {
-				comparators[i] = comparatorFactories[i]
-						.createBinaryComparator();
-			}
-			IOperatorNodePushable op = new AbstractUnaryOutputSourceOperatorNodePushable() {
-				/**
-				 * Input frames, one for each run file.
-				 */
-				private List<ByteBuffer> inFrames;
+        @Override
+        public IOperatorNodePushable createPushRuntime(final IHyracksStageletContext ctx,
+                final IOperatorEnvironment env, IRecordDescriptorProvider recordDescProvider, int partition,
+                int nPartitions) {
+            final IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
+            for (int i = 0; i < comparatorFactories.length; ++i) {
+                comparators[i] = comparatorFactories[i].createBinaryComparator();
+            }
+            IOperatorNodePushable op = new AbstractUnaryOutputSourceOperatorNodePushable() {
+                /**
+                 * Input frames, one for each run file.
+                 */
+                private List<ByteBuffer> inFrames;
 
-				/**
-				 * Output frame.
-				 */
-				private ByteBuffer outFrame;
+                /**
+                 * Output frame.
+                 */
+                private ByteBuffer outFrame;
 
-				/**
-				 * List of the run files to be merged
-				 */
-				LinkedList<RunFileReader> runs;
+                /**
+                 * List of the run files to be merged
+                 */
+                LinkedList<RunFileReader> runs;
 
-				/**
-				 * Tuple appender for the output frame {@link #outFrame}.
-				 */
-				private FrameTupleAppender outFrameAppender;
+                /**
+                 * Tuple appender for the output frame {@link #outFrame}.
+                 */
+                private FrameTupleAppender outFrameAppender;
 
-				private ISpillableAccumulatingAggregator visitingAggregator;
-				private ArrayTupleBuilder visitingKeyTuple;
+                private ISpillableAccumulatingAggregator visitingAggregator;
+                private ArrayTupleBuilder visitingKeyTuple;
 
-				@SuppressWarnings("unchecked")
-				@Override
-				public void initialize() throws HyracksDataException {
-					runs = (LinkedList<RunFileReader>) env.get(RUNS);
-					writer.open();
+                @SuppressWarnings("unchecked")
+                @Override
+                public void initialize() throws HyracksDataException {
+                    runs = (LinkedList<RunFileReader>) env.get(RUNS);
+                    writer.open();
 
-					try {
-						if (runs.size() <= 0) {
-							// If the aggregate results can be fit into
-							// memory...
-							SpillableGroupingHashTable gTable = (SpillableGroupingHashTable) env
-									.get(GROUPTABLES);
-							if (gTable != null) {
-								gTable.flushFrames(writer, sortOutput);
-							}
-							env.set(GROUPTABLES, null);
-						} else {
-							// Otherwise, merge the run files into a single file
-							inFrames = new ArrayList<ByteBuffer>();
-							outFrame = ctx.allocateFrame();
-							outFrameAppender = new FrameTupleAppender(
-									ctx.getFrameSize());
-							outFrameAppender.reset(outFrame, true);
-							for (int i = 0; i < framesLimit - 1; ++i) {
-								inFrames.add(ctx.allocateFrame());
-							}
-							int passCount = 0;
-							while (runs.size() > 0) {
-								passCount++;
-								try {
-									doPass(runs, passCount);
-								} catch (Exception e) {
-									throw new HyracksDataException(e);
-								}
-							}
-						}
+                    try {
+                        if (runs.size() <= 0) {
+                            // If the aggregate results can be fit into
+                            // memory...
+                            SpillableGroupingHashTable gTable = (SpillableGroupingHashTable) env.get(GROUPTABLES);
+                            if (gTable != null) {
+                                gTable.flushFrames(writer, sortOutput);
+                            }
+                            env.set(GROUPTABLES, null);
+                        } else {
+                            // Otherwise, merge the run files into a single file
+                            inFrames = new ArrayList<ByteBuffer>();
+                            outFrame = ctx.allocateFrame();
+                            outFrameAppender = new FrameTupleAppender(ctx.getFrameSize());
+                            outFrameAppender.reset(outFrame, true);
+                            for (int i = 0; i < framesLimit - 1; ++i) {
+                                inFrames.add(ctx.allocateFrame());
+                            }
+                            int passCount = 0;
+                            while (runs.size() > 0) {
+                                passCount++;
+                                try {
+                                    doPass(runs, passCount);
+                                } catch (Exception e) {
+                                    throw new HyracksDataException(e);
+                                }
+                            }
+                        }
 
-					} finally {
-						writer.close();
-					}
-					env.set(RUNS, null);
-				}
+                    } finally {
+                        writer.close();
+                    }
+                    env.set(RUNS, null);
+                }
 
-				/**
-				 * Merge the run files once.
-				 * 
-				 * @param runs
-				 * @param passCount
-				 * @throws HyracksDataException
-				 * @throws IOException
-				 */
-				private void doPass(LinkedList<RunFileReader> runs,
-						int passCount) throws HyracksDataException, IOException {
-					FileReference newRun = null;
-					IFrameWriter writer = this.writer;
-					boolean finalPass = false;
+                /**
+                 * Merge the run files once.
+                 * 
+                 * @param runs
+                 * @param passCount
+                 * @throws HyracksDataException
+                 * @throws IOException
+                 */
+                private void doPass(LinkedList<RunFileReader> runs, int passCount) throws HyracksDataException,
+                IOException {
+                    FileReference newRun = null;
+                    IFrameWriter writer = this.writer;
+                    boolean finalPass = false;
 
-					int[] storedKeys = new int[keyFields.length];
-					// Get the list of the fields in the stored records.
-					for (int i = 0; i < keyFields.length; ++i) {
-						storedKeys[i] = i;
-					}
+                    int[] storedKeys = new int[keyFields.length];
+                    // Get the list of the fields in the stored records.
+                    for (int i = 0; i < keyFields.length; ++i) {
+                        storedKeys[i] = i;
+                    }
 
-					// Release the space not used
-					if (runs.size() + 1 <= framesLimit) {
-						// If there are run files no more than the available
-						// frame slots...
-						// No run file to be generated, since the result can be
-						// directly
-						// outputted into the output frame for write.
-						finalPass = true;
-						for (int i = inFrames.size() - 1; i >= runs.size(); i--) {
-							inFrames.remove(i);
-						}
-					} else {
-						// Otherwise, a new run file will be created
-						newRun = ctx.getJobletContext().createWorkspaceFile(
-								ExternalHashGroupOperatorDescriptor.class
-										.getSimpleName());
-						writer = new RunFileWriter(newRun, ctx.getIOManager());
-						writer.open();
-					}
-					try {
-						// Create run file read handler for each input frame
-						RunFileReader[] runFileReaders = new RunFileReader[inFrames
-								.size()];
-						// Create input frame accessor
-						FrameTupleAccessor[] tupleAccessors = new FrameTupleAccessor[inFrames
-								.size()];
-						Comparator<ReferenceEntry> comparator = createEntryComparator(comparators);
-						ReferencedPriorityQueue topTuples = new ReferencedPriorityQueue(
-								ctx.getFrameSize(), recordDescriptors[0],
-								inFrames.size(), comparator);
-						// For the index of tuples visited in each frame.
-						int[] tupleIndexes = new int[inFrames.size()];
-						for (int i = 0; i < inFrames.size(); i++) {
-							tupleIndexes[i] = 0;
-							int runIndex = topTuples.peek().getRunid();
-							runFileReaders[runIndex] = runs.get(runIndex);
-							runFileReaders[runIndex].open();
-							// Load the first frame of the file into the main
-							// memory
-							if (runFileReaders[runIndex].nextFrame(inFrames
-									.get(runIndex))) {
-								// initialize the tuple accessor for the frame
-								tupleAccessors[runIndex] = new FrameTupleAccessor(
-										ctx.getFrameSize(),
-										recordDescriptors[0]);
-								tupleAccessors[runIndex].reset(inFrames
-										.get(runIndex));
-								setNextTopTuple(runIndex, tupleIndexes,
-										runFileReaders, tupleAccessors,
-										topTuples);
-							} else {
-								closeRun(runIndex, runFileReaders,
-										tupleAccessors);
-							}
-						}
-						// Merge
-						// Get a key holder for the current working
-						// aggregator keys
-						visitingAggregator = null;
-						visitingKeyTuple = null;
-						// Loop on all run files, and update the key
-						// holder.
-						while (!topTuples.areRunsExhausted()) {
-							// Get the top record
-							ReferenceEntry top = topTuples.peek();
-							int tupleIndex = top.getTupleIndex();
-							int runIndex = topTuples.peek().getRunid();
-							FrameTupleAccessor fta = top.getAccessor();
-							if (visitingAggregator == null) {
-								// Initialize the aggregator
-								visitingAggregator = aggregatorFactory
-										.createSpillableAggregator(ctx,
-												recordDescriptors[0],
-												recordDescriptors[0]);
-								// Initialize the partial aggregation result
-								visitingAggregator.initFromPartial(fta,
-										tupleIndex, keyFields);
-								visitingKeyTuple = new ArrayTupleBuilder(
-										recordDescriptors[0].getFields().length);
-								for (int i = 0; i < keyFields.length; i++) {
-									visitingKeyTuple.addField(fta, tupleIndex,
-											keyFields[i]);
-								}
-							} else {
-								if (compareTupleWithFrame(visitingKeyTuple,
-										fta, tupleIndex, storedKeys, keyFields,
-										comparators) == 0) {
-									// If the two partial results are on the
-									// same key
-									visitingAggregator.accumulatePartialResult(
-											fta, tupleIndex, keyFields);
-								} else {
-									// Otherwise, write the partial result back
-									// to the output frame
-									if (!visitingAggregator.output(
-											outFrameAppender, visitingKeyTuple)) {
-										FrameUtils.flushFrame(outFrame, writer);
-										outFrameAppender.reset(outFrame, true);
-										if (!visitingAggregator.output(
-												outFrameAppender,
-												visitingKeyTuple)) {
-											throw new IllegalStateException();
-										}
-									}
-									// Reset the partial aggregation result
-									visitingAggregator.initFromPartial(fta,
-											tupleIndex, keyFields);
-									visitingKeyTuple.reset();
-									for (int i = 0; i < keyFields.length; i++) {
-										visitingKeyTuple.addField(fta,
-												tupleIndex, keyFields[i]);
-									}
-								}
-							}
-							tupleIndexes[runIndex]++;
-							setNextTopTuple(runIndex, tupleIndexes,
-									runFileReaders, tupleAccessors, topTuples);
-						}
-						// Output the last aggregation result in the frame
-						if (visitingAggregator != null) {
-							if (!visitingAggregator.output(outFrameAppender,
-									visitingKeyTuple)) {
-								FrameUtils.flushFrame(outFrame, writer);
-								outFrameAppender.reset(outFrame, true);
-								if (!visitingAggregator.output(
-										outFrameAppender, visitingKeyTuple)) {
-									throw new IllegalStateException();
-								}
-							}
-						}
-						// Output data into run file writer after all tuples
-						// have been checked
-						if (outFrameAppender.getTupleCount() > 0) {
-							FrameUtils.flushFrame(outFrame, writer);
-							outFrameAppender.reset(outFrame, true);
-						}
-						// empty the input frames
-						runs.subList(0, inFrames.size()).clear();
-						// insert the new run file into the beginning of the run
-						// file list
-						if (!finalPass) {
-							runs.add(0, ((RunFileWriter) writer).createReader());
-						}
-					} catch (Exception ex) {
-						throw new HyracksDataException(ex);
-					} finally {
-						if (!finalPass) {
-							writer.close();
-						}
-					}
-				}
+                    // Release the space not used
+                    if (runs.size() + 1 <= framesLimit) {
+                        // If there are run files no more than the available
+                        // frame slots...
+                        // No run file to be generated, since the result can be
+                        // directly
+                        // outputted into the output frame for write.
+                        finalPass = true;
+                        for (int i = inFrames.size() - 1; i >= runs.size(); i--) {
+                            inFrames.remove(i);
+                        }
+                    } else {
+                        // Otherwise, a new run file will be created
+                        newRun = ctx.getJobletContext().createWorkspaceFile(
+                                ExternalHashGroupOperatorDescriptor.class.getSimpleName());
+                        writer = new RunFileWriter(newRun, ctx.getIOManager());
+                        writer.open();
+                    }
+                    try {
+                        // Create run file read handler for each input frame
+                        RunFileReader[] runFileReaders = new RunFileReader[inFrames.size()];
+                        // Create input frame accessor
+                        FrameTupleAccessor[] tupleAccessors = new FrameTupleAccessor[inFrames.size()];
+                        Comparator<ReferenceEntry> comparator = createEntryComparator(comparators);
+                        ReferencedPriorityQueue topTuples = new ReferencedPriorityQueue(ctx.getFrameSize(),
+                                recordDescriptors[0], inFrames.size(), comparator);
+                        // For the index of tuples visited in each frame.
+                        int[] tupleIndexes = new int[inFrames.size()];
+                        for (int i = 0; i < inFrames.size(); i++) {
+                            tupleIndexes[i] = 0;
+                            int runIndex = topTuples.peek().getRunid();
+                            runFileReaders[runIndex] = runs.get(runIndex);
+                            runFileReaders[runIndex].open();
+                            // Load the first frame of the file into the main
+                            // memory
+                            if (runFileReaders[runIndex].nextFrame(inFrames.get(runIndex))) {
+                                // initialize the tuple accessor for the frame
+                                tupleAccessors[runIndex] = new FrameTupleAccessor(ctx.getFrameSize(),
+                                        recordDescriptors[0]);
+                                tupleAccessors[runIndex].reset(inFrames.get(runIndex));
+                                setNextTopTuple(runIndex, tupleIndexes, runFileReaders, tupleAccessors, topTuples);
+                            } else {
+                                closeRun(runIndex, runFileReaders, tupleAccessors);
+                            }
+                        }
+                        // Merge
+                        // Get a key holder for the current working
+                        // aggregator keys
+                        visitingAggregator = null;
+                        visitingKeyTuple = null;
+                        // Loop on all run files, and update the key
+                        // holder.
+                        while (!topTuples.areRunsExhausted()) {
+                            // Get the top record
+                            ReferenceEntry top = topTuples.peek();
+                            int tupleIndex = top.getTupleIndex();
+                            int runIndex = topTuples.peek().getRunid();
+                            FrameTupleAccessor fta = top.getAccessor();
+                            if (visitingAggregator == null) {
+                                // Initialize the aggregator
+                                visitingAggregator = aggregatorFactory.createSpillableAggregator(ctx,
+                                        recordDescriptors[0], recordDescriptors[0]);
+                                // Initialize the partial aggregation result
+                                visitingAggregator.initFromPartial(fta, tupleIndex, keyFields);
+                                visitingKeyTuple = new ArrayTupleBuilder(recordDescriptors[0].getFields().length);
+                                for (int i = 0; i < keyFields.length; i++) {
+                                    visitingKeyTuple.addField(fta, tupleIndex, keyFields[i]);
+                                }
+                            } else {
+                                if (compareTupleWithFrame(visitingKeyTuple, fta, tupleIndex, storedKeys, keyFields,
+                                        comparators) == 0) {
+                                    // If the two partial results are on the
+                                    // same key
+                                    visitingAggregator.accumulatePartialResult(fta, tupleIndex, keyFields);
+                                } else {
+                                    // Otherwise, write the partial result back
+                                    // to the output frame
+                                    if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
+                                        FrameUtils.flushFrame(outFrame, writer);
+                                        outFrameAppender.reset(outFrame, true);
+                                        if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
+                                            throw new IllegalStateException();
+                                        }
+                                    }
+                                    // Reset the partial aggregation result
+                                    visitingAggregator.initFromPartial(fta, tupleIndex, keyFields);
+                                    visitingKeyTuple.reset();
+                                    for (int i = 0; i < keyFields.length; i++) {
+                                        visitingKeyTuple.addField(fta, tupleIndex, keyFields[i]);
+                                    }
+                                }
+                            }
+                            tupleIndexes[runIndex]++;
+                            setNextTopTuple(runIndex, tupleIndexes, runFileReaders, tupleAccessors, topTuples);
+                        }
+                        // Output the last aggregation result in the frame
+                        if (visitingAggregator != null) {
+                            if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
+                                FrameUtils.flushFrame(outFrame, writer);
+                                outFrameAppender.reset(outFrame, true);
+                                if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
+                                    throw new IllegalStateException();
+                                }
+                            }
+                        }
+                        // Output data into run file writer after all tuples
+                        // have been checked
+                        if (outFrameAppender.getTupleCount() > 0) {
+                            FrameUtils.flushFrame(outFrame, writer);
+                            outFrameAppender.reset(outFrame, true);
+                        }
+                        // empty the input frames
+                        runs.subList(0, inFrames.size()).clear();
+                        // insert the new run file into the beginning of the run
+                        // file list
+                        if (!finalPass) {
+                            runs.add(0, ((RunFileWriter) writer).createReader());
+                        }
+                    } catch (Exception ex) {
+                        throw new HyracksDataException(ex);
+                    } finally {
+                        if (!finalPass) {
+                            writer.close();
+                        }
+                    }
+                }
 
-				/**
-				 * Insert the tuple into the priority queue.
-				 * 
-				 * @param runIndex
-				 * @param tupleIndexes
-				 * @param runCursors
-				 * @param tupleAccessors
-				 * @param topTuples
-				 * @throws IOException
-				 */
-				private void setNextTopTuple(int runIndex, int[] tupleIndexes,
-						RunFileReader[] runCursors,
-						FrameTupleAccessor[] tupleAccessors,
-						ReferencedPriorityQueue topTuples) throws IOException {
-					boolean exists = hasNextTuple(runIndex, tupleIndexes,
-							runCursors, tupleAccessors);
-					if (exists) {
-						topTuples.popAndReplace(tupleAccessors[runIndex],
-								tupleIndexes[runIndex]);
-					} else {
-						topTuples.pop();
-						closeRun(runIndex, runCursors, tupleAccessors);
-					}
-				}
+                /**
+                 * Insert the tuple into the priority queue.
+                 * 
+                 * @param runIndex
+                 * @param tupleIndexes
+                 * @param runCursors
+                 * @param tupleAccessors
+                 * @param topTuples
+                 * @throws IOException
+                 */
+                private void setNextTopTuple(int runIndex, int[] tupleIndexes, RunFileReader[] runCursors,
+                        FrameTupleAccessor[] tupleAccessors, ReferencedPriorityQueue topTuples) throws IOException {
+                    boolean exists = hasNextTuple(runIndex, tupleIndexes, runCursors, tupleAccessors);
+                    if (exists) {
+                        topTuples.popAndReplace(tupleAccessors[runIndex], tupleIndexes[runIndex]);
+                    } else {
+                        topTuples.pop();
+                        closeRun(runIndex, runCursors, tupleAccessors);
+                    }
+                }
 
-				/**
-				 * Check whether there are any more tuples to be checked for the
-				 * given run file from the corresponding input frame.
-				 * 
-				 * If the input frame for this run file is exhausted, load a new
-				 * frame of the run file into the input frame.
-				 * 
-				 * @param runIndex
-				 * @param tupleIndexes
-				 * @param runCursors
-				 * @param tupleAccessors
-				 * @return
-				 * @throws IOException
-				 */
-				private boolean hasNextTuple(int runIndex, int[] tupleIndexes,
-						RunFileReader[] runCursors,
-						FrameTupleAccessor[] tupleAccessors) throws IOException {
+                /**
+                 * Check whether there are any more tuples to be checked for the
+                 * given run file from the corresponding input frame.
+                 * If the input frame for this run file is exhausted, load a new
+                 * frame of the run file into the input frame.
+                 * 
+                 * @param runIndex
+                 * @param tupleIndexes
+                 * @param runCursors
+                 * @param tupleAccessors
+                 * @return
+                 * @throws IOException
+                 */
+                private boolean hasNextTuple(int runIndex, int[] tupleIndexes, RunFileReader[] runCursors,
+                        FrameTupleAccessor[] tupleAccessors) throws IOException {
 
-					if (tupleAccessors[runIndex] == null
-							|| runCursors[runIndex] == null) {
-						/*
-						 * Return false if the targeting run file is not
-						 * available, or the frame for the run file is not
-						 * available.
-						 */
-						return false;
-					} else if (tupleIndexes[runIndex] >= tupleAccessors[runIndex]
-							.getTupleCount()) {
-						/*
-						 * If all tuples in the targeting frame have been
-						 * checked.
-						 */
-						ByteBuffer buf = tupleAccessors[runIndex].getBuffer(); // same-as-inFrames.get(runIndex)
-						// Refill the buffer with contents from the run file.
-						if (runCursors[runIndex].nextFrame(buf)) {
-							tupleIndexes[runIndex] = 0;
-							return hasNextTuple(runIndex, tupleIndexes,
-									runCursors, tupleAccessors);
-						} else {
-							return false;
-						}
-					} else {
-						return true;
-					}
-				}
+                    if (tupleAccessors[runIndex] == null || runCursors[runIndex] == null) {
+                        /*
+                         * Return false if the targeting run file is not
+                         * available, or the frame for the run file is not
+                         * available.
+                         */
+                        return false;
+                    } else if (tupleIndexes[runIndex] >= tupleAccessors[runIndex].getTupleCount()) {
+                        /*
+                         * If all tuples in the targeting frame have been
+                         * checked.
+                         */
+                        ByteBuffer buf = tupleAccessors[runIndex].getBuffer(); // same-as-inFrames.get(runIndex)
+                        // Refill the buffer with contents from the run file.
+                        if (runCursors[runIndex].nextFrame(buf)) {
+                            tupleIndexes[runIndex] = 0;
+                            return hasNextTuple(runIndex, tupleIndexes, runCursors, tupleAccessors);
+                        } else {
+                            return false;
+                        }
+                    } else {
+                        return true;
+                    }
+                }
 
-				/**
-				 * Close the run file, and also the corresponding readers and
-				 * input frame.
-				 * 
-				 * @param index
-				 * @param runCursors
-				 * @param tupleAccessor
-				 * @throws HyracksDataException
-				 */
-				private void closeRun(int index, RunFileReader[] runCursors,
-						IFrameTupleAccessor[] tupleAccessor)
-						throws HyracksDataException {
-					runCursors[index].close();
-					runCursors[index] = null;
-					tupleAccessor[index] = null;
-				}
+                /**
+                 * Close the run file, and also the corresponding readers and
+                 * input frame.
+                 * 
+                 * @param index
+                 * @param runCursors
+                 * @param tupleAccessor
+                 * @throws HyracksDataException
+                 */
+                private void closeRun(int index, RunFileReader[] runCursors, IFrameTupleAccessor[] tupleAccessor)
+                throws HyracksDataException {
+                    runCursors[index].close();
+                    runCursors[index] = null;
+                    tupleAccessor[index] = null;
+                }
 
-				/**
-				 * Compare a tuple (in the format of a {@link ArrayTupleBuilder}
-				 * ) with a record in a frame (in the format of a
-				 * {@link FrameTupleAccessor}). Comparing keys and comparators
-				 * are specified for this method as inputs.
-				 * 
-				 * @param tuple0
-				 * @param accessor1
-				 * @param tIndex1
-				 * @param keys0
-				 * @param keys1
-				 * @param comparators
-				 * @return
-				 */
-				private int compareTupleWithFrame(ArrayTupleBuilder tuple0,
-						FrameTupleAccessor accessor1, int tIndex1, int[] keys0,
-						int[] keys1, IBinaryComparator[] comparators) {
-					int tStart1 = accessor1.getTupleStartOffset(tIndex1);
-					int fStartOffset1 = accessor1.getFieldSlotsLength()
-							+ tStart1;
+                /**
+                 * Compare a tuple (in the format of a {@link ArrayTupleBuilder} ) with a record in a frame (in the format of a {@link FrameTupleAccessor}). Comparing keys and comparators
+                 * are specified for this method as inputs.
+                 * 
+                 * @param tuple0
+                 * @param accessor1
+                 * @param tIndex1
+                 * @param keys0
+                 * @param keys1
+                 * @param comparators
+                 * @return
+                 */
+                private int compareTupleWithFrame(ArrayTupleBuilder tuple0, FrameTupleAccessor accessor1, int tIndex1,
+                        int[] keys0, int[] keys1, IBinaryComparator[] comparators) {
+                    int tStart1 = accessor1.getTupleStartOffset(tIndex1);
+                    int fStartOffset1 = accessor1.getFieldSlotsLength() + tStart1;
 
-					for (int i = 0; i < keys0.length; ++i) {
-						int fIdx0 = keys0[i];
-						int fStart0 = (i == 0 ? 0
-								: tuple0.getFieldEndOffsets()[fIdx0 - 1]);
-						int fEnd0 = tuple0.getFieldEndOffsets()[fIdx0];
-						int fLen0 = fEnd0 - fStart0;
+                    for (int i = 0; i < keys0.length; ++i) {
+                        int fIdx0 = keys0[i];
+                        int fStart0 = (i == 0 ? 0 : tuple0.getFieldEndOffsets()[fIdx0 - 1]);
+                        int fEnd0 = tuple0.getFieldEndOffsets()[fIdx0];
+                        int fLen0 = fEnd0 - fStart0;
 
-						int fIdx1 = keys1[i];
-						int fStart1 = accessor1.getFieldStartOffset(tIndex1,
-								fIdx1);
-						int fEnd1 = accessor1.getFieldEndOffset(tIndex1, fIdx1);
-						int fLen1 = fEnd1 - fStart1;
+                        int fIdx1 = keys1[i];
+                        int fStart1 = accessor1.getFieldStartOffset(tIndex1, fIdx1);
+                        int fEnd1 = accessor1.getFieldEndOffset(tIndex1, fIdx1);
+                        int fLen1 = fEnd1 - fStart1;
 
-						int c = comparators[i].compare(tuple0.getByteArray(),
-								fStart0, fLen0, accessor1.getBuffer().array(),
-								fStart1 + fStartOffset1, fLen1);
-						if (c != 0) {
-							return c;
-						}
-					}
-					return 0;
-				}
-			};
-			return op;
-		}
+                        int c = comparators[i].compare(tuple0.getByteArray(), fStart0, fLen0, accessor1.getBuffer()
+                                .array(), fStart1 + fStartOffset1, fLen1);
+                        if (c != 0) {
+                            return c;
+                        }
+                    }
+                    return 0;
+                }
+            };
+            return op;
+        }
 
-		@Override
-		public IOperatorDescriptor getOwner() {
-			return ExternalHashGroupOperatorDescriptor.this;
-		}
+        @Override
+        public IOperatorDescriptor getOwner() {
+            return ExternalHashGroupOperatorDescriptor.this;
+        }
 
-		private Comparator<ReferenceEntry> createEntryComparator(
-				final IBinaryComparator[] comparators) {
-			return new Comparator<ReferenceEntry>() {
-				public int compare(ReferenceEntry tp1, ReferenceEntry tp2) {
-					FrameTupleAccessor fta1 = (FrameTupleAccessor) tp1
-							.getAccessor();
-					FrameTupleAccessor fta2 = (FrameTupleAccessor) tp2
-							.getAccessor();
-					int j1 = (Integer) tp1.getTupleIndex();
-					int j2 = (Integer) tp2.getTupleIndex();
-					byte[] b1 = fta1.getBuffer().array();
-					byte[] b2 = fta2.getBuffer().array();
-					for (int f = 0; f < keyFields.length; ++f) {
-						int fIdx = keyFields[f];
-						int s1 = fta1.getTupleStartOffset(j1)
-								+ fta1.getFieldSlotsLength()
-								+ fta1.getFieldStartOffset(j1, fIdx);
-						int l1 = fta1.getFieldEndOffset(j1, fIdx)
-								- fta1.getFieldStartOffset(j1, fIdx);
-						int s2 = fta2.getTupleStartOffset(j2)
-								+ fta2.getFieldSlotsLength()
-								+ fta2.getFieldStartOffset(j2, fIdx);
-						int l2 = fta2.getFieldEndOffset(j2, fIdx)
-								- fta2.getFieldStartOffset(j2, fIdx);
-						int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
-						if (c != 0) {
-							return c;
-						}
-					}
-					return 0;
-				}
-			};
-		}
+        private Comparator<ReferenceEntry> createEntryComparator(final IBinaryComparator[] comparators) {
+            return new Comparator<ReferenceEntry>() {
+                public int compare(ReferenceEntry tp1, ReferenceEntry tp2) {
+                    FrameTupleAccessor fta1 = (FrameTupleAccessor) tp1.getAccessor();
+                    FrameTupleAccessor fta2 = (FrameTupleAccessor) tp2.getAccessor();
+                    int j1 = (Integer) tp1.getTupleIndex();
+                    int j2 = (Integer) tp2.getTupleIndex();
+                    byte[] b1 = fta1.getBuffer().array();
+                    byte[] b2 = fta2.getBuffer().array();
+                    for (int f = 0; f < keyFields.length; ++f) {
+                        int fIdx = keyFields[f];
+                        int s1 = fta1.getTupleStartOffset(j1) + fta1.getFieldSlotsLength()
+                        + fta1.getFieldStartOffset(j1, fIdx);
+                        int l1 = fta1.getFieldEndOffset(j1, fIdx) - fta1.getFieldStartOffset(j1, fIdx);
+                        int s2 = fta2.getTupleStartOffset(j2) + fta2.getFieldSlotsLength()
+                        + fta2.getFieldStartOffset(j2, fIdx);
+                        int l2 = fta2.getFieldEndOffset(j2, fIdx) - fta2.getFieldStartOffset(j2, fIdx);
+                        int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
+                        if (c != 0) {
+                            return c;
+                        }
+                    }
+                    return 0;
+                }
+            };
+        }
 
-	}
+    }
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/IAccumulatingAggregatorFactory.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/IAccumulatingAggregatorFactory.java
index a2e8281..3fc7d79 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/IAccumulatingAggregatorFactory.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/IAccumulatingAggregatorFactory.java
@@ -21,11 +21,9 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 
 public interface IAccumulatingAggregatorFactory extends Serializable {
-	IAccumulatingAggregator createAggregator(IHyracksStageletContext ctx,
-			RecordDescriptor inRecordDesc, RecordDescriptor outRecordDescriptor)
-			throws HyracksDataException;
+    IAccumulatingAggregator createAggregator(IHyracksStageletContext ctx, RecordDescriptor inRecordDesc,
+            RecordDescriptor outRecordDescriptor) throws HyracksDataException;
 
-	ISpillableAccumulatingAggregator createSpillableAggregator(
-			IHyracksStageletContext ctx, RecordDescriptor inRecordDesc,
-			RecordDescriptor outRecordDescriptor) throws HyracksDataException;
+    ISpillableAccumulatingAggregator createSpillableAggregator(IHyracksStageletContext ctx,
+            RecordDescriptor inRecordDesc, RecordDescriptor outRecordDescriptor) throws HyracksDataException;
 }
\ No newline at end of file
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ISpillableAccumulatingAggregator.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ISpillableAccumulatingAggregator.java
index ab7b624..59c69eb 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ISpillableAccumulatingAggregator.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/ISpillableAccumulatingAggregator.java
@@ -22,23 +22,19 @@
 /**
  * An spillable version of the {@link IAccumulatingAggregator} supporting
  * external aggregation.
- * 
  */
-public interface ISpillableAccumulatingAggregator extends
-		IAccumulatingAggregator {
+public interface ISpillableAccumulatingAggregator extends IAccumulatingAggregator {
 
-	public void initFromPartial(IFrameTupleAccessor accessor, int tIndex,
-			int[] keyFieldIndexes) throws HyracksDataException;
+    public void initFromPartial(IFrameTupleAccessor accessor, int tIndex, int[] keyFieldIndexes)
+    throws HyracksDataException;
 
-	/**
-	 * 
-	 * @param accessor
-	 * @param tIndex
-	 * @throws HyracksDataException
-	 */
-	public void accumulatePartialResult(IFrameTupleAccessor accessor,
-			int tIndex, int[] keyFieldIndexes) throws HyracksDataException;
+    /**
+     * @param accessor
+     * @param tIndex
+     * @throws HyracksDataException
+     */
+    public void accumulatePartialResult(IFrameTupleAccessor accessor, int tIndex, int[] keyFieldIndexes)
+    throws HyracksDataException;
 
-	public boolean output(FrameTupleAppender appender, ArrayTupleBuilder tbder)
-			throws HyracksDataException;
+    public boolean output(FrameTupleAppender appender, ArrayTupleBuilder tbder) throws HyracksDataException;
 }
diff --git a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/SpillableGroupingHashTable.java b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/SpillableGroupingHashTable.java
index 16bd972..b3b9f24 100644
--- a/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/SpillableGroupingHashTable.java
+++ b/hyracks-dataflow-std/src/main/java/edu/uci/ics/hyracks/dataflow/std/group/SpillableGroupingHashTable.java
@@ -34,551 +34,536 @@
 import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
 
 /**
- * An in-mem hash table for spillable grouping operations. 
- * 
+ * An in-mem hash table for spillable grouping operations.
  * A table of {@link #Link}s are maintained in this object, and each row
- * of this table represents a hash partition. 
- * 
- * 
+ * of this table represents a hash partition.
  */
 public class SpillableGroupingHashTable {
 
-	/**
-	 * Context.
-	 */
-	private final IHyracksStageletContext ctx;
+    /**
+     * Context.
+     */
+    private final IHyracksStageletContext ctx;
 
-	/**
-	 * Columns for group-by
-	 */
-	private final int[] fields;
+    /**
+     * Columns for group-by
+     */
+    private final int[] fields;
 
-	/**
-	 * Key fields of records in the hash table (starting from 0
-	 * to the number of the key fields). 
-	 * 
-	 * This is different from the key fields in the input records, 
-	 * since these fields are extracted when being inserted into
-	 * the hash table.
-	 */
-	private final int[] storedKeys;
+    /**
+     * Key fields of records in the hash table (starting from 0
+     * to the number of the key fields).
+     * This is different from the key fields in the input records,
+     * since these fields are extracted when being inserted into
+     * the hash table.
+     */
+    private final int[] storedKeys;
 
-	/**
-	 * Comparators: one for each column in {@link #groupFields}
-	 */
-	private final IBinaryComparator[] comparators;
+    /**
+     * Comparators: one for each column in {@link #groupFields}
+     */
+    private final IBinaryComparator[] comparators;
 
-	/**
-	 * Record descriptor for the input tuple.
-	 */
-	private final RecordDescriptor inRecordDescriptor;
+    /**
+     * Record descriptor for the input tuple.
+     */
+    private final RecordDescriptor inRecordDescriptor;
 
-	/**
-	 * Record descriptor for the partial aggregation result.
-	 */
-	private final RecordDescriptor outputRecordDescriptor;
+    /**
+     * Record descriptor for the partial aggregation result.
+     */
+    private final RecordDescriptor outputRecordDescriptor;
 
-	/**
-	 * Accumulators in the main memory.
-	 */
-	private ISpillableAccumulatingAggregator[] accumulators;
+    /**
+     * Accumulators in the main memory.
+     */
+    private ISpillableAccumulatingAggregator[] accumulators;
 
-	/**
-	 * The hashing group table containing pointers to aggregators and also the
-	 * corresponding key tuples. So for each entry, there will be three integer
-	 * fields:
-	 * 
-	 * 1. The frame index containing the key tuple; 2. The tuple index inside of
-	 * the frame for the key tuple; 3. The index of the aggregator.
-	 * 
-	 * Note that each link in the table is a partition for the input records. Multiple
-	 * records in the same partition based on the {@link #tpc} are stored as
-	 * pointers.
-	 */
-	private final Link[] table;
+    /**
+     * The hashing group table containing pointers to aggregators and also the
+     * corresponding key tuples. So for each entry, there will be three integer
+     * fields:
+     * 1. The frame index containing the key tuple; 2. The tuple index inside of
+     * the frame for the key tuple; 3. The index of the aggregator.
+     * Note that each link in the table is a partition for the input records. Multiple
+     * records in the same partition based on the {@link #tpc} are stored as
+     * pointers.
+     */
+    private final Link[] table;
 
-	/**
-	 * Number of accumulators.
-	 */
-	private int accumulatorSize = 0;
+    /**
+     * Number of accumulators.
+     */
+    private int accumulatorSize = 0;
 
-	/**
-	 * Factory for the aggregators.
-	 */
-	private final IAccumulatingAggregatorFactory aggregatorFactory;
+    /**
+     * Factory for the aggregators.
+     */
+    private final IAccumulatingAggregatorFactory aggregatorFactory;
 
-	private final List<ByteBuffer> frames;
-	
-	private final ByteBuffer outFrame;
+    private final List<ByteBuffer> frames;
 
-	/**
-	 * Frame appender for output frames in {@link #frames}.
-	 */
-	private final FrameTupleAppender appender;
+    private final ByteBuffer outFrame;
 
-	/**
-	 * The count of used frames in the table.
-	 * 
-	 * Note that this cannot be replaced by {@link #frames} since frames will
-	 * not be removed after being created.
-	 */
-	private int dataFrameCount;
+    /**
+     * Frame appender for output frames in {@link #frames}.
+     */
+    private final FrameTupleAppender appender;
 
-	/**
-	 * Pointers for the sorted aggregators
-	 */
-	private int[] tPointers;
+    /**
+     * The count of used frames in the table.
+     * Note that this cannot be replaced by {@link #frames} since frames will
+     * not be removed after being created.
+     */
+    private int dataFrameCount;
 
-	private static final int INIT_ACCUMULATORS_SIZE = 8;
+    /**
+     * Pointers for the sorted aggregators
+     */
+    private int[] tPointers;
 
-	/**
-	 * The maximum number of frames available for this hashing group table.
-	 */
-	private final int framesLimit;
+    private static final int INIT_ACCUMULATORS_SIZE = 8;
 
-	private final FrameTuplePairComparator ftpc;
+    /**
+     * The maximum number of frames available for this hashing group table.
+     */
+    private final int framesLimit;
 
-	/**
-	 * A partition computer to partition the hashing group table.
-	 */
-	private final ITuplePartitionComputer tpc;
+    private final FrameTuplePairComparator ftpc;
 
-	/**
-	 * Accessors for the tuples. Two accessors are necessary during the sort.
-	 */
-	private final FrameTupleAccessor storedKeysAccessor1;
-	private final FrameTupleAccessor storedKeysAccessor2;
+    /**
+     * A partition computer to partition the hashing group table.
+     */
+    private final ITuplePartitionComputer tpc;
 
-	/**
-	 * Create a spillable grouping hash table. 
-	 * @param ctx					The context of the job.
-	 * @param fields				Fields of keys for grouping.
-	 * @param comparatorFactories	The comparators.
-	 * @param tpcf					The partitioners. These are used to partition the incoming records into proper partition of the hash table.
-	 * @param aggregatorFactory		The aggregators.
-	 * @param inRecordDescriptor	Record descriptor for input data.
-	 * @param outputRecordDescriptor	Record descriptor for output data.
-	 * @param framesLimit			The maximum number of frames usable in the memory for hash table.
-	 * @param tableSize				The size of the table, which specified the number of partitions of the table.
-	 */
-	public SpillableGroupingHashTable(IHyracksStageletContext ctx, int[] fields,
-			IBinaryComparatorFactory[] comparatorFactories,
-			ITuplePartitionComputerFactory tpcf,
-			IAccumulatingAggregatorFactory aggregatorFactory,
-			RecordDescriptor inRecordDescriptor,
-			RecordDescriptor outputRecordDescriptor, int framesLimit,
-			int tableSize) {
-		this.ctx = ctx;
-		this.fields = fields;
+    /**
+     * Accessors for the tuples. Two accessors are necessary during the sort.
+     */
+    private final FrameTupleAccessor storedKeysAccessor1;
+    private final FrameTupleAccessor storedKeysAccessor2;
 
-		storedKeys = new int[fields.length];
-		@SuppressWarnings("rawtypes")
-		ISerializerDeserializer[] storedKeySerDeser = new ISerializerDeserializer[fields.length];
-		
-		// Note that after storing a record into the hash table, the index for the fields should
-		// be updated. Here we assume that all these key fields are written at the beginning of 
-		// the record, so their index should start from 0 and end at the length of the key fields.
-		for (int i = 0; i < fields.length; ++i) {
-			storedKeys[i] = i;
-			storedKeySerDeser[i] = inRecordDescriptor.getFields()[fields[i]];
-		}
-		RecordDescriptor storedKeysRecordDescriptor = new RecordDescriptor(
-				storedKeySerDeser);
-		storedKeysAccessor1 = new FrameTupleAccessor(ctx.getFrameSize(),
-				storedKeysRecordDescriptor);
-		storedKeysAccessor2 = new FrameTupleAccessor(ctx.getFrameSize(),
-				storedKeysRecordDescriptor);
+    /**
+     * Create a spillable grouping hash table.
+     * 
+     * @param ctx
+     *            The context of the job.
+     * @param fields
+     *            Fields of keys for grouping.
+     * @param comparatorFactories
+     *            The comparators.
+     * @param tpcf
+     *            The partitioners. These are used to partition the incoming records into proper partition of the hash table.
+     * @param aggregatorFactory
+     *            The aggregators.
+     * @param inRecordDescriptor
+     *            Record descriptor for input data.
+     * @param outputRecordDescriptor
+     *            Record descriptor for output data.
+     * @param framesLimit
+     *            The maximum number of frames usable in the memory for hash table.
+     * @param tableSize
+     *            The size of the table, which specified the number of partitions of the table.
+     */
+    public SpillableGroupingHashTable(IHyracksStageletContext ctx, int[] fields,
+            IBinaryComparatorFactory[] comparatorFactories, ITuplePartitionComputerFactory tpcf,
+            IAccumulatingAggregatorFactory aggregatorFactory, RecordDescriptor inRecordDescriptor,
+            RecordDescriptor outputRecordDescriptor, int framesLimit, int tableSize) {
+        this.ctx = ctx;
+        this.fields = fields;
 
-		comparators = new IBinaryComparator[comparatorFactories.length];
-		for (int i = 0; i < comparatorFactories.length; ++i) {
-			comparators[i] = comparatorFactories[i].createBinaryComparator();
-		}
+        storedKeys = new int[fields.length];
+        @SuppressWarnings("rawtypes")
+        ISerializerDeserializer[] storedKeySerDeser = new ISerializerDeserializer[fields.length];
 
-		this.table = new Link[tableSize];
+        // Note that after storing a record into the hash table, the index for the fields should
+        // be updated. Here we assume that all these key fields are written at the beginning of 
+        // the record, so their index should start from 0 and end at the length of the key fields.
+        for (int i = 0; i < fields.length; ++i) {
+            storedKeys[i] = i;
+            storedKeySerDeser[i] = inRecordDescriptor.getFields()[fields[i]];
+        }
+        RecordDescriptor storedKeysRecordDescriptor = new RecordDescriptor(storedKeySerDeser);
+        storedKeysAccessor1 = new FrameTupleAccessor(ctx.getFrameSize(), storedKeysRecordDescriptor);
+        storedKeysAccessor2 = new FrameTupleAccessor(ctx.getFrameSize(), storedKeysRecordDescriptor);
 
-		this.aggregatorFactory = aggregatorFactory;
-		accumulators = new ISpillableAccumulatingAggregator[INIT_ACCUMULATORS_SIZE];
+        comparators = new IBinaryComparator[comparatorFactories.length];
+        for (int i = 0; i < comparatorFactories.length; ++i) {
+            comparators[i] = comparatorFactories[i].createBinaryComparator();
+        }
 
-		this.framesLimit = framesLimit;
+        this.table = new Link[tableSize];
 
-		// Tuple pair comparator
-		ftpc = new FrameTuplePairComparator(fields, storedKeys, comparators);
-		
-		// Partitioner
-		tpc = tpcf.createPartitioner();
+        this.aggregatorFactory = aggregatorFactory;
+        accumulators = new ISpillableAccumulatingAggregator[INIT_ACCUMULATORS_SIZE];
 
-		this.inRecordDescriptor = inRecordDescriptor;
-		this.outputRecordDescriptor = outputRecordDescriptor;
-		frames = new ArrayList<ByteBuffer>();
-		appender = new FrameTupleAppender(ctx.getFrameSize());
+        this.framesLimit = framesLimit;
 
-		dataFrameCount = -1;
-		
-		outFrame = ctx.allocateFrame();
-	}
+        // Tuple pair comparator
+        ftpc = new FrameTuplePairComparator(fields, storedKeys, comparators);
 
-	public void reset() {
-		dataFrameCount = -1;
-		tPointers = null;
-		// Reset the grouping hash table
-		for (int i = 0; i < table.length; i++) {
-			table[i] = new Link();
-		}
-	}
+        // Partitioner
+        tpc = tpcf.createPartitioner();
 
-	public int getFrameCount() {
-		return dataFrameCount;
-	}
+        this.inRecordDescriptor = inRecordDescriptor;
+        this.outputRecordDescriptor = outputRecordDescriptor;
+        frames = new ArrayList<ByteBuffer>();
+        appender = new FrameTupleAppender(ctx.getFrameSize());
 
-	/**
-	 * How to define pointers for the partial aggregation
-	 * 
-	 * @return
-	 */
-	public int[] getTPointers() {
-		return tPointers;
-	}
+        dataFrameCount = -1;
 
-	/**
-	 * Redefine the number of fields in the pointer. 
-	 * 
-	 * Only two pointers are necessary for external grouping: one is to the
-	 * index of the hash table, and the other is to the row index inside of the
-	 * hash table.
-	 * 
-	 * @return
-	 */
-	public int getPtrFields() {
-		return 2;
-	}
+        outFrame = ctx.allocateFrame();
+    }
 
-	public List<ByteBuffer> getFrames() {
-		return frames;
-	}
+    public void reset() {
+        dataFrameCount = -1;
+        tPointers = null;
+        // Reset the grouping hash table
+        for (int i = 0; i < table.length; i++) {
+            table[i] = new Link();
+        }
+    }
 
-	/**
-	 * Set the working frame to the next available frame in the
-	 * frame list. There are two cases:<br>
-	 * 
-	 * 1) If the next frame is not initialized, allocate
-	 * a new frame.
-	 * 
-	 * 2) When frames are already created, they are recycled.
-	 * 
-	 * @return Whether a new frame is added successfully.
-	 */
-	private boolean nextAvailableFrame() {
-		// Return false if the number of frames is equal to the limit.
-		if (dataFrameCount + 1 >= framesLimit)
-			return false;
-		
-		if (frames.size() < framesLimit) {
-			// Insert a new frame
-			ByteBuffer frame = ctx.allocateFrame();
-			frame.position(0);
-			frame.limit(frame.capacity());
-			frames.add(frame);
-			appender.reset(frame, true);
-			dataFrameCount++;
-		} else {
-			// Reuse an old frame
-			dataFrameCount++;
-			ByteBuffer frame = frames.get(dataFrameCount);
-			frame.position(0);
-			frame.limit(frame.capacity());
-			appender.reset(frame, true);
-		}
-		return true;
-	}
+    public int getFrameCount() {
+        return dataFrameCount;
+    }
 
-	/**
-	 * Insert a new record from the input frame.
-	 * 
-	 * @param accessor
-	 * @param tIndex
-	 * @return
-	 * @throws HyracksDataException
-	 */
-	public boolean insert(FrameTupleAccessor accessor, int tIndex)
-			throws HyracksDataException {
-		if (dataFrameCount < 0)
-			nextAvailableFrame();
-		// Get the partition for the inserting tuple
-		int entry = tpc.partition(accessor, tIndex, table.length);
-		Link link = table[entry];
-		if (link == null) {
-			link = table[entry] = new Link();
-		}
-		// Find the corresponding aggregator from existing aggregators
-		ISpillableAccumulatingAggregator aggregator = null;
-		for (int i = 0; i < link.size; i += 3) {
-			int sbIndex = link.pointers[i];
-			int stIndex = link.pointers[i + 1];
-			int saIndex = link.pointers[i + 2];
-			storedKeysAccessor1.reset(frames.get(sbIndex));
-			int c = ftpc
-					.compare(accessor, tIndex, storedKeysAccessor1, stIndex);
-			if (c == 0) {
-				aggregator = accumulators[saIndex];
-				break;
-			}
-		}
-		// Do insert
-		if (aggregator == null) {
-			// Did not find the aggregator. Insert a new aggregator entry
-			if (!appender.appendProjection(accessor, tIndex, fields)) {
-				if (!nextAvailableFrame()) {
-					// If buffer is full, return false to trigger a run file
-					// write
-					return false;
-				} else {
-					// Try to do insert after adding a new frame.
-					if (!appender.appendProjection(accessor, tIndex, fields)) {
-						throw new IllegalStateException();
-					}
-				}
-			}
-			int sbIndex = dataFrameCount;
-			int stIndex = appender.getTupleCount() - 1;
-			if (accumulatorSize >= accumulators.length) {
-				accumulators = Arrays.copyOf(accumulators,
-						accumulators.length * 2);
-			}
-			int saIndex = accumulatorSize++;
-			aggregator = accumulators[saIndex] = aggregatorFactory
-					.createSpillableAggregator(ctx, inRecordDescriptor,
-							outputRecordDescriptor);
-			aggregator.init(accessor, tIndex);
-			link.add(sbIndex, stIndex, saIndex);
-		}
-		aggregator.accumulate(accessor, tIndex);
-		return true;
-	}
+    /**
+     * How to define pointers for the partial aggregation
+     * 
+     * @return
+     */
+    public int[] getTPointers() {
+        return tPointers;
+    }
 
-	/**
-	 * Sort partial results
-	 */
-	public void sortFrames() {
-		int totalTCount = 0;
-		// Get the number of records
-		for (int i = 0; i < table.length; i++) {
-			if (table[i] == null)
-				continue;
-			totalTCount += table[i].size / 3;
-		}
-		// Start sorting:
-		/*
-		 * Based on the data structure for the partial aggregates, the
-		 * pointers should be initialized.
-		 */
-		tPointers = new int[totalTCount * getPtrFields()];
-		// Initialize pointers
-		int ptr = 0;
-		// Maintain two pointers to each entry of the hashing group table
-		for (int i = 0; i < table.length; i++) {
-			if (table[i] == null)
-				continue;
-			for (int j = 0; j < table[i].size; j = j + 3) {
-				tPointers[ptr * getPtrFields()] = i;
-				tPointers[ptr * getPtrFields() + 1] = j;
-				ptr++;
-			}
-		}
-		// Sort using quick sort
-		if (tPointers.length > 0) {
-			sort(tPointers, 0, totalTCount);
-		}
-	}
+    /**
+     * Redefine the number of fields in the pointer.
+     * Only two pointers are necessary for external grouping: one is to the
+     * index of the hash table, and the other is to the row index inside of the
+     * hash table.
+     * 
+     * @return
+     */
+    public int getPtrFields() {
+        return 2;
+    }
 
-	/**
-	 * 
-	 * @param writer
-	 * @throws HyracksDataException
-	 */
-	public void flushFrames(IFrameWriter writer, boolean sorted) throws HyracksDataException {
-		FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
+    public List<ByteBuffer> getFrames() {
+        return frames;
+    }
 
-		ISpillableAccumulatingAggregator aggregator = null;
-		writer.open();
-		appender.reset(outFrame, true);
-		if (sorted){
-			sortFrames();
-		}
-		if(tPointers == null){
-			// Not sorted
-			for (int i = 0; i < table.length; ++i) {
-	            Link link = table[i];
-	            if (link != null) {
-	                for (int j = 0; j < link.size; j += 3) {
-	                    int bIndex = link.pointers[j];
-	                    int tIndex = link.pointers[j + 1];
-	                    int aIndex = link.pointers[j + 2];
-	                    ByteBuffer keyBuffer = frames.get(bIndex);
-	                    storedKeysAccessor1.reset(keyBuffer);
-	                    aggregator = accumulators[aIndex];
-	                    while (!aggregator.output(appender, storedKeysAccessor1, tIndex, storedKeys)) {
-	                    	FrameUtils.flushFrame(outFrame, writer);
-	                    	appender.reset(outFrame, true);
-	                    }
-	                }
-	            }
-	        }
-	        if (appender.getTupleCount() != 0) {
-	        	FrameUtils.flushFrame(outFrame, writer);
-	        }
-	        return;
-		}
-		int n = tPointers.length / getPtrFields();
-		for (int ptr = 0; ptr < n; ptr++) {
-			int tableIndex = tPointers[ptr * 2];
-			int rowIndex = tPointers[ptr * 2 + 1];
-			int frameIndex = table[tableIndex].pointers[rowIndex];
-			int tupleIndex = table[tableIndex].pointers[rowIndex + 1];
-			int aggregatorIndex = table[tableIndex].pointers[rowIndex + 2];
-			// Get the frame containing the value
-			ByteBuffer buffer = frames.get(frameIndex);
-			storedKeysAccessor1.reset(buffer);
+    /**
+     * Set the working frame to the next available frame in the
+     * frame list. There are two cases:<br>
+     * 1) If the next frame is not initialized, allocate
+     * a new frame.
+     * 2) When frames are already created, they are recycled.
+     * 
+     * @return Whether a new frame is added successfully.
+     */
+    private boolean nextAvailableFrame() {
+        // Return false if the number of frames is equal to the limit.
+        if (dataFrameCount + 1 >= framesLimit)
+            return false;
 
-			// Get the aggregator
-			aggregator = accumulators[aggregatorIndex];
-			// Insert
-			if (!aggregator.output(appender, storedKeysAccessor1, tupleIndex,
-					fields)) {
-				FrameUtils.flushFrame(outFrame, writer);
-				appender.reset(outFrame, true);
-				if (!aggregator.output(appender, storedKeysAccessor1,
-						tupleIndex, fields)) {
-					throw new IllegalStateException();
-				} else {
-					accumulators[aggregatorIndex] = null;
-				}
-			} else {
-				accumulators[aggregatorIndex] = null;
-			}
-		}
-		if (appender.getTupleCount() > 0) {
-			FrameUtils.flushFrame(outFrame, writer);
-		}
-	}
+        if (frames.size() < framesLimit) {
+            // Insert a new frame
+            ByteBuffer frame = ctx.allocateFrame();
+            frame.position(0);
+            frame.limit(frame.capacity());
+            frames.add(frame);
+            appender.reset(frame, true);
+            dataFrameCount++;
+        } else {
+            // Reuse an old frame
+            dataFrameCount++;
+            ByteBuffer frame = frames.get(dataFrameCount);
+            frame.position(0);
+            frame.limit(frame.capacity());
+            appender.reset(frame, true);
+        }
+        return true;
+    }
 
-	private void sort(int[] tPointers, int offset, int length) {
-		int m = offset + (length >> 1);
-		// Get table index
-		int mTable = tPointers[m * 2];
-		int mRow = tPointers[m * 2 + 1];
-		// Get frame and tuple index
-		int mFrame = table[mTable].pointers[mRow];
-		int mTuple = table[mTable].pointers[mRow + 1];
-		storedKeysAccessor1.reset(frames.get(mFrame));
+    /**
+     * Insert a new record from the input frame.
+     * 
+     * @param accessor
+     * @param tIndex
+     * @return
+     * @throws HyracksDataException
+     */
+    public boolean insert(FrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
+        if (dataFrameCount < 0)
+            nextAvailableFrame();
+        // Get the partition for the inserting tuple
+        int entry = tpc.partition(accessor, tIndex, table.length);
+        Link link = table[entry];
+        if (link == null) {
+            link = table[entry] = new Link();
+        }
+        // Find the corresponding aggregator from existing aggregators
+        ISpillableAccumulatingAggregator aggregator = null;
+        for (int i = 0; i < link.size; i += 3) {
+            int sbIndex = link.pointers[i];
+            int stIndex = link.pointers[i + 1];
+            int saIndex = link.pointers[i + 2];
+            storedKeysAccessor1.reset(frames.get(sbIndex));
+            int c = ftpc.compare(accessor, tIndex, storedKeysAccessor1, stIndex);
+            if (c == 0) {
+                aggregator = accumulators[saIndex];
+                break;
+            }
+        }
+        // Do insert
+        if (aggregator == null) {
+            // Did not find the aggregator. Insert a new aggregator entry
+            if (!appender.appendProjection(accessor, tIndex, fields)) {
+                if (!nextAvailableFrame()) {
+                    // If buffer is full, return false to trigger a run file
+                    // write
+                    return false;
+                } else {
+                    // Try to do insert after adding a new frame.
+                    if (!appender.appendProjection(accessor, tIndex, fields)) {
+                        throw new IllegalStateException();
+                    }
+                }
+            }
+            int sbIndex = dataFrameCount;
+            int stIndex = appender.getTupleCount() - 1;
+            if (accumulatorSize >= accumulators.length) {
+                accumulators = Arrays.copyOf(accumulators, accumulators.length * 2);
+            }
+            int saIndex = accumulatorSize++;
+            aggregator = accumulators[saIndex] = aggregatorFactory.createSpillableAggregator(ctx, inRecordDescriptor,
+                    outputRecordDescriptor);
+            aggregator.init(accessor, tIndex);
+            link.add(sbIndex, stIndex, saIndex);
+        }
+        aggregator.accumulate(accessor, tIndex);
+        return true;
+    }
 
-		int a = offset;
-		int b = a;
-		int c = offset + length - 1;
-		int d = c;
-		while (true) {
-			while (b <= c) {
-				int bTable = tPointers[b * 2];
-				int bRow = tPointers[b * 2 + 1];
-				int bFrame = table[bTable].pointers[bRow];
-				int bTuple = table[bTable].pointers[bRow + 1];
-				storedKeysAccessor2.reset(frames.get(bFrame));
-				int cmp = ftpc.compare(storedKeysAccessor2, bTuple,
-						storedKeysAccessor1, mTuple);
-				// int cmp = compare(tPointers, b, mi, mj, mv);
-				if (cmp > 0) {
-					break;
-				}
-				if (cmp == 0) {
-					swap(tPointers, a++, b);
-				}
-				++b;
-			}
-			while (c >= b) {
-				int cTable = tPointers[c * 2];
-				int cRow = tPointers[c * 2 + 1];
-				int cFrame = table[cTable].pointers[cRow];
-				int cTuple = table[cTable].pointers[cRow + 1];
-				storedKeysAccessor2.reset(frames.get(cFrame));
-				int cmp = ftpc.compare(storedKeysAccessor2, cTuple,
-						storedKeysAccessor1, mTuple);
-				// int cmp = compare(tPointers, c, mi, mj, mv);
-				if (cmp < 0) {
-					break;
-				}
-				if (cmp == 0) {
-					swap(tPointers, c, d--);
-				}
-				--c;
-			}
-			if (b > c)
-				break;
-			swap(tPointers, b++, c--);
-		}
+    /**
+     * Sort partial results
+     */
+    public void sortFrames() {
+        int totalTCount = 0;
+        // Get the number of records
+        for (int i = 0; i < table.length; i++) {
+            if (table[i] == null)
+                continue;
+            totalTCount += table[i].size / 3;
+        }
+        // Start sorting:
+        /*
+         * Based on the data structure for the partial aggregates, the
+         * pointers should be initialized.
+         */
+        tPointers = new int[totalTCount * getPtrFields()];
+        // Initialize pointers
+        int ptr = 0;
+        // Maintain two pointers to each entry of the hashing group table
+        for (int i = 0; i < table.length; i++) {
+            if (table[i] == null)
+                continue;
+            for (int j = 0; j < table[i].size; j = j + 3) {
+                tPointers[ptr * getPtrFields()] = i;
+                tPointers[ptr * getPtrFields() + 1] = j;
+                ptr++;
+            }
+        }
+        // Sort using quick sort
+        if (tPointers.length > 0) {
+            sort(tPointers, 0, totalTCount);
+        }
+    }
 
-		int s;
-		int n = offset + length;
-		s = Math.min(a - offset, b - a);
-		vecswap(tPointers, offset, b - s, s);
-		s = Math.min(d - c, n - d - 1);
-		vecswap(tPointers, b, n - s, s);
+    /**
+     * @param writer
+     * @throws HyracksDataException
+     */
+    public void flushFrames(IFrameWriter writer, boolean sorted) throws HyracksDataException {
+        FrameTupleAppender appender = new FrameTupleAppender(ctx.getFrameSize());
 
-		if ((s = b - a) > 1) {
-			sort(tPointers, offset, s);
-		}
-		if ((s = d - c) > 1) {
-			sort(tPointers, n - s, s);
-		}
-	}
+        ISpillableAccumulatingAggregator aggregator = null;
+        writer.open();
+        appender.reset(outFrame, true);
+        if (sorted) {
+            sortFrames();
+        }
+        if (tPointers == null) {
+            // Not sorted
+            for (int i = 0; i < table.length; ++i) {
+                Link link = table[i];
+                if (link != null) {
+                    for (int j = 0; j < link.size; j += 3) {
+                        int bIndex = link.pointers[j];
+                        int tIndex = link.pointers[j + 1];
+                        int aIndex = link.pointers[j + 2];
+                        ByteBuffer keyBuffer = frames.get(bIndex);
+                        storedKeysAccessor1.reset(keyBuffer);
+                        aggregator = accumulators[aIndex];
+                        while (!aggregator.output(appender, storedKeysAccessor1, tIndex, storedKeys)) {
+                            FrameUtils.flushFrame(outFrame, writer);
+                            appender.reset(outFrame, true);
+                        }
+                    }
+                }
+            }
+            if (appender.getTupleCount() != 0) {
+                FrameUtils.flushFrame(outFrame, writer);
+            }
+            return;
+        }
+        int n = tPointers.length / getPtrFields();
+        for (int ptr = 0; ptr < n; ptr++) {
+            int tableIndex = tPointers[ptr * 2];
+            int rowIndex = tPointers[ptr * 2 + 1];
+            int frameIndex = table[tableIndex].pointers[rowIndex];
+            int tupleIndex = table[tableIndex].pointers[rowIndex + 1];
+            int aggregatorIndex = table[tableIndex].pointers[rowIndex + 2];
+            // Get the frame containing the value
+            ByteBuffer buffer = frames.get(frameIndex);
+            storedKeysAccessor1.reset(buffer);
 
-	private void swap(int x[], int a, int b) {
-		for (int i = 0; i < 2; ++i) {
-			int t = x[a * 2 + i];
-			x[a * 2 + i] = x[b * 2 + i];
-			x[b * 2 + i] = t;
-		}
-	}
+            // Get the aggregator
+            aggregator = accumulators[aggregatorIndex];
+            // Insert
+            if (!aggregator.output(appender, storedKeysAccessor1, tupleIndex, fields)) {
+                FrameUtils.flushFrame(outFrame, writer);
+                appender.reset(outFrame, true);
+                if (!aggregator.output(appender, storedKeysAccessor1, tupleIndex, fields)) {
+                    throw new IllegalStateException();
+                } else {
+                    accumulators[aggregatorIndex] = null;
+                }
+            } else {
+                accumulators[aggregatorIndex] = null;
+            }
+        }
+        if (appender.getTupleCount() > 0) {
+            FrameUtils.flushFrame(outFrame, writer);
+        }
+    }
 
-	private void vecswap(int x[], int a, int b, int n) {
-		for (int i = 0; i < n; i++, a++, b++) {
-			swap(x, a, b);
-		}
-	}
+    private void sort(int[] tPointers, int offset, int length) {
+        int m = offset + (length >> 1);
+        // Get table index
+        int mTable = tPointers[m * 2];
+        int mRow = tPointers[m * 2 + 1];
+        // Get frame and tuple index
+        int mFrame = table[mTable].pointers[mRow];
+        int mTuple = table[mTable].pointers[mRow + 1];
+        storedKeysAccessor1.reset(frames.get(mFrame));
 
-	/**
-	 * The pointers in the link store 3 int values for each entry in the
-	 * hashtable: (bufferIdx, tIndex, accumulatorIdx).
-	 * 
-	 * @author vinayakb
-	 */
-	private static class Link {
-		private static final int INIT_POINTERS_SIZE = 9;
+        int a = offset;
+        int b = a;
+        int c = offset + length - 1;
+        int d = c;
+        while (true) {
+            while (b <= c) {
+                int bTable = tPointers[b * 2];
+                int bRow = tPointers[b * 2 + 1];
+                int bFrame = table[bTable].pointers[bRow];
+                int bTuple = table[bTable].pointers[bRow + 1];
+                storedKeysAccessor2.reset(frames.get(bFrame));
+                int cmp = ftpc.compare(storedKeysAccessor2, bTuple, storedKeysAccessor1, mTuple);
+                // int cmp = compare(tPointers, b, mi, mj, mv);
+                if (cmp > 0) {
+                    break;
+                }
+                if (cmp == 0) {
+                    swap(tPointers, a++, b);
+                }
+                ++b;
+            }
+            while (c >= b) {
+                int cTable = tPointers[c * 2];
+                int cRow = tPointers[c * 2 + 1];
+                int cFrame = table[cTable].pointers[cRow];
+                int cTuple = table[cTable].pointers[cRow + 1];
+                storedKeysAccessor2.reset(frames.get(cFrame));
+                int cmp = ftpc.compare(storedKeysAccessor2, cTuple, storedKeysAccessor1, mTuple);
+                // int cmp = compare(tPointers, c, mi, mj, mv);
+                if (cmp < 0) {
+                    break;
+                }
+                if (cmp == 0) {
+                    swap(tPointers, c, d--);
+                }
+                --c;
+            }
+            if (b > c)
+                break;
+            swap(tPointers, b++, c--);
+        }
 
-		int[] pointers;
-		int size;
+        int s;
+        int n = offset + length;
+        s = Math.min(a - offset, b - a);
+        vecswap(tPointers, offset, b - s, s);
+        s = Math.min(d - c, n - d - 1);
+        vecswap(tPointers, b, n - s, s);
 
-		Link() {
-			pointers = new int[INIT_POINTERS_SIZE];
-			size = 0;
-		}
+        if ((s = b - a) > 1) {
+            sort(tPointers, offset, s);
+        }
+        if ((s = d - c) > 1) {
+            sort(tPointers, n - s, s);
+        }
+    }
 
-		void add(int bufferIdx, int tIndex, int accumulatorIdx) {
-			while (size + 3 > pointers.length) {
-				pointers = Arrays.copyOf(pointers, pointers.length * 2);
-			}
-			pointers[size++] = bufferIdx;
-			pointers[size++] = tIndex;
-			pointers[size++] = accumulatorIdx;
-		}
+    private void swap(int x[], int a, int b) {
+        for (int i = 0; i < 2; ++i) {
+            int t = x[a * 2 + i];
+            x[a * 2 + i] = x[b * 2 + i];
+            x[b * 2 + i] = t;
+        }
+    }
 
-		public String toString() {
-			StringBuilder sb = new StringBuilder();
-			sb.append("[Size=" + size + "]");
-			for (int i = 0; i < pointers.length; i = i + 3) {
-				sb.append(pointers[i] + ",");
-				sb.append(pointers[i + 1] + ",");
-				sb.append(pointers[i + 2] + "; ");
-			}
-			return sb.toString();
-		}
-	}
+    private void vecswap(int x[], int a, int b, int n) {
+        for (int i = 0; i < n; i++, a++, b++) {
+            swap(x, a, b);
+        }
+    }
+
+    /**
+     * The pointers in the link store 3 int values for each entry in the
+     * hashtable: (bufferIdx, tIndex, accumulatorIdx).
+     * 
+     * @author vinayakb
+     */
+    private static class Link {
+        private static final int INIT_POINTERS_SIZE = 9;
+
+        int[] pointers;
+        int size;
+
+        Link() {
+            pointers = new int[INIT_POINTERS_SIZE];
+            size = 0;
+        }
+
+        void add(int bufferIdx, int tIndex, int accumulatorIdx) {
+            while (size + 3 > pointers.length) {
+                pointers = Arrays.copyOf(pointers, pointers.length * 2);
+            }
+            pointers[size++] = bufferIdx;
+            pointers[size++] = tIndex;
+            pointers[size++] = accumulatorIdx;
+        }
+
+        public String toString() {
+            StringBuilder sb = new StringBuilder();
+            sb.append("[Size=" + size + "]");
+            for (int i = 0; i < pointers.length; i = i + 3) {
+                sb.append(pointers[i] + ",");
+                sb.append(pointers[i + 1] + ",");
+                sb.append(pointers[i + 2] + "; ");
+            }
+            return sb.toString();
+        }
+    }
 }
diff --git a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/spillable/ExternalAggregateTest.java b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/spillable/ExternalAggregateTest.java
index 421b47c..2621049 100644
--- a/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/spillable/ExternalAggregateTest.java
+++ b/hyracks-examples/hyracks-integration-tests/src/test/java/edu/uci/ics/hyracks/tests/spillable/ExternalAggregateTest.java
@@ -60,517 +60,371 @@
 
 /**
  * Test cases for external hash group operator.
- * 
  */
 public class ExternalAggregateTest extends AbstractIntegrationTest {
 
-	/**
-	 * Test 01: aggregate (count) on single field, on a simple data set.
-	 * 
-	 * @throws Exception
-	 */
-	@Test
-	public void externalAggregateTestSingleFieldSimpleData() throws Exception {
-		JobSpecification spec = new JobSpecification();
+    /**
+     * Test 01: aggregate (count) on single field, on a simple data set.
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void externalAggregateTestSingleFieldSimpleData() throws Exception {
+        JobSpecification spec = new JobSpecification();
 
-		IFileSplitProvider splitProvider = new ConstantFileSplitProvider(
-				new FileSplit[] {
-						new FileSplit(NC2_ID, new FileReference(new File(
-								"data/wordcount.tsv"))),
-						new FileSplit(NC1_ID, new FileReference(new File(
-								"data/wordcount.tsv"))) });
+        IFileSplitProvider splitProvider = new ConstantFileSplitProvider(new FileSplit[] {
+                new FileSplit(NC2_ID, new FileReference(new File("data/wordcount.tsv"))),
+                new FileSplit(NC1_ID, new FileReference(new File("data/wordcount.tsv"))) });
 
-		// Input format: a string field as the key
-		RecordDescriptor desc = new RecordDescriptor(
-				new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE });
+        // Input format: a string field as the key
+        RecordDescriptor desc = new RecordDescriptor(
+                new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE });
 
-		// Output format: a string field as the key, and an integer field as the
-		// count
-		RecordDescriptor outputRec = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE });
+        // Output format: a string field as the key, and an integer field as the
+        // count
+        RecordDescriptor outputRec = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE });
 
-		// Data set format: word(string),count(int)
-		FileScanOperatorDescriptor csvScanner = new FileScanOperatorDescriptor(
-				spec,
-				splitProvider,
-				new DelimitedDataTupleParserFactory(
-						new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE },
-						','), desc);
-		PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		csvScanner.setPartitionConstraint(csvPartitionConstraint);
+        // Data set format: word(string),count(int)
+        FileScanOperatorDescriptor csvScanner = new FileScanOperatorDescriptor(
+                spec,
+                splitProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE }, ','),
+                desc);
+        PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        csvScanner.setPartitionConstraint(csvPartitionConstraint);
 
-		int[] keys = new int[] { 0 };
-		int tableSize = 8;
+        int[] keys = new int[] { 0 };
+        int tableSize = 8;
 
-		ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(
-				spec, // Job conf
-				keys, // Group key
-				3, // Number of frames
-				false, // Whether to sort the output
-				// Hash partitioner
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
-				// Key comparator
-				new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
-				// Aggregator factory
-				new MultiAggregatorFactory(
-						new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-				outputRec, // Output format
-				tableSize // Size of the hashing table, which is used to control
-							// the partition when hashing
-		);
+        ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(
+                spec, // Job conf
+                keys, // Group key
+                3, // Number of frames
+                false, // Whether to sort the output
+                // Hash partitioner
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
+                        // Key comparator
+                        new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
+                        // Aggregator factory
+                        new MultiAggregatorFactory(new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
+                        outputRec, // Output format
+                        tableSize // Size of the hashing table, which is used to control
+                        // the partition when hashing
+        );
 
-		PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		grouper.setPartitionConstraint(grouperPartitionConstraint);
+        PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        grouper.setPartitionConstraint(grouperPartitionConstraint);
 
-		IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(
-				spec,
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
-		spec.connect(conn1, csvScanner, 0, grouper, 0);
+        IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(spec,
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
+        spec.connect(conn1, csvScanner, 0, grouper, 0);
 
-		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		printer.setPartitionConstraint(printerPartitionConstraint);
+        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+        PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        printer.setPartitionConstraint(printerPartitionConstraint);
 
-		IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
-		spec.connect(conn2, grouper, 0, printer, 0);
+        IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
+        spec.connect(conn2, grouper, 0, printer, 0);
 
-		spec.addRoot(printer);
-		runTest(spec);
-	}
+        spec.addRoot(printer);
+        runTest(spec);
+    }
 
-	/**
-	 * Test 02: Control experiment using in-memory aggregator, on the same data
-	 * set of {@link #externalAggregateTest01()}
-	 * 
-	 * @throws Exception
-	 */
-	@Test
-	public void externalAggregateTestSingleFieldSimpleDataInMemControl()
-			throws Exception {
-		JobSpecification spec = new JobSpecification();
+    /**
+     * Test 02: Control experiment using in-memory aggregator, on the same data
+     * set of {@link #externalAggregateTest01()}
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void externalAggregateTestSingleFieldSimpleDataInMemControl() throws Exception {
+        JobSpecification spec = new JobSpecification();
 
-		IFileSplitProvider splitProvider = new ConstantFileSplitProvider(
-				new FileSplit[] {
-						new FileSplit(NC2_ID, new FileReference(new File(
-								"data/wordcount.tsv"))),
-						new FileSplit(NC1_ID, new FileReference(new File(
-								"data/wordcount.tsv"))) });
+        IFileSplitProvider splitProvider = new ConstantFileSplitProvider(new FileSplit[] {
+                new FileSplit(NC2_ID, new FileReference(new File("data/wordcount.tsv"))),
+                new FileSplit(NC1_ID, new FileReference(new File("data/wordcount.tsv"))) });
 
-		RecordDescriptor desc = new RecordDescriptor(
-				new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE });
+        RecordDescriptor desc = new RecordDescriptor(
+                new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE });
 
-		RecordDescriptor outputRec = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE });
+        RecordDescriptor outputRec = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE });
 
-		FileScanOperatorDescriptor csvScanner = new FileScanOperatorDescriptor(
-				spec,
-				splitProvider,
-				new DelimitedDataTupleParserFactory(
-						new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE },
-						','), desc);
-		PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		csvScanner.setPartitionConstraint(csvPartitionConstraint);
+        FileScanOperatorDescriptor csvScanner = new FileScanOperatorDescriptor(
+                spec,
+                splitProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE }, ','),
+                desc);
+        PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        csvScanner.setPartitionConstraint(csvPartitionConstraint);
 
-		int[] keys = new int[] { 0 };
-		int tableSize = 8;
+        int[] keys = new int[] { 0 };
+        int tableSize = 8;
 
-		HashGroupOperatorDescriptor grouper = new HashGroupOperatorDescriptor(
-				spec,
-				keys,
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
-				new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
-				new MultiAggregatorFactory(
-						new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-				outputRec, tableSize);
+        HashGroupOperatorDescriptor grouper = new HashGroupOperatorDescriptor(
+                spec,
+                keys,
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
+                        new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
+                        new MultiAggregatorFactory(new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
+                        outputRec, tableSize);
 
-		PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		grouper.setPartitionConstraint(grouperPartitionConstraint);
+        PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        grouper.setPartitionConstraint(grouperPartitionConstraint);
 
-		IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(
-				spec,
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
-		spec.connect(conn1, csvScanner, 0, grouper, 0);
+        IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(spec,
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
+        spec.connect(conn1, csvScanner, 0, grouper, 0);
 
-		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] {
-						new AbsoluteLocationConstraint(NC2_ID),
-						new AbsoluteLocationConstraint(NC1_ID) });
-		printer.setPartitionConstraint(printerPartitionConstraint);
+        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+        PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(new LocationConstraint[] {
+                new AbsoluteLocationConstraint(NC2_ID), new AbsoluteLocationConstraint(NC1_ID) });
+        printer.setPartitionConstraint(printerPartitionConstraint);
 
-		IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
-		spec.connect(conn2, grouper, 0, printer, 0);
+        IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
+        spec.connect(conn2, grouper, 0, printer, 0);
 
-		spec.addRoot(printer);
-		runTest(spec);
-	}
+        spec.addRoot(printer);
+        runTest(spec);
+    }
 
-	/**
-	 * Test 03: aggregates on multiple fields
-	 * 
-	 * @throws Exception
-	 */
-	@Test
-	public void externalAggregateTestMultiAggFields() throws Exception {
-		JobSpecification spec = new JobSpecification();
+    /**
+     * Test 03: aggregates on multiple fields
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void externalAggregateTestMultiAggFields() throws Exception {
+        JobSpecification spec = new JobSpecification();
 
-		FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID,
-				new FileReference(new File("data/tpch0.001/lineitem.tbl"))) };
-		IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(
-				ordersSplits);
-		RecordDescriptor ordersDesc = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE });
+        FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID, new FileReference(new File(
+        "data/tpch0.001/lineitem.tbl"))) };
+        IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(ordersSplits);
+        RecordDescriptor ordersDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
 
-		FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(
-				spec, ordersSplitsProvider,
-				new DelimitedDataTupleParserFactory(new IValueParserFactory[] {
-						UTF8StringParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
-		PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(ordersPartitionConstraint);
+        FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(spec, ordersSplitsProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, FloatParserFactory.INSTANCE, FloatParserFactory.INSTANCE,
+                        FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
+        PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(ordersPartitionConstraint);
 
-		RecordDescriptor outputRec = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE });
+        RecordDescriptor outputRec = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE });
 
-		PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(csvPartitionConstraint);
+        PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(csvPartitionConstraint);
 
-		int[] keys = new int[] { 0 };
-		int tableSize = 8;
+        int[] keys = new int[] { 0 };
+        int tableSize = 8;
 
-		ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(
-				spec,
-				keys,
-				3,
-				false,
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
-				new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
-				new MultiAggregatorFactory(
-						new IFieldValueResultingAggregatorFactory[] {
-								new CountAggregatorFactory(),
-								new SumAggregatorFactory(4),
-								new MinMaxAggregatorFactory(true, 5) }),
-				outputRec, tableSize);
+        ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(spec, keys, 3, false,
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }),
+                        new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE },
+                        new MultiAggregatorFactory(new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory(),
+                                new SumAggregatorFactory(4), new MinMaxAggregatorFactory(true, 5) }), outputRec, tableSize);
 
-		PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		grouper.setPartitionConstraint(grouperPartitionConstraint);
+        PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        grouper.setPartitionConstraint(grouperPartitionConstraint);
 
-		IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(
-				spec,
-				new FieldHashPartitionComputerFactory(
-						keys,
-						new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
-		spec.connect(conn1, ordScanner, 0, grouper, 0);
+        IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(spec,
+                new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] { UTF8StringBinaryHashFunctionFactory.INSTANCE }));
+        spec.connect(conn1, ordScanner, 0, grouper, 0);
 
-		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		printer.setPartitionConstraint(printerPartitionConstraint);
+        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+        PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        printer.setPartitionConstraint(printerPartitionConstraint);
 
-		IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
-		spec.connect(conn2, grouper, 0, printer, 0);
+        IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
+        spec.connect(conn2, grouper, 0, printer, 0);
 
-		spec.addRoot(printer);
-		runTest(spec);
-	}
+        spec.addRoot(printer);
+        runTest(spec);
+    }
 
-	/**
-	 * Test 05: aggregate on multiple key fields
-	 * 
-	 * @throws Exception
-	 */
-	@Test
-	public void externalAggregateTestMultiKeys() throws Exception {
-		JobSpecification spec = new JobSpecification();
+    /**
+     * Test 05: aggregate on multiple key fields
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void externalAggregateTestMultiKeys() throws Exception {
+        JobSpecification spec = new JobSpecification();
 
-		FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID,
-				new FileReference(new File("data/tpch0.001/lineitem.tbl"))) };
-		IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(
-				ordersSplits);
-		RecordDescriptor ordersDesc = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE });
+        FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID, new FileReference(new File(
+        "data/tpch0.001/lineitem.tbl"))) };
+        IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(ordersSplits);
+        RecordDescriptor ordersDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
 
-		FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(
-				spec, ordersSplitsProvider,
-				new DelimitedDataTupleParserFactory(new IValueParserFactory[] {
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
-		PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(ordersPartitionConstraint);
+        FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(spec, ordersSplitsProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, FloatParserFactory.INSTANCE, FloatParserFactory.INSTANCE,
+                        FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
+        PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(ordersPartitionConstraint);
 
-		RecordDescriptor outputRec = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE });
+        RecordDescriptor outputRec = new RecordDescriptor(new ISerializerDeserializer[] {
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE });
 
-		PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(csvPartitionConstraint);
+        PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(csvPartitionConstraint);
 
-		// Group on two fields
-		int[] keys = new int[] { 0, 1 };
-		int tableSize = 8;
+        // Group on two fields
+        int[] keys = new int[] { 0, 1 };
+        int tableSize = 8;
 
-		ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(
-				spec,
-				keys,
-				3,
-				false,
-				new FieldHashPartitionComputerFactory(keys,
-						new IBinaryHashFunctionFactory[] {
-								UTF8StringBinaryHashFunctionFactory.INSTANCE,
-								UTF8StringBinaryHashFunctionFactory.INSTANCE }),
-				new IBinaryComparatorFactory[] {
-						UTF8StringBinaryComparatorFactory.INSTANCE,
-						UTF8StringBinaryComparatorFactory.INSTANCE },
-				new MultiAggregatorFactory(
-						new IFieldValueResultingAggregatorFactory[] {
-								new CountAggregatorFactory(),
-								new SumAggregatorFactory(4),
-								new MinMaxAggregatorFactory(true, 5) }),
-				outputRec, tableSize);
+        ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(spec, keys, 3, false,
+                new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                        UTF8StringBinaryHashFunctionFactory.INSTANCE, UTF8StringBinaryHashFunctionFactory.INSTANCE }),
+                        new IBinaryComparatorFactory[] { UTF8StringBinaryComparatorFactory.INSTANCE,
+            UTF8StringBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                    new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory(),
+                            new SumAggregatorFactory(4), new MinMaxAggregatorFactory(true, 5) }), outputRec,
+                            tableSize);
 
-		PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		grouper.setPartitionConstraint(grouperPartitionConstraint);
+        PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        grouper.setPartitionConstraint(grouperPartitionConstraint);
 
-		IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(
-				spec, new FieldHashPartitionComputerFactory(keys,
-						new IBinaryHashFunctionFactory[] {
-								UTF8StringBinaryHashFunctionFactory.INSTANCE,
-								UTF8StringBinaryHashFunctionFactory.INSTANCE }));
-		spec.connect(conn1, ordScanner, 0, grouper, 0);
+        IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(spec,
+                new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                        UTF8StringBinaryHashFunctionFactory.INSTANCE, UTF8StringBinaryHashFunctionFactory.INSTANCE }));
+        spec.connect(conn1, ordScanner, 0, grouper, 0);
 
-		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		printer.setPartitionConstraint(printerPartitionConstraint);
+        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+        PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        printer.setPartitionConstraint(printerPartitionConstraint);
 
-		IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
-		spec.connect(conn2, grouper, 0, printer, 0);
+        IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
+        spec.connect(conn2, grouper, 0, printer, 0);
 
-		spec.addRoot(printer);
-		runTest(spec);
-	}
+        spec.addRoot(printer);
+        runTest(spec);
+    }
 
-	/**
-	 * Test 06: tests on non-string key field
-	 * 
-	 * @throws Exception
-	 */
-	@Test
-	public void externalAggregateTestNonStringKey() throws Exception {
-		JobSpecification spec = new JobSpecification();
+    /**
+     * Test 06: tests on non-string key field
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void externalAggregateTestNonStringKey() throws Exception {
+        JobSpecification spec = new JobSpecification();
 
-		FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID,
-				new FileReference(new File("data/tpch0.001/lineitem.tbl"))) };
-		IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(
-				ordersSplits);
-		RecordDescriptor ordersDesc = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE });
+        FileSplit[] ordersSplits = new FileSplit[] { new FileSplit(NC2_ID, new FileReference(new File(
+        "data/tpch0.001/lineitem.tbl"))) };
+        IFileSplitProvider ordersSplitsProvider = new ConstantFileSplitProvider(ordersSplits);
+        RecordDescriptor ordersDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
 
-		FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(
-				spec, ordersSplitsProvider,
-				new DelimitedDataTupleParserFactory(new IValueParserFactory[] {
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						IntegerParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						FloatParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE,
-						UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
-		PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(ordersPartitionConstraint);
+        FileScanOperatorDescriptor ordScanner = new FileScanOperatorDescriptor(spec, ordersSplitsProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, FloatParserFactory.INSTANCE, FloatParserFactory.INSTANCE,
+                        FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, }, '|'), ordersDesc);
+        PartitionConstraint ordersPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(ordersPartitionConstraint);
 
-		RecordDescriptor outputRec = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE });
+        RecordDescriptor outputRec = new RecordDescriptor(new ISerializerDeserializer[] {
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE });
 
-		PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		ordScanner.setPartitionConstraint(csvPartitionConstraint);
+        PartitionConstraint csvPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        ordScanner.setPartitionConstraint(csvPartitionConstraint);
 
-		// Group on two fields
-		int[] keys = new int[] { 0, 1 };
-		int tableSize = 8;
+        // Group on two fields
+        int[] keys = new int[] { 0, 1 };
+        int tableSize = 8;
 
-		ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(
-				spec, keys, 3000, true, new FieldHashPartitionComputerFactory(
-						keys, new IBinaryHashFunctionFactory[] {
-								IntegerBinaryHashFunctionFactory.INSTANCE,
-								IntegerBinaryHashFunctionFactory.INSTANCE }),
-				new IBinaryComparatorFactory[] {
-						IntegerBinaryComparatorFactory.INSTANCE,
-						IntegerBinaryComparatorFactory.INSTANCE },
-				new MultiAggregatorFactory(
-						new IFieldValueResultingAggregatorFactory[] {
-								new CountAggregatorFactory(),
-								new SumAggregatorFactory(4),
-								new MinMaxAggregatorFactory(true, 5) }),
-				outputRec, tableSize);
+        ExternalHashGroupOperatorDescriptor grouper = new ExternalHashGroupOperatorDescriptor(spec, keys, 3000, true,
+                new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                        IntegerBinaryHashFunctionFactory.INSTANCE, IntegerBinaryHashFunctionFactory.INSTANCE }),
+                        new IBinaryComparatorFactory[] { IntegerBinaryComparatorFactory.INSTANCE,
+            IntegerBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                    new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory(),
+                            new SumAggregatorFactory(4), new MinMaxAggregatorFactory(true, 5) }), outputRec,
+                            tableSize);
 
-		PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		grouper.setPartitionConstraint(grouperPartitionConstraint);
+        PartitionConstraint grouperPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        grouper.setPartitionConstraint(grouperPartitionConstraint);
 
-		IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(
-				spec, new FieldHashPartitionComputerFactory(keys,
-						new IBinaryHashFunctionFactory[] {
-								IntegerBinaryHashFunctionFactory.INSTANCE,
-								IntegerBinaryHashFunctionFactory.INSTANCE }));
-		spec.connect(conn1, ordScanner, 0, grouper, 0);
+        IConnectorDescriptor conn1 = new MToNHashPartitioningConnectorDescriptor(spec,
+                new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                        IntegerBinaryHashFunctionFactory.INSTANCE, IntegerBinaryHashFunctionFactory.INSTANCE }));
+        spec.connect(conn1, ordScanner, 0, grouper, 0);
 
-		PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
-		PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
-				new LocationConstraint[] { new AbsoluteLocationConstraint(
-						NC1_ID) });
-		printer.setPartitionConstraint(printerPartitionConstraint);
+        PrinterOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);
+        PartitionConstraint printerPartitionConstraint = new ExplicitPartitionConstraint(
+                new LocationConstraint[] { new AbsoluteLocationConstraint(NC1_ID) });
+        printer.setPartitionConstraint(printerPartitionConstraint);
 
-		IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
-		spec.connect(conn2, grouper, 0, printer, 0);
+        IConnectorDescriptor conn2 = new OneToOneConnectorDescriptor(spec);
+        spec.connect(conn2, grouper, 0, printer, 0);
 
-		spec.addRoot(printer);
-		runTest(spec);
-	}
+        spec.addRoot(printer);
+        runTest(spec);
+    }
 }
diff --git a/hyracks-examples/text-example/textclient/src/main/java/edu/uci/ics/hyracks/examples/text/client/ExternalGroupClient.java b/hyracks-examples/text-example/textclient/src/main/java/edu/uci/ics/hyracks/examples/text/client/ExternalGroupClient.java
index 9b88062..f02c855 100644
--- a/hyracks-examples/text-example/textclient/src/main/java/edu/uci/ics/hyracks/examples/text/client/ExternalGroupClient.java
+++ b/hyracks-examples/text-example/textclient/src/main/java/edu/uci/ics/hyracks/examples/text/client/ExternalGroupClient.java
@@ -65,308 +65,243 @@
 /**
  * The application client for the performance tests of the external hash group
  * operator.
- * 
  */
 public class ExternalGroupClient {
-	private static class Options {
-		@Option(name = "-host", usage = "Hyracks Cluster Controller Host name", required = true)
-		public String host;
+    private static class Options {
+        @Option(name = "-host", usage = "Hyracks Cluster Controller Host name", required = true)
+        public String host;
 
-		@Option(name = "-port", usage = "Hyracks Cluster Controller Port (default: 1099)")
-		public int port = 1099;
+        @Option(name = "-port", usage = "Hyracks Cluster Controller Port (default: 1099)")
+        public int port = 1099;
 
-		@Option(name = "-app", usage = "Hyracks Application name", required = true)
-		public String app;
+        @Option(name = "-app", usage = "Hyracks Application name", required = true)
+        public String app;
 
-		@Option(name = "-infile-splits", usage = "Comma separated list of file-splits for the input. A file-split is <node-name>:<path>", required = true)
-		public String inFileSplits;
+        @Option(name = "-infile-splits", usage = "Comma separated list of file-splits for the input. A file-split is <node-name>:<path>", required = true)
+        public String inFileSplits;
 
-		@Option(name = "-outfile-splits", usage = "Comma separated list of file-splits for the output", required = true)
-		public String outFileSplits;
+        @Option(name = "-outfile-splits", usage = "Comma separated list of file-splits for the output", required = true)
+        public String outFileSplits;
 
-		@Option(name = "-hashtable-size", usage = "Hash table size (default: 8191)", required = false)
-		public int htSize = 8191;
+        @Option(name = "-hashtable-size", usage = "Hash table size (default: 8191)", required = false)
+        public int htSize = 8191;
 
-		@Option(name = "-frames-limit", usage = "Frame size (default: 32768)", required = false)
-		public int framesLimit = 32768;
+        @Option(name = "-frames-limit", usage = "Frame size (default: 32768)", required = false)
+        public int framesLimit = 32768;
 
-		@Option(name = "-sortbuffer-size", usage = "Sort buffer size in frames (default: 512)", required = false)
-		public int sbSize = 512;
+        @Option(name = "-sortbuffer-size", usage = "Sort buffer size in frames (default: 512)", required = false)
+        public int sbSize = 512;
 
-		@Option(name = "-sort-output", usage = "Whether to sort the output (default: true)", required = false)
-		public boolean sortOutput = false;
+        @Option(name = "-sort-output", usage = "Whether to sort the output (default: true)", required = false)
+        public boolean sortOutput = false;
 
-		@Option(name = "-out-plain", usage = "Whether to output plain text (default: true)", required = false)
-		public boolean outPlain = true;
-	}
+        @Option(name = "-out-plain", usage = "Whether to output plain text (default: true)", required = false)
+        public boolean outPlain = true;
+    }
 
-	/**
-	 * @param args
-	 */
-	public static void main(String[] args) throws Exception {
-		Options options = new Options();
-		CmdLineParser parser = new CmdLineParser(options);
-		parser.parseArgument(args);
+    /**
+     * @param args
+     */
+    public static void main(String[] args) throws Exception {
+        Options options = new Options();
+        CmdLineParser parser = new CmdLineParser(options);
+        parser.parseArgument(args);
 
-		IHyracksClientConnection hcc = new HyracksRMIConnection(options.host,
-				options.port);
+        IHyracksClientConnection hcc = new HyracksRMIConnection(options.host, options.port);
 
-		JobSpecification job;
+        JobSpecification job;
 
-		for (int i = 0; i < 3; i++) {
-			long start = System.currentTimeMillis();
-			job = createJob(parseFileSplits(options.inFileSplits),
-					parseFileSplits(options.outFileSplits, i % 2),
-					options.htSize, options.sbSize, options.framesLimit,
-					options.sortOutput, i % 2, options.outPlain);
+        for (int i = 0; i < 3; i++) {
+            long start = System.currentTimeMillis();
+            job = createJob(parseFileSplits(options.inFileSplits), parseFileSplits(options.outFileSplits, i % 2),
+                    options.htSize, options.sbSize, options.framesLimit, options.sortOutput, i % 2, options.outPlain);
 
-			System.out.print(i + "\t" + (System.currentTimeMillis() - start));
-			start = System.currentTimeMillis();
-			UUID jobId = hcc.createJob(options.app, job);
-			hcc.start(jobId);
-			hcc.waitForCompletion(jobId);
-			System.out.println("\t" + (System.currentTimeMillis() - start));
-		}
-	}
+            System.out.print(i + "\t" + (System.currentTimeMillis() - start));
+            start = System.currentTimeMillis();
+            UUID jobId = hcc.createJob(options.app, job);
+            hcc.start(jobId);
+            hcc.waitForCompletion(jobId);
+            System.out.println("\t" + (System.currentTimeMillis() - start));
+        }
+    }
 
-	private static FileSplit[] parseFileSplits(String fileSplits) {
-		String[] splits = fileSplits.split(",");
-		FileSplit[] fSplits = new FileSplit[splits.length];
-		for (int i = 0; i < splits.length; ++i) {
-			String s = splits[i].trim();
-			int idx = s.indexOf(':');
-			if (idx < 0) {
-				throw new IllegalArgumentException("File split " + s
-						+ " not well formed");
-			}
-			fSplits[i] = new FileSplit(s.substring(0, idx), new FileReference(
-					new File(s.substring(idx + 1))));
-		}
-		return fSplits;
-	}
+    private static FileSplit[] parseFileSplits(String fileSplits) {
+        String[] splits = fileSplits.split(",");
+        FileSplit[] fSplits = new FileSplit[splits.length];
+        for (int i = 0; i < splits.length; ++i) {
+            String s = splits[i].trim();
+            int idx = s.indexOf(':');
+            if (idx < 0) {
+                throw new IllegalArgumentException("File split " + s + " not well formed");
+            }
+            fSplits[i] = new FileSplit(s.substring(0, idx), new FileReference(new File(s.substring(idx + 1))));
+        }
+        return fSplits;
+    }
 
-	private static FileSplit[] parseFileSplits(String fileSplits, int count) {
-		String[] splits = fileSplits.split(",");
-		FileSplit[] fSplits = new FileSplit[splits.length];
-		for (int i = 0; i < splits.length; ++i) {
-			String s = splits[i].trim();
-			int idx = s.indexOf(':');
-			if (idx < 0) {
-				throw new IllegalArgumentException("File split " + s
-						+ " not well formed");
-			}
-			fSplits[i] = new FileSplit(s.substring(0, idx), new FileReference(
-					new File(s.substring(idx + 1) + "_" + count)));
-		}
-		return fSplits;
-	}
+    private static FileSplit[] parseFileSplits(String fileSplits, int count) {
+        String[] splits = fileSplits.split(",");
+        FileSplit[] fSplits = new FileSplit[splits.length];
+        for (int i = 0; i < splits.length; ++i) {
+            String s = splits[i].trim();
+            int idx = s.indexOf(':');
+            if (idx < 0) {
+                throw new IllegalArgumentException("File split " + s + " not well formed");
+            }
+            fSplits[i] = new FileSplit(s.substring(0, idx), new FileReference(new File(s.substring(idx + 1) + "_"
+                    + count)));
+        }
+        return fSplits;
+    }
 
-	private static JobSpecification createJob(FileSplit[] inSplits,
-			FileSplit[] outSplits, int htSize, int sbSize, int framesLimit,
-			boolean sortOutput, int alg, boolean outPlain) {
-		JobSpecification spec = new JobSpecification();
-		IFileSplitProvider splitsProvider = new ConstantFileSplitProvider(
-				inSplits);
+    private static JobSpecification createJob(FileSplit[] inSplits, FileSplit[] outSplits, int htSize, int sbSize,
+            int framesLimit, boolean sortOutput, int alg, boolean outPlain) {
+        JobSpecification spec = new JobSpecification();
+        IFileSplitProvider splitsProvider = new ConstantFileSplitProvider(inSplits);
 
-		RecordDescriptor inDesc = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						FloatSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE,
-						UTF8StringSerializerDeserializer.INSTANCE });
+        RecordDescriptor inDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                FloatSerializerDeserializer.INSTANCE, FloatSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE,
+                UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE });
 
-		FileScanOperatorDescriptor fileScanner = new FileScanOperatorDescriptor(
-				spec, splitsProvider, new DelimitedDataTupleParserFactory(
-						new IValueParserFactory[] {
-								IntegerParserFactory.INSTANCE,
-								IntegerParserFactory.INSTANCE,
-								IntegerParserFactory.INSTANCE,
-								IntegerParserFactory.INSTANCE,
-								IntegerParserFactory.INSTANCE,
-								FloatParserFactory.INSTANCE,
-								FloatParserFactory.INSTANCE,
-								FloatParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE,
-								UTF8StringParserFactory.INSTANCE, }, '|'),
-				inDesc);
+        FileScanOperatorDescriptor fileScanner = new FileScanOperatorDescriptor(spec, splitsProvider,
+                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE,
+                        IntegerParserFactory.INSTANCE, FloatParserFactory.INSTANCE, FloatParserFactory.INSTANCE,
+                        FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE,
+                        UTF8StringParserFactory.INSTANCE, }, '|'), inDesc);
 
-		fileScanner.setPartitionConstraint(createPartitionConstraint(inSplits));
+        fileScanner.setPartitionConstraint(createPartitionConstraint(inSplits));
 
-		// Output: each unique string with an integer count
-		RecordDescriptor outDesc = new RecordDescriptor(
-				new ISerializerDeserializer[] {
-						IntegerSerializerDeserializer.INSTANCE,
-						// IntegerSerializerDeserializer.INSTANCE,
-						IntegerSerializerDeserializer.INSTANCE });
+        // Output: each unique string with an integer count
+        RecordDescriptor outDesc = new RecordDescriptor(new ISerializerDeserializer[] {
+                IntegerSerializerDeserializer.INSTANCE,
+                // IntegerSerializerDeserializer.INSTANCE,
+                IntegerSerializerDeserializer.INSTANCE });
 
-		// Specify the grouping key, which will be the string extracted during
-		// the scan.
-		int[] keys = new int[] { 0,
-		// 1
-		};
+        // Specify the grouping key, which will be the string extracted during
+        // the scan.
+        int[] keys = new int[] { 0,
+                // 1
+        };
 
-		AbstractOperatorDescriptor grouper;
+        AbstractOperatorDescriptor grouper;
 
-		switch (alg) {
-		case 0: // External hash group
-			grouper = new ExternalHashGroupOperatorDescriptor(
-					spec,
-					keys,
-					framesLimit,
-					false,
-					new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }),
-					new IBinaryComparatorFactory[] {
-					// IntegerBinaryComparatorFactory.INSTANCE,
-					IntegerBinaryComparatorFactory.INSTANCE },
-					new MultiAggregatorFactory(
-							new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-					outDesc, htSize);
+        switch (alg) {
+            case 0: // External hash group
+                grouper = new ExternalHashGroupOperatorDescriptor(spec, keys, framesLimit, false,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }), new IBinaryComparatorFactory[] {
+                    // IntegerBinaryComparatorFactory.INSTANCE,
+                    IntegerBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                            new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }), outDesc,
+                            htSize);
 
-			grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
+                grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
 
-			// Connect scanner with the grouper
-			IConnectorDescriptor scanGroupConn = new MToNHashPartitioningConnectorDescriptor(
-					spec, new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }));
-			spec.connect(scanGroupConn, fileScanner, 0, grouper, 0);
-			break;
-		case 1: // External sort + pre-cluster
-			ExternalSortOperatorDescriptor sorter = new ExternalSortOperatorDescriptor(
-					spec, framesLimit, keys, new IBinaryComparatorFactory[] {
-					// IntegerBinaryComparatorFactory.INSTANCE,
-					IntegerBinaryComparatorFactory.INSTANCE }, inDesc);
-			sorter.setPartitionConstraint(createPartitionConstraint(inSplits));
+                // Connect scanner with the grouper
+                IConnectorDescriptor scanGroupConn = new MToNHashPartitioningConnectorDescriptor(spec,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }));
+                spec.connect(scanGroupConn, fileScanner, 0, grouper, 0);
+                break;
+            case 1: // External sort + pre-cluster
+                ExternalSortOperatorDescriptor sorter = new ExternalSortOperatorDescriptor(spec, framesLimit, keys,
+                        new IBinaryComparatorFactory[] {
+                        // IntegerBinaryComparatorFactory.INSTANCE,
+                        IntegerBinaryComparatorFactory.INSTANCE }, inDesc);
+                sorter.setPartitionConstraint(createPartitionConstraint(inSplits));
 
-			// Connect scan operator with the sorter
-			IConnectorDescriptor scanSortConn = new MToNHashPartitioningConnectorDescriptor(
-					spec, new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }));
-			spec.connect(scanSortConn, fileScanner, 0, sorter, 0);
+                // Connect scan operator with the sorter
+                IConnectorDescriptor scanSortConn = new MToNHashPartitioningConnectorDescriptor(spec,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }));
+                spec.connect(scanSortConn, fileScanner, 0, sorter, 0);
 
-			grouper = new PreclusteredGroupOperatorDescriptor(
-					spec,
-					keys,
-					new IBinaryComparatorFactory[] {
-					// IntegerBinaryComparatorFactory.INSTANCE,
-					IntegerBinaryComparatorFactory.INSTANCE },
-					new MultiAggregatorFactory(
-							new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-					outDesc);
+                grouper = new PreclusteredGroupOperatorDescriptor(spec, keys, new IBinaryComparatorFactory[] {
+                        // IntegerBinaryComparatorFactory.INSTANCE,
+                        IntegerBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                                new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }), outDesc);
 
-			grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
+                grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
 
-			// Connect sorter with the pre-cluster
-			OneToOneConnectorDescriptor sortGroupConn = new OneToOneConnectorDescriptor(
-					spec);
-			spec.connect(sortGroupConn, sorter, 0, grouper, 0);
-			break;
-		case 2: // In-memory hash group
-			grouper = new HashGroupOperatorDescriptor(
-					spec,
-					keys,
-					new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }),
-					new IBinaryComparatorFactory[] {
-					// IntegerBinaryComparatorFactory.INSTANCE,
-					IntegerBinaryComparatorFactory.INSTANCE },
-					new MultiAggregatorFactory(
-							new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-					outDesc, htSize);
+                // Connect sorter with the pre-cluster
+                OneToOneConnectorDescriptor sortGroupConn = new OneToOneConnectorDescriptor(spec);
+                spec.connect(sortGroupConn, sorter, 0, grouper, 0);
+                break;
+            case 2: // In-memory hash group
+                grouper = new HashGroupOperatorDescriptor(spec, keys, new FieldHashPartitionComputerFactory(keys,
+                        new IBinaryHashFunctionFactory[] {
+                        // IntegerBinaryHashFunctionFactory.INSTANCE,
+                        IntegerBinaryHashFunctionFactory.INSTANCE }), new IBinaryComparatorFactory[] {
+                    // IntegerBinaryComparatorFactory.INSTANCE,
+                    IntegerBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                            new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }), outDesc,
+                            htSize);
 
-			grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
+                grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
 
-			// Connect scanner with the grouper
-			IConnectorDescriptor scanConn = new MToNHashPartitioningConnectorDescriptor(
-					spec, new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }));
-			spec.connect(scanConn, fileScanner, 0, grouper, 0);
-			break;
-		default:
-			grouper = new ExternalHashGroupOperatorDescriptor(
-					spec,
-					keys,
-					framesLimit,
-					false,
-					new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }),
-					new IBinaryComparatorFactory[] {
-					// IntegerBinaryComparatorFactory.INSTANCE,
-					IntegerBinaryComparatorFactory.INSTANCE },
-					new MultiAggregatorFactory(
-							new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }),
-					outDesc, htSize);
+                // Connect scanner with the grouper
+                IConnectorDescriptor scanConn = new MToNHashPartitioningConnectorDescriptor(spec,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }));
+                spec.connect(scanConn, fileScanner, 0, grouper, 0);
+                break;
+            default:
+                grouper = new ExternalHashGroupOperatorDescriptor(spec, keys, framesLimit, false,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }), new IBinaryComparatorFactory[] {
+                    // IntegerBinaryComparatorFactory.INSTANCE,
+                    IntegerBinaryComparatorFactory.INSTANCE }, new MultiAggregatorFactory(
+                            new IFieldValueResultingAggregatorFactory[] { new CountAggregatorFactory() }), outDesc,
+                            htSize);
 
-			grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
+                grouper.setPartitionConstraint(createPartitionConstraint(outSplits));
 
-			// Connect scanner with the grouper
-			IConnectorDescriptor scanGroupConnDef = new MToNHashPartitioningConnectorDescriptor(
-					spec, new FieldHashPartitionComputerFactory(keys,
-							new IBinaryHashFunctionFactory[] {
-							// IntegerBinaryHashFunctionFactory.INSTANCE,
-							IntegerBinaryHashFunctionFactory.INSTANCE }));
-			spec.connect(scanGroupConnDef, fileScanner, 0, grouper, 0);
-		}
+                // Connect scanner with the grouper
+                IConnectorDescriptor scanGroupConnDef = new MToNHashPartitioningConnectorDescriptor(spec,
+                        new FieldHashPartitionComputerFactory(keys, new IBinaryHashFunctionFactory[] {
+                                // IntegerBinaryHashFunctionFactory.INSTANCE,
+                                IntegerBinaryHashFunctionFactory.INSTANCE }));
+                spec.connect(scanGroupConnDef, fileScanner, 0, grouper, 0);
+        }
 
-		IFileSplitProvider outSplitProvider = new ConstantFileSplitProvider(
-				outSplits);
+        IFileSplitProvider outSplitProvider = new ConstantFileSplitProvider(outSplits);
 
-		AbstractSingleActivityOperatorDescriptor writer;
+        AbstractSingleActivityOperatorDescriptor writer;
 
-		if (outPlain)
-			writer = new PlainFileWriterOperatorDescriptor(spec,
-					outSplitProvider, "|");
-		else
-			writer = new FrameFileWriterOperatorDescriptor(spec,
-					outSplitProvider);
+        if (outPlain)
+            writer = new PlainFileWriterOperatorDescriptor(spec, outSplitProvider, "|");
+        else
+            writer = new FrameFileWriterOperatorDescriptor(spec, outSplitProvider);
 
-		writer.setPartitionConstraint(createPartitionConstraint(outSplits));
+        writer.setPartitionConstraint(createPartitionConstraint(outSplits));
 
-		IConnectorDescriptor groupOutConn = new OneToOneConnectorDescriptor(
-				spec);
-		spec.connect(groupOutConn, grouper, 0, writer, 0);
+        IConnectorDescriptor groupOutConn = new OneToOneConnectorDescriptor(spec);
+        spec.connect(groupOutConn, grouper, 0, writer, 0);
 
-		spec.addRoot(writer);
-		return spec;
-	}
+        spec.addRoot(writer);
+        return spec;
+    }
 
-	private static PartitionConstraint createPartitionConstraint(
-			FileSplit[] splits) {
-		LocationConstraint[] lConstraints = new LocationConstraint[splits.length];
-		for (int i = 0; i < splits.length; ++i) {
-			lConstraints[i] = new AbsoluteLocationConstraint(
-					splits[i].getNodeName());
-		}
-		return new ExplicitPartitionConstraint(lConstraints);
-	}
+    private static PartitionConstraint createPartitionConstraint(FileSplit[] splits) {
+        LocationConstraint[] lConstraints = new LocationConstraint[splits.length];
+        for (int i = 0; i < splits.length; ++i) {
+            lConstraints[i] = new AbsoluteLocationConstraint(splits[i].getNodeName());
+        }
+        return new ExplicitPartitionConstraint(lConstraints);
+    }
 }