// 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.impala.planner;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Set;

import org.apache.impala.analysis.AggregateInfo;
import org.apache.impala.analysis.AnalyticInfo;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BaseTableRef;
import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.BinaryPredicate.Operator;
import org.apache.impala.analysis.CollectionTableRef;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ExprId;
import org.apache.impala.analysis.ExprSubstitutionMap;
import org.apache.impala.analysis.InlineViewRef;
import org.apache.impala.analysis.JoinOperator;
import org.apache.impala.analysis.MultiAggregateInfo;
import org.apache.impala.analysis.MultiAggregateInfo.AggPhase;
import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.QueryStmt;
import org.apache.impala.analysis.SelectStmt;
import org.apache.impala.analysis.SingularRowSrcTableRef;
import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotId;
import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.SortInfo;
import org.apache.impala.analysis.TableRef;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.analysis.TupleId;
import org.apache.impala.analysis.TupleIsNullPredicate;
import org.apache.impala.analysis.UnionStmt;
import org.apache.impala.analysis.UnionStmt.UnionOperand;
import org.apache.impala.catalog.ColumnStats;
import org.apache.impala.catalog.FeDataSourceTable;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.common.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

/**
 * Constructs a non-executable single-node plan from an analyzed parse tree.
 * The single-node plan does not contain data exchanges or data-reduction optimizations
 * such as local aggregations that are important for distributed execution.
 * The single-node plan needs to be wrapped in a plan fragment for it to be executable.
 */
public class SingleNodePlanner {
  private final static Logger LOG = LoggerFactory.getLogger(SingleNodePlanner.class);

  private final PlannerContext ctx_;

  public SingleNodePlanner(PlannerContext ctx) {
    ctx_ = ctx;
  }

  /**
   * Generates and returns the root of the single-node plan for the analyzed parse tree
   * in the planner context. The planning process recursively walks the parse tree and
   * performs the following actions.
   * In the top-down phase over query statements:
   * - Materialize the slots required for evaluating expressions of that statement.
   * - Migrate conjuncts from parent blocks into inline views and union operands.
   * In the bottom-up phase generate the plan tree for every query statement:
   * - Generate the plan for the FROM-clause of a select statement: The plan trees of
   *   absolute and uncorrelated table refs are connected via JoinNodes. The relative
   *   and correlated table refs are associated with one or more SubplanNodes.
   * - A new SubplanNode is placed on top of an existing plan node whenever the tuples
   *   materialized by that plan node enable evaluation of one or more relative or
   *   correlated table refs, i.e., SubplanNodes are placed at the lowest possible point
   *   in the plan, often right after a ScanNode materializing the (single) parent tuple.
   * - The right-hand side of each SubplanNode is a plan tree generated by joining a
   *   SingularRowSrcTableRef with those applicable relative and correlated refs.
   *   A SingularRowSrcTableRef represents the current row being processed by the
   *   SubplanNode from its input (first child).
   * - Connecting table ref plans via JoinNodes is done in a cost-based fashion
   *   (join-order optimization). All materialized slots, including those of tuples
   *   materialized inside a SubplanNode, must be known for an accurate estimate of row
   *   sizes needed for cost-based join ordering.
   * - The remaining aggregate/analytic/orderby portions of a select statement are added
   *   on top of the FROM-clause plan.
   * - Whenever a new node is added to the plan tree, assign conjuncts that can be
   *   evaluated at that node and compute the stats of that node (cardinality, etc.).
   * - Apply combined expression substitution map of child plan nodes; if a plan node
   *   re-maps its input, set a substitution map to be applied by parents.
   */
  public PlanNode createSingleNodePlan() throws ImpalaException {
    QueryStmt queryStmt = ctx_.getQueryStmt();
    // Use the stmt's analyzer which is not necessarily the root analyzer
    // to detect empty result sets.
    Analyzer analyzer = queryStmt.getAnalyzer();
    analyzer.computeValueTransferGraph();
    ctx_.getTimeline().markEvent("Value transfer graph computed");

    // Mark slots referenced by output exprs as materialized, prior to generating the
    // plan tree.
    // We need to mark the result exprs of the topmost select block as materialized, so
    // that PlanNode.init() can compute the final mem layout of materialized tuples
    // (the byte size of tuples is needed for cost computations).
    // TODO: instead of materializing everything produced by the plan root, derive
    // referenced slots from destination fragment and add a materialization node
    // if not all output is needed by destination fragment
    // TODO 2: should the materialization decision be cost-based?
    if (queryStmt.getBaseTblResultExprs() != null) {
      analyzer.materializeSlots(queryStmt.getBaseTblResultExprs());
    }

    if (LOG.isTraceEnabled()) {
      LOG.trace("desctbl: " + analyzer.getDescTbl().debugString());
    }
    PlanNode singleNodePlan = createQueryPlan(queryStmt, analyzer,
        ctx_.getQueryOptions().isDisable_outermost_topn());
    Preconditions.checkNotNull(singleNodePlan);
    return singleNodePlan;
  }

  /**
   * Checks that the given single-node plan is executable:
   * - It may not contain right or full outer joins with no equi-join conjuncts that
   *   are not inside the right child of a SubplanNode.
   * Throws a NotImplementedException if plan validation fails.
   */
  public void validatePlan(PlanNode planNode) throws NotImplementedException {
    // Any join can run in a single-node plan.
    if (ctx_.isSingleNodeExec()) return;

    if (planNode instanceof NestedLoopJoinNode) {
      JoinNode joinNode = (JoinNode) planNode;
      JoinOperator joinOp = joinNode.getJoinOp();
      if ((joinOp.isRightSemiJoin() || joinOp.isFullOuterJoin()
           || joinOp == JoinOperator.RIGHT_OUTER_JOIN)
          && joinNode.getEqJoinConjuncts().isEmpty()) {
        throw new NotImplementedException(String.format("Error generating a valid " +
            "execution plan for this query. A %s type with no equi-join " +
            "predicates can only be executed with a single node plan.",
            joinOp.toString()));
      }
    }

    if (planNode instanceof SubplanNode) {
      // Right and full outer joins with no equi-join conjuncts are ok in the right
      // child of a SubplanNode.
      validatePlan(planNode.getChild(0));
    } else {
      for (PlanNode child: planNode.getChildren()) {
        validatePlan(child);
      }
    }
  }

  /**
   * Returns true if there is a join in the plan outside of the right branch of a
   * subplan. This specific behaviour maintains compatibility with older
   * validatePlan() logic that allowed joins with mt_dop only in this specific case
   * (presumably by accident).
   */
  public boolean hasUnsupportedMtDopJoin(PlanNode planNode) {
    if (planNode instanceof JoinNode) return true;

    if (planNode instanceof SubplanNode) {
      return hasUnsupportedMtDopJoin(planNode.getChild(0));
    }

    for (PlanNode child : planNode.getChildren()) {
      if (hasUnsupportedMtDopJoin(child)) return true;
    }
    return false;
  }

  /**
   * Creates an EmptyNode that 'materializes' the tuples of the given stmt.
   * Marks all collection-typed slots referenced in stmt as non-materialized because
   * they are never unnested, and therefore the corresponding parent scan should not
   * materialize them.
   */
  private PlanNode createEmptyNode(QueryStmt stmt, Analyzer analyzer) {
    List<TupleId> tupleIds = new ArrayList<>();
    stmt.getMaterializedTupleIds(tupleIds);
    if (tupleIds.isEmpty()) {
      // Constant selects do not have materialized tuples at this stage.
      Preconditions.checkState(stmt instanceof SelectStmt,
          "Only constant selects should have no materialized tuples");
      SelectStmt selectStmt = (SelectStmt)stmt;
      Preconditions.checkState(selectStmt.getTableRefs().isEmpty());
      tupleIds.add(createResultTupleDescriptor(selectStmt, "empty", analyzer).getId());
    }
    unmarkCollectionSlots(stmt);
    EmptySetNode node = new EmptySetNode(ctx_.getNextNodeId(), tupleIds);
    node.init(analyzer);
    // Set the output smap to resolve exprs referencing inline views within stmt.
    // Not needed for a UnionStmt because it materializes its input operands.
    if (stmt instanceof SelectStmt) {
      node.setOutputSmap(((SelectStmt) stmt).getBaseTblSmap());
    }
    return node;
  }

  /**
   * Mark all collection-typed slots in stmt as non-materialized.
   */
  private void unmarkCollectionSlots(QueryStmt stmt) {
    List<TableRef> tblRefs = new ArrayList<>();
    stmt.collectFromClauseTableRefs(tblRefs);
    for (TableRef ref: tblRefs) {
      if (!ref.isRelative()) continue;
      Preconditions.checkState(ref instanceof CollectionTableRef);
      CollectionTableRef collTblRef = (CollectionTableRef) ref;
      Expr collExpr = collTblRef.getCollectionExpr();
      Preconditions.checkState(collExpr instanceof SlotRef);
      SlotRef collSlotRef = (SlotRef) collExpr;
      collSlotRef.getDesc().setIsMaterialized(false);
      // Re-compute the mem layout if necessary. The tuple may not have a mem layout if
      // no plan has been generated for the TableRef (e.g. due to limit 0 or similar).
      collSlotRef.getDesc().getParent().recomputeMemLayout();
    }
  }

