asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From buyin...@apache.org
Subject [2/4] incubator-asterixdb-hyracks git commit: Fix for ASTERIXDB-1018, ASTERIXDB-1017, ASTERIXDB-1019, ASTERIXDB-1020, ASTERIXDB-1029, ASTERIXDB-1030, ASTERIXDB-1034:
Date Thu, 31 Dec 2015 05:18:17 GMT
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java
deleted file mode 100644
index 9dce6c9..0000000
--- a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/PushSubplanIntoGroupByRule.java
+++ /dev/null
@@ -1,209 +0,0 @@
-/*
-
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 at
- *
- *   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 org.apache.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 org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.algebricks.common.utils.ListSet;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
-import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AggregateOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
-import org.apache.hyracks.algebricks.core.algebra.util.OperatorManipulationUtil;
-import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
-
-/**
- * This rule pushes an array of subplans on top of a group-by into the
- * nested plan of the group-by.
- * 
- * @author yingyib
- */
-
-public class PushSubplanIntoGroupByRule implements IAlgebraicRewriteRule {
-    /** Stores used variables above the current operator. */
-    private final Set<LogicalVariable> usedVarsSoFar = new HashSet<LogicalVariable>();
-
-    @Override
-    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
-            throws AlgebricksException {
-        return false;
-    }
-
-    @Override
-    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
-        ILogicalOperator parentOperator = opRef.getValue();
-        if (context.checkIfInDontApplySet(this, parentOperator)) {
-            return false;
-        }
-        context.addToDontApplySet(this, parentOperator);
-        VariableUtilities.getUsedVariables(parentOperator, usedVarsSoFar);
-        if (parentOperator.getInputs().size() <= 0) {
-            return false;
-        }
-        boolean changed = false;
-        GroupByOperator gby = null;
-        for (Mutable<ILogicalOperator> ref : parentOperator.getInputs()) {
-            AbstractLogicalOperator op = (AbstractLogicalOperator) ref.getValue();
-            /** Only processes subplan operator. */
-            List<SubplanOperator> subplans = new ArrayList<SubplanOperator>();
-            if (op.getOperatorTag() == LogicalOperatorTag.SUBPLAN) {
-                while (op.getOperatorTag() == LogicalOperatorTag.SUBPLAN) {
-                    SubplanOperator currentSubplan = (SubplanOperator) op;
-                    subplans.add(currentSubplan);
-                    op = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
-                }
-                /** Only processes the case a group-by operator is the input of the subplan operators. */
-                if (op.getOperatorTag() == LogicalOperatorTag.GROUP) {
-                    gby = (GroupByOperator) op;
-                    List<ILogicalPlan> newGbyNestedPlans = new ArrayList<ILogicalPlan>();
-                    for (SubplanOperator subplan : subplans) {
-                        List<ILogicalPlan> subplanNestedPlans = subplan.getNestedPlans();
-                        List<ILogicalPlan> gbyNestedPlans = gby.getNestedPlans();
-                        List<ILogicalPlan> subplanNestedPlansToRemove = new ArrayList<ILogicalPlan>();
-                        for (ILogicalPlan subplanNestedPlan : subplanNestedPlans) {
-                            List<Mutable<ILogicalOperator>> rootOpRefs = subplanNestedPlan.getRoots();
-                            List<Mutable<ILogicalOperator>> rootOpRefsToRemove = new ArrayList<Mutable<ILogicalOperator>>();
-                            for (Mutable<ILogicalOperator> rootOpRef : rootOpRefs) {
-                                /** Gets free variables in the root operator of a nested plan and its descent. */
-                                Set<LogicalVariable> freeVars = new ListSet<LogicalVariable>();
-                                VariableUtilities.getUsedVariablesInDescendantsAndSelf(rootOpRef.getValue(), freeVars);
-                                Set<LogicalVariable> producedVars = new ListSet<LogicalVariable>();
-                                VariableUtilities.getProducedVariablesInDescendantsAndSelf(rootOpRef.getValue(),
-                                        producedVars);
-                                freeVars.removeAll(producedVars);
-                                /** * Checks whether the above freeVars are all contained in live variables * of one nested plan inside the group-by operator. * If yes, then the subplan can be pushed into the nested plan of the group-by. */
-                                for (ILogicalPlan gbyNestedPlanOriginal : gbyNestedPlans) {
-                                    // add a subplan in the original gby
-                                    if (!newGbyNestedPlans.contains(gbyNestedPlanOriginal)) {
-                                        newGbyNestedPlans.add(gbyNestedPlanOriginal);
-                                    }
-
-                                    // add a pushed subplan
-                                    ILogicalPlan gbyNestedPlan = OperatorManipulationUtil.deepCopy(
-                                            gbyNestedPlanOriginal, context);
-                                    List<Mutable<ILogicalOperator>> gbyRootOpRefs = gbyNestedPlan.getRoots();
-                                    for (int rootIndex = 0; rootIndex < gbyRootOpRefs.size(); rootIndex++) {
-                                        //set the nts for a original subplan
-                                        Mutable<ILogicalOperator> originalGbyRootOpRef = gbyNestedPlanOriginal
-                                                .getRoots().get(rootIndex);
-                                        Mutable<ILogicalOperator> originalGbyNtsRef = downToNts(originalGbyRootOpRef);
-                                        NestedTupleSourceOperator originalNts = (NestedTupleSourceOperator) originalGbyNtsRef
-                                                .getValue();
-                                        originalNts.setDataSourceReference(new MutableObject<ILogicalOperator>(gby));
-
-                                        //push a new subplan if possible
-                                        Mutable<ILogicalOperator> gbyRootOpRef = gbyRootOpRefs.get(rootIndex);
-                                        Set<LogicalVariable> liveVars = new ListSet<LogicalVariable>();
-                                        VariableUtilities.getLiveVariables(gbyRootOpRef.getValue(), liveVars);
-                                        if (liveVars.containsAll(freeVars)) {
-                                            /** Does the actual push. */
-                                            Mutable<ILogicalOperator> ntsRef = downToNts(rootOpRef);
-                                            ntsRef.setValue(gbyRootOpRef.getValue());
-                                            // Removes unused vars.
-                                            AggregateOperator aggOp = (AggregateOperator) gbyRootOpRef.getValue();
-                                            for (int varIndex = aggOp.getVariables().size() - 1; varIndex >= 0; varIndex--) {
-                                                if (!freeVars.contains(aggOp.getVariables().get(varIndex))) {
-                                                    aggOp.getVariables().remove(varIndex);
-                                                    aggOp.getExpressions().remove(varIndex);
-                                                }
-                                            }
-
-                                            gbyRootOpRef.setValue(rootOpRef.getValue());
-                                            rootOpRefsToRemove.add(rootOpRef);
-
-                                            // Sets the nts for a new pushed plan.
-                                            Mutable<ILogicalOperator> oldGbyNtsRef = downToNts(gbyRootOpRef);
-                                            NestedTupleSourceOperator nts = (NestedTupleSourceOperator) oldGbyNtsRef
-                                                    .getValue();
-                                            nts.setDataSourceReference(new MutableObject<ILogicalOperator>(gby));
-
-                                            newGbyNestedPlans.add(gbyNestedPlan);
-                                            changed = true;
-                                            continue;
-                                        }
-                                    }
-                                }
-                            }
-                            rootOpRefs.removeAll(rootOpRefsToRemove);
-                            if (rootOpRefs.size() == 0) {
-                                subplanNestedPlansToRemove.add(subplanNestedPlan);
-                            }
-                        }
-                        subplanNestedPlans.removeAll(subplanNestedPlansToRemove);
-                    }
-                    if (changed) {
-                        ref.setValue(gby);
-                        gby.getNestedPlans().clear();
-                        gby.getNestedPlans().addAll(newGbyNestedPlans);
-                    }
-                }
-            }
-        }
-        if (changed) {
-            cleanup(gby);
-        }
-        return changed;
-    }
-
-    /**
-     * Removes unused aggregation variables (and expressions)
-     * 
-     * @param gby
-     * @throws AlgebricksException
-     */
-    private void cleanup(GroupByOperator gby) throws AlgebricksException {
-        for (ILogicalPlan nestedPlan : gby.getNestedPlans()) {
-            for (Mutable<ILogicalOperator> rootRef : nestedPlan.getRoots()) {
-                AggregateOperator aggOp = (AggregateOperator) rootRef.getValue();
-                for (int varIndex = aggOp.getVariables().size() - 1; varIndex >= 0; varIndex--) {
-                    if (!usedVarsSoFar.contains(aggOp.getVariables().get(varIndex))) {
-                        aggOp.getVariables().remove(varIndex);
-                        aggOp.getExpressions().remove(varIndex);
-                    }
-                }
-            }
-
-        }
-    }
-
-    private Mutable<ILogicalOperator> downToNts(Mutable<ILogicalOperator> opRef) {
-        Mutable<ILogicalOperator> currentOpRef = opRef;
-        while (currentOpRef.getValue().getInputs().size() > 0) {
-            currentOpRef = currentOpRef.getValue().getInputs().get(0);
-        }
-        return currentOpRef;
-    }
-
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SimpleUnnestToProductRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SimpleUnnestToProductRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SimpleUnnestToProductRule.java
index 50b4ea9..21e9c06 100644
--- a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SimpleUnnestToProductRule.java
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SimpleUnnestToProductRule.java
@@ -36,7 +36,6 @@ import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogi
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractScanOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.EmptyTupleSourceOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
 import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 
