diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/EnumerateJoinsRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/EnumerateJoinsRule.java
index b9d6050..8ea1452 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/EnumerateJoinsRule.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/EnumerateJoinsRule.java
@@ -58,6 +58,8 @@
 import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 import org.apache.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
+import org.apache.hyracks.api.exceptions.IWarningCollector;
+import org.apache.hyracks.api.exceptions.Warning;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -159,6 +161,8 @@
 
         PlanNode cheapestPlanNode = joinEnum.allPlans.get(cheapestPlan);
 
+        generateHintWarnings();
+
         if (numberOfFromTerms > 1) {
             buildNewTree(cheapestPlanNode, joinLeafInputsHashMap, joinOps, new MutableInt(0));
             printPlan(pp, (AbstractLogicalOperator) joinOps.get(0), "New Whole Plan after buildNewTree 1");
@@ -198,6 +202,19 @@
         return false;
     }
 
+    private void generateHintWarnings() {
+        for (Map.Entry<IExpressionAnnotation, Warning> mapElement : joinEnum.joinHints.entrySet()) {
+            IExpressionAnnotation annotation = mapElement.getKey();
+            Warning warning = mapElement.getValue();
+            if (warning != null) {
+                IWarningCollector warningCollector = joinEnum.optCtx.getWarningCollector();
+                if (warningCollector.shouldWarn()) {
+                    warningCollector.warn(warning);
+                }
+            }
+        }
+    }
+
     private boolean getCBOMode(IOptimizationContext context) {
         PhysicalOptimizationConfig physOptConfig = context.getPhysicalOptimizationConfig();
         return physOptConfig.getCBOMode();
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinEnum.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinEnum.java
index 9154bd9..4d9511e 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinEnum.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinEnum.java
@@ -22,6 +22,7 @@
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -54,6 +55,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.BroadcastExpressionAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.HashJoinExpressionAnnotation;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.UnnestingFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
@@ -68,6 +70,7 @@
 import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.algebra.util.OperatorManipulationUtil;
 import org.apache.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
+import org.apache.hyracks.api.exceptions.Warning;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -76,6 +79,7 @@
     private static final Logger LOGGER = LogManager.getLogger();
 
     protected List<JoinCondition> joinConditions; // "global" list of join conditions
+    protected Map<IExpressionAnnotation, Warning> joinHints;
     protected List<PlanNode> allPlans; // list of all plans
     protected JoinNode[] jnArray; // array of all join nodes
     protected int jnArraySize;
@@ -108,6 +112,7 @@
         this.singleDatasetPreds = new ArrayList<>();
         this.joinConditions = new ArrayList<>();
         this.assignOps = new ArrayList<>();
+        this.joinHints = new HashMap<>();
         this.allPlans = new ArrayList<>();
         this.numberOfTerms = numberOfFromTerms;
         this.cboMode = cboMode;
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinNode.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinNode.java
index 43247d4..bddf875 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinNode.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/JoinNode.java
@@ -50,6 +50,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.BroadcastExpressionAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.HashJoinExpressionAnnotation;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.expressions.PredicateCardinalityAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
@@ -494,7 +495,8 @@
             rightExchangeCost = joinEnum.getCostMethodsHandle().computeHJBuildExchangeCost(this);
             childCosts = allPlans.get(leftPlan).totalCost.costAdd(allPlans.get(rightPlan).totalCost);
             totalCost = hjCost.costAdd(leftExchangeCost).costAdd(rightExchangeCost).costAdd(childCosts);
-            if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)) {
+            if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)
+                    || hintHashJoin != null) {
                 pn = new PlanNode(allPlans.size(), joinEnum);
                 pn.jn = this;
                 pn.jnIndexes[0] = leftJn.jnArrayIndex;
@@ -502,9 +504,7 @@
                 pn.planIndexes[0] = leftPlan;
                 pn.planIndexes[1] = rightPlan;
                 pn.joinOp = PlanNode.JoinMethod.HYBRID_HASH_JOIN; // need to check that all the conditions have equality predicates ONLY.
-                if (hintHashJoin != null) {
-                    hintHashJoin.setBuildSide(HashJoinExpressionAnnotation.BuildSide.RIGHT);
-                }
+                pn.joinHint = hintHashJoin;
                 pn.side = HashJoinExpressionAnnotation.BuildSide.RIGHT;
                 pn.joinExpr = hashJoinExpr;
                 pn.opCost = hjCost;
@@ -512,10 +512,8 @@
                 pn.leftExchangeCost = leftExchangeCost;
                 pn.rightExchangeCost = rightExchangeCost;
                 allPlans.add(pn);
-                this.planIndexesArray.add(allPlans.size() - 1);
-                this.cheapestPlanCost = totalCost;
-                this.cheapestPlanIndex = allPlans.size() - 1;
-                return this.cheapestPlanIndex;
+                this.planIndexesArray.add(pn.allPlansIndex);
+                return pn.allPlansIndex;
             }
         }
 
