diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java
index 74adf5b..41711cb 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -104,12 +104,17 @@
                     }
                     case LOCAL_ORDER_PROPERTY: {
                         LocalOrderProperty lop = (LocalOrderProperty) lsp;
-                        OrderColumn oc = lop.getOrderColumn();
-                        LogicalVariable v2 = getLhsGbyVar(gby, oc.getColumn());
-                        if (v2 != null) {
-                            propsLocal.add(new LocalOrderProperty(new OrderColumn(v2, oc.getOrder())));
-                        } else {
-                            failed = true;
+                        List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
+                        for (OrderColumn oc : lop.getOrderColumns()) {
+                            LogicalVariable v2 = getLhsGbyVar(gby, oc.getColumn());
+                            if (v2 != null) {
+                                orderColumns.add(new OrderColumn(v2, oc.getOrder()));
+                            } else {
+                                failed = true;
+                            }
+                        }
+                        if (!failed) {
+                            propsLocal.add(new LocalOrderProperty(orderColumns));
                         }
                         break;
                     }
@@ -173,22 +178,26 @@
                         break;
                     }
                     LocalOrderProperty lop = (LocalOrderProperty) prop;
-                    LogicalVariable ord = lop.getColumn();
-                    Pair<LogicalVariable, Mutable<ILogicalExpression>> p = getGbyPairByRhsVar(gby, ord);
-                    if (p == null) {
-                        p = getDecorPairByRhsVar(gby, ord);
+                    List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
+                    List<OrderColumn> ords = lop.getOrderColumns();
+                    for (OrderColumn ord : ords) {
+                        Pair<LogicalVariable, Mutable<ILogicalExpression>> p = getGbyPairByRhsVar(gby, ord.getColumn());
                         if (p == null) {
-                            allOk = false;
-                            break;
+                            p = getDecorPairByRhsVar(gby, ord.getColumn());
+                            if (p == null) {
+                                allOk = false;
+                                break;
+                            }
                         }
+                        ILogicalExpression e = p.second.getValue();
+                        if (e.getExpressionTag() != LogicalExpressionTag.VARIABLE) {
+                            throw new IllegalStateException(
+                                    "Right hand side of group-by assignment should have been normalized to a variable reference.");
+                        }
+                        LogicalVariable v = ((VariableReferenceExpression) e).getVariableReference();
+                        orderColumns.add(new OrderColumn(v, ord.getOrder()));
                     }
-                    ILogicalExpression e = p.second.getValue();
-                    if (e.getExpressionTag() != LogicalExpressionTag.VARIABLE) {
-                        throw new IllegalStateException(
-                                "Right hand side of group-by assignment should have been normalized to a variable reference.");
-                    }
-                    LogicalVariable v = ((VariableReferenceExpression) e).getVariableReference();
-                    props.add(new LocalOrderProperty(new OrderColumn(v, lop.getOrder())));
+                    props.add(new LocalOrderProperty(orderColumns));
                 }
                 List<FunctionalDependency> fdList = new ArrayList<FunctionalDependency>();
                 for (Pair<LogicalVariable, Mutable<ILogicalExpression>> decorPair : gby.getDecorList()) {
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPropagatePropertiesForUsedVariablesPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPropagatePropertiesForUsedVariablesPOperator.java
index 1ce563c..c77222b 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPropagatePropertiesForUsedVariablesPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractPropagatePropertiesForUsedVariablesPOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -34,11 +34,11 @@
         for (ILocalStructuralProperty lsp : downPropsLocal) {
             LinkedList<LogicalVariable> cols = new LinkedList<LogicalVariable>();
             lsp.getColumns(cols);
-            if (usedVariables.containsAll(cols)) {
-                propsLocal.add(lsp);
+            ILocalStructuralProperty propagatedProp = lsp.retainVariables(usedVariables);
+            if (propagatedProp != null) {
+                propsLocal.add(propagatedProp);
             }
         }
         deliveredProperties = new StructuralPropertiesVector(pp, propsLocal);
     }
-
 }
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractStableSortPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractStableSortPOperator.java
index 8cfe067..4b85c56 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractStableSortPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/AbstractStableSortPOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -14,12 +14,12 @@
  */
 package edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 import org.apache.commons.lang3.mutable.Mutable;
 
-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.ILogicalExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
@@ -30,7 +30,6 @@
 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.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPartitioningProperty;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPartitioningRequirementsCoordinator;
@@ -43,7 +42,7 @@
 public abstract class AbstractStableSortPOperator extends AbstractPhysicalOperator {
 
     protected OrderColumn[] sortColumns;
-    protected List<ILocalStructuralProperty> orderProps;
+    protected ILocalStructuralProperty orderProp;
 
     public AbstractStableSortPOperator() {
     }
@@ -60,7 +59,8 @@
         // }
         AbstractLogicalOperator op2 = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
         StructuralPropertiesVector childProp = (StructuralPropertiesVector) op2.getDeliveredPhysicalProperties();
-        deliveredProperties = new StructuralPropertiesVector(childProp.getPartitioningProperty(), orderProps);
+        deliveredProperties = new StructuralPropertiesVector(childProp.getPartitioningProperty(),
+                Collections.singletonList(orderProp));
     }
 
     @Override