  /**
   * Create plan tree for single-node execution. Generates PlanNodes for the
   * Select/Project/Join/Union [All]/Group by/Having/Order by clauses of the query stmt.
   */
  private PlanNode createQueryPlan(QueryStmt stmt, Analyzer analyzer, boolean disableTopN)
      throws ImpalaException {
    if (analyzer.hasEmptyResultSet()) return createEmptyNode(stmt, analyzer);

    PlanNode root;
    if (stmt instanceof SelectStmt) {
      SelectStmt selectStmt = (SelectStmt) stmt;
      root = createSelectPlan(selectStmt, analyzer);

      // insert possible AnalyticEvalNode before SortNode
      if (((SelectStmt) stmt).getAnalyticInfo() != null) {
        AnalyticInfo analyticInfo = selectStmt.getAnalyticInfo();
        AnalyticPlanner analyticPlanner =
            new AnalyticPlanner(analyticInfo, analyzer, ctx_);
        MultiAggregateInfo multiAggInfo = selectStmt.getMultiAggInfo();
        List<Expr> groupingExprs;
        if (multiAggInfo != null) {
          groupingExprs = multiAggInfo.getSubstGroupingExprs();
          Preconditions.checkState(groupingExprs != null);
        } else {
          groupingExprs = Collections.emptyList();
        }
        List<Expr> inputPartitionExprs = new ArrayList<>();
        root = analyticPlanner.createSingleNodePlan(
            root, groupingExprs, inputPartitionExprs);
        if (multiAggInfo != null && !inputPartitionExprs.isEmpty()
            && multiAggInfo.getMaterializedAggClasses().size() == 1) {
          // analytic computation will benefit from a partition on inputPartitionExprs
          multiAggInfo.getMaterializedAggClass(0).setPartitionExprs(inputPartitionExprs);
        }
      }
    } else {
      Preconditions.checkState(stmt instanceof UnionStmt);
      root = createUnionPlan((UnionStmt) stmt, analyzer);
    }

    // Avoid adding a sort node if the sort tuple has no materialized slots.
    boolean sortHasMaterializedSlots = false;
    if (stmt.evaluateOrderBy()) {
      for (SlotDescriptor sortSlotDesc:
        stmt.getSortInfo().getSortTupleDescriptor().getSlots()) {
        if (sortSlotDesc.isMaterialized()) {
          sortHasMaterializedSlots = true;
          break;
        }
      }
    }

    if (stmt.evaluateOrderBy() && sortHasMaterializedSlots) {
      root = createSortNode(analyzer, root, stmt.getSortInfo(), stmt.getLimit(),
          stmt.getOffset(), stmt.hasLimit(), disableTopN);
    } else {
      root.setLimit(stmt.getLimit());
      root.computeStats(analyzer);
    }

    return root;
  }

  /**
   * Creates and initializes either a SortNode or a TopNNode depending on various
   * heuristics and configuration parameters.
   */
  private SortNode createSortNode(Analyzer analyzer, PlanNode root, SortInfo sortInfo,
      long limit, long offset, boolean hasLimit, boolean disableTopN)
      throws ImpalaException {
    SortNode sortNode;
    long topNBytesLimit = ctx_.getQueryOptions().topn_bytes_limit;

    if (hasLimit && !disableTopN) {
      if (topNBytesLimit <= 0) {
        sortNode =
            SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
      } else {
        long topNCardinality = PlanNode.capCardinalityAtLimit(root.cardinality_, limit);
        long estimatedTopNMaterializedSize =
            sortInfo.estimateTopNMaterializedSize(topNCardinality, offset);

        if (estimatedTopNMaterializedSize < topNBytesLimit) {
          sortNode =
              SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
        } else {
          sortNode =
              SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
        }
      }
    } else {
      sortNode =
          SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
    }
    Preconditions.checkState(sortNode.hasValidStats());
    sortNode.setLimit(limit);
    sortNode.init(analyzer);
    return sortNode;
  }

  /**
   * If there are unassigned conjuncts that are bound by tupleIds or if there are slot
   * equivalences for tupleIds that have not yet been enforced, returns a SelectNode on
   * top of root that evaluates those conjuncts; otherwise returns root unchanged.
   * TODO: change this to assign the unassigned conjuncts to root itself, if that is
   * semantically correct
   */
  private PlanNode addUnassignedConjuncts(
      Analyzer analyzer, List<TupleId> tupleIds, PlanNode root) {
    // No point in adding SelectNode on top of an EmptyNode.
    if (root instanceof EmptySetNode) return root;
    Preconditions.checkNotNull(root);
    // Gather unassigned conjuncts and generate predicates to enforce
    // slot equivalences for each tuple id.
    List<Expr> conjuncts = analyzer.getUnassignedConjuncts(root);
    if (LOG.isTraceEnabled()) {
      LOG.trace(String.format("unassigned conjuncts for (Node %s): %s",
          root.getDisplayLabel(), Expr.debugString(conjuncts)));
      LOG.trace("all conjuncts: " + analyzer.conjunctAssignmentsDebugString());
    }
    for (TupleId tid: tupleIds) {
      analyzer.createEquivConjuncts(tid, conjuncts);
    }
    if (conjuncts.isEmpty()) return root;
    // evaluate conjuncts in SelectNode
    SelectNode selectNode = new SelectNode(ctx_.getNextNodeId(), root, conjuncts);
    // init() marks conjuncts as assigned
    selectNode.init(analyzer);
    Preconditions.checkState(selectNode.hasValidStats());
    return selectNode;
  }

  /**
   * Return the cheapest plan that materializes the joins of all TableRefs in
   * parentRefPlans and the subplans of all applicable TableRefs in subplanRefs.
   * Assumes that parentRefPlans are in the order as they originally appeared in
   * the query.
   * For this plan:
   * - the plan is executable, ie, all non-cross joins have equi-join predicates
   * - the leftmost scan is over the largest of the inputs for which we can still
   *   construct an executable plan
   * - from bottom to top, all rhs's are in increasing order of selectivity (percentage
   *   of surviving rows)
   * - outer/cross/semi joins: rhs serialized size is < lhs serialized size;
   *   enforced via join inversion, if necessary
   * - SubplanNodes are placed as low as possible in the plan tree - as soon as the
   *   required tuple ids of one or more TableRefs in subplanRefs are materialized
   * Returns null if we can't create an executable plan.
   */
  private PlanNode createCheapestJoinPlan(Analyzer analyzer,
      List<Pair<TableRef, PlanNode>> parentRefPlans, List<SubplanRef> subplanRefs)
      throws ImpalaException {
    LOG.trace("createCheapestJoinPlan");
    if (parentRefPlans.size() == 1) return parentRefPlans.get(0).second;

    // collect eligible candidates for the leftmost input; list contains
    // (plan, materialized size)
    List<Pair<TableRef, Long>> candidates = new ArrayList<>();
    for (Pair<TableRef, PlanNode> entry: parentRefPlans) {
      TableRef ref = entry.first;
      JoinOperator joinOp = ref.getJoinOp();

      // Avoid reordering outer/semi joins which is generally incorrect.
      // TODO: Allow the rhs of any cross join as the leftmost table. This needs careful
      // consideration of the joinOps that result from such a re-ordering (IMPALA-1281).
      if (joinOp.isOuterJoin() || joinOp.isSemiJoin() || joinOp.isCrossJoin()) continue;

      PlanNode plan = entry.second;
      if (plan.getCardinality() == -1) {
        // use 0 for the size to avoid it becoming the leftmost input
        // TODO: Consider raw size of scanned partitions in the absence of stats.
        candidates.add(new Pair<TableRef, Long>(ref, new Long(0)));
        if (LOG.isTraceEnabled()) {
          LOG.trace("candidate " + ref.getUniqueAlias() + ": 0");
        }
        continue;
      }
      Preconditions.checkState(ref.isAnalyzed());
      long materializedSize =
          (long) Math.ceil(plan.getAvgRowSize() * (double) plan.getCardinality());
      candidates.add(new Pair<TableRef, Long>(ref, new Long(materializedSize)));
      if (LOG.isTraceEnabled()) {
        LOG.trace(
            "candidate " + ref.getUniqueAlias() + ": " + Long.toString(materializedSize));
      }
    }
    if (candidates.isEmpty()) return null;

    // order candidates by descending materialized size; we want to minimize the memory
    // consumption of the materialized hash tables required for the join sequence
    Collections.sort(candidates,
        new Comparator<Pair<TableRef, Long>>() {
          @Override
          public int compare(Pair<TableRef, Long> a, Pair<TableRef, Long> b) {
            long diff = b.second - a.second;
            return (diff < 0 ? -1 : (diff > 0 ? 1 : 0));
          }
        });

    for (Pair<TableRef, Long> candidate: candidates) {
      PlanNode result = createJoinPlan(analyzer, candidate.first, parentRefPlans, subplanRefs);
      if (result != null) return result;
    }
    return null;
  }