@@ -556,7 +554,8 @@
             rightExchangeCost = joinEnum.getCostMethodsHandle().computeBHJBuildExchangeCost(this);
             childCosts = allPlans.get(leftPlan).totalCost.costAdd(allPlans.get(rightPlan).totalCost);
             totalCost = bcastHjCost.costAdd(rightExchangeCost).costAdd(childCosts);
-            if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)) {
+            if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)
+                    || hintBroadcastHashJoin != null) {
                 pn = new PlanNode(allPlans.size(), joinEnum);
                 pn.jn = this;
                 pn.jnIndexes[0] = leftJn.jnArrayIndex;
@@ -564,9 +563,7 @@
                 pn.planIndexes[0] = leftPlan;
                 pn.planIndexes[1] = rightPlan;
                 pn.joinOp = PlanNode.JoinMethod.BROADCAST_HASH_JOIN; // need to check that all the conditions have equality predicates ONLY.
-                if (hintBroadcastHashJoin != null) {
-                    hintBroadcastHashJoin.setBroadcastSide(BroadcastExpressionAnnotation.BroadcastSide.RIGHT);
-                }
+                pn.joinHint = hintBroadcastHashJoin;
                 pn.side = HashJoinExpressionAnnotation.BuildSide.RIGHT;
                 pn.joinExpr = hashJoinExpr;
                 pn.opCost = bcastHjCost;
@@ -575,18 +572,16 @@
                 pn.rightExchangeCost = rightExchangeCost;
 
                 allPlans.add(pn);
-                this.planIndexesArray.add(allPlans.size() - 1);
-                this.cheapestPlanCost = totalCost;
-                this.cheapestPlanIndex = allPlans.size() - 1;
-                return this.cheapestPlanIndex;
+                this.planIndexesArray.add(pn.allPlansIndex);
+                return pn.allPlansIndex;
             }
         }
 
         return PlanNode.NO_PLAN;
     }
 
-    protected int buildNLJoinPlan(JoinNode leftJn, JoinNode rightJn, ILogicalExpression nestedLoopJoinExpr)
-            throws AlgebricksException {
+    protected int buildNLJoinPlan(JoinNode leftJn, JoinNode rightJn, ILogicalExpression nestedLoopJoinExpr,
+            IndexedNLJoinExpressionAnnotation hintNLJoin) throws AlgebricksException {
         // Build a nested loops plan, first check if it is possible
         // left right order must be preserved and right side should be a single data set
         List<PlanNode> allPlans = joinEnum.allPlans;
@@ -612,7 +607,8 @@
         rightExchangeCost = joinEnum.getCostHandle().zeroCost();
         childCosts = allPlans.get(leftPlan).totalCost;
         totalCost = nljCost.costAdd(leftExchangeCost).costAdd(childCosts);
-        if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)) {
+        if (this.cheapestPlanIndex == PlanNode.NO_PLAN || totalCost.costLT(this.cheapestPlanCost)
+                || hintNLJoin != null) {
             pn = new PlanNode(allPlans.size(), joinEnum);
             pn.jn = this;
             pn.jnIndexes[0] = leftJn.jnArrayIndex;
@@ -620,17 +616,15 @@
             pn.planIndexes[0] = leftPlan;
             pn.planIndexes[1] = rightPlan;
             pn.joinOp = PlanNode.JoinMethod.INDEX_NESTED_LOOP_JOIN;
+            pn.joinHint = hintNLJoin;
             pn.joinExpr = nestedLoopJoinExpr; // save it so can be used to add the NESTED annotation in getNewTree.
             pn.opCost = nljCost;
             pn.totalCost = totalCost;
             pn.leftExchangeCost = leftExchangeCost;
             pn.rightExchangeCost = rightExchangeCost;
-
             allPlans.add(pn);
-            this.planIndexesArray.add(allPlans.size() - 1);
-            this.cheapestPlanCost = totalCost;
-            this.cheapestPlanIndex = allPlans.size() - 1;
-            return allPlans.size() - 1;
+            this.planIndexesArray.add(pn.allPlansIndex);
+            return pn.allPlansIndex;
         }
         return PlanNode.NO_PLAN;
     }