@@ -68,11 +68,11 @@
             IPhysicalPropertiesVector reqdByParent) {
         AbstractLogicalOperator op = (AbstractLogicalOperator) iop;
         if (op.getExecutionMode() == AbstractLogicalOperator.ExecutionMode.PARTITIONED) {
-            if (orderProps == null) {
+            if (orderProp == null) {
                 computeLocalProperties(op);
             }
             StructuralPropertiesVector[] r = new StructuralPropertiesVector[] { new StructuralPropertiesVector(
-                    IPartitioningProperty.UNPARTITIONED, orderProps) };
+                    IPartitioningProperty.UNPARTITIONED, Collections.singletonList(orderProp)) };
             return new PhysicalRequirements(r, IPartitioningRequirementsCoordinator.NO_COORDINATION);
         } else {
             return emptyUnaryRequirements();
@@ -80,50 +80,32 @@
     }
 
     public void computeLocalProperties(ILogicalOperator op) {
-        orderProps = new LinkedList<ILocalStructuralProperty>();
-
         OrderOperator ord = (OrderOperator) op;
+        List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
         for (Pair<IOrder, Mutable<ILogicalExpression>> p : ord.getOrderExpressions()) {
             ILogicalExpression expr = p.second.getValue();
             if (expr.getExpressionTag() == LogicalExpressionTag.VARIABLE) {
                 VariableReferenceExpression varRef = (VariableReferenceExpression) expr;
                 LogicalVariable var = varRef.getVariableReference();
-                switch (p.first.getKind()) {
-                    case ASC: {
-                        orderProps.add(new LocalOrderProperty(new OrderColumn(var, OrderKind.ASC)));
-                        break;
-                    }
-                    case DESC: {
-                        orderProps.add(new LocalOrderProperty(new OrderColumn(var, OrderKind.DESC)));
-                        break;
-                    }
-                    default: {
-                        throw new NotImplementedException();
-                    }
-                }
+                orderColumns.add(new OrderColumn(var, p.first.getKind()));
             } else {
                 throw new IllegalStateException();
             }
         }
-
-        int n = orderProps.size();
-        sortColumns = new OrderColumn[n];
-        int i = 0;
-        for (ILocalStructuralProperty prop : orderProps) {
-            sortColumns[i++] = ((LocalOrderProperty) prop).getOrderColumn();
-        }
+        sortColumns = orderColumns.toArray(new OrderColumn[orderColumns.size()]);
+        orderProp = new LocalOrderProperty(orderColumns);
     }
 