  /**
   * Returns a plan with leftmostRef's plan as its leftmost input; the joins
   * are in decreasing order of selectiveness (percentage of rows they eliminate).
   * Creates and adds subplan nodes as soon as the tuple ids required by at least one
   * subplan ref are materialized by a join node added during plan generation.
   */
  private PlanNode createJoinPlan(Analyzer analyzer, TableRef leftmostRef,
      List<Pair<TableRef, PlanNode>> refPlans, List<SubplanRef> subplanRefs)
      throws ImpalaException {

    if (LOG.isTraceEnabled()) {
      LOG.trace("createJoinPlan: " + leftmostRef.getUniqueAlias());
    }
    // the refs that have yet to be joined
    List<Pair<TableRef, PlanNode>> remainingRefs = new ArrayList<>();
    PlanNode root = null;  // root of accumulated join plan
    for (Pair<TableRef, PlanNode> entry: refPlans) {
      if (entry.first == leftmostRef) {
        root = entry.second;
      } else {
        remainingRefs.add(entry);
      }
    }
    Preconditions.checkNotNull(root);

    // Maps from a TableRef in refPlans with an outer/semi join op to the set of
    // TableRefs that precede it refPlans (i.e., in FROM-clause order).
    // The map is used to place outer/semi joins at a fixed position in the plan tree
    // (IMPALA-860), s.t. all the tables appearing to the left/right of an outer/semi
    // join in the original query still remain to the left/right after join ordering.
    // This prevents join re-ordering across outer/semi joins which is generally wrong.
    Map<TableRef, Set<TableRef>> precedingRefs = new HashMap<>();
    List<TableRef> tmpTblRefs = new ArrayList<>();
    for (Pair<TableRef, PlanNode> entry: refPlans) {
      TableRef tblRef = entry.first;
      if (tblRef.getJoinOp().isOuterJoin() || tblRef.getJoinOp().isSemiJoin()) {
        precedingRefs.put(tblRef, Sets.newHashSet(tmpTblRefs));
      }
      tmpTblRefs.add(tblRef);
    }

    // Refs that have been joined. The union of joinedRefs and the refs in remainingRefs
    // are the set of all table refs.
    Set<TableRef> joinedRefs = Sets.newHashSet(leftmostRef);
    long numOps = 0;
    int i = 0;
    while (!remainingRefs.isEmpty()) {
      // We minimize the resulting cardinality at each step in the join chain,
      // which minimizes the total number of hash table lookups.
      PlanNode newRoot = null;
      Pair<TableRef, PlanNode> minEntry = null;
      for (Pair<TableRef, PlanNode> entry: remainingRefs) {
        TableRef ref = entry.first;
        JoinOperator joinOp = ref.getJoinOp();

        // Place outer/semi joins at a fixed position in the plan tree.
        Set<TableRef> requiredRefs = precedingRefs.get(ref);
        if (requiredRefs != null) {
          Preconditions.checkState(joinOp.isOuterJoin() || joinOp.isSemiJoin());
          // If the required table refs have not been placed yet, do not even consider
          // the remaining table refs to prevent incorrect re-ordering of tables across
          // outer/semi joins.
          if (!requiredRefs.equals(joinedRefs)) break;
        }

        analyzer.setAssignedConjuncts(root.getAssignedConjuncts());
        PlanNode candidate = createJoinNode(root, entry.second, ref, analyzer);
        if (candidate == null) continue;
        if (LOG.isTraceEnabled()) {
          LOG.trace("cardinality=" + Long.toString(candidate.getCardinality()));
        }

        // Use 'candidate' as the new root; don't consider any other table refs at this
        // position in the plan.
        if (joinOp.isOuterJoin() || joinOp.isSemiJoin()) {
          newRoot = candidate;
          minEntry = entry;
          break;
        }

        // Always prefer Hash Join over Nested-Loop Join due to limited costing
        // infrastructure.
        if (newRoot == null
            || (candidate.getClass().equals(newRoot.getClass())
                && candidate.getCardinality() < newRoot.getCardinality())
            || (candidate instanceof HashJoinNode
                && newRoot instanceof NestedLoopJoinNode)) {
          newRoot = candidate;
          minEntry = entry;
        }
      }
      if (newRoot == null) {
        // Could not generate a valid plan.
        return null;
      }

      // we need to insert every rhs row into the hash table and then look up
      // every lhs row
      long lhsCardinality = root.getCardinality();
      long rhsCardinality = minEntry.second.getCardinality();
      numOps += lhsCardinality + rhsCardinality;
      if (LOG.isTraceEnabled()) {
        LOG.trace(Integer.toString(i) + " chose " + minEntry.first.getUniqueAlias()
            + " #lhs=" + Long.toString(lhsCardinality)
            + " #rhs=" + Long.toString(rhsCardinality)
            + " #ops=" + Long.toString(numOps));
      }
      remainingRefs.remove(minEntry);
      joinedRefs.add(minEntry.first);
      root = newRoot;
      // Create a Subplan on top of the new root for all the subplan refs that can be
      // evaluated at this point.
      // TODO: Once we have stats on nested collections, we should consider the join
      // order in conjunction with the placement of SubplanNodes, i.e., move the creation
      // of SubplanNodes into the join-ordering loop above.
      root = createSubplan(root, subplanRefs, false, analyzer);
      // assign node ids after running through the possible choices in order to end up
      // with a dense sequence of node ids
      if (root instanceof SubplanNode) root.getChild(0).setId(ctx_.getNextNodeId());
      root.setId(ctx_.getNextNodeId());
      analyzer.setAssignedConjuncts(root.getAssignedConjuncts());
      ++i;
    }

    return root;
  }

  /**
   * Return a plan with joins in the order of parentRefPlans (= FROM clause order).
   * Adds coalesced SubplanNodes based on the FROM-clause order of subplanRefs.
   */
  private PlanNode createFromClauseJoinPlan(Analyzer analyzer,
      List<Pair<TableRef, PlanNode>> parentRefPlans, List<SubplanRef> subplanRefs)
      throws ImpalaException {
    // create left-deep sequence of binary hash joins; assign node ids as we go along
    Preconditions.checkState(!parentRefPlans.isEmpty());
    PlanNode root = parentRefPlans.get(0).second;
    for (int i = 1; i < parentRefPlans.size(); ++i) {
      TableRef innerRef = parentRefPlans.get(i).first;
      PlanNode innerPlan = parentRefPlans.get(i).second;
      root = createJoinNode(root, innerPlan, innerRef, analyzer);
      if (root != null) root = createSubplan(root, subplanRefs, false, analyzer);
      if (root instanceof SubplanNode) root.getChild(0).setId(ctx_.getNextNodeId());
      root.setId(ctx_.getNextNodeId());
    }
    return root;
  }

  /**
   * Create tree of PlanNodes that implements the Select/Project/Join/Group by/Having
   * of the selectStmt query block.
   */
  private PlanNode createSelectPlan(SelectStmt selectStmt, Analyzer analyzer)
      throws ImpalaException {
    // no from clause -> materialize the select's exprs with a UnionNode
    if (selectStmt.getTableRefs().isEmpty()) {
      return createConstantSelectPlan(selectStmt, analyzer);
    }

    // Slot materialization:
    // We need to mark all slots as materialized that are needed during the execution
    // of selectStmt, and we need to do that prior to creating plans for the TableRefs
    // (because createTableRefNode() might end up calling computeMemLayout() on one or
    // more TupleDescriptors, at which point all referenced slots need to be marked).
    //
    // For non-join predicates, slots are marked as follows:
    // - for base table scan predicates, this is done directly by ScanNode.init(), which
    //   can do a better job because it doesn't need to materialize slots that are only
    //   referenced for partition pruning, for instance
    // - for inline views, non-join predicates are pushed down, at which point the
    //   process repeats itself.
    selectStmt.materializeRequiredSlots(analyzer);

    List<TupleId> rowTuples = new ArrayList<>();
    // collect output tuples of subtrees
    for (TableRef tblRef: selectStmt.getTableRefs()) {
      rowTuples.addAll(tblRef.getMaterializedTupleIds());
    }

    // If the selectStmt's select-project-join portion returns an empty result set
    // create a plan that feeds the aggregation of selectStmt with an empty set.
    // Make sure the slots of the aggregation exprs and the tuples that they reference
    // are materialized (see IMPALA-1960). Marks all collection-typed slots referenced
    // in this select stmt as non-materialized because they are never unnested. Note that
    // this creates extra unused space in the tuple since the mem layout has already been
    // computed.
    if (analyzer.hasEmptySpjResultSet()) {
      unmarkCollectionSlots(selectStmt);
      PlanNode emptySetNode = new EmptySetNode(ctx_.getNextNodeId(), rowTuples);
      emptySetNode.init(analyzer);
      emptySetNode.setOutputSmap(selectStmt.getBaseTblSmap());
      return createAggregationPlan(selectStmt, analyzer, emptySetNode);
    }

    // Separate table refs into parent refs (uncorrelated or absolute) and
    // subplan refs (correlated or relative), and generate their plan.
    List<TableRef> parentRefs = new ArrayList<>();
    List<SubplanRef> subplanRefs = new ArrayList<>();
    computeParentAndSubplanRefs(
        selectStmt.getTableRefs(), analyzer.isStraightJoin(), parentRefs, subplanRefs);
    MultiAggregateInfo multiAggInfo = selectStmt.getMultiAggInfo();
    // Only optimize scan/agg plan if there is a single aggregation class.
    AggregateInfo scanAggInfo = null;
    if (multiAggInfo != null && multiAggInfo.getMaterializedAggClasses().size() == 1) {
      scanAggInfo = multiAggInfo.getMaterializedAggClass(0);
    }
    PlanNode root = createTableRefsPlan(parentRefs, subplanRefs, scanAggInfo, analyzer);
    // Add aggregation, if any.
    if (multiAggInfo != null) {
      // Apply substitution for optimized scan/agg plan,
      if (scanAggInfo != null) {
        if (root instanceof HdfsScanNode) {
          scanAggInfo.substitute(((HdfsScanNode) root).getOptimizedAggSmap(), analyzer);
          scanAggInfo.getMergeAggInfo().substitute(
              ((HdfsScanNode) root).getOptimizedAggSmap(), analyzer);
        } else if (root instanceof KuduScanNode) {
          scanAggInfo.substitute(((KuduScanNode) root).getOptimizedAggSmap(), analyzer);
          scanAggInfo.getMergeAggInfo().substitute(
              ((KuduScanNode) root).getOptimizedAggSmap(), analyzer);
        }
      }
      root = createAggregationPlan(selectStmt, analyzer, root);
    }

    // All the conjuncts_ should be assigned at this point.
    // TODO: Re-enable this check here and/or elswehere.
    //Preconditions.checkState(!analyzer.hasUnassignedConjuncts());
    return root;
  }

  /**
   * Holds a table ref that must be evaluated inside a subplan (i.e., a relative or
   * correlated ref), along with the materialized tuple ids and table ref ids that
   * are required for this table ref to be correctly evaluated inside a SubplanNode.
   *
   * Required materialized tuple ids:
   * These ensure that the SubplanNode evaluating this table ref is placed only once all
   * root tuples needed by this table ref or relative refs contained in this table ref
   * are materialized.
   *
   * Required table ref ids:
   * These ensure that the SubplanNode evaluating this table ref is placed correctly
   * with respect to join ordering, in particular, that the SubplanNode is not ordered
   * across semi/outer joins.
   */
  private static class SubplanRef {
    // Relative or correlated table ref.
    public final TableRef tblRef;

    // List of tuple ids that must be materialized before 'tblRef' can be
    // correctly evaluated inside a SubplanNode.
    public final List<TupleId> requiredTids;

    // List of table ref ids that a plan tree must contain before 'tblRef'
    // can be correctly evaluated inside a SubplanNode.
    public final List<TupleId> requiredTblRefIds;

    public SubplanRef(TableRef tblRef, List<TupleId> requiredTids,
        List<TupleId> requiredTblRefIds) {
      Preconditions.checkState(tblRef.isRelative() || tblRef.isCorrelated());
      this.tblRef = tblRef;
      this.requiredTids = requiredTids;
      this.requiredTblRefIds = requiredTblRefIds;
    }
  }