@@ -683,12 +677,9 @@
             pn.totalCost = totalCost;
             pn.leftExchangeCost = leftExchangeCost;
             pn.rightExchangeCost = rightExchangeCost;
-
             allPlans.add(pn);
-            this.planIndexesArray.add(allPlans.size() - 1);
-            this.cheapestPlanCost = totalCost;
-            this.cheapestPlanIndex = allPlans.size() - 1;
-            return allPlans.size() - 1;
+            this.planIndexesArray.add(pn.allPlansIndex);
+            return pn.allPlansIndex;
         }
         return PlanNode.NO_PLAN;
     }
@@ -746,6 +737,7 @@
                 validBuildOrProbeObject = true;
             }
             if (validBuildOrProbeObject) {
+                joinEnum.joinHints.put(hintHashJoin, null);
                 if ((build && (rightJn.datasetNames.contains(buildOrProbeObject)
                         || rightJn.aliases.contains(buildOrProbeObject)))
                         || (probe && (leftJn.datasetNames.contains(buildOrProbeObject)
@@ -757,16 +749,18 @@
                                 || rightJn.aliases.contains(buildOrProbeObject)))) {
                     commutativeHjPlan = buildHashJoinPlan(rightJn, leftJn, hashJoinExpr, hintHashJoin);
                 }
-            } else {
+            }
+            if (hjPlan == PlanNode.NO_PLAN && commutativeHjPlan == PlanNode.NO_PLAN) {
                 // Hints are attached to predicates, so newJoinConditions should not be empty, but adding the check to be safe.
                 if (!joinEnum.getJoinConditions().isEmpty() && !newJoinConditions.isEmpty()) {
                     IWarningCollector warningCollector = joinEnum.optCtx.getWarningCollector();
-                    if (warningCollector.shouldWarn()) {
-                        warningCollector.warn(Warning.of(
-                                joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
-                                        .getSourceLocation(),
-                                ErrorCode.INAPPLICABLE_HINT, "hash join",
-                                (build ? "build " : "probe ") + "with " + buildOrProbeObject));
+                    if (!joinEnum.joinHints.containsKey(hintHashJoin)) {
+                        joinEnum.joinHints.put(hintHashJoin,
+                                Warning.of(
+                                        joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
+                                                .getSourceLocation(),
+                                        ErrorCode.INAPPLICABLE_HINT, "hash join",
+                                        (build ? "build " : "probe ") + "with " + buildOrProbeObject));
                     }
                 }
                 hjPlan = buildHashJoinPlan(leftJn, rightJn, hashJoinExpr, null);
@@ -777,9 +771,9 @@
                 if (!joinEnum.forceJoinOrderMode) {
                     commutativeBcastHjPlan = buildBroadcastHashJoinPlan(rightJn, leftJn, hashJoinExpr, null);
                 }
-                nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr);
+                nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr, null);
                 if (!joinEnum.forceJoinOrderMode) {
-                    commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr);
+                    commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr, null);
                 }
                 cpPlan = buildCPJoinPlan(leftJn, rightJn, hashJoinExpr, nestedLoopJoinExpr);
                 if (!joinEnum.forceJoinOrderMode) {
@@ -795,6 +789,7 @@
                 validBroadcastObject = true;
             }
             if (validBroadcastObject) {
+                joinEnum.joinHints.put(hintBroadcastHashJoin, null);
                 if (rightJn.datasetNames.contains(broadcastObject) || rightJn.aliases.contains(broadcastObject)) {
                     bcastHjPlan = buildBroadcastHashJoinPlan(leftJn, rightJn, hashJoinExpr, hintBroadcastHashJoin);
                 } else if (leftJn.datasetNames.contains(broadcastObject) || leftJn.aliases.contains(broadcastObject)) {
@@ -802,20 +797,24 @@
                             buildBroadcastHashJoinPlan(rightJn, leftJn, hashJoinExpr, hintBroadcastHashJoin);
                 }
             } else if (broadcastObject == null) {
+                joinEnum.joinHints.put(hintBroadcastHashJoin, null);
                 bcastHjPlan = buildBroadcastHashJoinPlan(leftJn, rightJn, hashJoinExpr, hintBroadcastHashJoin);
                 if (!joinEnum.forceJoinOrderMode) {
                     commutativeBcastHjPlan =
                             buildBroadcastHashJoinPlan(rightJn, leftJn, hashJoinExpr, hintBroadcastHashJoin);
                 }
-            } else {
+            }
+            if (bcastHjPlan == PlanNode.NO_PLAN && commutativeBcastHjPlan == PlanNode.NO_PLAN) {
                 // Hints are attached to predicates, so newJoinConditions should not be empty, but adding the check to be safe.
                 if (!joinEnum.getJoinConditions().isEmpty() && !newJoinConditions.isEmpty()) {
                     IWarningCollector warningCollector = joinEnum.optCtx.getWarningCollector();
-                    if (warningCollector.shouldWarn()) {
-                        warningCollector.warn(Warning.of(
-                                joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
-                                        .getSourceLocation(),
-                                ErrorCode.INAPPLICABLE_HINT, "broadcast hash join", "broadcast " + broadcastObject));
+                    if (!joinEnum.joinHints.containsKey(hintBroadcastHashJoin)) {
+                        joinEnum.joinHints.put(hintBroadcastHashJoin,
+                                Warning.of(
+                                        joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
+                                                .getSourceLocation(),
+                                        ErrorCode.INAPPLICABLE_HINT, "broadcast hash join",
+                                        "broadcast " + broadcastObject));
                     }
                 }
 
@@ -827,9 +826,9 @@
                 if (!joinEnum.forceJoinOrderMode) {
                     commutativeBcastHjPlan = buildBroadcastHashJoinPlan(rightJn, leftJn, hashJoinExpr, null);
                 }
-                nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr);
+                nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr, null);
                 if (!joinEnum.forceJoinOrderMode) {
-                    commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr);
+                    commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr, null);
                 }
                 cpPlan = buildCPJoinPlan(leftJn, rightJn, hashJoinExpr, nestedLoopJoinExpr);
                 if (!joinEnum.forceJoinOrderMode) {
@@ -837,19 +836,21 @@
                 }
             }
         } else if (hintNLJoin != null) {
-            nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr);
+            joinEnum.joinHints.put(hintNLJoin, null);
+            nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr, hintNLJoin);
             if (!joinEnum.forceJoinOrderMode) {
-                commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr);
+                commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr, hintNLJoin);
             }
             if (nljPlan == PlanNode.NO_PLAN && commutativeNljPlan == PlanNode.NO_PLAN) {
                 // Hints are attached to predicates, so newJoinConditions should not be empty, but adding the check to be safe.
                 if (!joinEnum.getJoinConditions().isEmpty() && !newJoinConditions.isEmpty()) {
                     IWarningCollector warningCollector = joinEnum.optCtx.getWarningCollector();
-                    if (warningCollector.shouldWarn()) {
-                        warningCollector.warn(Warning.of(
-                                joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
-                                        .getSourceLocation(),
-                                ErrorCode.INAPPLICABLE_HINT, "index nested loop join", "ignored"));
+                    if (!joinEnum.joinHints.containsKey(hintNLJoin)) {
+                        joinEnum.joinHints.put(hintNLJoin,
+                                Warning.of(
+                                        joinEnum.getJoinConditions().get(newJoinConditions.get(0)).joinCondition
+                                                .getSourceLocation(),
+                                        ErrorCode.INAPPLICABLE_HINT, "index nested loop join", "ignored"));
                     }
                 }
                 hjPlan = buildHashJoinPlan(leftJn, rightJn, hashJoinExpr, null);
@@ -874,9 +875,9 @@
             if (!joinEnum.forceJoinOrderMode) {
                 commutativeBcastHjPlan = buildBroadcastHashJoinPlan(rightJn, leftJn, hashJoinExpr, null);
             }
-            nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr);
+            nljPlan = buildNLJoinPlan(leftJn, rightJn, nestedLoopJoinExpr, null);
             if (!joinEnum.forceJoinOrderMode) {
-                commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr);
+                commutativeNljPlan = buildNLJoinPlan(rightJn, leftJn, nestedLoopJoinExpr, null);
             }
             cpPlan = buildCPJoinPlan(leftJn, rightJn, hashJoinExpr, nestedLoopJoinExpr);
             if (!joinEnum.forceJoinOrderMode) {
@@ -895,9 +896,39 @@
         this.leftJn = leftJn;
         this.rightJn = rightJn;
 
+        PlanNode cheapestPlan = findCheapestPlan();
+        this.cheapestPlanCost = cheapestPlan.totalCost;
+        this.cheapestPlanIndex = cheapestPlan.allPlansIndex;
+
         return new Pair<>(this.cheapestPlanIndex, this.cheapestPlanCost);
     }
 