-    public List<ILocalStructuralProperty> getOrderProperties() {
-        return orderProps;
+    public ILocalStructuralProperty getOrderProperty() {
+        return orderProp;
     }
 
     @Override
     public String toString() {
-        if (orderProps == null) {
+        if (orderProp == null) {
             return getOperatorTag().toString();
         } else {
-            return getOperatorTag().toString() + " " + orderProps;
+            return getOperatorTag().toString() + " " + orderProp;
         }
     }
 
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/HashPartitionMergeExchangePOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/HashPartitionMergeExchangePOperator.java
index 3f1aae2..fc4f116 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/HashPartitionMergeExchangePOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/HashPartitionMergeExchangePOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -56,9 +56,9 @@
 
 public class HashPartitionMergeExchangePOperator extends AbstractExchangePOperator {
 
-    private List<OrderColumn> orderColumns;
-    private List<LogicalVariable> partitionFields;
-    private INodeDomain domain;
+    private final List<OrderColumn> orderColumns;
+    private final List<LogicalVariable> partitionFields;
+    private final INodeDomain domain;
 
     public HashPartitionMergeExchangePOperator(List<OrderColumn> orderColumns, List<LogicalVariable> partitionFields,
             INodeDomain domain) {
@@ -98,22 +98,12 @@
     public PhysicalRequirements getRequiredPropertiesForChildren(ILogicalOperator op,
             IPhysicalPropertiesVector reqdByParent) {
         List<ILocalStructuralProperty> orderProps = new LinkedList<ILocalStructuralProperty>();
+        List<OrderColumn> columns = new ArrayList<OrderColumn>();
         for (OrderColumn oc : orderColumns) {
             LogicalVariable var = oc.getColumn();
-            switch (oc.getOrder()) {
-                case ASC: {
-                    orderProps.add(new LocalOrderProperty(new OrderColumn(var, OrderKind.ASC)));
-                    break;
-                }
-                case DESC: {
-                    orderProps.add(new LocalOrderProperty(new OrderColumn(var, OrderKind.DESC)));
-                    break;
-                }
-                default: {
-                    throw new IllegalStateException();
-                }
-            }
+            columns.add(new OrderColumn(var, oc.getOrder()));
         }
+        orderProps.add(new LocalOrderProperty(columns));
         StructuralPropertiesVector[] r = new StructuralPropertiesVector[] { new StructuralPropertiesVector(null,
                 orderProps) };
         return new PhysicalRequirements(r, IPartitioningRequirementsCoordinator.NO_COORDINATION);
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
index e0d8d79..b6b3ef6 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
@@ -67,16 +67,16 @@
         IPhysicalPropertiesVector physicalProps = dataSourceIndex.getDataSource().getPropertiesProvider()
                 .computePropertiesVector(scanVariables);
         List<ILocalStructuralProperty> localProperties = new ArrayList<>();
+        List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
         // Data needs to be sorted based on the [token, number of token, PK]
         // OR [token, PK] if the index is not partitioned
         for (LogicalVariable skVar : secondaryKeys) {
-            localProperties.add(new LocalOrderProperty(new OrderColumn(skVar,
-                    OrderKind.ASC)));
+            orderColumns.add(new OrderColumn(skVar, OrderKind.ASC));
         }
         for (LogicalVariable pkVar : primaryKeys) {
-            localProperties.add(new LocalOrderProperty(new OrderColumn(pkVar,
-                    OrderKind.ASC)));
+            orderColumns.add(new OrderColumn(pkVar, OrderKind.ASC));
         }
+        localProperties.add(new LocalOrderProperty(orderColumns));
         StructuralPropertiesVector spv = new StructuralPropertiesVector(physicalProps.getPartitioningProperty(),
                 localProperties);
         return new PhysicalRequirements(new IPhysicalPropertiesVector[] { spv },
@@ -87,7 +87,7 @@
     public void computeDeliveredProperties(ILogicalOperator op, IOptimizationContext context)
             throws AlgebricksException {
         AbstractLogicalOperator op2 = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
-        deliveredProperties = (StructuralPropertiesVector) op2.getDeliveredPhysicalProperties().clone();
+        deliveredProperties = op2.getDeliveredPhysicalProperties().clone();
     }
 
     @Override
@@ -122,5 +122,4 @@
         return false;
     }
 
-
 }
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
index f68b327..4fdfa32 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
@@ -14,16 +14,22 @@
  */
 package edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical;
 
+import java.util.ArrayList;
+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.IHyracksJobBuilder;
 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.PhysicalOperatorTag;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.IPhysicalPropertiesVector;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.PhysicalRequirements;
 import edu.uci.ics.hyracks.algebricks.core.algebra.properties.StructuralPropertiesVector;
@@ -52,7 +58,29 @@
         Mutable<ILogicalOperator> dataSource = ((NestedTupleSourceOperator) op).getDataSourceReference();
         AbstractLogicalOperator op2 = (AbstractLogicalOperator) dataSource.getValue().getInputs().get(0).getValue();
         IPhysicalPropertiesVector inheritedProps = op2.getDeliveredPhysicalProperties();
-        deliveredProperties = (StructuralPropertiesVector) inheritedProps.clone();
+        AbstractLogicalOperator parent = (AbstractLogicalOperator) dataSource.getValue();
+        if (parent.getOperatorTag() == LogicalOperatorTag.GROUP) {
+            // The following part computes the data property regarding to each particular group.
+            // TODO(buyingyi): we need to add the original data property as well. But currently
+            // there are places assuming there is only one LocalOrderProperty and one
+            // LocalGroupingProperty delivered by an operator.
+            GroupByOperator gby = (GroupByOperator) parent;
+            List<ILocalStructuralProperty> originalLocalProperties = inheritedProps.getLocalProperties();
+            List<ILocalStructuralProperty> newLocalProperties = null;
+            if (originalLocalProperties != null) {
+                newLocalProperties = new ArrayList<ILocalStructuralProperty>();
+                for (ILocalStructuralProperty lsp : inheritedProps.getLocalProperties()) {
+                    ILocalStructuralProperty newLsp = lsp.regardToGroup(gby.getGbyVarList());
+                    if (newLsp != null) {
+                        newLocalProperties.add(newLsp);
+                    }
+                }
+            }
+            deliveredProperties = new StructuralPropertiesVector(inheritedProps.getPartitioningProperty(),
+                    newLocalProperties);
+        } else {
+            deliveredProperties = inheritedProps.clone();
+        }
     }
 
     @Override
@@ -67,7 +95,8 @@
             throws AlgebricksException {
         propagatedSchema.addAllVariables(outerPlanSchema);
         NestedTupleSourceRuntimeFactory runtime = new NestedTupleSourceRuntimeFactory();
-        RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema, context);
+        RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema,
+                context);
         builder.contributeMicroOperator(op, runtime, recDesc);
     }
 
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
index 3f78793..a24c48d 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -76,10 +76,12 @@
     public PhysicalRequirements getRequiredPropertiesForChildren(ILogicalOperator op,
             IPhysicalPropertiesVector reqdByParent) {
         StructuralPropertiesVector[] pv = new StructuralPropertiesVector[1];
-        List<ILocalStructuralProperty> localProps = new ArrayList<ILocalStructuralProperty>(columnList.size());
+        List<ILocalStructuralProperty> localProps = new ArrayList<ILocalStructuralProperty>();
+        List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
         for (LogicalVariable column : columnList) {
-            localProps.add(new LocalOrderProperty(new OrderColumn(column, OrderKind.ASC)));
+            orderColumns.add(new OrderColumn(column, OrderKind.ASC));
         }
+        localProps.add(new LocalOrderProperty(orderColumns));
         IPartitioningProperty pp = null;
         AbstractLogicalOperator aop = (AbstractLogicalOperator) op;
         if (aop.getExecutionMode() == ExecutionMode.PARTITIONED) {
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
index 48a0fa6..6ad1fab 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -111,13 +111,15 @@
 
         GroupByOperator gOp = (GroupByOperator) op;
         Set<LogicalVariable> columnSet = new ListSet<LogicalVariable>();
+        List<OrderColumn> ocs = new ArrayList<OrderColumn>();
 
         if (!columnSet.isEmpty()) {
             propsLocal.add(new LocalGroupingProperty(columnSet));
         }
         for (OrderColumn oc : orderColumns) {
-            propsLocal.add(new LocalOrderProperty(oc));
+            ocs.add(oc);
         }
+        propsLocal.add(new LocalOrderProperty(ocs));
         for (ILogicalPlan p : gOp.getNestedPlans()) {
             for (Mutable<ILogicalOperator> r : p.getRoots()) {
                 ILogicalOperator rOp = r.getValue();
@@ -195,13 +197,16 @@
         }
 
         List<LogicalVariable> keyAndDecVariables = new ArrayList<LogicalVariable>();
-        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> p : gby.getGroupByList())
+        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> p : gby.getGroupByList()) {
             keyAndDecVariables.add(p.first);
-        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> p : gby.getDecorList())
+        }
+        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> p : gby.getDecorList()) {
             keyAndDecVariables.add(GroupByOperator.getDecorVariable(p));
+        }
 