  /**
   * Separates tblRefs into the following two lists.
   *
   * parentRefs:
   * Uncorrelated and non-relative table refs. These are the 'regular' table refs whose
   * plans are connected by join nodes, and are not placed inside a Subplan. The returned
   * parentRefs are self-contained with respect to TableRef linking, i.e., each returned
   * TableRef has its left TableRef link set to the TableRef preceding it in parentRefs.
   *
   * subplanRefs:
   * Correlated and relative table refs. The plan of such refs must be put inside a
   * Subplan. See SubplanRef for details. The left TableRef link of the TableRefs in
   * returned SubplanRefs are set to null.
   * If isStraightJoin is true, then the required tuple ids and table ref ids of a
   * correlated or relative ref are simply those of all table refs preceding it in
   * the FROM-clause order.
   *
   * If this function is called when generating the right-hand side of a SubplanNode,
   * then correlated and relative table refs that require only tuples produced by the
   * SubplanNode's input are placed inside parentRefs.
   */
  private void computeParentAndSubplanRefs(List<TableRef> tblRefs,
      boolean isStraightJoin, List<TableRef> parentRefs, List<SubplanRef> subplanRefs) {
    // List of table ref ids materialized so far during plan generation, including those
    // from the subplan context, if any. We append the ids of table refs placed into
    // parentRefs to this list to satisfy the ordering requirement of subsequent
    // table refs that should also be put into parentRefs. Consider this example:
    // FROM t, (SELECT ... FROM t.c1 LEFT JOIN t.c2 ON(...) JOIN t.c3 ON (...)) v
    // Table ref t.c3 has an ordering dependency on t.c2 due to the outer join, but t.c3
    // must be placed into the subplan that materializes t.c1 and t.c2.
    List<TupleId> planTblRefIds = new ArrayList<>();

    // List of materialized tuple ids in the subplan context, if any. This list must
    // remain constant in this function because the subplan context is fixed. Any
    // relative or correlated table ref that requires a materialized tuple id produced
    // by an element in tblRefs should be placed into subplanRefs because it requires
    // a new subplan context. Otherwise, it should be placed into parentRefs.
    List<TupleId> subplanTids = Collections.emptyList();

    if (ctx_.hasSubplan()) {
      // Add all table ref ids from the subplan context.
      planTblRefIds.addAll(ctx_.getSubplan().getChild(0).getTblRefIds());
      subplanTids =
          Collections.unmodifiableList(ctx_.getSubplan().getChild(0).getTupleIds());
    }

    // Table ref representing the last outer or semi join we have seen.
    TableRef lastSemiOrOuterJoin = null;
    for (TableRef ref: tblRefs) {
      boolean isParentRef = true;
      if (ref.isRelative() || ref.isCorrelated()) {
        List<TupleId> requiredTids = new ArrayList<>();
        List<TupleId> requiredTblRefIds = new ArrayList<>();
        if (ref.isCorrelated()) {
          requiredTids.addAll(ref.getCorrelatedTupleIds());
        } else {
          CollectionTableRef collectionTableRef = (CollectionTableRef) ref;
          requiredTids.add(collectionTableRef.getResolvedPath().getRootDesc().getId());
        }
        // Add all plan table ref ids as an ordering dependency for straight_join.
        if (isStraightJoin) requiredTblRefIds.addAll(planTblRefIds);
        if (lastSemiOrOuterJoin != null) {
          // Prevent incorrect join re-ordering across outer/semi joins by requiring all
          // table ref ids to the left and including the last outer/semi join.
          // TODO: Think about when we can allow re-ordering across semi/outer joins
          // in subplans.
          requiredTblRefIds.addAll(lastSemiOrOuterJoin.getAllTableRefIds());
        }
        if (!subplanTids.containsAll(requiredTids)) {
          isParentRef = false;
          // Outer and semi joins are placed at a fixed position in the join order.
          // They require that all tables to their left are materialized.
          if (ref.getJoinOp().isOuterJoin() || ref.getJoinOp().isSemiJoin()) {
            requiredTblRefIds.addAll(ref.getAllTableRefIds());
            requiredTblRefIds.remove(ref.getId());
          }
          subplanRefs.add(new SubplanRef(ref, requiredTids, requiredTblRefIds));
        }
      }
      if (isParentRef) {
        parentRefs.add(ref);
        planTblRefIds.add(ref.getId());
      }
      if (ref.getJoinOp().isOuterJoin() || ref.getJoinOp().isSemiJoin()) {
        lastSemiOrOuterJoin = ref;
      }
    }
    Preconditions.checkState(tblRefs.size() == parentRefs.size() + subplanRefs.size());

    // Fix the chain of parent table refs and set the left table of all subplanRefs to
    // null. This step needs to be done outside of the loop above because the left links
    // are required for getAllTupleIds() used for determining the requiredTblRefIds.
    parentRefs.get(0).setLeftTblRef(null);
    for (int i = 1; i < parentRefs.size(); ++i) {
      parentRefs.get(i).setLeftTblRef(parentRefs.get(i - 1));
    }
    for (SubplanRef subplanRef: subplanRefs) subplanRef.tblRef.setLeftTblRef(null);
  }

  /**
   * Returns a plan tree for evaluating the given parentRefs and subplanRefs.
   */
  private PlanNode createTableRefsPlan(List<TableRef> parentRefs,
      List<SubplanRef> subplanRefs, AggregateInfo aggInfo, Analyzer analyzer)
      throws ImpalaException {
    // create plans for our table refs; use a list here instead of a map to
    // maintain a deterministic order of traversing the TableRefs during join
    // plan generation (helps with tests)
    List<Pair<TableRef, PlanNode>> parentRefPlans = new ArrayList<>();
    for (TableRef ref: parentRefs) {
      PlanNode root = createTableRefNode(ref, aggInfo, analyzer);
      Preconditions.checkNotNull(root);
      root = createSubplan(root, subplanRefs, true, analyzer);
      parentRefPlans.add(new Pair<TableRef, PlanNode>(ref, root));
    }
    // save state of conjunct assignment; needed for join plan generation
    for (Pair<TableRef, PlanNode> entry: parentRefPlans) {
      entry.second.setAssignedConjuncts(analyzer.getAssignedConjuncts());
    }

    PlanNode root = null;
    if (!analyzer.isStraightJoin()) {
      Set<ExprId> assignedConjuncts = analyzer.getAssignedConjuncts();
      root = createCheapestJoinPlan(analyzer, parentRefPlans, subplanRefs);
      // If createCheapestJoinPlan() failed to produce an executable plan, then we need
      // to restore the original state of conjunct assignment for the straight-join plan
      // to not incorrectly miss conjuncts.
      if (root == null) analyzer.setAssignedConjuncts(assignedConjuncts);
    }
    if (analyzer.isStraightJoin() || root == null) {
      // we didn't have enough stats to do a cost-based join plan, or the STRAIGHT_JOIN
      // keyword was in the select list: use the FROM clause order instead
      root = createFromClauseJoinPlan(analyzer, parentRefPlans, subplanRefs);
      Preconditions.checkNotNull(root);
    }
    return root;
  }

  /**
   * Places a SubplanNode on top of 'root' that evaluates all the subplan refs that can
   * be correctly evaluated from 'root's materialized tuple ids. Returns 'root' if there
   * are no applicable subplan refs.
   * Assigns the returned SubplanNode a new node id unless assignId is false.
   *
   * If applicable, the SubplanNode is created as follows:
   * - 'root' is the input of the SubplanNode (first child)
   * - the second child is the plan tree generated from these table refs:
   *   1. a SingularRowSrcTableRef that represents the current row being processed
   *      by the SubplanNode to be joined with
   *   2. all applicable subplan refs
   * - the second child plan tree is generated as usual with createTableRefsPlan()
   * - the plans of the applicable subplan refs are generated as usual, without a
   *   SingularRowSrcTableRef
   * - nested SubplanNodes are generated recursively inside createTableRefsPlan() by
   *   passing in the remaining subplanRefs that are not applicable after 'root'; some
   *   of those subplanRefs may become applicable inside the second child plan tree of
   *   the SubplanNode generated here
   */
  private PlanNode createSubplan(PlanNode root, List<SubplanRef> subplanRefs,
      boolean assignId, Analyzer analyzer) throws ImpalaException {
    Preconditions.checkNotNull(root);
    List<TableRef> applicableRefs = extractApplicableRefs(root, subplanRefs);
    if (applicableRefs.isEmpty()) return root;

    // Prepend a SingularRowSrcTableRef representing the current row being processed
    // by the SubplanNode from its input (first child).
    Preconditions.checkState(applicableRefs.get(0).getLeftTblRef() == null);
    applicableRefs.add(0, new SingularRowSrcTableRef(root));
    applicableRefs.get(1).setLeftTblRef(applicableRefs.get(0));

    // Construct an incomplete SubplanNode that only knows its input so we can push it
    // into the planner context. The subplan is set after the subplan tree has been
    // constructed.
    SubplanNode subplanNode = new SubplanNode(root);
    if (assignId) subplanNode.setId(ctx_.getNextNodeId());

    // Push the SubplanNode such that UnnestNodes and SingularRowSrcNodes can pick up
    // their containing SubplanNode. Also, further plan generation relies on knowing
    // whether we are in a subplan context or not (see computeParentAndSubplanRefs()).
    ctx_.pushSubplan(subplanNode);
    PlanNode subplan = createTableRefsPlan(applicableRefs, subplanRefs, null, analyzer);
    ctx_.popSubplan();
    subplanNode.setSubplan(subplan);
    subplanNode.init(analyzer);
    return subplanNode;
  }