+    private PlanNode findCheapestPlan() {
+        List<PlanNode> allPlans = joinEnum.allPlans;
+        ICost cheapestCost = joinEnum.getCostHandle().maxCost();
+        PlanNode cheapestPlanNode = null;
+        IExpressionAnnotation cheapestPlanJoinHint = null;
+
+        for (int planIndex : this.planIndexesArray) {
+            PlanNode plan = allPlans.get(planIndex);
+            if (plan.joinHint != null && cheapestPlanJoinHint == null) {
+                // The hinted plan wins!
+                cheapestPlanNode = plan;
+                cheapestCost = plan.totalCost;
+                cheapestPlanJoinHint = plan.joinHint;
+            } else if (plan.joinHint != null || cheapestPlanJoinHint == null) {
+                // Either both plans are hinted, or both are non-hinted.
+                // Cost is the decider.
+                if (plan.totalCost.costLT(cheapestCost)) {
+                    cheapestPlanNode = plan;
+                    cheapestCost = plan.totalCost;
+                    cheapestPlanJoinHint = plan.joinHint;
+                }
+            }
+        }
+        return cheapestPlanNode;
+    }
+
     @Override
     public String toString() {
         if (planIndexesArray.isEmpty()) {
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/PlanNode.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/PlanNode.java
index 7e9c3ee..1e91347 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/PlanNode.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/cbo/PlanNode.java
@@ -23,6 +23,7 @@
 import org.apache.hyracks.algebricks.common.utils.Pair;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.HashJoinExpressionAnnotation;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionAnnotation;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.EmptyTupleSourceOperator;
 
@@ -41,6 +42,7 @@
     ICost leftExchangeCost;
     ICost rightExchangeCost;
     JoinMethod joinOp;
+    IExpressionAnnotation joinHint;
     // Used to indicate which side to build for HJ and which side to broadcast for BHJ.
     HashJoinExpressionAnnotation.BuildSide side;
     ScanMethod scanOp;
diff --git a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/joins/nlj_partitioning_property_1.plan b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/joins/nlj_partitioning_property_1.plan
index 91a6aca..5f3c681 100644
--- a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/joins/nlj_partitioning_property_1.plan
+++ b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/joins/nlj_partitioning_property_1.plan
@@ -6,26 +6,24 @@
           -- RANDOM_MERGE_EXCHANGE  |PARTITIONED|
             -- AGGREGATE  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                -- NESTED_LOOP  |PARTITIONED|
+                -- HYBRID_HASH_JOIN [$$76][$$78]  |PARTITIONED|
                   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                    -- STREAM_PROJECT  |PARTITIONED|
+                    -- NESTED_LOOP  |PARTITIONED|
                       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                        -- HYBRID_HASH_JOIN [$$76][$$78]  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                            -- STREAM_PROJECT  |PARTITIONED|
+                            -- DATASOURCE_SCAN (tpch.Supplier)  |PARTITIONED|
                               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                -- DATASOURCE_SCAN (tpch.Supplier)  |PARTITIONED|
-                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                          -- BROADCAST_EXCHANGE  |PARTITIONED|
-                            -- STREAM_PROJECT  |PARTITIONED|
+                                -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                      -- BROADCAST_EXCHANGE  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
+                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                            -- DATASOURCE_SCAN (tpch.Part)  |PARTITIONED|
                               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                -- DATASOURCE_SCAN (tpch.Partsupp)  |PARTITIONED|
-                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
                   -- BROADCAST_EXCHANGE  |PARTITIONED|
                     -- STREAM_PROJECT  |PARTITIONED|
                       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                        -- DATASOURCE_SCAN (tpch.Part)  |PARTITIONED|
+                        -- DATASOURCE_SCAN (tpch.Partsupp)  |PARTITIONED|
                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                             -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping.plan b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping.plan
index f1672b9..c1f2aff 100644
--- a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping.plan
+++ b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping.plan
@@ -17,20 +17,20 @@
                 -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                   -- STREAM_PROJECT  |PARTITIONED|
                     -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                      -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
+                      -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
                         -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                           -- STREAM_PROJECT  |PARTITIONED|
-                            -- ASSIGN  |PARTITIONED|
-                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
-                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                        -- BROADCAST_EXCHANGE  |PARTITIONED|
-                          -- STREAM_PROJECT  |PARTITIONED|
                             -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                              -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
+                              -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
                                 -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                   -- STREAM_PROJECT  |PARTITIONED|
+                                    -- ASSIGN  |PARTITIONED|
+                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                        -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
+                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                            -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                  -- STREAM_PROJECT  |PARTITIONED|
                                     -- STREAM_SELECT  |PARTITIONED|
                                       -- ASSIGN  |PARTITIONED|
                                         -- STREAM_PROJECT  |PARTITIONED|
@@ -38,6 +38,6 @@
                                             -- DATASOURCE_SCAN (tpch.LineItem)  |PARTITIONED|
                                               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                 -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                                -- BROADCAST_EXCHANGE  |PARTITIONED|
-                                  -- UNNEST  |UNPARTITIONED|
-                                    -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                        -- BROADCAST_EXCHANGE  |PARTITIONED|
+                          -- UNNEST  |UNPARTITIONED|
+                            -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
diff --git a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping_ps.plan b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping_ps.plan
index 922ecc4..b926894 100644
--- a/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping_ps.plan
+++ b/asterixdb/asterix-app/src/test/resources/optimizerts/results_cbo/tpch/q12_shipping_ps.plan
@@ -23,20 +23,20 @@
                             -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                               -- STREAM_PROJECT  |PARTITIONED|
                                 -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                  -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
+                                  -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
                                     -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                       -- STREAM_PROJECT  |PARTITIONED|
-                                        -- ASSIGN  |PARTITIONED|
-                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                            -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
-                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                                    -- BROADCAST_EXCHANGE  |PARTITIONED|
-                                      -- STREAM_PROJECT  |PARTITIONED|
                                         -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                          -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
+                                          -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
                                             -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                               -- STREAM_PROJECT  |PARTITIONED|
+                                                -- ASSIGN  |PARTITIONED|
+                                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                    -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
+                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                        -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                              -- STREAM_PROJECT  |PARTITIONED|
                                                 -- STREAM_SELECT  |PARTITIONED|
                                                   -- ASSIGN  |PARTITIONED|
                                                     -- STREAM_PROJECT  |PARTITIONED|
@@ -44,9 +44,9 @@
                                                         -- DATASOURCE_SCAN (tpch.LineItem)  |PARTITIONED|
                                                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                             -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
-                                              -- UNNEST  |UNPARTITIONED|
-                                                -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                    -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                      -- UNNEST  |UNPARTITIONED|
+                                        -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
                 -- BROADCAST_EXCHANGE  |PARTITIONED|
                   -- AGGREGATE  |UNPARTITIONED|
                     -- RANDOM_MERGE_EXCHANGE  |PARTITIONED|
@@ -69,20 +69,20 @@
                                       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                         -- STREAM_PROJECT  |PARTITIONED|
                                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                            -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
+                                            -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
                                               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                 -- STREAM_PROJECT  |PARTITIONED|
-                                                  -- ASSIGN  |PARTITIONED|
-                                                    -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                      -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
-                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                          -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
-                                                -- STREAM_PROJECT  |PARTITIONED|
                                                   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                    -- HYBRID_HASH_JOIN [$$118][$$124]  |PARTITIONED|
+                                                    -- HYBRID_HASH_JOIN [$$122][$$126]  |PARTITIONED|
                                                       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                         -- STREAM_PROJECT  |PARTITIONED|
+                                                          -- ASSIGN  |PARTITIONED|
+                                                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                              -- DATASOURCE_SCAN (tpch.Orders)  |PARTITIONED|
+                                                                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                      -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                        -- STREAM_PROJECT  |PARTITIONED|
                                                           -- STREAM_SELECT  |PARTITIONED|
                                                             -- ASSIGN  |PARTITIONED|
                                                               -- STREAM_PROJECT  |PARTITIONED|
@@ -90,6 +90,6 @@
                                                                   -- DATASOURCE_SCAN (tpch.LineItem)  |PARTITIONED|
                                                                     -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                                       -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
-                                                      -- BROADCAST_EXCHANGE  |PARTITIONED|
-                                                        -- UNNEST  |UNPARTITIONED|
-                                                          -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                -- UNNEST  |UNPARTITIONED|
+                                                  -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|