diff --git a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java
index d130d4c..ac2ae5c 100644
--- a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java
+++ b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java
@@ -233,7 +233,7 @@
                     AlgebricksPartitionConstraint pc = partitionConstraintMap.get(parentOp);
                     if (pc != null) {
                         opConstraint = pc;
-                    } else if (opInputs == null || opInputs.size() == 0) {
+                    } else if ((opInputs == null || opInputs.size() == 0) && finalPass) {
                         opConstraint = new AlgebricksCountPartitionConstraint(1);
                     }
                 }
diff --git a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
index 7b04bd5..2a28d2e 100644
--- a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
+++ b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
@@ -35,7 +35,7 @@
 
 public class HeuristicOptimizer {
 
-    public static PhysicalOperatorTag[] hyraxOperators = new PhysicalOperatorTag[] {
+    public static PhysicalOperatorTag[] hyracksOperators = new PhysicalOperatorTag[] {
             PhysicalOperatorTag.DATASOURCE_SCAN, PhysicalOperatorTag.BTREE_SEARCH,
             PhysicalOperatorTag.EXTERNAL_GROUP_BY, PhysicalOperatorTag.HASH_GROUP_BY, PhysicalOperatorTag.HDFS_READER,
             PhysicalOperatorTag.HYBRID_HASH_JOIN, PhysicalOperatorTag.IN_MEMORY_HASH_JOIN,
@@ -44,8 +44,8 @@
             PhysicalOperatorTag.UNION_ALL };
     public static PhysicalOperatorTag[] hyraxOperatorsBelowWhichJobGenIsDisabled = new PhysicalOperatorTag[] {};
 
-    public static boolean isHyraxOp(PhysicalOperatorTag opTag) {
-        for (PhysicalOperatorTag t : hyraxOperators) {
+    public static boolean isHyracksOp(PhysicalOperatorTag opTag) {
+        for (PhysicalOperatorTag t : hyracksOperators) {
             if (t == opTag) {
                 return true;
             }
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
index f13187f..3b31f6d 100644
--- a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
@@ -47,6 +47,7 @@
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.OneToOneExchangePOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.ReplicatePOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.StreamProjectPOperator;
+import org.apache.hyracks.algebricks.core.rewriter.base.HeuristicOptimizer;
 import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 
 public class ExtractCommonOperatorsRule implements IAlgebraicRewriteRule {
@@ -67,8 +68,9 @@
                 && op.getOperatorTag() != LogicalOperatorTag.DISTRIBUTE_RESULT) {
             return false;
         }
-        if (!roots.contains(op))
+        if (!roots.contains(op)) {
             roots.add(new MutableObject<ILogicalOperator>(op));
+        }
         return false;
     }
 
@@ -89,10 +91,12 @@
                 topDownMaterialization(roots);
                 genCandidates(context);
                 removeTrivialShare();
-                if (equivalenceClasses.size() > 0)
+                if (equivalenceClasses.size() > 0) {
                     changed = rewrite(context);
-                if (!rewritten)
+                }
+                if (!rewritten) {
                     rewritten = changed;
+                }
                 equivalenceClasses.clear();
                 childrenToParents.clear();
                 opToCandidateInputs.clear();
@@ -110,22 +114,27 @@
             for (int i = candidates.size() - 1; i >= 0; i--) {
                 Mutable<ILogicalOperator> opRef = candidates.get(i);
                 AbstractLogicalOperator aop = (AbstractLogicalOperator) opRef.getValue();
-                if (aop.getOperatorTag() == LogicalOperatorTag.EXCHANGE)
+                if (aop.getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
                     aop = (AbstractLogicalOperator) aop.getInputs().get(0).getValue();
-                if (aop.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE)
+                }
+                if (aop.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) {
                     candidates.remove(i);
+                }
             }
         }
-        for (int i = equivalenceClasses.size() - 1; i >= 0; i--)
-            if (equivalenceClasses.get(i).size() < 2)
+        for (int i = equivalenceClasses.size() - 1; i >= 0; i--) {
+            if (equivalenceClasses.get(i).size() < 2) {
                 equivalenceClasses.remove(i);
+            }
+        }
     }
 
     private boolean rewrite(IOptimizationContext context) throws AlgebricksException {
         boolean changed = false;
         for (List<Mutable<ILogicalOperator>> members : equivalenceClasses) {
-            if (rewriteForOneEquivalentClass(members, context))
+            if (rewriteForOneEquivalentClass(members, context)) {
                 changed = true;
+            }
         }
         return changed;
     }
@@ -191,11 +200,13 @@
             List<LogicalVariable> liveVarsNew = new ArrayList<LogicalVariable>();
             VariableUtilities.getLiveVariables(candidate.getValue(), liveVarsNew);
             ArrayList<Mutable<ILogicalExpression>> assignExprs = new ArrayList<Mutable<ILogicalExpression>>();
-            for (LogicalVariable liveVar : liveVarsNew)
+            for (LogicalVariable liveVar : liveVarsNew) {
                 assignExprs.add(new MutableObject<ILogicalExpression>(new VariableReferenceExpression(liveVar)));
+            }
             for (Mutable<ILogicalOperator> ref : group) {
-                if (ref.equals(candidate))
+                if (ref.equals(candidate)) {
                     continue;
+                }
                 ArrayList<LogicalVariable> liveVars = new ArrayList<LogicalVariable>();
                 Map<LogicalVariable, LogicalVariable> variableMappingBack = new HashMap<LogicalVariable, LogicalVariable>();
                 IsomorphismUtilities.mapVariablesTopDown(ref.getValue(), candidate.getValue(), variableMappingBack);
@@ -227,20 +238,13 @@
                 for (Mutable<ILogicalOperator> parentOpRef : parentOpList) {
                     AbstractLogicalOperator parentOp = (AbstractLogicalOperator) parentOpRef.getValue();
                     int index = parentOp.getInputs().indexOf(ref);
-                    if (parentOp.getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
-                        AbstractLogicalOperator parentOpNext = (AbstractLogicalOperator) childrenToParents
-                                .get(parentOpRef).get(0).getValue();
-                        if (parentOpNext.isMap()) {
-                            index = parentOpNext.getInputs().indexOf(parentOpRef);
-                            parentOp = parentOpNext;
-                        }
-                    }
-
                     ILogicalOperator childOp = parentOp.getOperatorTag() == LogicalOperatorTag.PROJECT ? assignOperator
                             : projectOperator;
-                    if (parentOp.isMap()) {
+                    if (!HeuristicOptimizer.isHyracksOp(parentOp.getPhysicalOperator().getOperatorTag())) {
                         parentOp.getInputs().set(index, new MutableObject<ILogicalOperator>(childOp));
                     } else {
+                        // If the parent operator is a hyracks operator,
+                        // an extra one-to-one exchange is needed.
                         AbstractLogicalOperator exchg = new ExchangeOperator();
                         exchg.setPhysicalOperator(new OneToOneExchangePOperator());
                         exchg.setExecutionMode(childOp.getExecutionMode());
@@ -270,16 +274,19 @@
                 if (candidates.size() > 0) {
                     for (Mutable<ILogicalOperator> opRef : candidates) {
                         List<Mutable<ILogicalOperator>> refs = childrenToParents.get(opRef);
-                        if (refs != null)
+                        if (refs != null) {
                             currentLevelOpRefs.addAll(refs);
+                        }
                     }
                 }
-                if (currentLevelOpRefs.size() == 0)
+                if (currentLevelOpRefs.size() == 0) {
                     continue;
+                }
                 candidatesGrow(currentLevelOpRefs, candidates);
             }
-            if (currentLevelOpRefs.size() == 0)
+            if (currentLevelOpRefs.size() == 0) {
                 break;
+            }
             prune(context);
         }
         if (equivalenceClasses.size() < 1 && previousEquivalenceClasses.size() > 0) {
@@ -301,8 +308,9 @@
                 }
                 opRefList.add(op);
             }
-            if (op.getValue().getInputs().size() == 0)
+            if (op.getValue().getInputs().size() == 0) {
                 candidates.add(op);
+            }
         }
         if (equivalenceClasses.size() > 0) {
             equivalenceClasses.get(0).addAll(candidates);
@@ -344,10 +352,12 @@
                     }
                 }
             }
-            if (!validCandidate)
+            if (!validCandidate) {
                 continue;
-            if (!candidates.contains(op))
+            }
+            if (!candidates.contains(op)) {
                 candidates.add(op);
+            }
         }
     }
 
@@ -361,8 +371,9 @@
         equivalenceClasses.clear();
         for (List<Mutable<ILogicalOperator>> candidates : previousEquivalenceClasses) {
             boolean[] reserved = new boolean[candidates.size()];
-            for (int i = 0; i < reserved.length; i++)
+            for (int i = 0; i < reserved.length; i++) {
                 reserved[i] = false;
+            }
             for (int i = candidates.size() - 1; i >= 0; i--) {
                 if (reserved[i] == false) {
                     List<Mutable<ILogicalOperator>> equivalentClass = new ArrayList<Mutable<ILogicalOperator>>();