  /**
   * Returns a new list with all table refs from subplanRefs that can be correctly
   * evaluated inside a SubplanNode placed after the given plan root.
   * The returned table refs have their left-table links properly set, and the
   * corresponding SubplanRefs are removed from subplanRefs.
   */
  private List<TableRef> extractApplicableRefs(PlanNode root,
      List<SubplanRef> subplanRefs) {
    // List of table ref ids in 'root' as well as the table ref ids of all table refs
    // placed in 'subplanRefs' so far.
    List<TupleId> tblRefIds = Lists.newArrayList(root.getTblRefIds());
    List<TableRef> result = new ArrayList<>();
    Iterator<SubplanRef> subplanRefIt = subplanRefs.iterator();
    TableRef leftTblRef = null;
    while (subplanRefIt.hasNext()) {
      SubplanRef subplanRef = subplanRefIt.next();
      // Ensure that 'root' materializes all required tuples (first condition), and that
      // correct join ordering is obeyed (second condition).
      if (root.getTupleIds().containsAll(subplanRef.requiredTids) &&
          tblRefIds.containsAll(subplanRef.requiredTblRefIds)) {
        subplanRef.tblRef.setLeftTblRef(leftTblRef);
        result.add(subplanRef.tblRef);
        leftTblRef = subplanRef.tblRef;
        subplanRefIt.remove();
        // Add the table ref id such that other subplan refs that can be evaluated inside
        // the same SubplanNode but only after this ref are returned as well.
        tblRefIds.add(subplanRef.tblRef.getId());
      }
    }
    return result;
  }

  /**
   * Returns a new AggregationNode that materializes the aggregation of the given stmt.
   * Assigns conjuncts from the Having clause to the returned node.
   */
  private AggregationNode createAggregationPlan(
      SelectStmt selectStmt, Analyzer analyzer, PlanNode root) throws ImpalaException {
    MultiAggregateInfo multiAggInfo =
        Preconditions.checkNotNull(selectStmt.getMultiAggInfo());
    AggregationNode result = createAggregationPlan(root, multiAggInfo, analyzer);
    ExprSubstitutionMap simplifiedAggSmap = multiAggInfo.getSimplifiedAggSmap();
    if (simplifiedAggSmap == null) return result;

    // Fix up aggregations that simplified to a single class after
    // materializeRequiredSlots().

    // Collect conjuncts and then re-assign them to the top-most aggregation node
    // of the simplified plan.
    AggregationNode dummyAgg = new AggregationNode(
        ctx_.getNextNodeId(), result, multiAggInfo, AggPhase.TRANSPOSE);
    dummyAgg.init(analyzer);
    List<Expr> conjuncts =
        Expr.substituteList(dummyAgg.getConjuncts(), simplifiedAggSmap, analyzer, true);
    // Validate conjuncts after substitution.
    for (Expr c : conjuncts) {
      Preconditions.checkState(c.isBound(result.getTupleIds().get(0)));
    }
    result.getConjuncts().addAll(conjuncts);

    // Apply simplification substitution in ancestors.
    result.setOutputSmap(
        ExprSubstitutionMap.compose(result.getOutputSmap(), simplifiedAggSmap, analyzer));
    return result;
  }

  private AggregationNode createAggregationPlan(PlanNode root,
      MultiAggregateInfo multiAggInfo, Analyzer analyzer) throws InternalException {
    Preconditions.checkNotNull(multiAggInfo);
    AggregationNode firstPhaseAgg =
        new AggregationNode(ctx_.getNextNodeId(), root, multiAggInfo, AggPhase.FIRST);
    firstPhaseAgg.init(analyzer);
    if (!multiAggInfo.hasSecondPhase()) return firstPhaseAgg;

    firstPhaseAgg.unsetNeedsFinalize();
    firstPhaseAgg.setIntermediateTuple();

    AggregationNode secondPhaseAgg = new AggregationNode(
        ctx_.getNextNodeId(), firstPhaseAgg, multiAggInfo, AggPhase.SECOND);
    secondPhaseAgg.init(analyzer);
    if (!multiAggInfo.hasTransposePhase()) return secondPhaseAgg;

    AggregationNode transposePhaseAgg = new AggregationNode(
        ctx_.getNextNodeId(), secondPhaseAgg, multiAggInfo, AggPhase.TRANSPOSE);
    transposePhaseAgg.init(analyzer);
    return transposePhaseAgg;
  }

 /**
  * Returns a UnionNode that materializes the exprs of the constant selectStmt.
  * Replaces the resultExprs of the selectStmt with SlotRefs into the materialized tuple.
  */
  private PlanNode createConstantSelectPlan(SelectStmt selectStmt, Analyzer analyzer)
      throws InternalException {
    Preconditions.checkState(selectStmt.getTableRefs().isEmpty());
    List<Expr> resultExprs = selectStmt.getResultExprs();
    // Create tuple descriptor for materialized tuple.
    TupleDescriptor tupleDesc = createResultTupleDescriptor(selectStmt, "union", analyzer);
    UnionNode unionNode = new UnionNode(ctx_.getNextNodeId(), tupleDesc.getId());
    // Analysis guarantees that selects without a FROM clause only have constant exprs.
    unionNode.addConstExprList(Lists.newArrayList(resultExprs));

    // Replace the select stmt's resultExprs with SlotRefs into tupleDesc.
    for (int i = 0; i < resultExprs.size(); ++i) {
      SlotRef slotRef = new SlotRef(tupleDesc.getSlots().get(i));
      resultExprs.set(i, slotRef);
    }
    // UnionNode.init() needs tupleDesc to have been initialized
    unionNode.init(analyzer);
    return unionNode;
  }

  /**
   * Create tuple descriptor that can hold the results of the given SelectStmt, with one
   * slot per result expr.
   */
  private TupleDescriptor createResultTupleDescriptor(SelectStmt selectStmt,
      String debugName, Analyzer analyzer) {
    TupleDescriptor tupleDesc = analyzer.getDescTbl().createTupleDescriptor(
        debugName);
    tupleDesc.setIsMaterialized(true);

    List<Expr> resultExprs = selectStmt.getResultExprs();
    List<String> colLabels = selectStmt.getColLabels();
    for (int i = 0; i < resultExprs.size(); ++i) {
      Expr resultExpr = resultExprs.get(i);
      String colLabel = colLabels.get(i);
      SlotDescriptor slotDesc = analyzer.addSlotDescriptor(tupleDesc);
      slotDesc.setLabel(colLabel);
      slotDesc.setSourceExpr(resultExpr);
      slotDesc.setType(resultExpr.getType());
      slotDesc.setStats(ColumnStats.fromExpr(resultExpr));
      slotDesc.setIsMaterialized(true);
    }
    tupleDesc.computeMemLayout();
    return tupleDesc;
  }

  /**
   * Returns plan tree for an inline view ref:
   * - predicates from the enclosing scope that can be evaluated directly within
   *   the inline-view plan are pushed down
   * - predicates that cannot be evaluated directly within the inline-view plan
   *   but only apply to the inline view are evaluated in a SelectNode placed
   *   on top of the inline view plan
   * - all slots that are referenced by predicates from the enclosing scope that cannot
   *   be pushed down are marked as materialized (so that when computeMemLayout() is
   *   called on the base table descriptors materialized by the inline view it has a
   *   complete picture)
   */
  private PlanNode createInlineViewPlan(Analyzer analyzer, InlineViewRef inlineViewRef)
      throws ImpalaException {
    // If possible, "push down" view predicates; this is needed in order to ensure
    // that predicates such as "x + y = 10" are evaluated in the view's plan tree
    // rather than a SelectNode grafted on top of that plan tree.
    // This doesn't prevent predicate propagation, because predicates like
    // "x = 10" that get pushed down are still connected to equivalent slots
    // via the equality predicates created for the view's select list.
    // Include outer join conjuncts here as well because predicates from the
    // On-clause of an outer join may be pushed into the inline view as well.
    migrateConjunctsToInlineView(analyzer, inlineViewRef);

    // Turn a constant select into a UnionNode that materializes the exprs.
    // TODO: unify this with createConstantSelectPlan(), this is basically the
    // same thing
    QueryStmt viewStmt = inlineViewRef.getViewStmt();
    if (viewStmt instanceof SelectStmt) {
      SelectStmt selectStmt = (SelectStmt) viewStmt;
      if (selectStmt.getTableRefs().isEmpty()) {
        if (inlineViewRef.getAnalyzer().hasEmptyResultSet()) {
          PlanNode emptySetNode = createEmptyNode(viewStmt, inlineViewRef.getAnalyzer());
          // Still substitute exprs in parent nodes with the inline-view's smap to make
          // sure no exprs reference the non-materialized inline view slots. No wrapping
          // with TupleIsNullPredicates is necessary here because we do not migrate
          // conjuncts into outer-joined inline views, so hasEmptyResultSet() cannot be
          // true for an outer-joined inline view that has no table refs.
          Preconditions.checkState(!analyzer.isOuterJoined(inlineViewRef.getId()));
          emptySetNode.setOutputSmap(inlineViewRef.getSmap());
          // The tblRef materialized by this node is still the 'inlineViewRef'.
          emptySetNode.setTblRefIds(Lists.newArrayList(inlineViewRef.getId()));
          return emptySetNode;
        }
        // Analysis should have generated a tuple id into which to materialize the exprs.
        Preconditions.checkState(inlineViewRef.getMaterializedTupleIds().size() == 1);
        // we need to materialize all slots of our inline view tuple
        analyzer.getTupleDesc(inlineViewRef.getId()).materializeSlots();
        UnionNode unionNode = new UnionNode(ctx_.getNextNodeId(),
            inlineViewRef.getMaterializedTupleIds().get(0));
        if (analyzer.hasEmptyResultSet()) return unionNode;
        unionNode.setTblRefIds(Lists.newArrayList(inlineViewRef.getId()));
        unionNode.addConstExprList(selectStmt.getBaseTblResultExprs());
        unionNode.init(analyzer);
        return unionNode;
      }
    }

    PlanNode rootNode =
        createQueryPlan(inlineViewRef.getViewStmt(), inlineViewRef.getAnalyzer(), false);
    // TODO: we should compute the "physical layout" of the view's descriptor, so that
    // the avg row size is available during optimization; however, that means we need to
    // select references to its resultExprs from the enclosing scope(s)
    rootNode.setTblRefIds(Lists.newArrayList(inlineViewRef.getId()));

    // The output smap is the composition of the inline view's smap and the output smap
    // of the inline view's plan root. This ensures that all downstream exprs referencing
    // the inline view are replaced with exprs referencing the physical output of the
    // inline view's plan.
    ExprSubstitutionMap outputSmap = ExprSubstitutionMap.compose(
        inlineViewRef.getSmap(), rootNode.getOutputSmap(), analyzer);
    if (analyzer.isOuterJoined(inlineViewRef.getId())) {
      // Exprs against non-matched rows of an outer join should always return NULL.
      // Make the rhs exprs of the output smap nullable, if necessary. This expr wrapping
      // must be performed on the composed smap, and not on the the inline view's smap,
      // because the rhs exprs must first be resolved against the physical output of
      // 'planRoot' to correctly determine whether wrapping is necessary.
      List<Expr> nullableRhs = TupleIsNullPredicate.wrapExprs(
          outputSmap.getRhs(), rootNode.getTupleIds(), analyzer);
      outputSmap = new ExprSubstitutionMap(outputSmap.getLhs(), nullableRhs);
    }
    // Set output smap of rootNode *before* creating a SelectNode for proper resolution.
    rootNode.setOutputSmap(outputSmap);

    // Add runtime cardinality check if needed
    if (inlineViewRef.getViewStmt().isRuntimeScalar()) {
      rootNode = new CardinalityCheckNode(ctx_.getNextNodeId(), rootNode,
          inlineViewRef.getViewStmt().getOrigSqlString());
      rootNode.setOutputSmap(outputSmap);
      rootNode.init(ctx_.getRootAnalyzer());
    }

    // If the inline view has a LIMIT/OFFSET or unassigned conjuncts due to analytic
    // functions, we may have conjuncts that need to be assigned to a SELECT node on
    // top of the current plan root node.
    //
    // TODO: This check is also repeated in migrateConjunctsToInlineView() because we
    // need to make sure that equivalences are not enforced multiple times. Consolidate
    // the assignment of conjuncts and the enforcement of equivalences into a single
    // place.
    if (!canMigrateConjuncts(inlineViewRef)) {
      rootNode = addUnassignedConjuncts(
          analyzer, inlineViewRef.getDesc().getId().asList(), rootNode);
    }
    return rootNode;
  }