@@ -80,19 +79,41 @@ public class SimpleUnnestToProductRule implements IAlgebraicRewriteRule {
         Mutable<ILogicalOperator> tupleSourceOpRef = currentOpRef;
         currentOpRef = opRef;
         if (tupleSourceOpRef.getValue().getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
-            NestedTupleSourceOperator nts = (NestedTupleSourceOperator) tupleSourceOpRef.getValue();
-            // If the subplan input is a trivial plan, do not do the rewriting.
-            if (nts.getSourceOperator().getOperatorTag() != LogicalOperatorTag.EMPTYTUPLESOURCE) {
-                while (currentOpRef.getValue().getInputs().size() == 1
-                        && currentOpRef.getValue() instanceof AbstractScanOperator
-                        && descOrSelfIsSourceScan((AbstractLogicalOperator) currentOpRef.getValue())) {
-                    if (opsAreIndependent(currentOpRef.getValue(), tupleSourceOpRef.getValue())) {
-                        /** move down the boundary if the operator is independent of the tuple source */
-                        boundaryOpRef = currentOpRef.getValue().getInputs().get(0);
-                    } else {
-                        break;
-                    }
-                    currentOpRef = currentOpRef.getValue().getInputs().get(0);
+            while (currentOpRef.getValue().getInputs().size() == 1
+                    /*
+                     * When this rule is fired,
+                     * Unnests with a dataset function have been rewritten to DataSourceScans and
+                     * AccessMethod related rewriting hasn't been done. Therefore, we only need
+                     * to check if currentOpRef holds a DataSourceScanOperator.
+                     */
+                    && currentOpRef.getValue().getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN
+                    && descOrSelfIsSourceScan((AbstractLogicalOperator) currentOpRef.getValue())) {
+                if (opsAreIndependent(currentOpRef.getValue(), tupleSourceOpRef.getValue())) {
+                    /** move down the boundary if the operator is independent of the tuple source */
+                    boundaryOpRef = currentOpRef.getValue().getInputs().get(0);
+                } else {
+                    break;
+                }
+                currentOpRef = currentOpRef.getValue().getInputs().get(0);
+            }
+        } else {
+            //Move the boundary below any top const assigns.
+            boundaryOpRef = opRef.getValue().getInputs().get(0);
+            while (boundaryOpRef.getValue().getInputs().size() == 1
+                    /*
+                     * When this rule is fired,
+                     * Unnests with a dataset function have been rewritten to DataSourceScans and
+                     * AccessMethod related rewriting hasn't been done. Therefore, we only need
+                     * to check if boundaryOpRef holds a DataSourceScanOperator.
+                     */
+                    && boundaryOpRef.getValue().getOperatorTag() != LogicalOperatorTag.DATASOURCESCAN) {
+                List<LogicalVariable> opUsedVars = new ArrayList<LogicalVariable>();
+                VariableUtilities.getUsedVariables(boundaryOpRef.getValue(), opUsedVars);
+                if (opUsedVars.size() == 0) {
+                    // move down the boundary if the operator is a const assigns.
+                    boundaryOpRef = boundaryOpRef.getValue().getInputs().get(0);
+                } else {
+                    break;
                 }
             }
         }

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SubplanOutOfGroupRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SubplanOutOfGroupRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SubplanOutOfGroupRule.java
deleted file mode 100644
index d0d3dd7..0000000
--- a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/SubplanOutOfGroupRule.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 at
- *
- *   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 org.apache.hyracks.algebricks.rewriter.rules;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-import org.apache.commons.lang3.mutable.Mutable;
-import org.apache.commons.lang3.mutable.MutableObject;
-
-import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
-import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
-import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
-import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
-
-/**
- * Looks for a nested group-by plan ending in
- * subplan {
- * ...
- * }
- * select (function-call: algebricks:not, Args:[function-call:
- * algebricks:is-null, Args:[...]])
- * nested tuple source -- |UNPARTITIONED|
- */
-
-public class SubplanOutOfGroupRule 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 op0 = (AbstractLogicalOperator) opRef.getValue();
-        if (op0.getOperatorTag() != LogicalOperatorTag.GROUP) {
-            return false;
-        }
-        GroupByOperator gby = (GroupByOperator) op0;
-
-        Iterator<ILogicalPlan> plansIter = gby.getNestedPlans().iterator();
-        ILogicalPlan p = null;
-        while (plansIter.hasNext()) {
-            p = plansIter.next();
-        }
-        if (p == null) {
-            return false;
-        }
-        if (p.getRoots().size() != 1) {
-            return false;
-        }
-        Mutable<ILogicalOperator> op1Ref = p.getRoots().get(0);
-        AbstractLogicalOperator op1 = (AbstractLogicalOperator) op1Ref.getValue();
-        boolean found = false;
-        while (op1.getInputs().size() == 1) {
-            if (op1.getOperatorTag() == LogicalOperatorTag.SUBPLAN) {
-                SubplanOperator subplan = (SubplanOperator) op1;
-                AbstractLogicalOperator op2 = (AbstractLogicalOperator) subplan.getInputs().get(0).getValue();
-                if (OperatorPropertiesUtil.isNullTest(op2)) {
-                    if (subplan.getNestedPlans().size() == 1) {
-                        ILogicalPlan p1 = subplan.getNestedPlans().get(0);
-                        if (p1.getRoots().size() == 1) {
-                            AbstractLogicalOperator r1 = (AbstractLogicalOperator) p1.getRoots().get(0).getValue();
-                            if (r1.getOperatorTag() == LogicalOperatorTag.INNERJOIN
-                                    || r1.getOperatorTag() == LogicalOperatorTag.LEFTOUTERJOIN) {
-                                // now, check that it propagates all variables,
-                                // so it can be pushed
-                                List<LogicalVariable> op2Vars = new ArrayList<LogicalVariable>();
-                                VariableUtilities.getLiveVariables(op2, op2Vars);
-                                List<LogicalVariable> op1Vars = new ArrayList<LogicalVariable>();
-                                VariableUtilities.getLiveVariables(subplan, op1Vars);
-                                if (op1Vars.containsAll(op2Vars)) {
-                                    found = true;
-                                    break;
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-            op1Ref = op1.getInputs().get(0);
-            op1 = (AbstractLogicalOperator) op1Ref.getValue();
-        }
-        if (!found) {
-            return false;
-        }
-
-        ILogicalOperator subplan = op1;
-        ILogicalOperator op2 = op1.getInputs().get(0).getValue();
-        op1Ref.setValue(op2);
-        Mutable<ILogicalOperator> opUnderRef = gby.getInputs().get(0);
-        ILogicalOperator opUnder = opUnderRef.getValue();
-        subplan.getInputs().clear();
-        subplan.getInputs().add(new MutableObject<ILogicalOperator>(opUnder));
-        opUnderRef.setValue(subplan);
-
-        return true;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanRule.java
new file mode 100644
index 0000000..32c7e03
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanRule.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+import java.util.LinkedList;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorManipulationUtil;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.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.
+     */
+    @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 {
+                    InnerJoinOperator j = new InnerJoinOperator(
+                            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/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanWithInputCardinalityOneRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanWithInputCardinalityOneRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanWithInputCardinalityOneRule.java
new file mode 100644
index 0000000..f7b6dea
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/EliminateSubplanWithInputCardinalityOneRule.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.common.utils.ListSet;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.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/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceGroupByForSubplanRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceGroupByForSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceGroupByForSubplanRule.java
new file mode 100644
index 0000000..3e65d91
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceGroupByForSubplanRule.java
@@ -0,0 +1,337 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.common.utils.ListSet;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
+import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
+import org.apache.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AggregateOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.ProjectOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import org.apache.hyracks.algebricks.core.algebra.plan.ALogicalPlanImpl;
+import org.apache.hyracks.algebricks.core.algebra.properties.FunctionalDependency;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorManipulationUtil;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.hyracks.algebricks.core.config.AlgebricksConfig;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+import org.apache.hyracks.algebricks.rewriter.util.PhysicalOptimizationsUtil;
+
+/**
+ * The rule searches for SUBPLAN operator with a optional PROJECT operator and
+ * an AGGREGATE followed by a join operator.
+ *
+ * <pre>
+ * Before
+ *
+ *   plan__parent
+ *   SUBPLAN {
+ *     PROJECT?
+ *     AGGREGATE
+ *     plan__nested_A
+ *     INNER_JOIN | LEFT_OUTER_JOIN ($condition, $left, $right)
+ *       plan__nested_B
+ *   }
+ *   plan__child
+ *
+ *   where $condition does not equal a constant true.
+ *
+ * After (This is a general application of the rule, specifics may vary based on the query plan.)
+ *
+ *   plan__parent
+ *   GROUP_BY {
+ *     PROJECT?
+ *     AGGREGATE
+ *     plan__nested_A
+ *     SELECT( algebricks:not( is_null( $right ) ) )
+ *     NESTED_TUPLE_SOURCE
+ *   }
+ *   SUBPLAN {
+ *     INNER_JOIN | LEFT_OUTER_JOIN ($condition, $left, $right)
+ *       plan__nested_B
+ *   }
+ *   plan__child
+ * </pre>
+ *
+ * @author prestonc
+ */
+
+public class IntroduceGroupByForSubplanRule 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 op0 = (AbstractLogicalOperator) opRef.getValue();
+        if (op0.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+            return false;
+        }
+        SubplanOperator subplan = (SubplanOperator) op0;
+
+        Iterator<ILogicalPlan> plansIter = subplan.getNestedPlans().iterator();
+        ILogicalPlan p = null;
+        while (plansIter.hasNext()) {
+            p = plansIter.next();
+        }
+        if (p == null) {
+            return false;
+        }
+        if (p.getRoots().size() != 1) {
+            return false;
+        }
+        Mutable<ILogicalOperator> subplanRoot = p.getRoots().get(0);
+        AbstractLogicalOperator op1 = (AbstractLogicalOperator) subplanRoot.getValue();
+
+        Mutable<ILogicalOperator> botRef = subplanRoot;
+        AbstractLogicalOperator op2;
+        // Project is optional
+        if (op1.getOperatorTag() != LogicalOperatorTag.PROJECT) {
+            op2 = op1;
+        } else {
+            ProjectOperator project = (ProjectOperator) op1;
+            botRef = project.getInputs().get(0);
+            op2 = (AbstractLogicalOperator) botRef.getValue();
+        }
+        if (op2.getOperatorTag() != LogicalOperatorTag.AGGREGATE) {
+            return false;
+        }
+        AggregateOperator aggregate = (AggregateOperator) op2;
+
+        Set<LogicalVariable> free = new HashSet<LogicalVariable>();
+        VariableUtilities.getUsedVariables(aggregate, free);
+
+        Mutable<ILogicalOperator> op3Ref = aggregate.getInputs().get(0);
+        AbstractLogicalOperator op3 = (AbstractLogicalOperator) op3Ref.getValue();
+
+        while (op3.getInputs().size() == 1) {
+            Set<LogicalVariable> prod = new HashSet<LogicalVariable>();
+            VariableUtilities.getProducedVariables(op3, prod);
+            free.removeAll(prod);
+            VariableUtilities.getUsedVariables(op3, free);
+            botRef = op3Ref;
+            op3Ref = op3.getInputs().get(0);
+            op3 = (AbstractLogicalOperator) op3Ref.getValue();
+        }
+
+        if (op3.getOperatorTag() != LogicalOperatorTag.INNERJOIN
+                && op3.getOperatorTag() != LogicalOperatorTag.LEFTOUTERJOIN) {
+            return false;
+        }
+        AbstractBinaryJoinOperator join = (AbstractBinaryJoinOperator) op3;
+        if (join.getCondition().getValue() == ConstantExpression.TRUE) {
+            return false;
+        }
+        VariableUtilities.getUsedVariables(join, free);
+
+        AbstractLogicalOperator b0 = (AbstractLogicalOperator) join.getInputs().get(0).getValue();
+        // see if there's an NTS at the end of the pipeline
+        NestedTupleSourceOperator outerNts = getNts(b0);
+        if (outerNts == null) {
+            AbstractLogicalOperator b1 = (AbstractLogicalOperator) join.getInputs().get(1).getValue();
+            outerNts = getNts(b1);
+            if (outerNts == null) {
+                return false;
+            }
+        }
+
+        Set<LogicalVariable> pkVars = computeGbyVars(outerNts, free, context);
+        if (pkVars == null || pkVars.size() < 1) {
+            // there is no non-trivial primary key, group-by keys are all live variables
+            // that were produced by descendant or self
+            ILogicalOperator subplanInput = subplan.getInputs().get(0).getValue();
+            pkVars = new HashSet<LogicalVariable>();
+            //get live variables
+            VariableUtilities.getLiveVariables(subplanInput, pkVars);
+
+            //get produced variables
+            Set<LogicalVariable> producedVars = new HashSet<LogicalVariable>();
+            VariableUtilities.getProducedVariablesInDescendantsAndSelf(subplanInput, producedVars);
+
+            //retain the intersection
+            pkVars.retainAll(producedVars);
+        }
+        AlgebricksConfig.ALGEBRICKS_LOGGER.fine("Found FD for introducing group-by: " + pkVars);
+
+        Mutable<ILogicalOperator> rightRef = join.getInputs().get(1);
+        LogicalVariable testForNull = null;
+        AbstractLogicalOperator right = (AbstractLogicalOperator) rightRef.getValue();
+        switch (right.getOperatorTag()) {
+            case UNNEST: {
+                UnnestOperator innerUnnest = (UnnestOperator) right;
+                // Select [ $y != null ]
+                testForNull = innerUnnest.getVariable();
+                break;
+            }
+            case RUNNINGAGGREGATE: {
+                ILogicalOperator inputToRunningAggregate = right.getInputs().get(0).getValue();
+                Set<LogicalVariable> producedVars = new ListSet<LogicalVariable>();
+                VariableUtilities.getProducedVariables(inputToRunningAggregate, producedVars);
+                if (!producedVars.isEmpty()) {
+                    // Select [ $y != null ]
+                    testForNull = producedVars.iterator().next();
+                }
+                break;
+            }
+            case DATASOURCESCAN: {
+                DataSourceScanOperator innerScan = (DataSourceScanOperator) right;
+                // Select [ $y != null ]
+                if (innerScan.getVariables().size() == 1) {
+                    testForNull = innerScan.getVariables().get(0);
+                }
+                break;
+            }
+            default:
+                break;
+        }
+        if (testForNull == null) {
+            testForNull = context.newVar();
+            AssignOperator tmpAsgn = new AssignOperator(testForNull,
+                    new MutableObject<ILogicalExpression>(ConstantExpression.TRUE));
+            tmpAsgn.getInputs().add(new MutableObject<ILogicalOperator>(rightRef.getValue()));
+            rightRef.setValue(tmpAsgn);
+            context.computeAndSetTypeEnvironmentForOperator(tmpAsgn);
+        }
+
+        IFunctionInfo finfoEq = context.getMetadataProvider().lookupFunction(AlgebricksBuiltinFunctions.IS_NULL);
+        ILogicalExpression isNullTest = new ScalarFunctionCallExpression(finfoEq,
+                new MutableObject<ILogicalExpression>(new VariableReferenceExpression(testForNull)));
+        IFunctionInfo finfoNot = context.getMetadataProvider().lookupFunction(AlgebricksBuiltinFunctions.NOT);
+        ScalarFunctionCallExpression nonNullTest = new ScalarFunctionCallExpression(finfoNot,
+                new MutableObject<ILogicalExpression>(isNullTest));
+        SelectOperator selectNonNull = new SelectOperator(new MutableObject<ILogicalExpression>(nonNullTest), false,
+                null);
+        GroupByOperator g = new GroupByOperator();
+        Mutable<ILogicalOperator> newSubplanRef = new MutableObject<ILogicalOperator>(subplan);
+        NestedTupleSourceOperator nts = new NestedTupleSourceOperator(new MutableObject<ILogicalOperator>(g));
+        opRef.setValue(g);
+        selectNonNull.getInputs().add(new MutableObject<ILogicalOperator>(nts));
+
+        List<Mutable<ILogicalOperator>> prodInpList = botRef.getValue().getInputs();
+        prodInpList.clear();
+        prodInpList.add(new MutableObject<ILogicalOperator>(selectNonNull));
+
+        ILogicalPlan gPlan = new ALogicalPlanImpl(new MutableObject<ILogicalOperator>(subplanRoot.getValue()));
+        g.getNestedPlans().add(gPlan);
+        subplanRoot.setValue(op3Ref.getValue());
+        g.getInputs().add(newSubplanRef);
+
+        HashSet<LogicalVariable> underVars = new HashSet<LogicalVariable>();
+        VariableUtilities.getLiveVariables(subplan.getInputs().get(0).getValue(), underVars);
+        underVars.removeAll(pkVars);
+        Map<LogicalVariable, LogicalVariable> mappedVars = buildVarExprList(pkVars, context, g, g.getGroupByList());
+        context.updatePrimaryKeys(mappedVars);
+        for (LogicalVariable uv : underVars) {
+            g.getDecorList().add(new Pair<LogicalVariable, Mutable<ILogicalExpression>>(null,
+                    new MutableObject<ILogicalExpression>(new VariableReferenceExpression(uv))));
+        }
+        OperatorPropertiesUtil.typeOpRec(subplanRoot, context);
+        OperatorPropertiesUtil.typeOpRec(gPlan.getRoots().get(0), context);
+        context.computeAndSetTypeEnvironmentForOperator(g);
+        return true;
+    }
+
+    private NestedTupleSourceOperator getNts(AbstractLogicalOperator op) {
+        AbstractLogicalOperator alo = op;
+        do {
+            if (alo.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+                return (NestedTupleSourceOperator) alo;
+            }
+            if (alo.getInputs().size() != 1) {
+                return null;
+            }
+            alo = (AbstractLogicalOperator) alo.getInputs().get(0).getValue();
+        } while (true);
+    }
+
+    protected Set<LogicalVariable> computeGbyVars(AbstractLogicalOperator op, Set<LogicalVariable> freeVars,
+            IOptimizationContext context) throws AlgebricksException {
+        PhysicalOptimizationsUtil.computeFDsAndEquivalenceClasses(op, context);
+        List<FunctionalDependency> fdList = context.getFDList(op);
+        if (fdList == null) {
+            return null;
+        }
+        // check if any of the FDs is a key
+        List<LogicalVariable> all = new ArrayList<LogicalVariable>();
+        VariableUtilities.getLiveVariables(op, all);
+        all.retainAll(freeVars);
+        for (FunctionalDependency fd : fdList) {
+            if (fd.getTail().containsAll(all)) {
+                return new HashSet<LogicalVariable>(fd.getHead());
+            }
+        }
+        return null;
+    }
+
+    private Map<LogicalVariable, LogicalVariable> buildVarExprList(Collection<LogicalVariable> vars,
+            IOptimizationContext context, GroupByOperator g,
+            List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> outVeList) throws AlgebricksException {
+        Map<LogicalVariable, LogicalVariable> m = new HashMap<LogicalVariable, LogicalVariable>();
+        for (LogicalVariable ov : vars) {
+            LogicalVariable newVar = context.newVar();
+            ILogicalExpression varExpr = new VariableReferenceExpression(newVar);
+            outVeList.add(new Pair<LogicalVariable, Mutable<ILogicalExpression>>(ov,
+                    new MutableObject<ILogicalExpression>(varExpr)));
+            for (ILogicalPlan p : g.getNestedPlans()) {
+                for (Mutable<ILogicalOperator> r : p.getRoots()) {
+                    OperatorManipulationUtil.substituteVarRec((AbstractLogicalOperator) r.getValue(), ov, newVar, true,
+                            context);
+                }
+            }
+            AbstractLogicalOperator opUnder = (AbstractLogicalOperator) g.getInputs().get(0).getValue();
+            OperatorManipulationUtil.substituteVarRec(opUnder, ov, newVar, true, context);
+            m.put(ov, newVar);
+        }
+        return m;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceLeftOuterJoinForSubplanRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceLeftOuterJoinForSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceLeftOuterJoinForSubplanRule.java
new file mode 100644
index 0000000..daf27ac
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/IntroduceLeftOuterJoinForSubplanRule.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.LeftOuterJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+public class IntroduceLeftOuterJoinForSubplanRule 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 op0 = (AbstractLogicalOperator) opRef.getValue();
+        if (op0.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+            return false;
+        }
+        SubplanOperator subplan = (SubplanOperator) op0;
+
+        Iterator<ILogicalPlan> plansIter = subplan.getNestedPlans().iterator();
+        ILogicalPlan p = null;
+        while (plansIter.hasNext()) {
+            p = plansIter.next();
+        }
+        if (p == null) {
+            return false;
+        }
+        if (p.getRoots().size() != 1) {
+            return false;
+        }
+        Mutable<ILogicalOperator> subplanRoot = p.getRoots().get(0);
+        AbstractLogicalOperator op1 = (AbstractLogicalOperator) subplanRoot.getValue();
+        Mutable<ILogicalOperator> opUnder = subplan.getInputs().get(0);
+
+        if (OperatorPropertiesUtil.isNullTest((AbstractLogicalOperator) opUnder.getValue())) {
+            return false;
+        }
+
+        switch (op1.getOperatorTag()) {
+            case INNERJOIN: {
+                InnerJoinOperator join = (InnerJoinOperator) op1;
+                Mutable<ILogicalOperator> leftRef = join.getInputs().get(0);
+                Mutable<ILogicalOperator> rightRef = join.getInputs().get(1);
+                Mutable<ILogicalOperator> ntsRef = getNtsAtEndOfPipeline(leftRef);
+                if (ntsRef == null) {
+                    ntsRef = getNtsAtEndOfPipeline(rightRef);
+                    if (ntsRef == null) {
+                        return false;
+                    } else {
+                        Mutable<ILogicalOperator> t = leftRef;
+                        leftRef = rightRef;
+                        rightRef = t;
+                    }
+                }
+                ntsRef.setValue(opUnder.getValue());
+                LeftOuterJoinOperator loj = new LeftOuterJoinOperator(join.getCondition());
+                loj.getInputs().add(leftRef);
+                loj.getInputs().add(rightRef);
+                opRef.setValue(loj);
+                context.computeAndSetTypeEnvironmentForOperator(loj);
+                return true;
+            }
+            case LEFTOUTERJOIN: {
+                LeftOuterJoinOperator join = (LeftOuterJoinOperator) op1;
+                Mutable<ILogicalOperator> leftRef = join.getInputs().get(0);
+                Mutable<ILogicalOperator> ntsRef = getNtsAtEndOfPipeline(leftRef);
+                if (ntsRef == null) {
+                    return false;
+                }
+                ntsRef.setValue(opUnder.getValue());
+                opRef.setValue(join);
+                context.computeAndSetTypeEnvironmentForOperator(join);
+                return true;
+            }
+            default: {
+                return false;
+            }
+        }
+    }
+
+    private Mutable<ILogicalOperator> getNtsAtEndOfPipeline(Mutable<ILogicalOperator> opRef) {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+            return opRef;
+        }
+        if (op.getInputs().size() != 1) {
+            return null;
+        }
+        return getNtsAtEndOfPipeline(op.getInputs().get(0));
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/MoveFreeVariableOperatorOutOfSubplanRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/MoveFreeVariableOperatorOutOfSubplanRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/MoveFreeVariableOperatorOutOfSubplanRule.java
new file mode 100644
index 0000000..959466a
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/MoveFreeVariableOperatorOutOfSubplanRule.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+import java.util.HashSet;
+import java.util.ListIterator;
+import java.util.Set;
+
+import org.apache.commons.lang3.mutable.Mutable;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.hyracks.algebricks.rewriter.rules.AbstractDecorrelationRule;
+
+/**
+ * The rule searches for operators that can be moved outside the subplan.
+ *
+ * <pre>
+ * Before
+ * 
+ *   %PARENT_PLAN
+ *   SUBPLAN{
+ *     %NESTED_OPERATORS_B+
+ *     ASSIGN || %SUBPLAN
+ *     %NESTED_OPERATORS_A*
+ *     NESTEDTUPLESOURCE
+ *   }
+ *   %CHILD_PLAN
+ * 
+ *   where
+ *     %SUBPLAN has one nested plan with a root AGGREGATE operator.
+ * 
+ * After
+ * 
+ *   %PARENT_PLAN
+ *   SUBPLAN{
+ *     %NESTED_OPERATORS_B+
+ *     %NESTED_OPERATORS_A*
+ *     NESTEDTUPLESOURCE
+ *   }
+ *   ASSIGN || %SUBPLAN
+ *   %CHILD_PLAN
+ * </pre>
+ */
+public class MoveFreeVariableOperatorOutOfSubplanRule extends AbstractDecorrelationRule {
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        AbstractLogicalOperator op0 = (AbstractLogicalOperator) opRef.getValue();
+        if (op0.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+            return false;
+        }
+        SubplanOperator subplan = (SubplanOperator) op0;
+
+        Mutable<ILogicalOperator> leftRef = subplan.getInputs().get(0);
+        if (((AbstractLogicalOperator) leftRef.getValue()).getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) {
+            return false;
+        }
+
+        ListIterator<ILogicalPlan> plansIter = subplan.getNestedPlans().listIterator();
+        ILogicalPlan p = null;
+        while (plansIter.hasNext()) {
+            p = plansIter.next();
+        }
+        if (p == null) {
+            return false;
+        }
+        if (p.getRoots().size() != 1) {
+            return false;
+        }
+        Mutable<ILogicalOperator> opRef1 = p.getRoots().get(0);
+
+        //The root operator will not be movable. Start with the second op
+        AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef1.getValue();
+        if (op1.getInputs().size() != 1) {
+            return false;
+        }
+        Mutable<ILogicalOperator> op2Ref = op1.getInputs().get(0);
+
+        //Get all variables that come from outside of the loop
+        Set<LogicalVariable> free = new HashSet<LogicalVariable>();
+        OperatorPropertiesUtil.getFreeVariablesInSelfOrDesc(op1, free);
+
+        while (op2Ref != null) {
+            //Get the operator that we want to look at
+            AbstractLogicalOperator op2 = (AbstractLogicalOperator) op2Ref.getValue();
+
+            //Make sure we are looking at subplan with a scan/join
+            if (op2.getInputs().size() != 1 || !descOrSelfIsScanOrJoin(op2)) {
+                return false;
+            }
+            boolean notApplicable = false;
+
+            //Get its used variables
+            Set<LogicalVariable> used = new HashSet<LogicalVariable>();
+
+            //not movable if the operator is not an assign or subplan
+            //Might be helpful in the future for other operations in the future
+            if (movableOperator(op2.getOperatorTag())) {
+                if (op2.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
+                    VariableUtilities.getUsedVariables(op2, used);
+                } else if (op2.getOperatorTag() == LogicalOperatorTag.SUBPLAN) {
+                    // Nested plan must have an aggregate root.
+                    ListIterator<ILogicalPlan> subplansIter = ((SubplanOperator) op2).getNestedPlans().listIterator();
+                    ILogicalPlan plan = null;
+                    while (subplansIter.hasNext()) {
+                        plan = subplansIter.next();
+                    }
+                    if (plan == null) {
+                        return false;
+                    }
+                    if (plan.getRoots().size() != 1) {
+                        return false;
+                    }
+                    ILogicalOperator op3 = plan.getRoots().get(0).getValue();
+                    if (op3.getOperatorTag() != LogicalOperatorTag.AGGREGATE) {
+                        return false;
+                    }
+                    // Used variables do not include ones created in the subplan.
+                    VariableUtilities.getUsedVariables(op2, used);
+                    Set<LogicalVariable> subplanProducedAndDown = new HashSet<LogicalVariable>();
+                    VariableUtilities.getProducedVariablesInDescendantsAndSelf(op3, subplanProducedAndDown);
+                    used.removeAll(subplanProducedAndDown);
+                } else {
+                    notApplicable = true;
+                }
+            } else {
+                notApplicable = true;
+            }
+
+            //Make sure that all of its used variables come from outside
+            for (LogicalVariable var : used) {
+                if (!free.contains(var)) {
+                    notApplicable = true;
+                }
+            }
+
+            if (notApplicable) {
+                op2Ref = op2.getInputs().get(0);
+            } else {
+                //Make the input of op2 be the input of op1
+                op2Ref.setValue(op2.getInputs().get(0).getValue());
+
+                //Make the outside of the subplan the input of op2
+                Mutable<ILogicalOperator> outsideRef = op2.getInputs().get(0);
+                outsideRef.setValue(op0.getInputs().get(0).getValue());
+
+                //Make op2 the input of the subplan
+                Mutable<ILogicalOperator> op2OutsideRef = op0.getInputs().get(0);
+                op2OutsideRef.setValue(op2);
+
+                return true;
+            }
+
+        }
+        return false;
+    }
+
+    protected boolean movableOperator(LogicalOperatorTag operatorTag) {
+        return (operatorTag == LogicalOperatorTag.ASSIGN || operatorTag == LogicalOperatorTag.SUBPLAN);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/7dd47992/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/NestedSubplanToJoinRule.java
----------------------------------------------------------------------
diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/NestedSubplanToJoinRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/NestedSubplanToJoinRule.java
new file mode 100644
index 0000000..35c7e4e
--- /dev/null
+++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/subplan/NestedSubplanToJoinRule.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *   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 org.apache.hyracks.algebricks.rewriter.rules.subplan;
+
+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 org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractOperatorWithNestedPlans;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.EmptyTupleSourceOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.util.OperatorPropertiesUtil;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+/**
+ * replace Subplan operators with nested loop joins where the join condition is true, if the Subplan
+ * does not contain free variables (does not have correlations to the input stream).
+ *
+ * @author yingyib
+ */
+public class NestedSubplanToJoinRule 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 {
+        if (context.checkIfInDontApplySet(this, opRef.getValue()))
+            return false;
+        context.addToDontApplySet(this, opRef.getValue());
+
+        ILogicalOperator op1 = opRef.getValue();
+        if (op1.getInputs().size() == 0) {
+            return false;
+        }
+
+        boolean rewritten = false;
+        for (int index = 0; index < op1.getInputs().size(); index++) {
+            AbstractLogicalOperator child = (AbstractLogicalOperator) op1.getInputs().get(index).getValue();
+            if (child.getOperatorTag() != LogicalOperatorTag.SUBPLAN) {
+                continue;
+            }
+
+            AbstractOperatorWithNestedPlans subplan = (AbstractOperatorWithNestedPlans) child;
+            Set<LogicalVariable> freeVars = new HashSet<LogicalVariable>();
+            OperatorPropertiesUtil.getFreeVariablesInSubplans(subplan, freeVars);
+            if (!freeVars.isEmpty()) {
+                /**
+                 * the subplan is correlated with the outer plan, other rules can deal with it
+                 */
+                continue;
+            }
+
+            /** get the input operator of the subplan operator */
+            ILogicalOperator subplanInput = subplan.getInputs().get(0).getValue();
+            AbstractLogicalOperator subplanInputOp = (AbstractLogicalOperator) subplanInput;
+
+            /** If the other join branch is a trivial plan, do not do the rewriting. */
+            if (subplanInputOp.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) {
+                continue;
+            }
+
+            /** get all nested top operators */
+            List<ILogicalPlan> nestedPlans = subplan.getNestedPlans();
+            List<Mutable<ILogicalOperator>> nestedRoots = new ArrayList<Mutable<ILogicalOperator>>();
+            for (ILogicalPlan nestedPlan : nestedPlans) {
+                nestedRoots.addAll(nestedPlan.getRoots());
+            }
+            if (nestedRoots.size() == 0) {
+                /** there is no nested top operators */
+                continue;
+            }
+
+            /**
+             * Expends the input and roots into a DAG of nested loop joins.
+             * Though joins should be left-outer joins, a left-outer join with condition TRUE is equivalent to an inner join.
+             **/
+            Mutable<ILogicalExpression> expr = new MutableObject<ILogicalExpression>(ConstantExpression.TRUE);
+            Mutable<ILogicalOperator> nestedRootRef = nestedRoots.get(0);
+            ILogicalOperator join = new InnerJoinOperator(expr, new MutableObject<ILogicalOperator>(subplanInput),
+                    nestedRootRef);
+
+            /** rewrite the nested tuple source to be empty tuple source */
+            rewriteNestedTupleSource(nestedRootRef, context);
+
+            for (int i = 1; i < nestedRoots.size(); i++) {
+                join = new InnerJoinOperator(expr, new MutableObject<ILogicalOperator>(join), nestedRoots.get(i));
+            }
+            op1.getInputs().get(index).setValue(join);
+            context.computeAndSetTypeEnvironmentForOperator(join);
+            rewritten = true;
+        }
+        return rewritten;
+    }
+
+    /**
+     * rewrite NestedTupleSource operators to EmptyTupleSource operators
+     *
+     * @param nestedRootRef
+     */
+    private void rewriteNestedTupleSource(Mutable<ILogicalOperator> nestedRootRef, IOptimizationContext context)
+            throws AlgebricksException {
+        AbstractLogicalOperator nestedRoot = (AbstractLogicalOperator) nestedRootRef.getValue();
+        if (nestedRoot.getOperatorTag() == LogicalOperatorTag.NESTEDTUPLESOURCE) {
+            ILogicalOperator ets = new EmptyTupleSourceOperator();
+            nestedRootRef.setValue(ets);
+            context.computeAndSetTypeEnvironmentForOperator(ets);
+        }
+        List<Mutable<ILogicalOperator>> inputs = nestedRoot.getInputs();
+        for (Mutable<ILogicalOperator> input : inputs) {
+            rewriteNestedTupleSource(input, context);
+        }
+    }
+}


Mime
View raw message