Mailing-List: contact commits-help@asterixdb.incubator.apache.org;
 run by ezmlm
Precedence: bulk
Reply-To: dev@asterixdb.incubator.apache.org
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: imaxon@apache.org
To: commits@asterixdb.incubator.apache.org
Date: Tue, 25 Aug 2015 16:41:22 -0000
Message-Id: <3c2d0002c0bd4bad9e2093fb15e3487c@git.apache.org>
In-Reply-To: <339f74e1fc5c4aabb18d5e46ecf6ba11@git.apache.org>
References: <339f74e1fc5c4aabb18d5e46ecf6ba11@git.apache.org>
Subject: [09/51] [partial] incubator-asterixdb-hyracks git commit: Change
 folder structure for Java repackage

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/CopyLimitDownRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/CopyLimitDownRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/CopyLimitDownRule.java
new file mode 100644
index 0000000..2f080fb
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/CopyLimitDownRule.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.LimitOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.StreamLimitPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+public class CopyLimitDownRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() != LogicalOperatorTag.LIMIT) {
+            return false;
+        }
+        LimitOperator limitOp = (LimitOperator) op;
+        if (!limitOp.isTopmostLimitOp()) {
+            return false;
+        }
+
+        List<LogicalVariable> limitUsedVars = new ArrayList<>();
+        VariableUtilities.getUsedVariables(limitOp, limitUsedVars);
+
+        Mutable<ILogicalOperator> safeOpRef = null;
+        Mutable<ILogicalOperator> candidateOpRef = limitOp.getInputs().get(0);
+
+        List<LogicalVariable> candidateProducedVars = new ArrayList<>();
+        while (true) {
+            candidateProducedVars.clear();
+            ILogicalOperator candidateOp = candidateOpRef.getValue();
+            LogicalOperatorTag candidateOpTag = candidateOp.getOperatorTag();
+            if (candidateOp.getInputs().size() > 1 || !candidateOp.isMap()
+                    || candidateOpTag == LogicalOperatorTag.SELECT || candidateOpTag == LogicalOperatorTag.LIMIT
+                    || !OperatorPropertiesUtil.disjoint(limitUsedVars, candidateProducedVars)) {
+                break;
+            }
+
+            safeOpRef = candidateOpRef;
+            candidateOpRef = safeOpRef.getValue().getInputs().get(0);
+        }
+
+        if (safeOpRef != null) {
+            ILogicalOperator safeOp = safeOpRef.getValue();
+            Mutable<ILogicalOperator> unsafeOpRef = safeOp.getInputs().get(0);
+            ILogicalOperator unsafeOp = unsafeOpRef.getValue();
+            LimitOperator limitCloneOp = null;
+            if (limitOp.getOffset().getValue() == null) {
+                limitCloneOp = new LimitOperator(limitOp.getMaxObjects().getValue(), false);
+            } else {
+                IFunctionInfo finfoAdd = context.getMetadataProvider().lookupFunction(
+                        AlgebricksBuiltinFunctions.NUMERIC_ADD);
+                List<Mutable<ILogicalExpression>> addArgs = new ArrayList<>();
+                addArgs.add(new MutableObject<ILogicalExpression>(limitOp.getMaxObjects().getValue().cloneExpression()));
+                addArgs.add(new MutableObject<ILogicalExpression>(limitOp.getOffset().getValue().cloneExpression()));
+                ScalarFunctionCallExpression maxPlusOffset = new ScalarFunctionCallExpression(finfoAdd, addArgs);
+                limitCloneOp = new LimitOperator(maxPlusOffset, false);
+            }
+            limitCloneOp.setPhysicalOperator(new StreamLimitPOperator());
+            limitCloneOp.getInputs().add(new MutableObject<ILogicalOperator>(unsafeOp));
+            limitCloneOp.setExecutionMode(unsafeOp.getExecutionMode());
+            limitCloneOp.recomputeSchema();
+            unsafeOpRef.setValue(limitCloneOp);
+            context.computeAndSetTypeEnvironmentForOperator(limitCloneOp);
+            context.addToDontApplySet(this, limitOp);
+        }
+
+        return safeOpRef != null;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateGroupByEmptyKeyRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateGroupByEmptyKeyRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateGroupByEmptyKeyRule.java
new file mode 100644
index 0000000..e93fdd1
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateGroupByEmptyKeyRule.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.List;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * This rule lift out the aggregate operator out from a group-by operator
+ * if the gby operator groups-by on empty key, e.g., the group-by variables are empty.
+ * 
+ * @author yingyib
+ */
+public class EliminateGroupByEmptyKeyRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() != LogicalOperatorTag.GROUP) {
+            return false;
+        }
+        GroupByOperator groupOp = (GroupByOperator) op;
+        List<LogicalVariable> groupVars = groupOp.getGbyVarList();
+        if (groupVars.size() > 0) {
+            return false;
+        }
+        List<ILogicalPlan> nestedPlans = groupOp.getNestedPlans();
+        if (nestedPlans.size() > 1) {
+            return false;
+        }
+        ILogicalPlan nestedPlan = nestedPlans.get(0);
+        if (nestedPlan.getRoots().size() > 1) {
+            return false;
+        }
+        Mutable<ILogicalOperator> topOpRef = nestedPlan.getRoots().get(0);
+        ILogicalOperator topOp = nestedPlan.getRoots().get(0).getValue();
+        Mutable<ILogicalOperator> nestedTupleSourceRef = getNestedTupleSourceReference(topOpRef);
+        /**
+         * connect nested top op into the plan
+         */
+        opRef.setValue(topOp);
+        /**
+         * connect child op into the plan
+         */
+        nestedTupleSourceRef.setValue(groupOp.getInputs().get(0).getValue());
+        return true;
+    }
+
+    private Mutable<ILogicalOperator> getNestedTupleSourceReference(Mutable<ILogicalOperator> nestedTopOperatorRef) {
+        Mutable<ILogicalOperator> currentOpRef = nestedTopOperatorRef;
+        while (currentOpRef.getValue().getInputs() != null && currentOpRef.getValue().getInputs().size() > 0) {
+            currentOpRef = currentOpRef.getValue().getInputs().get(0);
+        }
+        return currentOpRef;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanRule.java
new file mode 100644
index 0000000..8a381f7
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanRule.java
@@ -0,0 +1,127 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.LinkedList;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.LeftOuterJoinOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorManipulationUtil;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+public class EliminateSubplanRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) {
+        return false;
+    }
+
+    /**
+     * Eliminate Subplan above ETS
+     * and Subplan that has only ops. with one input and no free vars. (could we
+     * modify it to consider free vars which are sources of Unnest or Assign, if
+     * there are no aggregates?)
+     */
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+            return false;
+        }
+        SubplanOperator subplan = (SubplanOperator) op;
+
+        Mutable<ILogicalOperator> outerRef = subplan.getInputs().get(0);
+        AbstractLogicalOperator outerRefOp = (AbstractLogicalOperator) outerRef.getValue();
+        if (outerRefOp.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) {
+            elimSubplanOverEts(opRef, context);
+            return true;
+        }
+        if (subplan.getNestedPlans().size() == 1 && subplan.getNestedPlans().get(0).getRoots().size() == 1
+                && !OperatorPropertiesUtil.hasFreeVariables(subplan)) {
+            if (elimOneSubplanWithNoFreeVars(opRef)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    private boolean elimOneSubplanWithNoFreeVars(Mutable<ILogicalOperator> opRef) {
+        SubplanOperator subplan = (SubplanOperator) opRef.getValue();
+        AbstractLogicalOperator rootOp = (AbstractLogicalOperator) subplan.getNestedPlans().get(0).getRoots().get(0)
+                .getValue();
+        if (rootOp.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE
+                || rootOp.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+            opRef.setValue(subplan.getInputs().get(0).getValue());
+            return true;
+        } else {
+            AbstractLogicalOperator botOp = rootOp;
+            if (botOp.getInputs().size() != 1) {
+                return false;
+            }
+            do {
+                Mutable<ILogicalOperator> botRef = botOp.getInputs().get(0);
+                botOp = (AbstractLogicalOperator) botRef.getValue();
+                if (botOp.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) {
+                    botRef.setValue(subplan.getInputs().get(0).getValue());
+                    opRef.setValue(rootOp);
+                    return true;
+                }
+            } while (botOp.getInputs().size() == 1);
+            return false;
+        }
+    }
+
+    private void elimSubplanOverEts(Mutable<ILogicalOperator> opRef, IOptimizationContext ctx)
+            throws AlgebricksException {
+        SubplanOperator subplan = (SubplanOperator) opRef.getValue();
+        for (ILogicalPlan p : subplan.getNestedPlans()) {
+            for (Mutable<ILogicalOperator> r : p.getRoots()) {
+                OperatorManipulationUtil.ntsToEts(r, ctx);
+            }
+        }
+        LinkedList<Mutable<ILogicalOperator>> allRoots = subplan.allRootsInReverseOrder();
+        if (allRoots.size() == 1) {
+            opRef.setValue(allRoots.get(0).getValue());
+        } else {
+            ILogicalOperator topOp = null;
+            for (Mutable<ILogicalOperator> r : allRoots) {
+                if (topOp == null) {
+                    topOp = r.getValue();
+                } else {
+                    LeftOuterJoinOperator j = new LeftOuterJoinOperator(new MutableObject<ILogicalExpression>(
+                            ConstantExpression.TRUE));
+                    j.getInputs().add(new MutableObject<ILogicalOperator>(topOp));
+                    j.getInputs().add(r);
+                    ctx.setOutputTypeEnvironment(j, j.computeOutputTypeEnvironment(ctx));
+                    topOp = j;
+                }
+            }
+            opRef.setValue(topOp);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java
new file mode 100644
index 0000000..1397956
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EliminateSubplanWithInputCardinalityOneRule.java
@@ -0,0 +1,199 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * This rule eliminates a subplan with the following pattern:
+ * -- SUBPLAN
+ * -- OP (where OP produces exactly one tuple)
+ * The live variables at OP will not be used after SUBPLAN.
+ * Note: This rule must be applied after
+ * the RemoveRedundantVariablesRule (to avoid the lineage analysis of variable cardinality).
+ * 
+ * @author yingyib
+ */
+public class EliminateSubplanWithInputCardinalityOneRule implements IAlgebraicRewriteRule {
+    /** The pointer to the topmost operator */
+    private Mutable<ILogicalOperator> rootRef;
+    /** Whether the rule has even been invoked */
+    private boolean invoked = false;
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        if (!invoked) {
+            rootRef = opRef;
+            invoked = true;
+        }
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getInputs().size() <= 0) {
+            return false;
+        }
+        boolean changed = false;
+        for (Mutable<ILogicalOperator> subplanRef : op.getInputs()) {
+            AbstractLogicalOperator op1 = (AbstractLogicalOperator) subplanRef.getValue();
+            if (op1.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+                continue;
+            }
+
+            SubplanOperator subplan = (SubplanOperator) op1;
+            Set<LogicalVariable> usedVarsUp = new ListSet<LogicalVariable>();
+            OperatorPropertiesUtil.getFreeVariablesInPath(rootRef.getValue(), subplan, usedVarsUp);
+            // TODO(buyingyi): figure out the rewriting for subplan operators with multiple subplans.
+            if (subplan.getNestedPlans().size() != 1) {
+                continue;
+            }
+
+            ILogicalOperator subplanInputOperator = subplan.getInputs().get(0).getValue();
+            Set<LogicalVariable> subplanInputVars = new ListSet<LogicalVariable>();
+            VariableUtilities.getLiveVariables(subplanInputOperator, subplanInputVars);
+            int subplanInputVarSize = subplanInputVars.size();
+            subplanInputVars.removeAll(usedVarsUp);
+            // Makes sure the free variables are only used in the subplan.
+            if (subplanInputVars.size() < subplanInputVarSize) {
+                continue;
+            }
+            Set<LogicalVariable> freeVars = new ListSet<LogicalVariable>();
+            OperatorPropertiesUtil.getFreeVariablesInSubplans(subplan, freeVars);
+            boolean cardinalityOne = isCardinalityOne(subplan.getInputs().get(0), freeVars);
+            if (cardinalityOne) {
+                /** If the cardinality of freeVars in the subplan is one, the subplan can be removed. */
+                ILogicalPlan plan = subplan.getNestedPlans().get(0);
+
+                List<Mutable<ILogicalOperator>> rootRefs = plan.getRoots();
+                // TODO(buyingyi): investigate the case of multi-root plans.
+                if (rootRefs.size() != 1) {
+                    continue;
+                }
+                Set<Mutable<ILogicalOperator>> ntsSet = new ListSet<Mutable<ILogicalOperator>>();
+                findNts(rootRefs.get(0), ntsSet);
+
+                /** Replaces nts with the input operator of the subplan. */
+                for (Mutable<ILogicalOperator> nts : ntsSet) {
+                    nts.setValue(subplanInputOperator);
+                }
+                subplanRef.setValue(rootRefs.get(0).getValue());
+                changed = true;
+            } else {
+                continue;
+            }
+        }
+        return changed;
+    }
+
+    /**
+     * Whether the cardinality of the input free variables are one.
+     * 
+     * @param opRef
+     *            the operator to be checked (including its input operators)
+     * @param freeVars
+     *            variables to be checked for produced operators
+     * @return true if every input variable has cardinality one; false otherwise.
+     * @throws AlgebricksException
+     */
+    private boolean isCardinalityOne(Mutable<ILogicalOperator> opRef, Set<LogicalVariable> freeVars)
+            throws AlgebricksException {
+        Set<LogicalVariable> varsWithCardinalityOne = new ListSet<LogicalVariable>();
+        Set<LogicalVariable> varsLiveAtUnnestAndJoin = new ListSet<LogicalVariable>();
+        isCardinalityOne(opRef, freeVars, varsWithCardinalityOne, varsLiveAtUnnestAndJoin);
+        varsWithCardinalityOne.removeAll(varsLiveAtUnnestAndJoin);
+        return varsWithCardinalityOne.equals(freeVars);
+    }
+
+    /**
+     * Recursively adding variables which has cardinality one and in int the input free variable set.
+     * 
+     * @param opRef
+     *            , the current operator reference.
+     * @param freeVars
+     *            , a set of variables.
+     * @param varsWithCardinalityOne
+     *            , variables in the free variable set with cardinality one at the time they are created.
+     * @param varsLiveAtUnnestAndJoin
+     *            , live variables at Unnest and Join. The cardinalities of those variables can become more than one
+     *            even if their cardinalities were one at the time those variables were created.
+     * @throws AlgebricksException
+     */
+    private void isCardinalityOne(Mutable<ILogicalOperator> opRef, Set<LogicalVariable> freeVars,
+            Set<LogicalVariable> varsWithCardinalityOne, Set<LogicalVariable> varsLiveAtUnnestAndJoin)
+            throws AlgebricksException {
+        AbstractLogicalOperator operator = (AbstractLogicalOperator) opRef.getValue();
+        List<LogicalVariable> producedVars = new ArrayList<LogicalVariable>();
+        VariableUtilities.getProducedVariables(operator, producedVars);
+        if (operator.getOperatorTag() == LogicalOperatorTag.UNNEST
+                || operator.getOperatorTag() == LogicalOperatorTag.INNERJOIN
+                || operator.getOperatorTag() == LogicalOperatorTag.LEFTOUTERJOIN) {
+            VariableUtilities.getLiveVariables(operator, varsLiveAtUnnestAndJoin);
+        }
+        if (operator.getOperatorTag() == LogicalOperatorTag.AGGREGATE) {
+            for (LogicalVariable producedVar : producedVars) {
+                if (freeVars.contains(producedVar)) {
+                    varsWithCardinalityOne.add(producedVar);
+                }
+            }
+        }
+        if (varsWithCardinalityOne.size() == freeVars.size()) {
+            return;
+        }
+        for (Mutable<ILogicalOperator> childRef : operator.getInputs()) {
+            isCardinalityOne(childRef, freeVars, varsWithCardinalityOne, varsLiveAtUnnestAndJoin);
+        }
+    }
+
+    /**
+     * Find the NestedTupleSource operator in the direct/undirect input operators of opRef.
+     * 
+     * @param opRef
+     *            , the current operator reference.
+     * @param ntsSet
+     *            , the set NestedTupleSource operator references.
+     */
+    private void findNts(Mutable<ILogicalOperator> opRef, Set<Mutable<ILogicalOperator>> ntsSet) {
+        int childSize = opRef.getValue().getInputs().size();
+        if (childSize == 0) {
+            AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+            if (op.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+                ntsSet.add(opRef);
+            }
+            return;
+        }
+        for (Mutable<ILogicalOperator> childRef : opRef.getValue().getInputs()) {
+            findNts(childRef, ntsSet);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceOrderByAfterSubplan.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceOrderByAfterSubplan.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceOrderByAfterSubplan.java
new file mode 100644
index 0000000..d4834a3
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceOrderByAfterSubplan.java
@@ -0,0 +1,165 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractLogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * If there is any ordering property before the subplan operator, the ordering should
+ * be kept after the subplan.
+ * This rule adds a redundant order operator after those cases, to guarantee the correctness.
+ * 
+ * @author yingyib
+ */
+public class EnforceOrderByAfterSubplan implements IAlgebraicRewriteRule {
+    /** a set of order-breaking operators */
+    private final Set<LogicalOperatorTag> orderBreakingOps = new HashSet<LogicalOperatorTag>();
+    /** a set of order-sensitive operators */
+    private final Set<LogicalOperatorTag> orderSensitiveOps = new HashSet<LogicalOperatorTag>();
+
+    public EnforceOrderByAfterSubplan() {
+        /** add operators that break the ordering */
+        orderBreakingOps.add(LogicalOperatorTag.INNERJOIN);
+        orderBreakingOps.add(LogicalOperatorTag.LEFTOUTERJOIN);
+        orderBreakingOps.add(LogicalOperatorTag.UNIONALL);
+        orderBreakingOps.add(LogicalOperatorTag.AGGREGATE);
+
+        /** add operators that are sensitive to the ordering */
+        orderSensitiveOps.add(LogicalOperatorTag.LIMIT);
+    }
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef.getValue();
+        if (context.checkIfInDontApplySet(this, op1)) {
+            return false;
+        }
+        List<Mutable<ILogicalOperator>> inputs = op1.getInputs();
+        context.addToDontApplySet(this, op1);
+        if (op1.getOperatorTag() == LogicalOperatorTag.ORDER || inputs == null) {
+            /**
+             * does not apply if
+             * 1. there is yet-another order operator on-top-of the subplan, because the downstream order operator's ordering will be broken anyway
+             * 2. the input operator(s) is null
+             */
+            return false;
+        }
+        boolean changed = false;
+        for (int i = 0; i < inputs.size(); i++) {
+            Mutable<ILogicalOperator> inputOpRef = inputs.get(i);
+            AbstractLogicalOperator op = (AbstractLogicalOperator) inputOpRef.getValue();
+            context.addToDontApplySet(this, op);
+            if (op.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+                continue;
+            }
+
+            /**
+             * check the order operators whose ordering is not broken before the subplan operator, and then
+             * duplicate them on-top-of the subplan operator
+             */
+            boolean foundTarget = true;
+            boolean orderSensitive = false;
+            Mutable<ILogicalOperator> childRef = op.getInputs().get(0);
+            AbstractLogicalOperator child = (AbstractLogicalOperator) childRef.getValue();
+            while (child.getOperatorTag() != LogicalOperatorTag.ORDER) {
+                context.addToDontApplySet(this, child);
+                if (orderBreakingOps.contains(child.getOperatorTag())) {
+                    foundTarget = false;
+                    break;
+                }
+                if(child.getOperatorTag() == LogicalOperatorTag.GROUP){
+                    foundTarget = false;
+                    break;
+                }
+                if (orderSensitiveOps.contains(child.getOperatorTag())) {
+                    orderSensitive = true;
+                }
+                List<Mutable<ILogicalOperator>> childInputs = child.getInputs();
+                if (childInputs == null || childInputs.size() > 2 || childInputs.size() < 1) {
+                    foundTarget = false;
+                    break;
+                } else {
+                    childRef = childInputs.get(0);
+                    child = (AbstractLogicalOperator) childRef.getValue();
+                }
+            }
+            /** the target order-by operator has not been found. */
+            if (!foundTarget) {
+                return false;
+            }
+
+            /** copy the original order-by operator and insert on-top-of the subplan operator */
+            context.addToDontApplySet(this, child);
+            OrderOperator sourceOrderOp = (OrderOperator) child;
+            for (Pair<IOrder, Mutable<ILogicalExpression>> expr : sourceOrderOp.getOrderExpressions()) {
+                if (!expr.second.getValue().isFunctional()) {
+                    return false;
+                }
+            }
+            List<Pair<IOrder, Mutable<ILogicalExpression>>> orderExprs = deepCopyOrderAndExpression(sourceOrderOp
+                    .getOrderExpressions());
+            OrderOperator newOrderOp = new OrderOperator(orderExprs);
+            context.addToDontApplySet(this, newOrderOp);
+            inputs.set(i, new MutableObject<ILogicalOperator>(newOrderOp));
+            newOrderOp.getInputs().add(inputOpRef);
+            context.computeAndSetTypeEnvironmentForOperator(newOrderOp);
+
+            if (!orderSensitive) {
+                /** remove the original order-by */
+                childRef.setValue(sourceOrderOp.getInputs().get(0).getValue());
+            }
+            changed = true;
+        }
+        return changed;
+    }
+
+    private Mutable<ILogicalExpression> deepCopyExpressionRef(Mutable<ILogicalExpression> oldExpr) {
+        return new MutableObject<ILogicalExpression>(((AbstractLogicalExpression) oldExpr.getValue()).cloneExpression());
+    }
+
+    private List<Pair<IOrder, Mutable<ILogicalExpression>>> deepCopyOrderAndExpression(
+            List<Pair<IOrder, Mutable<ILogicalExpression>>> ordersAndExprs) {
+        List<Pair<IOrder, Mutable<ILogicalExpression>>> newOrdersAndExprs = new ArrayList<Pair<IOrder, Mutable<ILogicalExpression>>>();
+        for (Pair<IOrder, Mutable<ILogicalExpression>> pair : ordersAndExprs)
+            newOrdersAndExprs.add(new Pair<IOrder, Mutable<ILogicalExpression>>(pair.first,
+                    deepCopyExpressionRef(pair.second)));
+        return newOrdersAndExprs;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/9939b48e/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
new file mode 100644
index 0000000..ebd7da1
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
@@ -0,0 +1,614 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.algebricks.rewriter.rules;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.EquivalenceClass;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IPhysicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.OperatorAnnotations;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractOperatorWithNestedPlans;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.DistinctOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.ExchangeOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.FDsAndEquivClassesVisitor;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.AbstractStableSortPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.BroadcastPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.ExternalGroupByPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.HashPartitionExchangePOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.HashPartitionMergeExchangePOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.InMemoryStableSortPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.PreSortedDistinctByPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.PreclusteredGroupByPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.RandomMergeExchangePOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.RandomPartitionPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.RangePartitionMergePOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.RangePartitionPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.SortMergeExchangePOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical.StableSortPOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
+import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.DefaultNodeGroupDomain;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.FunctionalDependency;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty.PropertyType;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.INodeDomain;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPartitioningProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPartitioningProperty.PartitioningType;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPartitioningRequirementsCoordinator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPhysicalPropertiesVector;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.LocalGroupingProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.LocalOrderProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.OrderColumn;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.OrderedPartitionedProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.PhysicalRequirements;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.PropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.RandomPartitioningProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.StructuralPropertiesVector;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.UnorderedPartitionedProperty;
+import edu.uci.ics.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import edu.uci.ics.hyracks.algebricks.core.config.AlgebricksConfig;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+import edu.uci.ics.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
+import edu.uci.ics.hyracks.algebricks.rewriter.util.PhysicalOptimizationsUtil;
+import edu.uci.ics.hyracks.dataflow.common.data.partition.range.IRangeMap;
+
+public class EnforceStructuralPropertiesRule implements IAlgebraicRewriteRule {
+
+    private static final INodeDomain DEFAULT_DOMAIN = new DefaultNodeGroupDomain("__DEFAULT");
+
+    private PhysicalOptimizationConfig physicalOptimizationConfig;
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        // wait for the physical operators to be set first
+        if (op.getPhysicalOperator() == null) {
+            return false;
+        }
+        if (context.checkIfInDontApplySet(this, op)) {
+            return false;
+        }
+
+        List<FunctionalDependency> fds = context.getFDList(op);
+        if (fds != null && !fds.isEmpty()) {
+            return false;
+        }
+        // These are actually logical constraints, so they could be pre-computed
+        // somewhere else, too.
+
+        physicalOptimizationConfig = context.getPhysicalOptimizationConfig();
+        AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Optimizing operator " + op.getPhysicalOperator() + ".\n");
+
+        PhysicalOptimizationsUtil.computeFDsAndEquivalenceClasses(op, context);
+
+        StructuralPropertiesVector pvector = new StructuralPropertiesVector(new RandomPartitioningProperty(null),
+                new LinkedList<ILocalStructuralProperty>());
+        boolean changed = physOptimizeOp(opRef, pvector, false, context);
+        op.computeDeliveredPhysicalProperties(context);
+        AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Structural properties for " + op.getPhysicalOperator() + ": "
+                + op.getDeliveredPhysicalProperties() + "\n");
+
+        context.addToDontApplySet(this, opRef.getValue());
+
+        return changed;
+    }
+
+    private boolean physOptimizePlan(ILogicalPlan plan, IPhysicalPropertiesVector pvector, boolean nestedPlan,
+            IOptimizationContext context) throws AlgebricksException {
+        boolean changed = false;
+        for (Mutable<ILogicalOperator> root : plan.getRoots()) {
+            if (physOptimizeOp(root, pvector, nestedPlan, context)) {
+                changed = true;
+            }
+            AbstractLogicalOperator op = (AbstractLogicalOperator) root.getValue();
+            op.computeDeliveredPhysicalProperties(context);
+            AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Structural properties for " + op.getPhysicalOperator()
+                    + ": " + op.getDeliveredPhysicalProperties() + "\n");
+        }
+        return changed;
+    }
+
+    private boolean physOptimizeOp(Mutable<ILogicalOperator> opRef, IPhysicalPropertiesVector required,
+            boolean nestedPlan, IOptimizationContext context) throws AlgebricksException {
+
+        boolean changed = false;
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        optimizeUsingConstraintsAndEquivClasses(op);
+        PhysicalRequirements pr = op.getRequiredPhysicalPropertiesForChildren(required);
+        IPhysicalPropertiesVector[] reqdProperties = null;
+        if (pr != null) {
+            reqdProperties = pr.getRequiredProperties();
+        }
+        boolean opIsRedundantSort = false;
+
+        // compute properties and figure out the domain
+        INodeDomain childrenDomain = null;
+        {
+            int j = 0;
+            for (Mutable<ILogicalOperator> childRef : op.getInputs()) {
+                AbstractLogicalOperator child = (AbstractLogicalOperator) childRef.getValue();
+                // recursive call
+                if (physOptimizeOp(childRef, reqdProperties[j], nestedPlan, context)) {
+                    changed = true;
+                }
+                child.computeDeliveredPhysicalProperties(context);
+                IPhysicalPropertiesVector delivered = child.getDeliveredPhysicalProperties();
+                if (childrenDomain == null) {
+                    childrenDomain = delivered.getPartitioningProperty().getNodeDomain();
+                } else {
+                    INodeDomain dom2 = delivered.getPartitioningProperty().getNodeDomain();
+                    if (!childrenDomain.sameAs(dom2)) {
+                        childrenDomain = DEFAULT_DOMAIN;
+                    }
+                }
+                j++;
+            }
+        }
+
+        if (reqdProperties != null) {
+            for (int k = 0; k < reqdProperties.length; k++) {
+                IPhysicalPropertiesVector pv = reqdProperties[k];
+                IPartitioningProperty pp = pv.getPartitioningProperty();
+                if (pp != null && pp.getNodeDomain() == null) {
+                    pp.setNodeDomain(childrenDomain);
+                }
+            }
+        }
+
+        IPartitioningProperty firstDeliveredPartitioning = null;
+        int i = 0;
+        for (Mutable<ILogicalOperator> childRef : op.getInputs()) {
+            AbstractLogicalOperator child = (AbstractLogicalOperator) childRef.getValue();
+            IPhysicalPropertiesVector delivered = child.getDeliveredPhysicalProperties();
+
+            AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Properties delivered by " + child.getPhysicalOperator()
+                    + ": " + delivered + "\n");
+            IPartitioningRequirementsCoordinator prc = pr.getPartitioningCoordinator();
+            Pair<Boolean, IPartitioningProperty> pbpp = prc.coordinateRequirements(
+                    reqdProperties[i].getPartitioningProperty(), firstDeliveredPartitioning, op, context);
+            boolean mayExpandPartitioningProperties = pbpp.first;
+            IPhysicalPropertiesVector rqd = new StructuralPropertiesVector(pbpp.second,
+                    reqdProperties[i].getLocalProperties());
+
+            AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Required properties for " + child.getPhysicalOperator()
+                    + ": " + rqd + "\n");
+            IPhysicalPropertiesVector diff = delivered.getUnsatisfiedPropertiesFrom(rqd,
+                    mayExpandPartitioningProperties, context.getEquivalenceClassMap(child), context.getFDList(child));
+
+            if (isRedundantSort(opRef, delivered, diff, context)) {
+                opIsRedundantSort = true;
+            }
+
+            if (diff != null) {
+                changed = true;
+                addEnforcers(op, i, diff, rqd, delivered, childrenDomain, nestedPlan, context);
+
+                AbstractLogicalOperator newChild = ((AbstractLogicalOperator) op.getInputs().get(i).getValue());
+
+                if (newChild != child) {
+                    delivered = newChild.getDeliveredPhysicalProperties();
+                    IPhysicalPropertiesVector newDiff = newPropertiesDiff(newChild, rqd,
+                            mayExpandPartitioningProperties, context);
+                    AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> New properties diff: " + newDiff + "\n");
+
+                    if (isRedundantSort(opRef, delivered, newDiff, context)) {
+                        opIsRedundantSort = true;
+                        break;
+                    }
+                }
+
+            }
+            if (firstDeliveredPartitioning == null) {
+                IPartitioningProperty dpp = delivered.getPartitioningProperty();
+                if (dpp.getPartitioningType() == PartitioningType.ORDERED_PARTITIONED
+                        || dpp.getPartitioningType() == PartitioningType.UNORDERED_PARTITIONED) {
+                    firstDeliveredPartitioning = dpp;
+                }
+            }
+
+            i++;
+        }
+
+        if (op.hasNestedPlans()) {
+            AbstractOperatorWithNestedPlans nested = (AbstractOperatorWithNestedPlans) op;
+            for (ILogicalPlan p : nested.getNestedPlans()) {
+                if (physOptimizePlan(p, required, true, context)) {
+                    changed = true;
+                }
+            }
+        }
+
+        if (opIsRedundantSort) {
+            if (AlgebricksConfig.DEBUG) {
+                AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Removing redundant SORT operator "
+                        + op.getPhysicalOperator() + "\n");
+                printOp(op);
+            }
+            changed = true;
+            AbstractLogicalOperator nextOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
+            if (nextOp.getOperatorTag() == LogicalOperatorTag.PROJECT) {
+                nextOp = (AbstractLogicalOperator) nextOp.getInputs().get(0).getValue();
+            }
+            opRef.setValue(nextOp);
+            // Now, transfer annotations from the original sort op. to this one.
+            AbstractLogicalOperator transferTo = nextOp;
+            if (transferTo.getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
+                //
+                // remove duplicate exchange operator
+                transferTo = (AbstractLogicalOperator) transferTo.getInputs().get(0).getValue();
+            }
+            transferTo.getAnnotations().putAll(op.getAnnotations());
+            physOptimizeOp(opRef, required, nestedPlan, context);
+        }
+        return changed;
+    }
+
+    private IPhysicalPropertiesVector newPropertiesDiff(AbstractLogicalOperator newChild,
+            IPhysicalPropertiesVector required, boolean mayExpandPartitioningProperties, IOptimizationContext context)
+            throws AlgebricksException {
+        IPhysicalPropertiesVector newDelivered = newChild.getDeliveredPhysicalProperties();
+
+        Map<LogicalVariable, EquivalenceClass> newChildEqClasses = context.getEquivalenceClassMap(newChild);
+        List<FunctionalDependency> newChildFDs = context.getFDList(newChild);
+        if (newChildEqClasses == null || newChildFDs == null) {
+            FDsAndEquivClassesVisitor fdsVisitor = new FDsAndEquivClassesVisitor();
+            newChild.accept(fdsVisitor, context);
+            newChildEqClasses = context.getEquivalenceClassMap(newChild);
+            newChildFDs = context.getFDList(newChild);
+        }
+        AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Required properties for new op. "
+                + newChild.getPhysicalOperator() + ": " + required + "\n");
+
+        return newDelivered.getUnsatisfiedPropertiesFrom(required, mayExpandPartitioningProperties, newChildEqClasses,
+                newChildFDs);
+    }
+
+    private void optimizeUsingConstraintsAndEquivClasses(AbstractLogicalOperator op) {
+        IPhysicalOperator pOp = op.getPhysicalOperator();
+        switch (pOp.getOperatorTag()) {
+            case HASH_GROUP_BY:
+            case EXTERNAL_GROUP_BY: {
+                GroupByOperator gby = (GroupByOperator) op;
+                ExternalGroupByPOperator hgbyOp = (ExternalGroupByPOperator) pOp;
+                hgbyOp.computeColumnSet(gby.getGroupByList());
+                break;
+            }
+            case PRE_CLUSTERED_GROUP_BY: {
+                GroupByOperator gby = (GroupByOperator) op;
+                PreclusteredGroupByPOperator preSortedGby = (PreclusteredGroupByPOperator) pOp;
+                preSortedGby.setGbyColumns(gby.getGbyVarList());
+                break;
+            }
+            case PRE_SORTED_DISTINCT_BY: {
+                DistinctOperator d = (DistinctOperator) op;
+                PreSortedDistinctByPOperator preSortedDistinct = (PreSortedDistinctByPOperator) pOp;
+                preSortedDistinct.setDistinctByColumns(d.getDistinctByVarList());
+                break;
+            }
+        }
+    }
+
+    private List<OrderColumn> getOrderColumnsFromGroupingProperties(List<ILocalStructuralProperty> reqd,
+            List<ILocalStructuralProperty> dlvd) {
+        List<OrderColumn> returnedProperties = new ArrayList<OrderColumn>();
+        List<LogicalVariable> rqdCols = new ArrayList<LogicalVariable>();
+        List<LogicalVariable> dlvdCols = new ArrayList<LogicalVariable>();
+        for (ILocalStructuralProperty r : reqd) {
+            r.getVariables(rqdCols);
+        }
+        for (ILocalStructuralProperty d : dlvd) {
+            d.getVariables(dlvdCols);
+        }
+
+        int prefix = dlvdCols.size() - 1;
+        for (; prefix >= 0;) {
+            if (!rqdCols.contains(dlvdCols.get(prefix))) {
+                prefix--;
+            } else {
+                break;
+            }
+        }
+
+        LocalOrderProperty orderProp = (LocalOrderProperty) dlvd.get(0);
+        List<OrderColumn> orderColumns = orderProp.getOrderColumns();
+        for (int j = 0; j <= prefix; j++) {
+            returnedProperties.add(new OrderColumn(orderColumns.get(j).getColumn(), orderColumns.get(j).getOrder()));
+        }
+        // maintain other order columns after the required order columns
+        if (returnedProperties.size() != 0) {
+            for (int j = prefix + 1; j < dlvdCols.size(); j++) {
+                OrderColumn oc = orderColumns.get(j);
+                returnedProperties.add(new OrderColumn(oc.getColumn(), oc.getOrder()));
+            }
+        }
+        return returnedProperties;
+    }
+
+    /*
+     * We assume delivered to be already normalized.
+     */
+    private boolean isRedundantSort(Mutable<ILogicalOperator> opRef, IPhysicalPropertiesVector delivered,
+            IPhysicalPropertiesVector diffOfProperties, IOptimizationContext context) {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() != LogicalOperatorTag.ORDER
+                || (op.getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.STABLE_SORT && op
+                        .getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.IN_MEMORY_STABLE_SORT)
+                || delivered.getLocalProperties() == null) {
+            return false;
+        }
+        AbstractStableSortPOperator sortOp = (AbstractStableSortPOperator) op.getPhysicalOperator();
+        sortOp.computeLocalProperties(op);
+        ILocalStructuralProperty orderProp = sortOp.getOrderProperty();
+        return PropertiesUtil.matchLocalProperties(Collections.singletonList(orderProp),
+                delivered.getLocalProperties(), context.getEquivalenceClassMap(op), context.getFDList(op));
+    }
+
+    private void addEnforcers(AbstractLogicalOperator op, int childIndex,
+            IPhysicalPropertiesVector diffPropertiesVector, IPhysicalPropertiesVector required,
+            IPhysicalPropertiesVector deliveredByChild, INodeDomain domain, boolean nestedPlan,
+            IOptimizationContext context) throws AlgebricksException {
+
+        IPartitioningProperty pp = diffPropertiesVector.getPartitioningProperty();
+        if (pp == null || pp.getPartitioningType() == PartitioningType.UNPARTITIONED) {
+            addLocalEnforcers(op, childIndex, diffPropertiesVector.getLocalProperties(), nestedPlan, context);
+            IPhysicalPropertiesVector deliveredByNewChild = ((AbstractLogicalOperator) op.getInputs().get(0).getValue())
+                    .getDeliveredPhysicalProperties();
+            addPartitioningEnforcers(op, childIndex, pp, required, deliveredByNewChild, domain, context);
+        } else {
+            addPartitioningEnforcers(op, childIndex, pp, required, deliveredByChild, domain, context);
+            AbstractLogicalOperator newChild = (AbstractLogicalOperator) op.getInputs().get(childIndex).getValue();
+            IPhysicalPropertiesVector newDiff = newPropertiesDiff(newChild, required, true, context);
+            AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> New properties diff: " + newDiff + "\n");
+            if (newDiff != null) {
+                addLocalEnforcers(op, childIndex, newDiff.getLocalProperties(), nestedPlan, context);
+            }
+        }
+    }
+
+    private void addLocalEnforcers(AbstractLogicalOperator op, int i, List<ILocalStructuralProperty> localProperties,
+            boolean nestedPlan, IOptimizationContext context) throws AlgebricksException {
+        if (AlgebricksConfig.DEBUG) {
+            AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Adding local enforcers for local props = " + localProperties
+                    + "\n");
+        }
+
+        if (localProperties == null || localProperties.isEmpty()) {
+            return;
+        }
+
+        Mutable<ILogicalOperator> topOp = new MutableObject<ILogicalOperator>();
+        topOp.setValue(op.getInputs().get(i).getValue());
+        LinkedList<LocalOrderProperty> oList = new LinkedList<LocalOrderProperty>();
+
+        for (ILocalStructuralProperty prop : localProperties) {
+            switch (prop.getPropertyType()) {
+                case LOCAL_ORDER_PROPERTY: {
+                    oList.add((LocalOrderProperty) prop);
+                    break;
+                }
+                case LOCAL_GROUPING_PROPERTY: {
+                    LocalGroupingProperty g = (LocalGroupingProperty) prop;
+                    Collection<LogicalVariable> vars = (g.getPreferredOrderEnforcer() != null) ? g
+                            .getPreferredOrderEnforcer() : g.getColumnSet();
+                    List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
+                    for (LogicalVariable v : vars) {
+                        OrderColumn oc = new OrderColumn(v, OrderKind.ASC);
+                        orderColumns.add(oc);
+                    }
+                    LocalOrderProperty lop = new LocalOrderProperty(orderColumns);
+                    oList.add(lop);
+                    break;
+                }
+                default: {
+                    throw new IllegalStateException();
+                }
+            }
+        }
+        if (!oList.isEmpty()) {
+            topOp = enforceOrderProperties(oList, topOp, nestedPlan, context);
+        }
+
+        op.getInputs().set(i, topOp);
+        OperatorPropertiesUtil.computeSchemaAndPropertiesRecIfNull((AbstractLogicalOperator) topOp.getValue(), context);
+        printOp((AbstractLogicalOperator) topOp.getValue());
+    }
+
+    private Mutable<ILogicalOperator> enforceOrderProperties(List<LocalOrderProperty> oList,
+            Mutable<ILogicalOperator> topOp, boolean isMicroOp, IOptimizationContext context)
+            throws AlgebricksException {
+        List<Pair<IOrder, Mutable<ILogicalExpression>>> oe = new LinkedList<Pair<IOrder, Mutable<ILogicalExpression>>>();
+        for (LocalOrderProperty orderProperty : oList) {
+            for (OrderColumn oc : orderProperty.getOrderColumns()) {
+                IOrder ordType = (oc.getOrder() == OrderKind.ASC) ? OrderOperator.ASC_ORDER : OrderOperator.DESC_ORDER;
+                Pair<IOrder, Mutable<ILogicalExpression>> pair = new Pair<IOrder, Mutable<ILogicalExpression>>(ordType,
+                        new MutableObject<ILogicalExpression>(new VariableReferenceExpression(oc.getColumn())));
+                oe.add(pair);
+            }
+        }
+        OrderOperator oo = new OrderOperator(oe);
+        oo.setExecutionMode(AbstractLogicalOperator.ExecutionMode.LOCAL);
+        if (isMicroOp) {
+            oo.setPhysicalOperator(new InMemoryStableSortPOperator());
+        } else {
+            oo.setPhysicalOperator(new StableSortPOperator(physicalOptimizationConfig.getMaxFramesExternalSort()));
+        }
+        oo.getInputs().add(topOp);
+        context.computeAndSetTypeEnvironmentForOperator(oo);
+        if (AlgebricksConfig.DEBUG) {
+            AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Added sort enforcer " + oo.getPhysicalOperator() + ".\n");
+        }
+        return new MutableObject<ILogicalOperator>(oo);
+    }
+
+    private void addPartitioningEnforcers(ILogicalOperator op, int i, IPartitioningProperty pp,
+            IPhysicalPropertiesVector required, IPhysicalPropertiesVector deliveredByChild, INodeDomain domain,
+            IOptimizationContext context) throws AlgebricksException {
+        if (pp != null) {
+            IPhysicalOperator pop;
+            switch (pp.getPartitioningType()) {
+                case UNPARTITIONED: {
+                    List<OrderColumn> ordCols = computeOrderColumns(deliveredByChild);
+                    if (ordCols == null || ordCols.size() == 0) {
+                        pop = new RandomMergeExchangePOperator();
+                    } else {
+                        if (op.getAnnotations().containsKey(OperatorAnnotations.USE_RANGE_CONNECTOR)) {
+                            IRangeMap rangeMap = (IRangeMap) op.getAnnotations().get(
+                                    OperatorAnnotations.USE_RANGE_CONNECTOR);
+                            pop = new RangePartitionMergePOperator(ordCols, domain, rangeMap);
+                        } else {
+                            OrderColumn[] sortColumns = new OrderColumn[ordCols.size()];
+                            sortColumns = ordCols.toArray(sortColumns);
+                            pop = new SortMergeExchangePOperator(sortColumns);
+                        }
+                    }
+                    break;
+                }
+                case UNORDERED_PARTITIONED: {
+                    List<LogicalVariable> varList = new ArrayList<LogicalVariable>(
+                            ((UnorderedPartitionedProperty) pp).getColumnSet());
+                    List<ILocalStructuralProperty> cldLocals = deliveredByChild.getLocalProperties();
+                    List<ILocalStructuralProperty> reqdLocals = required.getLocalProperties();
+                    boolean propWasSet = false;
+                    pop = null;
+                    if (reqdLocals != null && cldLocals != null && allAreOrderProps(cldLocals)) {
+                        AbstractLogicalOperator c = (AbstractLogicalOperator) op.getInputs().get(i).getValue();
+                        Map<LogicalVariable, EquivalenceClass> ecs = context.getEquivalenceClassMap(c);
+                        List<FunctionalDependency> fds = context.getFDList(c);
+                        if (PropertiesUtil.matchLocalProperties(reqdLocals, cldLocals, ecs, fds)) {
+                            List<OrderColumn> orderColumns = getOrderColumnsFromGroupingProperties(reqdLocals,
+                                    cldLocals);
+                            pop = new HashPartitionMergeExchangePOperator(orderColumns, varList, domain);
+                            propWasSet = true;
+                        }
+                    }
+                    if (!propWasSet) {
+                        pop = new HashPartitionExchangePOperator(varList, domain);
+                    }
+                    break;
+                }
+                case ORDERED_PARTITIONED: {
+                    pop = new RangePartitionPOperator(((OrderedPartitionedProperty) pp).getOrderColumns(), domain, null);
+                    break;
+                }
+                case BROADCAST: {
+                    pop = new BroadcastPOperator(domain);
+                    break;
+                }
+                case RANDOM: {
+                    RandomPartitioningProperty rpp = (RandomPartitioningProperty) pp;
+                    INodeDomain nd = rpp.getNodeDomain();
+                    pop = new RandomPartitionPOperator(nd);
+                    break;
+                }
+                default: {
+                    throw new NotImplementedException("Enforcer for " + pp.getPartitioningType()
+                            + " partitioning type has not been implemented.");
+                }
+            }
+            Mutable<ILogicalOperator> ci = op.getInputs().get(i);
+            ExchangeOperator exchg = new ExchangeOperator();
+            exchg.setPhysicalOperator(pop);
+            setNewOp(ci, exchg, context);
+            exchg.setExecutionMode(AbstractLogicalOperator.ExecutionMode.PARTITIONED);
+            OperatorPropertiesUtil.computeSchemaAndPropertiesRecIfNull(exchg, context);
+            context.computeAndSetTypeEnvironmentForOperator(exchg);
+            if (AlgebricksConfig.DEBUG) {
+                AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Added partitioning enforcer "
+                        + exchg.getPhysicalOperator() + ".\n");
+                printOp((AbstractLogicalOperator) op);
+            }
+        }
+    }
+
+    private boolean allAreOrderProps(List<ILocalStructuralProperty> cldLocals) {
+        for (ILocalStructuralProperty lsp : cldLocals) {
+            if (lsp.getPropertyType() != PropertyType.LOCAL_ORDER_PROPERTY) {
+                return false;
+            }
+        }
+        return !cldLocals.isEmpty();
+    }
+
+    private void printOp(AbstractLogicalOperator op) throws AlgebricksException {
+        StringBuilder sb = new StringBuilder();
+        LogicalOperatorPrettyPrintVisitor pvisitor = new LogicalOperatorPrettyPrintVisitor();
+        PlanPrettyPrinter.printOperator(op, sb, pvisitor, 0);
+        AlgebricksConfig.ALGEBRICKS_LOGGER.fine(sb.toString());
+    }
+
+    private List<OrderColumn> computeOrderColumns(IPhysicalPropertiesVector pv) {
+        List<OrderColumn> ordCols = new ArrayList<OrderColumn>();
+        List<ILocalStructuralProperty> localProps = pv.getLocalProperties();
+        if (localProps == null || localProps.size() == 0) {
+            return null;
+        } else {
+            for (ILocalStructuralProperty p : localProps) {
+                if (p.getPropertyType() == PropertyType.LOCAL_ORDER_PROPERTY) {
+                    LocalOrderProperty lop = (LocalOrderProperty) p;
+                    ordCols.addAll(lop.getOrderColumns());
+                } else {
+                    return null;
+                }
+            }
+            return ordCols;
+        }
+
+    }
+
+    private void setNewOp(Mutable<ILogicalOperator> opRef, AbstractLogicalOperator newOp, IOptimizationContext context)
+            throws AlgebricksException {
+        ILogicalOperator oldOp = opRef.getValue();
+        opRef.setValue(newOp);
+        newOp.getInputs().add(new MutableObject<ILogicalOperator>(oldOp));
+        newOp.recomputeSchema();
+        newOp.computeDeliveredPhysicalProperties(context);
+        context.computeAndSetTypeEnvironmentForOperator(newOp);
+        AlgebricksConfig.ALGEBRICKS_LOGGER.finest(">>>> Structural properties for " + newOp.getPhysicalOperator()
+                + ": " + newOp.getDeliveredPhysicalProperties() + "\n");
+
+        PhysicalOptimizationsUtil.computeFDsAndEquivalenceClasses(newOp, context);
+    }
+
+}