  /**
   * Migrates unassigned conjuncts into an inline view. Conjuncts are not
   * migrated into the inline view if the view has a LIMIT/OFFSET clause or if the
   * view's stmt computes analytic functions (see IMPALA-1243/IMPALA-1900).
   * The reason is that analytic functions compute aggregates over their entire input,
   * and applying filters from the enclosing scope *before* the aggregate computation
   * would alter the results. This is unlike regular aggregate computation, which only
   * makes the *output* of the computation visible to the enclosing scope, so that
   * filters from the enclosing scope can be safely applied (to the grouping cols, say).
   */
  public void migrateConjunctsToInlineView(final Analyzer analyzer,
      final InlineViewRef inlineViewRef) throws ImpalaException {
    List<Expr> unassignedConjuncts =
        analyzer.getUnassignedConjuncts(inlineViewRef.getId().asList(), true);
    if (LOG. isTraceEnabled()) {
      LOG.trace("unassignedConjuncts: " + Expr.debugString(unassignedConjuncts));
    }
    if (!canMigrateConjuncts(inlineViewRef)) {
      // mark (fully resolve) slots referenced by unassigned conjuncts as
      // materialized
      List<Expr> substUnassigned = Expr.substituteList(unassignedConjuncts,
          inlineViewRef.getBaseTblSmap(), analyzer, false);
      analyzer.materializeSlots(substUnassigned);
      return;
    }

    List<Expr> preds = new ArrayList<>();
    for (Expr e: unassignedConjuncts) {
      if (analyzer.canEvalPredicate(inlineViewRef.getId().asList(), e)) {
        preds.add(e);
        if (LOG. isTraceEnabled()) {
          LOG.trace(String.format("Can evaluate %s in inline view %s", e.debugString(),
                  inlineViewRef.getExplicitAlias()));
        }
      }
    }
    unassignedConjuncts.removeAll(preds);
    // Migrate the conjuncts by marking the original ones as assigned. They will either
    // be ignored if they are identity predicates (e.g. a = a), or be substituted into
    // new ones (viewPredicates below). The substituted ones will be re-registered.
    analyzer.markConjunctsAssigned(preds);
    // Generate predicates to enforce equivalences among slots of the inline view
    // tuple. These predicates are also migrated into the inline view.
    analyzer.createEquivConjuncts(inlineViewRef.getId(), preds);

    // Remove unregistered predicates that finally resolved to predicates reference
    // the same slot on both sides (e.g. a = a). Such predicates have been generated from
    // slot equivalences and may incorrectly reject rows with nulls
    // (IMPALA-1412/IMPALA-2643/IMPALA-8386).
    Predicate<Expr> isIdentityPredicate = new Predicate<Expr>() {
      @Override
      public boolean apply(Expr e) {
        if (!org.apache.impala.analysis.Predicate.isEquivalencePredicate(e)
            || !((BinaryPredicate) e).isInferred()) {
          return false;
        }
        try {
          BinaryPredicate finalExpr = (BinaryPredicate) e.trySubstitute(
              inlineViewRef.getBaseTblSmap(), analyzer, false);
          boolean isIdentity = finalExpr.hasIdenticalOperands();

          // Verity that "smap[e1] == smap[e2]" => "baseTblSmap[e1] == baseTblSmap[e2]"
          // in case we have bugs in generating baseTblSmap.
          BinaryPredicate midExpr = (BinaryPredicate) e.trySubstitute(
              inlineViewRef.getSmap(), analyzer, false);
          Preconditions.checkState(!midExpr.hasIdenticalOperands() || isIdentity);

          if (LOG.isTraceEnabled() && isIdentity) {
            LOG.trace("Removed identity predicate: " + finalExpr.debugString());
          }
          return isIdentity;
        } catch (Exception ex) {
          throw new IllegalStateException(
                  "Failed analysis after expr substitution.", ex);
        }
      }
    };
    Iterables.removeIf(preds, isIdentityPredicate);

    // create new predicates against the inline view's unresolved result exprs, not
    // the resolved result exprs, in order to avoid skipping scopes (and ignoring
    // limit clauses on the way)
    List<Expr> viewPredicates =
        Expr.substituteList(preds, inlineViewRef.getSmap(), analyzer, false);

    // perform any post-processing of the predicates before registering
    removeDisqualifyingInferredPreds(inlineViewRef.getAnalyzer(), viewPredicates);

    // Unset the On-clause flag of the migrated conjuncts because the migrated conjuncts
    // apply to the post-join/agg/analytic result of the inline view.
    for (Expr e: viewPredicates) e.setIsOnClauseConjunct(false);
    inlineViewRef.getAnalyzer().registerConjuncts(viewPredicates);

    // mark (fully resolve) slots referenced by remaining unassigned conjuncts as
    // materialized
    List<Expr> substUnassigned = Expr.substituteList(unassignedConjuncts,
        inlineViewRef.getBaseTblSmap(), analyzer, false);
    analyzer.materializeSlots(substUnassigned);
  }

  /**
   * Analyze the predicates in the context of the inline view for certain disqualifying
   * conditions and remove such predicates from the input list. One such condition is
   * the predicate is an inferred predicate AND either its left or right SlotRef
   * references the output of an outer join. Note that although such predicates
   * may have been detected at the time of creating the values transfer graph
   * (in the Analyzer), we do this check here anyways as a safety in case any such
   * predicate 'fell through' to this stage.
   */
  private void removeDisqualifyingInferredPreds(Analyzer analyzer, List<Expr> preds) {
    ListIterator<Expr> iter = preds.listIterator();
    while (iter.hasNext()) {
      Expr e = iter.next();
      if (e instanceof BinaryPredicate && ((BinaryPredicate)e).isInferred()) {
        BinaryPredicate p = (BinaryPredicate)e;
        Pair<SlotId, SlotId> slots = p.getEqSlots();
        if (slots == null) continue;
        TupleId leftParent = analyzer.getTupleId(slots.first);
        TupleId rightParent = analyzer.getTupleId(slots.second);
        // check if either the left parent or right parent is an outer joined tuple
        // Note: strictly, we may be ok to check only for the null producing
        // side but we are being conservative here to check both sides. With
        // additional testing we could potentially relax this.
        if (analyzer.isOuterJoined(leftParent) ||
                analyzer.isOuterJoined(rightParent)) {
          iter.remove();
          LOG.warn("Removed inferred predicate " + p.toSql() + " from the list of " +
                  "predicates considered for inline view because either the left " +
                  "or right side is derived from an outer join output.");
        }
      }
    }
  }

  /**
   * Checks if conjuncts can be migrated into an inline view.
   */
  private boolean canMigrateConjuncts(InlineViewRef inlineViewRef) {
    return !inlineViewRef.getViewStmt().hasLimit()
        && !inlineViewRef.getViewStmt().hasOffset()
        && (!(inlineViewRef.getViewStmt() instanceof SelectStmt)
            || !((SelectStmt) inlineViewRef.getViewStmt()).hasAnalyticInfo());
  }