-        for (LogicalVariable var : keyAndDecVariables)
+        for (LogicalVariable var : keyAndDecVariables) {
             aggOpInputEnv.setVarType(var, outputEnv.getVarType(var));
+        }
 
         compileSubplans(inputSchemas[0], gby, opSchema, context);
         IOperatorDescriptorRegistry spec = builder.getJobSpec();
@@ -234,10 +239,12 @@
         for (Object type : intermediateTypes) {
             aggOpInputEnv.setVarType(usedVars.get(i++), type);
         }
-        for (LogicalVariable keyVar : keyAndDecVariables)
+        for (LogicalVariable keyVar : keyAndDecVariables) {
             localInputSchemas[0].addVariable(keyVar);
-        for (LogicalVariable usedVar : usedVars)
+        }
+        for (LogicalVariable usedVar : usedVars) {
             localInputSchemas[0].addVariable(usedVar);
+        }
         for (i = 0; i < n; i++) {
             AggregateFunctionCallExpression mergeFun = (AggregateFunctionCallExpression) aggOp.getMergeExpressions()
                     .get(i).getValue();
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortMergeExchangePOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortMergeExchangePOperator.java
index b03b99d..7f29375 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortMergeExchangePOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/SortMergeExchangePOperator.java
@@ -3,9 +3,9 @@
  * 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.
@@ -15,6 +15,7 @@
 package edu.uci.ics.hyracks.algebricks.core.algebra.operators.physical;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -51,7 +52,7 @@
 
 public class SortMergeExchangePOperator extends AbstractExchangePOperator {
 
-    private OrderColumn[] sortColumns;
+    private final OrderColumn[] sortColumns;
 
     public SortMergeExchangePOperator(OrderColumn[] sortColumns) {
         this.sortColumns = sortColumns;
@@ -88,24 +89,28 @@
             inp1.computeDeliveredPhysicalProperties(context);
             pv1 = inp1.getDeliveredPhysicalProperties();
         }
-        int sortCol = 0;
+
+        List<OrderColumn> orderColumns = new ArrayList<OrderColumn>();
         List<ILocalStructuralProperty> localProps = new ArrayList<ILocalStructuralProperty>(sortColumns.length);
         for (ILocalStructuralProperty prop : pv1.getLocalProperties()) {
             if (prop.getPropertyType() == PropertyType.LOCAL_ORDER_PROPERTY) {
                 LocalOrderProperty lop = (LocalOrderProperty) prop;
-                if (lop.getOrderColumn().equals(sortColumns[sortCol])) {
-                    localProps.add(lop);
-                    sortCol++;
-                    if (sortCol == sortColumns.length) {
+                for (OrderColumn oc : lop.getOrderColumns()) {
+                    if (oc.equals(sortColumns[orderColumns.size()])) {
+                        orderColumns.add(oc);
+                        if (orderColumns.size() == sortColumns.length) {
+                            break;
+                        }
+                    } else {
                         break;
                     }
                 }
             } else {
-                break;
+                continue;
             }
         }
-        if (sortCol < sortColumns.length) {
-            localProps = null;
+        if (orderColumns.size() > 0) {
+            localProps.add(new LocalOrderProperty(orderColumns));
         }
         this.deliveredProperties = new StructuralPropertiesVector(IPartitioningProperty.UNPARTITIONED, localProps);
     }
@@ -114,9 +119,7 @@
     public PhysicalRequirements getRequiredPropertiesForChildren(ILogicalOperator op,
             IPhysicalPropertiesVector reqdByParent) {
         List<ILocalStructuralProperty> localProps = new ArrayList<ILocalStructuralProperty>(sortColumns.length);
-        for (OrderColumn oc : sortColumns) {
-            localProps.add(new LocalOrderProperty(oc));
-        }
+        localProps.add(new LocalOrderProperty(Arrays.asList(sortColumns)));
         StructuralPropertiesVector[] r = new StructuralPropertiesVector[] { new StructuralPropertiesVector(null,
                 localProps) };
         return new PhysicalRequirements(r, IPartitioningRequirementsCoordinator.NO_COORDINATION);
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/ILocalStructuralProperty.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/ILocalStructuralProperty.java
index 921b067..c32fd67 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/ILocalStructuralProperty.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/ILocalStructuralProperty.java
@@ -27,4 +27,24 @@
     public void getVariables(Collection<LogicalVariable> variables);
 
     public PropertyType getPropertyType();
+
+    /**
+     * Returns the retained property regarding to a collection of variables,
+     * e.g., some variables used in the property may not exist in the input
+     * collection and hence the data property changes.
+     * 
+     * @param vars
+     *            , an input collection of variables
+     * @return the retained data property.
+     */
+    public ILocalStructuralProperty retainVariables(Collection<LogicalVariable> vars);
+
+    /**
+     * Returns the additional data property within each group, which is dictated by the group keys.
+     * 
+     * @param vars
+     *            , group keys.
+     * @return the additional data property within each group.
+     */
+    public ILocalStructuralProperty regardToGroup(Collection<LogicalVariable> groupKeys);
 }
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalGroupingProperty.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalGroupingProperty.java
index 6f6ea64..c267364 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalGroupingProperty.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalGroupingProperty.java
@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Set;
 
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
 
 public class LocalGroupingProperty extends AbstractGroupingProperty implements ILocalStructuralProperty {
@@ -58,4 +59,45 @@
     public List<LogicalVariable> getPreferredOrderEnforcer() {
         return preferredOrderEnforcer;
     }
+
+    @Override
+    public ILocalStructuralProperty retainVariables(Collection<LogicalVariable> vars) {
+        Set<LogicalVariable> newVars = new ListSet<LogicalVariable>();
+        newVars.addAll(vars);
+        newVars.retainAll(columnSet);
+        if (columnSet.equals(newVars)) {
+            return new LocalGroupingProperty(columnSet, preferredOrderEnforcer);
+        }
+        // Column set for the retained grouping property
+        Set<LogicalVariable> newColumns = new ListSet<LogicalVariable>();
+        // Matches the prefix of the original column set.
+        for (LogicalVariable v : columnSet) {
+            if (newVars.contains(v)) {
+                newColumns.add(v);
+            } else {
+                break;
+            }
+        }
+        if (newColumns.size() > 0) {
+            return new LocalGroupingProperty(newColumns, preferredOrderEnforcer.subList(0, newColumns.size()));
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public ILocalStructuralProperty regardToGroup(Collection<LogicalVariable> groupKeys) {
+        Set<LogicalVariable> newColumns = new ListSet<LogicalVariable>();
+        for (LogicalVariable v : columnSet) {
+            if (!groupKeys.contains(v)) {
+                newColumns.add(v);
+            }
+        }
+        if (newColumns.size() > 0) {
+            return new LocalGroupingProperty(newColumns, preferredOrderEnforcer.subList(groupKeys.size(),
+                    newColumns.size()));
+        } else {
+            return null;
+        }
+    }
 }
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalOrderProperty.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalOrderProperty.java
index 6cb318a..6d568f4 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalOrderProperty.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/LocalOrderProperty.java
@@ -3,9 +3,9 @@
  * 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.
@@ -14,33 +14,48 @@
  */
 package edu.uci.ics.hyracks.algebricks.core.algebra.properties;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
+import edu.uci.ics.hyracks.algebricks.common.utils.ListSet;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.EquivalenceClass;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
 
 public final class LocalOrderProperty implements ILocalStructuralProperty {
 
-    private OrderColumn orderColumn;
+    private List<OrderColumn> orderColumns;
 
-    public LocalOrderProperty(OrderColumn orderColumn) {
-        this.orderColumn = orderColumn;
+    public LocalOrderProperty(List<OrderColumn> orderColumn) {
+        this.orderColumns = orderColumn;
     }
 
-    public OrderColumn getOrderColumn() {
-        return orderColumn;
+    public List<OrderColumn> getOrderColumns() {
+        return orderColumns;
     }
 
-    public void setOrderColumn(OrderColumn orderColumn) {
-        this.orderColumn = orderColumn;
+    public void setOrderColumns(List<OrderColumn> orderColumn) {
+        this.orderColumns = orderColumn;
     }
 
-    public LogicalVariable getColumn() {
-        return orderColumn.getColumn();
+    public List<LogicalVariable> getColumns() {
+        List<LogicalVariable> orderVars = new ArrayList<LogicalVariable>();
+        for (OrderColumn oc : orderColumns) {
+            orderVars.add(oc.getColumn());
+        }
+        return orderVars;
     }
 
-    public OrderKind getOrder() {
-        return orderColumn.getOrder();
+    public List<OrderKind> getOrders() {
+        List<OrderKind> orderKinds = new ArrayList<OrderKind>();
+        for (OrderColumn oc : orderColumns) {
+            orderKinds.add(oc.getOrder());
+        }
+        return orderKinds;
     }
 
     @Override
@@ -50,17 +65,146 @@
 
     @Override
     public void getColumns(Collection<LogicalVariable> columns) {
-        columns.add(getColumn());
+        columns.addAll(getColumns());
     }
 
     @Override
     public String toString() {
-        return orderColumn.toString();
+        return orderColumns.toString();
     }
 
     @Override
     public void getVariables(Collection<LogicalVariable> variables) {
-        variables.add(orderColumn.getColumn());
+        variables.addAll(getColumns());
     }
 
+    @Override
+    public boolean equals(Object object) {
+        LocalOrderProperty lop = (LocalOrderProperty) object;
+        return orderColumns.equals(lop.orderColumns);
+    }
+
+    /**
+     * Whether current property implies the required property
+     * 
+     * @param required
+     *            , a required property
+     * @return true if the current property satisfies the required property; otherwise false.
+     */
+    public final boolean implies(ILocalStructuralProperty required) {
+        if (required.getPropertyType() != PropertyType.LOCAL_ORDER_PROPERTY) {
+            return false;
+        }
+        LocalOrderProperty requiredOrderProperty = (LocalOrderProperty) required;
+        Iterator<OrderColumn> requiredColumnIterator = requiredOrderProperty.getOrderColumns().iterator();
+        Iterator<OrderColumn> currentColumnIterator = orderColumns.iterator();
+
+        // Returns true if requiredColumnIterator is a prefix of currentColumnIterator.
+        return isPrefixOf(requiredColumnIterator, currentColumnIterator);
+    }
+
+    private <T> boolean isPrefixOf(Iterator<T> requiredColumnIterator, Iterator<T> currentColumnIterator) {
+        while (requiredColumnIterator.hasNext()) {
+            T oc = requiredColumnIterator.next();
+            if (!currentColumnIterator.hasNext()) {
+                return false;
+            }
+            if (!oc.equals(currentColumnIterator.next())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public final void normalizeOrderingColumns(Map<LogicalVariable, EquivalenceClass> equivalenceClasses,
+            List<FunctionalDependency> fds) {
+        replaceOrderingColumnsByEqClasses(equivalenceClasses);
+        applyFDsToOrderingColumns(fds);
+    }
+
+    private void replaceOrderingColumnsByEqClasses(Map<LogicalVariable, EquivalenceClass> equivalenceClasses) {
+        if (equivalenceClasses == null || equivalenceClasses.isEmpty()) {
+            return;
+        }
+        List<OrderColumn> norm = new ArrayList<OrderColumn>();
+        for (OrderColumn oc : orderColumns) {
+            LogicalVariable v = oc.getColumn();
+            EquivalenceClass ec = equivalenceClasses.get(v);
+            if (ec == null) {
+                norm.add(new OrderColumn(v, oc.getOrder()));
+            } else {
+                if (ec.representativeIsConst()) {
+                    // trivially satisfied, so the var. can be removed
+                } else {
+                    norm.add(new OrderColumn(ec.getVariableRepresentative(), oc.getOrder()));
+                }
+            }
+        }
+        orderColumns = norm;
+    }
+
+    private void applyFDsToOrderingColumns(List<FunctionalDependency> fds) {
+        if (fds == null || fds.isEmpty()) {
+            return;
+        }
+        Set<LogicalVariable> norm = new ListSet<LogicalVariable>();
+        List<LogicalVariable> columns = getColumns();
+        for (LogicalVariable v : columns) {
+            boolean isImpliedByAnFD = false;
+            for (FunctionalDependency fdep : fds) {
+                if (columns.containsAll(fdep.getHead()) && fdep.getTail().contains(v)) {
+                    isImpliedByAnFD = true;
+                    norm.addAll(fdep.getHead());
+                    break;
+                }
+
+            }
+            if (!isImpliedByAnFD) {
+                norm.add(v);
+            }
+        }
+        Set<OrderColumn> impliedColumns = new ListSet<OrderColumn>();
+        for (OrderColumn oc : orderColumns) {
+            if (!norm.contains(oc.getColumn())) {
+                impliedColumns.add(oc);
+            }
+        }
+        orderColumns.removeAll(impliedColumns);
+    }
+
+    @Override
+    public ILocalStructuralProperty retainVariables(Collection<LogicalVariable> vars) {
+        List<LogicalVariable> columns = getColumns();
+        List<LogicalVariable> newVars = new ArrayList<LogicalVariable>();
+        newVars.addAll(vars);
+        newVars.retainAll(columns);
+        List<OrderColumn> newColumns = new ArrayList<OrderColumn>();
+        for (OrderColumn oc : orderColumns) {
+            if (newVars.contains(oc.getColumn())) {
+                newColumns.add(oc);
+            } else {
+                break;
+            }
+        }
+        if (newColumns.size() > 0) {
+            return new LocalOrderProperty(newColumns);
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public ILocalStructuralProperty regardToGroup(Collection<LogicalVariable> groupKeys) {
+        List<OrderColumn> newColumns = new ArrayList<OrderColumn>();
+        for (OrderColumn oc : orderColumns) {
+            if (!groupKeys.contains(oc.getColumn())) {
+                newColumns.add(oc);
+            }
+        }
+        if (newColumns.size() > 0) {
+            return new LocalOrderProperty(newColumns);
+        } else {
+            return null;
+        }
+    }
 }
diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/PropertiesUtil.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/PropertiesUtil.java
index 606c8cb..b651e59 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/PropertiesUtil.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/properties/PropertiesUtil.java
@@ -3,9 +3,9 @@
  * 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.
@@ -81,8 +81,7 @@
                             return false;
                         }
                         LocalOrderProperty lop = (LocalOrderProperty) d;
-                        if (lop.getColumn() == ((LocalOrderProperty) r).getColumn()
-                                && lop.getOrder() == ((LocalOrderProperty) r).getOrder()) {
+                        if (lop.implies(r)) {
                             implied = true;
                         } else {
                             return false;
@@ -92,15 +91,10 @@
                     case LOCAL_GROUPING_PROPERTY: {
                         dlvdCols.clear();
                         d.getColumns(dlvdCols);
-                        for (LogicalVariable v : dlvdCols) {
-                            if (rqdCols.contains(v)) {
-                                rqdCols.remove(v);
-                            } else {
-                                return false;
-                            }
-                        }
-                        if (rqdCols.isEmpty()) {
-                            implied = true;
+                        if (d.getPropertyType() == PropertyType.LOCAL_ORDER_PROPERTY) {
+                            implied = isPrefixOf(rqdCols.iterator(), dlvdCols.iterator());
+                        } else {
+                            implied = rqdCols.equals(dlvdCols) || isPrefixOf(rqdCols.iterator(), dlvdCols.iterator());
                         }
                         break;
                     }
@@ -114,12 +108,10 @@
             }
         }
         return true;
-
     }
 
     public static boolean matchPartitioningProps(IPartitioningProperty reqd, IPartitioningProperty dlvd,
             boolean mayExpandProperties) {
-
         INodeDomain dom1 = reqd.getNodeDomain();
         INodeDomain dom2 = dlvd.getNodeDomain();
         if (dom1 != null && dom2 != null && !dom1.sameAs(dom2)) {
@@ -137,7 +129,7 @@
                         UnorderedPartitionedProperty ur = (UnorderedPartitionedProperty) reqd;
                         UnorderedPartitionedProperty ud = (UnorderedPartitionedProperty) dlvd;
                         if (mayExpandProperties) {
-                            return ur.getColumnSet().containsAll(ud.getColumnSet());
+                            return isPrefixOf(ud.getColumnSet().iterator(), ur.getColumnSet().iterator());
                         } else {
                             return ur.getColumnSet().equals(ud.getColumnSet());
                         }
@@ -146,7 +138,8 @@
                         UnorderedPartitionedProperty ur = (UnorderedPartitionedProperty) reqd;
                         OrderedPartitionedProperty od = (OrderedPartitionedProperty) dlvd;
                         if (mayExpandProperties) {
-                            return ur.getColumnSet().containsAll(od.getOrderColumns());
+                            List<LogicalVariable> dlvdSortColumns = orderColumnsToVariables(od.getOrderColumns());
+                            return isPrefixOf(dlvdSortColumns.iterator(), ur.getColumnSet().iterator());
                         } else {
                             return ur.getColumnSet().containsAll(od.getOrderColumns())
                                     && od.getOrderColumns().containsAll(ur.getColumnSet());
@@ -163,7 +156,7 @@
                         OrderedPartitionedProperty or = (OrderedPartitionedProperty) reqd;
                         OrderedPartitionedProperty od = (OrderedPartitionedProperty) dlvd;
                         if (mayExpandProperties) {
-                            return isPrefixOf(od.getOrderColumns(), or.getOrderColumns());
+                            return isPrefixOf(od.getOrderColumns().iterator(), or.getOrderColumns().iterator());
                         } else {
                             return od.getOrderColumns().equals(or.getOrderColumns());
                         }
@@ -180,17 +173,32 @@
     }
 
     /**
+     * Converts a list of OrderColumns to a list of LogicalVariables.
+     * 
+     * @param orderColumns
+     *            , a list of OrderColumns
+     * @return the list of LogicalVariables
+     */
+    private static List<LogicalVariable> orderColumnsToVariables(List<OrderColumn> orderColumns) {
+        List<LogicalVariable> columns = new ArrayList<LogicalVariable>();
+        for (OrderColumn oc : orderColumns) {
+            columns.add(oc.getColumn());
+        }
+        return columns;
+    }
+
+    /**
      * @param pref
      * @param target
      * @return true iff pref is a prefix of target
      */
-    private static boolean isPrefixOf(List<OrderColumn> pref, List<OrderColumn> target) {
-        Iterator<OrderColumn> iter = target.iterator();
-        for (OrderColumn v : pref) {
-            if (!iter.hasNext()) {
+    private static <T> boolean isPrefixOf(Iterator<T> pref, Iterator<T> target) {
+        while (pref.hasNext()) {
+            T v = pref.next();
+            if (!target.hasNext()) {
                 return false;
             }
-            if (!v.equals(iter.next())) {
+            if (!v.equals(target.next())) {
                 return false;
             }
         }
@@ -276,18 +284,8 @@
                 ((LocalGroupingProperty) p).normalizeGroupingColumns(equivalenceClasses, fds);
                 pos++;
             } else {
-                LocalOrderProperty ord = (LocalOrderProperty) p;
-                EquivalenceClass ec = equivalenceClasses.get(ord.getColumn());
-                if (ec != null) {
-                    if (ec.representativeIsConst()) {
-                        propIter.remove();
-                    } else {
-                        ord.getOrderColumn().setColumn(ec.getVariableRepresentative());
-                        pos++;
-                    }
-                } else {
-                    pos++;
-                }
+                ((LocalOrderProperty) p).normalizeOrderingColumns(equivalenceClasses, fds);
+                pos++;
             }
         }
 
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
index 810defc..9d72881 100644
--- a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
@@ -3,9 +3,9 @@
  * 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.
@@ -16,6 +16,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -259,7 +260,7 @@
             if (AlgebricksConfig.DEBUG) {
                 AlgebricksConfig.ALGEBRICKS_LOGGER.fine(">>>> Removing redundant SORT operator "
                         + op.getPhysicalOperator() + "\n");
-                printOp((AbstractLogicalOperator) op);
+                printOp(op);
             }
             changed = true;
             AbstractLogicalOperator nextOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
@@ -270,7 +271,7 @@
             // 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();
             }
@@ -338,20 +339,24 @@
         }
 
         int prefix = dlvdCols.size() - 1;
-        for (; prefix >= 0;)
-            if (!rqdCols.contains(dlvdCols.get(prefix)))
+        for (; prefix >= 0;) {
+            if (!rqdCols.contains(dlvdCols.get(prefix))) {
                 prefix--;
-            else
+            } else {
                 break;
+            }
+        }
+
+        LocalOrderProperty orderProp = (LocalOrderProperty) dlvd.get(0);
+        List<OrderColumn> orderColumns = orderProp.getOrderColumns();
         for (int j = 0; j <= prefix; j++) {
-            LocalOrderProperty orderProp = (LocalOrderProperty) dlvd.get(j);
-            returnedProperties.add(new OrderColumn(orderProp.getColumn(), orderProp.getOrder()));
+            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++){
-                LocalOrderProperty orderProp = (LocalOrderProperty) dlvd.get(j);
-                returnedProperties.add(new OrderColumn(orderProp.getColumn(), orderProp.getOrder()));
+        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;
@@ -371,9 +376,9 @@
         }
         AbstractStableSortPOperator sortOp = (AbstractStableSortPOperator) op.getPhysicalOperator();
         sortOp.computeLocalProperties(op);
-        List<ILocalStructuralProperty> orderProps = sortOp.getOrderProperties();
-        return PropertiesUtil.matchLocalProperties(orderProps, delivered.getLocalProperties(),
-                context.getEquivalenceClassMap(op), context.getFDList(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,
@@ -423,11 +428,13 @@
                     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);
-                        LocalOrderProperty lop = new LocalOrderProperty(oc);
-                        oList.add(lop);
+                        orderColumns.add(oc);
                     }
+                    LocalOrderProperty lop = new LocalOrderProperty(orderColumns);
+                    oList.add(lop);
                     break;
                 }
                 default: {
@@ -448,11 +455,13 @@
             Mutable<ILogicalOperator> topOp, boolean isMicroOp, IOptimizationContext context)
             throws AlgebricksException {
         List<Pair<IOrder, Mutable<ILogicalExpression>>> oe = new LinkedList<Pair<IOrder, Mutable<ILogicalExpression>>>();
-        for (LocalOrderProperty o : oList) {
-            IOrder ordType = (o.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(o.getColumn())));
-            oe.add(pair);
+        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);
@@ -579,7 +588,7 @@
             for (ILocalStructuralProperty p : localProps) {
                 if (p.getPropertyType() == PropertyType.LOCAL_ORDER_PROPERTY) {
                     LocalOrderProperty lop = (LocalOrderProperty) p;
-                    ordCols.add(lop.getOrderColumn());
+                    ordCols.addAll(lop.getOrderColumns());
                 } else {
                     return null;
                 }