  /**
   * Create a node to materialize the slots in the given HdfsTblRef.
   *
   * The given 'aggInfo' is used for detecting and applying optimizations that span both
   * the scan and aggregation.
   */
  private PlanNode createHdfsScanPlan(TableRef hdfsTblRef, AggregateInfo aggInfo,
      List<Expr> conjuncts, Analyzer analyzer) throws ImpalaException {
    TupleDescriptor tupleDesc = hdfsTblRef.getDesc();

    // Do partition pruning before deciding which slots to materialize because we might
    // end up removing some predicates.
    HdfsPartitionPruner pruner = new HdfsPartitionPruner(tupleDesc);
    Pair<List<? extends FeFsPartition>, List<Expr>> pair =
        pruner.prunePartitions(analyzer, conjuncts, false);
    List<? extends FeFsPartition> partitions = pair.first;

    // Mark all slots referenced by the remaining conjuncts as materialized.
    analyzer.materializeSlots(conjuncts);

    // TODO: Remove this section, once DATE type is supported across all fileformats.
    // Check if there are any partitions for which DATE is not supported.
    FeFsPartition part = findUnsupportedDateFsPartition(partitions);
    if (part != null) {
      FeFsTable table = (FeFsTable)hdfsTblRef.getTable();
      HdfsFileFormat ff = part.getFileFormat();
      // Throw an exception if tupleDesc contains a non-clustering, materialized
      // DATE slot.
      for (SlotDescriptor slotDesc: tupleDesc.getMaterializedSlots()) {
        if (slotDesc.getColumn() != null
            && !table.isClusteringColumn(slotDesc.getColumn())
            && slotDesc.getType() == ScalarType.DATE) {
          throw new NotImplementedException(
              "Scanning DATE values in table '" + table.getFullName() +
              "' is not supported for fileformat " + ff);
        }
      }
    }

    // For queries which contain partition columns only, we may use the metadata instead
    // of table scans. This is only feasible if all materialized aggregate expressions
    // have distinct semantics. Please see createHdfsScanPlan() for details.
    boolean fastPartitionKeyScans =
        analyzer.getQueryCtx().client_request.query_options.optimize_partition_key_scans &&
        aggInfo != null && aggInfo.hasAllDistinctAgg();

    // If the optimization for partition key scans with metadata is enabled,
    // try evaluating with metadata first. If not, fall back to scanning.
    if (fastPartitionKeyScans && tupleDesc.hasClusteringColsOnly()) {
      Set<List<Expr>> uniqueExprs = new HashSet<>();

      for (FeFsPartition partition: partitions) {
        // Ignore empty partitions to match the behavior of the scan based approach.
        if (partition.getSize() == 0) continue;
        List<Expr> exprs = new ArrayList<>();
        for (SlotDescriptor slotDesc: tupleDesc.getSlots()) {
          // UnionNode.init() will go through all the slots in the tuple descriptor so
          // there needs to be an entry in 'exprs' for each slot. For unmaterialized
          // slots, use dummy null values. UnionNode will filter out unmaterialized slots.
          if (!slotDesc.isMaterialized()) {
            exprs.add(NullLiteral.create(slotDesc.getType()));
          } else {
            int pos = slotDesc.getColumn().getPosition();
            exprs.add(partition.getPartitionValue(pos));
          }
        }
        uniqueExprs.add(exprs);
      }

      // Create a UNION node with all unique partition keys.
      UnionNode unionNode = new UnionNode(ctx_.getNextNodeId(), tupleDesc.getId());
      for (List<Expr> exprList: uniqueExprs) {
        unionNode.addConstExprList(exprList);
      }
      unionNode.init(analyzer);
      return unionNode;
    } else {
      HdfsScanNode scanNode =
          new HdfsScanNode(ctx_.getNextNodeId(), tupleDesc, conjuncts, partitions,
              hdfsTblRef, aggInfo, pair.second);
      scanNode.init(analyzer);
      return scanNode;
    }
  }

  /**
   * Looks for a filesystem-based partition in 'partitions' with no DATE support and
   * returns the first one it finds. Right now, scanning DATE values is only supported for
   * TEXT, PARQUET and AVRO fileformats.
   *
   * Returns null otherwise.
   */
  private FeFsPartition findUnsupportedDateFsPartition(
      List<? extends FeFsPartition> partitions) {
    for (FeFsPartition part: partitions) {
      HdfsFileFormat ff = part.getFileFormat();
      if (!ff.isDateTypeSupported()) return part;
    }
    return null;
  }

  /**
   * Create node for scanning all data files of a particular table.
   *
   * The given 'aggInfo' is used for detecting and applying optimizations that span both
   * the scan and aggregation. Only applicable to HDFS and Kudu table refs.
   *
   * Throws if a PlanNode.init() failed or if planning of the given
   * table ref is not implemented.
   */
  private PlanNode createScanNode(TableRef tblRef, AggregateInfo aggInfo,
      Analyzer analyzer) throws ImpalaException {
    ScanNode scanNode = null;

    // Get all predicates bound by the tuple.
    List<Expr> conjuncts = new ArrayList<>();
    TupleId tid = tblRef.getId();
    conjuncts.addAll(analyzer.getBoundPredicates(tid));

    // Also add remaining unassigned conjuncts
    List<Expr> unassigned = analyzer.getUnassignedConjuncts(tid.asList());
    conjuncts.addAll(unassigned);
    analyzer.markConjunctsAssigned(unassigned);
    analyzer.createEquivConjuncts(tid, conjuncts);

    // Perform constant propagation and optimization if rewriting is enabled
    if (analyzer.getQueryCtx().client_request.query_options.enable_expr_rewrites) {
      if (!Expr.optimizeConjuncts(conjuncts, analyzer)) {
        // Conjuncts implied false; convert to EmptySetNode
        EmptySetNode node = new EmptySetNode(ctx_.getNextNodeId(), tid.asList());
        node.init(analyzer);
        return node;
      }
    } else {
      Expr.removeDuplicates(conjuncts);
    }

    // TODO(todd) introduce FE interfaces for DataSourceTable, HBaseTable, KuduTable
    FeTable table = tblRef.getTable();
    if (table instanceof FeFsTable) {
      return createHdfsScanPlan(tblRef, aggInfo, conjuncts, analyzer);
    } else if (table instanceof FeDataSourceTable) {
      scanNode = new DataSourceScanNode(ctx_.getNextNodeId(), tblRef.getDesc(),
          conjuncts);
      scanNode.init(analyzer);
      return scanNode;
    } else if (table instanceof FeHBaseTable) {
      // HBase table
      scanNode = new HBaseScanNode(ctx_.getNextNodeId(), tblRef.getDesc());
      scanNode.addConjuncts(conjuncts);
      scanNode.init(analyzer);
      return scanNode;
    } else if (tblRef.getTable() instanceof FeKuduTable) {
      scanNode = new KuduScanNode(ctx_.getNextNodeId(), tblRef.getDesc(), conjuncts,
          aggInfo);
      scanNode.init(analyzer);
      return scanNode;
    } else {
      throw new NotImplementedException(
          "Planning not implemented for table ref class: " + tblRef.getClass());
    }
  }

  /**
   * Returns all applicable conjuncts for join between two plan trees 'materializing' the
   * given left-hand and right-hand side table ref ids. The conjuncts either come from
   * the analyzer or are generated based on equivalence classes, if necessary. The
   * returned conjuncts are marked as assigned.
   * The conjuncts can be used for hash table lookups.
   * - for inner joins, those are equi-join predicates in which one side is fully bound
   *   by lhsTblRefIds and the other by rhsTblRefIds
   * - for outer joins: same type of conjuncts as inner joins, but only from the
   *   ON or USING clause
   * Predicates that are redundant based on equivalence classes are intentionally
   * returned by this function because the removal of redundant predicates and the
   * creation of new predicates for enforcing slot equivalences go hand-in-hand
   * (see analyzer.createEquivConjuncts()).
   */
  private List<BinaryPredicate> getHashLookupJoinConjuncts(
      List<TupleId> lhsTblRefIds, List<TupleId> rhsTblRefIds, Analyzer analyzer) {
    List<BinaryPredicate> result = new ArrayList<>();
    List<Expr> candidates = analyzer.getEqJoinConjuncts(lhsTblRefIds, rhsTblRefIds);
    Preconditions.checkNotNull(candidates);
    for (Expr e: candidates) {
      if (!(e instanceof BinaryPredicate)) continue;
      BinaryPredicate normalizedJoinConjunct =
          getNormalizedEqPred(e, lhsTblRefIds, rhsTblRefIds, analyzer);
      if (normalizedJoinConjunct == null) continue;
      analyzer.markConjunctAssigned(e);
      result.add(normalizedJoinConjunct);
    }
    if (!result.isEmpty()) return result;

    // Construct join conjuncts derived from equivalence class membership.
    Set<TupleId> lhsTblRefIdsHs = new HashSet<>(lhsTblRefIds);
    for (TupleId rhsId: rhsTblRefIds) {
      TableRef rhsTblRef = analyzer.getTableRef(rhsId);
      Preconditions.checkNotNull(rhsTblRef);
      for (SlotDescriptor slotDesc: rhsTblRef.getDesc().getSlots()) {
        SlotId rhsSid = slotDesc.getId();
        for (SlotId lhsSid : analyzer.getEquivClass(rhsSid)) {
          if (lhsTblRefIdsHs.contains(analyzer.getTupleId(lhsSid))) {
            result.add(analyzer.createInferredEqPred(lhsSid, rhsSid));
            break;
          }
        }
      }
    }
    return result;
  }

  /**
   * Returns a normalized version of a binary equality predicate 'expr' where the lhs
   * child expr is bound by some tuple in 'lhsTids' and the rhs child expr is bound by
   * some tuple in 'rhsTids'. Returns 'expr' if this predicate is already normalized.
   * Returns null in any of the following cases:
   * 1. It is not an equality predicate
   * 2. One of the operands is a constant
   * 3. Both children of this predicate are the same expr
   * 4. Cannot be normalized
   */
  public static BinaryPredicate getNormalizedEqPred(Expr expr, List<TupleId> lhsTids,
      List<TupleId> rhsTids, Analyzer analyzer) {
    if (!(expr instanceof BinaryPredicate)) return null;
    BinaryPredicate pred = (BinaryPredicate) expr;
    if (!pred.getOp().isEquivalence() && pred.getOp() != Operator.NULL_MATCHING_EQ) {
      return null;
    }
    if (pred.getChild(0).isConstant() || pred.getChild(1).isConstant()) return null;

    Expr lhsExpr = Expr.getFirstBoundChild(pred, lhsTids);
    Expr rhsExpr = Expr.getFirstBoundChild(pred, rhsTids);
    if (lhsExpr == null || rhsExpr == null || lhsExpr == rhsExpr) return null;

    BinaryPredicate result = new BinaryPredicate(pred.getOp(), lhsExpr, rhsExpr);
    result.analyzeNoThrow(analyzer);
    return result;
  }

  /**
   * Creates a new node to join outer with inner. Collects and assigns join conjunct
   * as well as regular conjuncts. Calls init() on the new join node.
   * Throws if the JoinNode.init() fails.
   */
  private PlanNode createJoinNode(PlanNode outer, PlanNode inner,
      TableRef innerRef, Analyzer analyzer) throws ImpalaException {
    // get eq join predicates for the TableRefs' ids (not the PlanNodes' ids, which
    // are materialized)
    List<BinaryPredicate> eqJoinConjuncts = getHashLookupJoinConjuncts(
        outer.getTblRefIds(), inner.getTblRefIds(), analyzer);
    // Outer joins should only use On-clause predicates as eqJoinConjuncts.
    if (!innerRef.getJoinOp().isOuterJoin()) {
      analyzer.createEquivConjuncts(outer.getTblRefIds(), inner.getTblRefIds(),
          eqJoinConjuncts);
    }
    if (!eqJoinConjuncts.isEmpty() && innerRef.getJoinOp() == JoinOperator.CROSS_JOIN) {
      innerRef.setJoinOp(JoinOperator.INNER_JOIN);
    }

    List<Expr> otherJoinConjuncts = new ArrayList<>();
    if (innerRef.getJoinOp().isOuterJoin()) {
      // Also assign conjuncts from On clause. All remaining unassigned conjuncts
      // that can be evaluated by this join are assigned in createSelectPlan().
      otherJoinConjuncts = analyzer.getUnassignedOjConjuncts(innerRef);
    } else if (innerRef.getJoinOp().isSemiJoin()) {
      // Unassigned conjuncts bound by the invisible tuple id of a semi join must have
      // come from the join's On-clause, and therefore, must be added to the other join
      // conjuncts to produce correct results.
      // TODO This doesn't handle predicates specified in the On clause which are not
      // bound by any tuple id (e.g. ON (true))
      List<TupleId> tblRefIds = Lists.newArrayList(outer.getTblRefIds());
      tblRefIds.addAll(inner.getTblRefIds());
      otherJoinConjuncts = analyzer.getUnassignedConjuncts(tblRefIds, false);
      if (innerRef.getJoinOp().isNullAwareLeftAntiJoin()) {
        boolean hasNullMatchingEqOperator = false;
        // Keep only the null-matching eq conjunct in the eqJoinConjuncts and move
        // all the others in otherJoinConjuncts. The BE relies on this
        // separation for correct execution of the null-aware left anti join.
        Iterator<BinaryPredicate> it = eqJoinConjuncts.iterator();
        while (it.hasNext()) {
          BinaryPredicate conjunct = it.next();
          if (!conjunct.isNullMatchingEq()) {
            otherJoinConjuncts.add(conjunct);
            it.remove();
          } else {
            // Only one null-matching eq conjunct is allowed
            Preconditions.checkState(!hasNullMatchingEqOperator);
            hasNullMatchingEqOperator = true;
          }
        }
        Preconditions.checkState(hasNullMatchingEqOperator);
      }
    }
    analyzer.markConjunctsAssigned(otherJoinConjuncts);

    // Use a nested-loop join if there are no equi-join conjuncts, or if the inner
    // (build side) is a singular row src. A singular row src has a cardinality of 1, so
    // a nested-loop join is certainly cheaper than a hash join.
    JoinNode result = null;
    Preconditions.checkState(!innerRef.getJoinOp().isNullAwareLeftAntiJoin()
        || !(inner instanceof SingularRowSrcNode));
    if (eqJoinConjuncts.isEmpty() || inner instanceof SingularRowSrcNode) {
      otherJoinConjuncts.addAll(eqJoinConjuncts);
      result = new NestedLoopJoinNode(outer, inner, analyzer.isStraightJoin(),
          innerRef.getDistributionMode(), innerRef.getJoinOp(), otherJoinConjuncts);
    } else {
      result = new HashJoinNode(outer, inner, analyzer.isStraightJoin(),
          innerRef.getDistributionMode(), innerRef.getJoinOp(), eqJoinConjuncts,
          otherJoinConjuncts);
    }
    result.init(analyzer);
    return result;
  }

  /**
   * Create a tree of PlanNodes for the given tblRef, which can be a BaseTableRef,
   * CollectionTableRef or an InlineViewRef.
   *
   * The given 'aggInfo' is used for detecting and applying optimizations that span both
   * the scan and aggregation. Only applicable to HDFS and Kudu table refs.
   *
   * Throws if a PlanNode.init() failed or if planning of the given
   * table ref is not implemented.
   */
  private PlanNode createTableRefNode(TableRef tblRef, AggregateInfo aggInfo,
      Analyzer analyzer) throws ImpalaException {
    PlanNode result = null;
    if (tblRef instanceof BaseTableRef) {
      result = createScanNode(tblRef, aggInfo, analyzer);
    } else if (tblRef instanceof CollectionTableRef) {
      if (tblRef.isRelative()) {
        Preconditions.checkState(ctx_.hasSubplan());
        result = new UnnestNode(ctx_.getNextNodeId(), ctx_.getSubplan(),
            (CollectionTableRef) tblRef);
        result.init(analyzer);
      } else {
        result = createScanNode(tblRef, null, analyzer);
      }
    } else if (tblRef instanceof InlineViewRef) {
      result = createInlineViewPlan(analyzer, (InlineViewRef) tblRef);
    } else if (tblRef instanceof SingularRowSrcTableRef) {
      Preconditions.checkState(ctx_.hasSubplan());
      result = new SingularRowSrcNode(ctx_.getNextNodeId(), ctx_.getSubplan());
      result.init(analyzer);
    } else {
      throw new NotImplementedException(
          "Planning not implemented for table ref class: " + tblRef.getClass());
    }
    return result;
  }

  /**
   * Create a plan tree corresponding to 'unionOperands' for the given unionStmt.
   * The individual operands' plan trees are attached to a single UnionNode.
   * If unionDistinctPlan is not null, it is expected to contain the plan for the
   * distinct portion of the given unionStmt. The unionDistinctPlan is then added
   * as a child of the returned UnionNode.
   */
  private UnionNode createUnionPlan(
      Analyzer analyzer, UnionStmt unionStmt, List<UnionOperand> unionOperands,
      PlanNode unionDistinctPlan)
      throws ImpalaException {
    UnionNode unionNode = new UnionNode(ctx_.getNextNodeId(), unionStmt.getTupleId(),
        unionStmt.getUnionResultExprs(), ctx_.hasSubplan());
    for (UnionOperand op: unionOperands) {
      if (op.getAnalyzer().hasEmptyResultSet()) {
        unmarkCollectionSlots(op.getQueryStmt());
        continue;
      }
      QueryStmt queryStmt = op.getQueryStmt();
      if (queryStmt instanceof SelectStmt) {
        SelectStmt selectStmt = (SelectStmt) queryStmt;
        if (selectStmt.getTableRefs().isEmpty()) {
          unionNode.addConstExprList(selectStmt.getResultExprs());
          continue;
        }
      }
      PlanNode opPlan = createQueryPlan(queryStmt, op.getAnalyzer(), false);
      // There may still be unassigned conjuncts if the operand has an order by + limit.
      // Place them into a SelectNode on top of the operand's plan.
      opPlan = addUnassignedConjuncts(analyzer, opPlan.getTupleIds(), opPlan);
      if (opPlan instanceof EmptySetNode) continue;
      unionNode.addChild(opPlan, op.getQueryStmt().getResultExprs());
    }

    if (unionDistinctPlan != null) {
      Preconditions.checkState(unionStmt.hasDistinctOps());
      Preconditions.checkState(unionDistinctPlan instanceof AggregationNode);
      unionNode.addChild(unionDistinctPlan,
          unionStmt.getDistinctAggInfo().getGroupingExprs());
    }
    unionNode.init(analyzer);
    return unionNode;
  }

  /**
   * Returns plan tree for unionStmt:
   * - distinctOperands' plan trees are collected in a single UnionNode
   *   and duplicates removed via distinct aggregation
   * - the output of that plus the allOperands' plan trees are collected in
   *   another UnionNode which materializes the result of unionStmt
   * - if any of the union operands contains analytic exprs, we avoid pushing
   *   predicates directly into the operands and instead evaluate them
   *   *after* the final UnionNode (see createInlineViewPlan() for the reasoning)
   *   TODO: optimize this by still pushing predicates into the union operands
   *   that don't contain analytic exprs and evaluating the conjuncts in Select
   *   directly above the AnalyticEvalNodes
   * TODO: Simplify the plan of unions with empty operands using an empty set node.
   * TODO: Simplify the plan of unions with only a single non-empty operand to not
   *       use a union node (this is tricky because a union materializes a new tuple).
   */
  private PlanNode createUnionPlan(UnionStmt unionStmt, Analyzer analyzer)
      throws ImpalaException {
    List<Expr> conjuncts =
        analyzer.getUnassignedConjuncts(unionStmt.getTupleId().asList(), false);
    if (!unionStmt.hasAnalyticExprs()) {
      // Turn unassigned predicates for unionStmt's tupleId_ into predicates for
      // the individual operands.
      // Do this prior to creating the operands' plan trees so they get a chance to
      // pick up propagated predicates.
      for (UnionOperand op: unionStmt.getOperands()) {
        List<Expr> opConjuncts =
            Expr.substituteList(conjuncts, op.getSmap(), analyzer, false);
        op.getAnalyzer().registerConjuncts(opConjuncts);
      }
      analyzer.markConjunctsAssigned(conjuncts);
    } else {
      // mark slots referenced by the yet-unassigned conjuncts
      analyzer.materializeSlots(conjuncts);
    }
    // mark slots after predicate propagation but prior to plan tree generation
    unionStmt.materializeRequiredSlots(analyzer);

    PlanNode result = null;
    // create DISTINCT tree
    if (unionStmt.hasDistinctOps()) {
      result = createUnionPlan(
          analyzer, unionStmt, unionStmt.getDistinctOperands(), null);
      result = new AggregationNode(
          ctx_.getNextNodeId(), result, unionStmt.getDistinctAggInfo(), AggPhase.FIRST);
      result.init(analyzer);
    }
    // create ALL tree
    if (unionStmt.hasAllOps()) {
      result = createUnionPlan(analyzer, unionStmt, unionStmt.getAllOperands(), result);
    }

    if (unionStmt.hasAnalyticExprs()) {
      result = addUnassignedConjuncts(
          analyzer, unionStmt.getTupleId().asList(), result);
    }
    return result;
  }
}