StmtRewriter.java

Go to the documentation of this file.

// Copyright 2014 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License 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 com.cloudera.impala.analysis;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.cloudera.impala.analysis.AnalysisContext.AnalysisResult;
import com.cloudera.impala.analysis.UnionStmt.UnionOperand;
import com.cloudera.impala.common.AnalysisException;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.collect.Lists;

public class StmtRewriter {
  private final static Logger LOG = LoggerFactory.getLogger(StmtRewriter.class);

  public static StatementBase rewrite(AnalysisResult analysisResult)
      throws AnalysisException {
    StatementBase rewrittenStmt = null;
    if (analysisResult.getStmt() instanceof QueryStmt) {
      QueryStmt analyzedStmt = (QueryStmt)analysisResult.getStmt();
      rewriteQueryStatement(analyzedStmt, analysisResult.getAnalyzer());
      rewrittenStmt = analyzedStmt.clone();
    } else if (analysisResult.getStmt() instanceof InsertStmt) {
      // For an InsertStmt, rewrites are performed during its analysis.
      // Clone the insert stmt to reset its analysis state.
      rewrittenStmt = ((InsertStmt)analysisResult.getStmt()).clone();
    } else if (analysisResult.getStmt() instanceof CreateTableAsSelectStmt) {
      // For a CTAS, rewrites are performed during its analysis.
      CreateTableAsSelectStmt ctasStmt =
          (CreateTableAsSelectStmt)analysisResult.getStmt();
      // Create a new CTAS from the original create statement and the
      // rewritten insert statement.
      Preconditions.checkNotNull(analysisResult.getTmpCreateTableStmt());
      rewrittenStmt = new CreateTableAsSelectStmt(analysisResult.getTmpCreateTableStmt(),
          ctasStmt.getQueryStmt().clone());
    } else {
      throw new AnalysisException("Unsupported statement containing subqueries: " +
          analysisResult.getStmt().toSql());
    }
    return rewrittenStmt;
  }

  public static void rewriteQueryStatement(QueryStmt stmt, Analyzer analyzer)
      throws AnalysisException {
    Preconditions.checkNotNull(stmt);
    if (stmt instanceof SelectStmt) {
      rewriteSelectStatement((SelectStmt)stmt, analyzer);
    } else if (stmt instanceof UnionStmt) {
      rewriteUnionStatement((UnionStmt)stmt, analyzer);
    } else {
      throw new AnalysisException("Subqueries not supported for " +
          stmt.getClass().getSimpleName() + " statements");
    }
  }

  private static void rewriteSelectStatement(SelectStmt stmt, Analyzer analyzer)
      throws AnalysisException {
    // Rewrite all the subqueries in the FROM clause.
    for (TableRef tblRef: stmt.tableRefs_) {
      if (!(tblRef instanceof InlineViewRef)) continue;
      InlineViewRef inlineViewRef = (InlineViewRef)tblRef;
      rewriteQueryStatement(inlineViewRef.getViewStmt(), inlineViewRef.getAnalyzer());
      // Reset the state of the underlying stmt since it was rewritten
      inlineViewRef.setRewrittenViewStmt(inlineViewRef.getViewStmt().clone());
    }
    // Rewrite all the subqueries in the WHERE clause.
    if (stmt.hasWhereClause()) {
      // Push negation to leaf operands.
      stmt.whereClause_ = Expr.pushNegationToOperands(stmt.whereClause_);
      // Check if we can rewrite the subqueries in the WHERE clause. OR predicates with
      // subqueries are not supported.
      if (hasSubqueryInDisjunction(stmt.whereClause_)) {
        throw new AnalysisException("Subqueries in OR predicates are not supported: " +
            stmt.whereClause_.toSql());
      }
      rewriteWhereClauseSubqueries(stmt, analyzer);
    }
    stmt.sqlString_ = null;
    LOG.trace("rewritten stmt: " + stmt.toSql());
  }

  private static void rewriteUnionStatement(UnionStmt stmt, Analyzer analyzer)
      throws AnalysisException {
    for (UnionOperand operand: stmt.getOperands()) {
      Preconditions.checkState(operand.getQueryStmt() instanceof SelectStmt);
      StmtRewriter.rewriteSelectStatement(
          (SelectStmt)operand.getQueryStmt(), operand.getAnalyzer());
    }
  }

  private static boolean hasSubqueryInDisjunction(Expr expr) {
    if (!(expr instanceof CompoundPredicate)) return false;
    if (Expr.IS_OR_PREDICATE.apply(expr)) {
      return expr.contains(Subquery.class);
    }
    for (Expr child: expr.getChildren()) {
      if (hasSubqueryInDisjunction(child)) return true;
    }
    return false;
  }

  private static void rewriteWhereClauseSubqueries(SelectStmt stmt, Analyzer analyzer)
     throws AnalysisException {
    int numTableRefs = stmt.tableRefs_.size();
    ArrayList<Expr> exprsWithSubqueries = Lists.newArrayList();
    ExprSubstitutionMap smap = new ExprSubstitutionMap();
    // Replace all BetweenPredicates that contain subqueries with their
    // equivalent compound predicates.
    stmt.whereClause_ = replaceBetweenPredicates(stmt.whereClause_);
    // Check if all the conjuncts in the WHERE clause that contain subqueries
    // can currently be rewritten as a join.
    for (Expr conjunct: stmt.whereClause_.getConjuncts()) {
      List<Subquery> subqueries = Lists.newArrayList();
      conjunct.collectAll(Predicates.instanceOf(Subquery.class), subqueries);
      if (subqueries.size() == 0) continue;
      if (subqueries.size() > 1) {
        throw new AnalysisException("Multiple subqueries are not supported in " +
            "expression: " + conjunct.toSql());
      }
      if (!(conjunct instanceof InPredicate) && !(conjunct instanceof ExistsPredicate) &&
          !(conjunct instanceof BinaryPredicate) &&
          !conjunct.contains(Expr.IS_SCALAR_SUBQUERY)) {
        throw new AnalysisException("Non-scalar subquery is not supported in " +
            "expression: " + conjunct.toSql());
      }

      if (conjunct instanceof ExistsPredicate) {
        // Check if we can determine the result of an ExistsPredicate during analysis.
        // If so, replace the predicate with a BoolLiteral predicate and remove it from
        // the list of predicates to be rewritten.
        BoolLiteral boolLiteral = replaceExistsPredicate((ExistsPredicate) conjunct);
        if (boolLiteral != null) {
          boolLiteral.analyze(analyzer);
          smap.put(conjunct, boolLiteral);
          continue;
        }
      }

      // Replace all the supported exprs with subqueries with true BoolLiterals
      // using an smap.
      BoolLiteral boolLiteral = new BoolLiteral(true);
      boolLiteral.analyze(analyzer);
      smap.put(conjunct, boolLiteral);
      exprsWithSubqueries.add(conjunct);
    }
    stmt.whereClause_ = stmt.whereClause_.substitute(smap, analyzer, false);

    boolean hasNewVisibleTuple = false;
    // Recursively rewrite all the exprs that contain subqueries and merge them
    // with 'stmt'.
    for (Expr expr: exprsWithSubqueries) {
      if (mergeExpr(stmt, rewriteExpr(expr, analyzer), analyzer)) {
        hasNewVisibleTuple = true;
      }
    }
    if (canEliminate(stmt.whereClause_)) stmt.whereClause_ = null;
    if (hasNewVisibleTuple) replaceUnqualifiedStarItems(stmt, numTableRefs);
  }

  private static BoolLiteral replaceExistsPredicate(ExistsPredicate predicate) {
    Subquery subquery = predicate.getSubquery();
    Preconditions.checkNotNull(subquery);
    SelectStmt subqueryStmt = (SelectStmt) subquery.getStatement();
    BoolLiteral boolLiteral = null;
    if (subqueryStmt.getAnalyzer().hasEmptyResultSet()) {
      boolLiteral = new BoolLiteral(predicate.isNotExists());
    } else if (subqueryStmt.hasAggInfo() && subqueryStmt.getAggInfo().hasAggregateExprs()
          && !subqueryStmt.hasAnalyticInfo() && subqueryStmt.getHavingPred() == null) {
      boolLiteral = new BoolLiteral(!predicate.isNotExists());
    }
    return boolLiteral;
  }

  private static Expr replaceBetweenPredicates(Expr expr) {
    if (expr instanceof BetweenPredicate && expr.contains(Subquery.class)) {
      return ((BetweenPredicate)expr).getRewrittenPredicate();
    }
    for (int i = 0; i < expr.getChildren().size(); ++i) {
      expr.setChild(i, replaceBetweenPredicates(expr.getChild(i)));
    }
    return expr;
  }

  private static Expr rewriteExpr(Expr expr, Analyzer analyzer)
      throws AnalysisException {
    // Extract the subquery and rewrite it.
    Subquery subquery = expr.getSubquery();
    Preconditions.checkNotNull(subquery);
    rewriteSelectStatement((SelectStmt) subquery.getStatement(), subquery.getAnalyzer());
    // Create a new Subquery with the rewritten stmt and use a substitution map
    // to replace the original subquery from the expr.
    Subquery newSubquery = new Subquery(subquery.getStatement().clone());
    newSubquery.analyze(analyzer);
    ExprSubstitutionMap smap = new ExprSubstitutionMap();
    smap.put(subquery, newSubquery);
    return expr.substitute(smap, analyzer, false);
  }

  private static boolean mergeExpr(SelectStmt stmt, Expr expr,
      Analyzer analyzer) throws AnalysisException {
    Preconditions.checkNotNull(expr);
    Preconditions.checkNotNull(analyzer);
    boolean updateSelectList = false;

    SelectStmt subqueryStmt = (SelectStmt)expr.getSubquery().getStatement();
    // Create a new inline view from the subquery stmt. The inline view will be added
    // to the stmt's table refs later. Explicitly set the inline view's column labels
    // to eliminate any chance that column aliases from the parent query could reference
    // select items from the inline view after the rewrite.
    List<String> colLabels = Lists.newArrayList();
    for (int i = 0; i < subqueryStmt.getColLabels().size(); ++i) {
      colLabels.add(subqueryStmt.getColumnAliasGenerator().getNextAlias());
    }
    InlineViewRef inlineView = new InlineViewRef(
        stmt.getTableAliasGenerator().getNextAlias(), subqueryStmt, colLabels);

    // Extract all correlated predicates from the subquery.
    List<Expr> onClauseConjuncts = extractCorrelatedPredicates(subqueryStmt);
    if (!onClauseConjuncts.isEmpty()) {
      canRewriteCorrelatedSubquery(expr);
      // For correlated subqueries that are eligible for rewrite by transforming
      // into a join, a LIMIT clause has no effect on the results, so we can
      // safely remove it.
      subqueryStmt.limitElement_ = null;
    }

    if (expr instanceof ExistsPredicate) {
      // For uncorrelated subqueries, we limit the number of rows returned by the
      // subquery.
      if (onClauseConjuncts.isEmpty()) subqueryStmt.setLimit(1);
    }

    // Update the subquery's select list and/or its GROUP BY clause by adding
    // exprs from the extracted correlated predicates.
    boolean updateGroupBy = expr.getSubquery().isScalarSubquery() ||
        (expr instanceof ExistsPredicate &&
         subqueryStmt.hasAggInfo() && !subqueryStmt.getSelectList().isDistinct());
    List<Expr> lhsExprs = Lists.newArrayList();
    List<Expr> rhsExprs = Lists.newArrayList();
    for (Expr conjunct: onClauseConjuncts) {
      updateInlineView(inlineView, conjunct, stmt.getTableRefIds(),
          lhsExprs, rhsExprs, updateGroupBy);
    }

    // Analyzing the inline view trigger reanalysis of the subquery's select statement.
    // However the statement is already analyzed and since statement analysis is not
    // idempotent, the analysis needs to be reset (by a call to clone()).
    inlineView = (InlineViewRef)inlineView.clone();
    inlineView.analyze(analyzer);
    inlineView.setLeftTblRef(stmt.tableRefs_.get(stmt.tableRefs_.size() - 1));
    stmt.tableRefs_.add(inlineView);
    JoinOperator joinOp = JoinOperator.LEFT_SEMI_JOIN;

    // Create a join conjunct from the expr that contains a subquery.
    Expr joinConjunct = createJoinConjunct(expr, inlineView, analyzer,
        !onClauseConjuncts.isEmpty());
    if (joinConjunct != null) {
      SelectListItem firstItem =
          ((SelectStmt) inlineView.getViewStmt()).getSelectList().getItems().get(0);
      if (!onClauseConjuncts.isEmpty() &&
          firstItem.getExpr().contains(Expr.NON_NULL_EMPTY_AGG)) {
        // Correlated subqueries with an aggregate function that returns non-null on
        // an empty input are rewritten using a LEFT OUTER JOIN because we
        // need to ensure that there is one agg value for every tuple of 'stmt'
        // (parent select block), even for those tuples of 'stmt' that get rejected
        // by the subquery due to some predicate. The new join conjunct is added to
        // stmt's WHERE clause because it needs to be applied to the result of the
        // LEFT OUTER JOIN (both matched and unmatched tuples).
        //
        // TODO Handle other aggregate functions and UDAs that return a non-NULL value
        // on an empty set.
        // TODO Handle count aggregate functions in an expression in subqueries
        // select list.
        stmt.whereClause_ =
            CompoundPredicate.createConjunction(joinConjunct, stmt.whereClause_);
        joinConjunct = null;
        joinOp = JoinOperator.LEFT_OUTER_JOIN;
        updateSelectList = true;
      }

      if (joinConjunct != null) onClauseConjuncts.add(joinConjunct);
    }

    // Create the ON clause from the extracted correlated predicates.
    Expr onClausePredicate =
        CompoundPredicate.createConjunctivePredicate(onClauseConjuncts);

    if (onClausePredicate == null) {
      Preconditions.checkState(expr instanceof ExistsPredicate);
      // TODO: Remove this when we support independent subquery evaluation.
      if (((ExistsPredicate)expr).isNotExists()) {
        throw new AnalysisException("Unsupported uncorrelated NOT EXISTS subquery: " +
            subqueryStmt.toSql());
      }
      // We don't have an ON clause predicate to create an equi-join. Rewrite the
      // subquery using a CROSS JOIN.
      // TODO This is very expensive. Remove it when we implement independent
      // subquery evaluation.
      inlineView.setJoinOp(JoinOperator.CROSS_JOIN);
      LOG.warn("uncorrelated subquery rewritten using a cross join");
      // Indicate that new visible tuples may be added in stmt's select list.
      return true;
    }

    // Create an smap from the original select-list exprs of the select list to
    // the corresponding inline-view columns.
    ExprSubstitutionMap smap = new ExprSubstitutionMap();
    Preconditions.checkState(lhsExprs.size() == rhsExprs.size());
    for (int i = 0; i < lhsExprs.size(); ++i) {
      Expr lhsExpr = lhsExprs.get(i);
      Expr rhsExpr = rhsExprs.get(i);
      rhsExpr.analyze(analyzer);
      smap.put(lhsExpr, rhsExpr);
    }
    onClausePredicate = onClausePredicate.substitute(smap, analyzer, false);

    // Check for references to ancestor query blocks (cycles in the dependency
    // graph of query blocks are not supported).
    if (!onClausePredicate.isBoundByTupleIds(stmt.getTableRefIds())) {
      throw new AnalysisException("Unsupported correlated subquery: " +
          subqueryStmt.toSql());
    }

    // Check if we have a valid ON clause for an equi-join.
    boolean hasEqJoinPred = false;
    for (Expr conjunct: onClausePredicate.getConjuncts()) {
      if (!(conjunct instanceof BinaryPredicate) ||
          ((BinaryPredicate)conjunct).getOp() != BinaryPredicate.Operator.EQ) {
        continue;
      }
      List<TupleId> lhsTupleIds = Lists.newArrayList();
      conjunct.getChild(0).getIds(lhsTupleIds, null);
      if (lhsTupleIds.isEmpty()) continue;
      List<TupleId> rhsTupleIds = Lists.newArrayList();
      conjunct.getChild(1).getIds(rhsTupleIds, null);
      if (rhsTupleIds.isEmpty()) continue;
      // Check if columns from the outer query block (stmt) appear in both sides
      // of the binary predicate.
      if ((lhsTupleIds.contains(inlineView.getDesc().getId()) && lhsTupleIds.size() > 1)
          || (rhsTupleIds.contains(inlineView.getDesc().getId())
              && rhsTupleIds.size() > 1)) {
        continue;
      }
      hasEqJoinPred = true;
      break;
    }

    if (!hasEqJoinPred) {
      // TODO: Remove this when independent subquery evaluation is implemented.
      // TODO: Requires support for non-equi joins.
      boolean hasGroupBy = ((SelectStmt) inlineView.getViewStmt()).hasGroupByClause();
      if (!expr.getSubquery().isScalarSubquery() ||
          (!(hasGroupBy && stmt.selectList_.isDistinct()) && hasGroupBy)) {
        throw new AnalysisException("Unsupported predicate with subquery: " +
            expr.toSql());
      }

      // We can rewrite the aggregate subquery using a cross join. All conjuncts
      // that were extracted from the subquery are added to stmt's WHERE clause.
      stmt.whereClause_ =
          CompoundPredicate.createConjunction(onClausePredicate, stmt.whereClause_);
      inlineView.setJoinOp(JoinOperator.CROSS_JOIN);
      // Indicate that the CROSS JOIN may add a new visible tuple to stmt's
      // select list (if the latter contains an unqualified star item '*')
      return true;
    }

    // We have a valid equi-join conjunct.
    if (expr instanceof InPredicate && ((InPredicate)expr).isNotIn() ||
        expr instanceof ExistsPredicate && ((ExistsPredicate)expr).isNotExists()) {
      // For the case of a NOT IN with an eq join conjunct, replace the join
      // conjunct with a conjunct that uses the null-matching eq operator.
      if (expr instanceof InPredicate) {
        joinOp = JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN;
        List<TupleId> tIds = Lists.newArrayList();
        joinConjunct.getIds(tIds, null);
        if (tIds.size() <= 1 || !tIds.contains(inlineView.getDesc().getId())) {
          throw new AnalysisException("Unsupported NOT IN predicate with subquery: " +
              expr.toSql());
        }
        // Replace the EQ operator in the generated join conjunct with a
        // null-matching EQ operator.
        for (Expr conjunct: onClausePredicate.getConjuncts()) {
          if (conjunct.equals(joinConjunct)) {
            Preconditions.checkState(conjunct instanceof BinaryPredicate);
            Preconditions.checkState(((BinaryPredicate)conjunct).getOp() ==
                BinaryPredicate.Operator.EQ);
            ((BinaryPredicate)conjunct).setOp(BinaryPredicate.Operator.NULL_MATCHING_EQ);
            break;
          }
        }
      } else {
        joinOp = JoinOperator.LEFT_ANTI_JOIN;
      }
    }
    inlineView.setJoinOp(joinOp);
    inlineView.setOnClause(onClausePredicate);
    return updateSelectList;
  }

  private static void replaceUnqualifiedStarItems(SelectStmt stmt, int tableIdx) {
    Preconditions.checkState(tableIdx < stmt.tableRefs_.size());
    ArrayList<SelectListItem> newItems = Lists.newArrayList();
    for (int i = 0; i < stmt.selectList_.getItems().size(); ++i) {
      SelectListItem item = stmt.selectList_.getItems().get(i);
      if (!item.isStar() || item.getRawPath() != null) {
        newItems.add(item);
        continue;
      }
      // '*' needs to be replaced by tbl1.*,...,tbln.*, where
      // tbl1,...,tbln are the visible tableRefs in stmt.
      for (int j = 0; j < tableIdx; ++j) {
        TableRef tableRef = stmt.tableRefs_.get(j);
        if (tableRef.getJoinOp() == JoinOperator.LEFT_SEMI_JOIN ||
            tableRef.getJoinOp() == JoinOperator.LEFT_ANTI_JOIN) {
          continue;
        }
        newItems.add(SelectListItem.createStarItem(
            Lists.newArrayList(tableRef.getUniqueAlias())));
      }
    }
    Preconditions.checkState(!newItems.isEmpty());
    boolean isDistinct = stmt.selectList_.isDistinct();
    stmt.selectList_ =
        new SelectList(newItems, isDistinct, stmt.selectList_.getPlanHints());
  }

  private static boolean canEliminate(Expr expr) {
    for (Expr conjunct: expr.getConjuncts()) {
      if (!Expr.IS_TRUE_LITERAL.apply(conjunct)) return false;
    }
    return true;
  }

  private static ArrayList<Expr> extractCorrelatedPredicates(SelectStmt subqueryStmt)
      throws AnalysisException {
    List<TupleId> subqueryTupleIds = subqueryStmt.getTableRefIds();
    ArrayList<Expr> correlatedPredicates = Lists.newArrayList();

    if (subqueryStmt.hasWhereClause()) {
      if (!canExtractCorrelatedPredicates(subqueryStmt.getWhereClause(),
          subqueryTupleIds)) {
        throw new AnalysisException("Disjunctions with correlated predicates " +
            "are not supported: " + subqueryStmt.getWhereClause().toSql());
      }
      // Extract the correlated predicates from the subquery's WHERE clause and
      // replace them with true BoolLiterals.
      Expr newWhereClause = extractCorrelatedPredicates(subqueryStmt.getWhereClause(),
          subqueryTupleIds, correlatedPredicates);
      if (canEliminate(newWhereClause)) newWhereClause = null;
      subqueryStmt.setWhereClause(newWhereClause);
    }

    // Process all correlated predicates from subquery's ON clauses.
    for (TableRef tableRef: subqueryStmt.getTableRefs()) {
      if (tableRef.getOnClause() == null) continue;

      ArrayList<Expr> onClauseCorrelatedPreds = Lists.newArrayList();
      Expr newOnClause = extractCorrelatedPredicates(tableRef.getOnClause(),
          subqueryTupleIds, onClauseCorrelatedPreds);
      if (onClauseCorrelatedPreds.isEmpty()) continue;

      correlatedPredicates.addAll(onClauseCorrelatedPreds);
      if (canEliminate(newOnClause)) {
        // After the extraction of correlated predicates from an ON clause,
        // the latter may only contain conjunctions of True BoolLiterals. In
        // this case, we can eliminate the ON clause and set the join type to
        // CROSS JOIN.
        tableRef.setJoinOp(JoinOperator.CROSS_JOIN);
        tableRef.setOnClause(null);
      } else {
        tableRef.setOnClause(newOnClause);
      }
    }
    return correlatedPredicates;
  }

  private static Expr extractCorrelatedPredicates(Expr root, List<TupleId> tupleIds,
      ArrayList<Expr> matches) {
    if (isCorrelatedPredicate(root, tupleIds)) {
      matches.add(root);
      return new BoolLiteral(true);
    }
    for (int i = 0; i < root.getChildren().size(); ++i) {
      root.getChildren().set(i, extractCorrelatedPredicates(root.getChild(i), tupleIds,
          matches));
    }
    return root;
  }

  private static void canRewriteCorrelatedSubquery(Expr expr)
        throws AnalysisException {
    Preconditions.checkNotNull(expr);
    Preconditions.checkState(expr.contains(Subquery.class));
    SelectStmt stmt = (SelectStmt) expr.getSubquery().getStatement();
    Preconditions.checkNotNull(stmt);
    // Grouping and/or aggregation (including analytic functions) is only
    // allowed on correlated EXISTS subqueries
    if ((expr instanceof BinaryPredicate
          && (stmt.hasGroupByClause() || stmt.hasAnalyticInfo()))
        || (expr instanceof InPredicate
            && (stmt.hasAggInfo() || stmt.hasAnalyticInfo()))) {
      throw new AnalysisException("Unsupported correlated subquery with grouping " +
          "and/or aggregation: " + stmt.toSql());
    }
    // The following correlated subqueries with a limit clause are supported:
    // 1. EXISTS subqueries
    // 2. Scalar subqueries with aggregation
    if (stmt.hasLimit() &&
        (!(expr instanceof BinaryPredicate) || !stmt.hasAggInfo() ||
         stmt.selectList_.isDistinct()) &&
        !(expr instanceof ExistsPredicate)) {
      throw new AnalysisException("Unsupported correlated subquery with a " +
          "LIMIT clause: " + stmt.toSql());
    }
  }

  private static void updateInlineView(InlineViewRef inlineView, Expr expr,
      List<TupleId> parentQueryTids, List<Expr> lhsExprs, List<Expr> rhsExprs,
      boolean updateGroupBy) throws AnalysisException {
    SelectStmt stmt = (SelectStmt)inlineView.getViewStmt();
    List<TupleId> subqueryTblIds = stmt.getTableRefIds();
    ArrayList<Expr> groupByExprs = null;
    if (updateGroupBy) groupByExprs = Lists.newArrayList();

    List<SelectListItem> items = stmt.selectList_.getItems();
    // Collect all the SlotRefs from 'expr' and identify those that are bound by
    // subquery tuple ids.
    ArrayList<Expr> slotRefs = Lists.newArrayList();
    expr.collectAll(Predicates.instanceOf(SlotRef.class), slotRefs);
    List<Expr> exprsBoundBySubqueryTids = Lists.newArrayList();
    for (Expr slotRef: slotRefs) {
      if (slotRef.isBoundByTupleIds(subqueryTblIds)) {
        exprsBoundBySubqueryTids.add(slotRef);
      }
    }
    // The correlated predicate only references slots from a parent block,
    // no need to update the subquery's select or group by list.
    if (exprsBoundBySubqueryTids.isEmpty()) return;
    if (updateGroupBy) {
      Preconditions.checkState(expr instanceof BinaryPredicate);
      Expr exprBoundBySubqueryTids = null;
      if (exprsBoundBySubqueryTids.size() > 1) {
        // If the predicate contains multiple SlotRefs bound by subquery tuple
        // ids, they must all be on the same side of that predicate.
        if (expr.getChild(0).isBoundByTupleIds(subqueryTblIds) &&
           expr.getChild(1).isBoundByTupleIds(parentQueryTids)) {
          exprBoundBySubqueryTids = expr.getChild(0);
        } else if (expr.getChild(0).isBoundByTupleIds(parentQueryTids) &&
            expr.getChild(1).isBoundByTupleIds(subqueryTblIds)) {
          exprBoundBySubqueryTids = expr.getChild(1);
        } else {
          throw new AnalysisException("All subquery columns " +
              "that participate in a predicate must be on the same side of " +
              "that predicate: " + expr.toSql());
        }
      } else {
        Preconditions.checkState(exprsBoundBySubqueryTids.size() == 1);
        exprBoundBySubqueryTids = exprsBoundBySubqueryTids.get(0);
      }
      exprsBoundBySubqueryTids.clear();
      exprsBoundBySubqueryTids.add(exprBoundBySubqueryTids);
    }

    // Add the exprs bound by subquery tuple ids to the select list and
    // register it for substitution. We use a temporary substitution map
    // because we cannot at this point analyze the new select list expr. Once
    // the new inline view is analyzed, the entries from this map will be
    // added to an ExprSubstitutionMap.
    for (Expr boundExpr: exprsBoundBySubqueryTids) {
      String colAlias = stmt.getColumnAliasGenerator().getNextAlias();
      items.add(new SelectListItem(boundExpr, null));
      inlineView.getExplicitColLabels().add(colAlias);
      lhsExprs.add(boundExpr);
      rhsExprs.add(new SlotRef(Lists.newArrayList(inlineView.getUniqueAlias(), colAlias)));
      if (groupByExprs != null) groupByExprs.add(boundExpr);
    }

    // Update the subquery's select list.
    boolean isDistinct = stmt.selectList_.isDistinct();
    stmt.selectList_ = new SelectList(
        items, isDistinct, stmt.selectList_.getPlanHints());
    // Update subquery's GROUP BY clause
    if (groupByExprs != null && !groupByExprs.isEmpty()) {
      if (stmt.hasGroupByClause()) {
        stmt.groupingExprs_.addAll(groupByExprs);
      } else {
        stmt.groupingExprs_ = groupByExprs;
      }
    }
  }

  private static boolean canExtractCorrelatedPredicates(Expr expr,
      List<TupleId> subqueryTupleIds) {
    if (!(expr instanceof CompoundPredicate)) return true;
    if (Expr.IS_OR_PREDICATE.apply(expr)) {
      return !containsCorrelatedPredicate(expr, subqueryTupleIds);
    }
    for (Expr child: expr.getChildren()) {
      if (!canExtractCorrelatedPredicates(child, subqueryTupleIds)) {
        return false;
      }
    }
    return true;
  }

  private static boolean containsCorrelatedPredicate(Expr root, List<TupleId> tupleIds) {
    if (isCorrelatedPredicate(root, tupleIds)) return true;
    for (Expr child: root.getChildren()) {
      if (containsCorrelatedPredicate(child, tupleIds)) return true;
    }
    return false;
  }

  private static boolean isCorrelatedPredicate(Expr expr, List<TupleId> tupleIds) {
    return (expr instanceof BinaryPredicate || expr instanceof SlotRef)
        && !expr.isBoundByTupleIds(tupleIds);
  }

  private static Expr createJoinConjunct(Expr exprWithSubquery, InlineViewRef inlineView,
      Analyzer analyzer, boolean isCorrelated) throws AnalysisException {
    Preconditions.checkNotNull(exprWithSubquery);
    Preconditions.checkNotNull(inlineView);
    Preconditions.checkState(exprWithSubquery.contains(Subquery.class));
    if (exprWithSubquery instanceof ExistsPredicate) return null;
    // Create a SlotRef from the first item of inlineView's select list
    SlotRef slotRef = new SlotRef(Lists.newArrayList(
        inlineView.getUniqueAlias(), inlineView.getColLabels().get(0)));
    slotRef.analyze(analyzer);
    Expr subquerySubstitute = slotRef;
    if (exprWithSubquery instanceof InPredicate) {
      BinaryPredicate pred = new BinaryPredicate(BinaryPredicate.Operator.EQ,
          exprWithSubquery.getChild(0), slotRef);
      pred.analyze(analyzer);
      return pred;
    }
    // Only scalar subqueries are supported
    Subquery subquery = exprWithSubquery.getSubquery();
    if (!subquery.isScalarSubquery()) {
      throw new AnalysisException("Unsupported predicate with a non-scalar subquery: "
          + subquery.toSql());
    }
    ExprSubstitutionMap smap = new ExprSubstitutionMap();
    SelectListItem item =
      ((SelectStmt) inlineView.getViewStmt()).getSelectList().getItems().get(0);
    if (isCorrelated && !item.getExpr().contains(Expr.IS_BUILTIN_AGG_FN)) {
      throw new AnalysisException("UDAs are not supported in the select list of " +
          "correlated subqueries: " + subquery.toSql());
    }
    if (isCorrelated && item.getExpr().contains(Expr.NON_NULL_EMPTY_AGG)) {
      // TODO: Add support for multiple agg functions that return non-null on an
      // empty input, by wrapping them with zeroifnull functions before the inline
      // view is analyzed.
      if (!Expr.NON_NULL_EMPTY_AGG.apply(item.getExpr()) &&
        (!(item.getExpr() instanceof CastExpr) ||
         !Expr.NON_NULL_EMPTY_AGG.apply(item.getExpr().getChild(0)))) {
        throw new AnalysisException("Aggregate function that returns non-null on " +
          "an empty input cannot be used in an expression in a " +
          "correlated subquery's select list: " + subquery.toSql());
      }

      List<Expr> aggFns = Lists.newArrayList();
      item.getExpr().collectAll(Expr.NON_NULL_EMPTY_AGG, aggFns);
      // TODO Generalize this by making the aggregate functions aware of the
      // literal expr that they return on empty input, e.g. max returns a
      // NullLiteral whereas count returns a NumericLiteral.
      if (((FunctionCallExpr)aggFns.get(0)).getReturnType().isNumericType()) {
        FunctionCallExpr zeroIfNull = new FunctionCallExpr("zeroifnull",
            Lists.newArrayList((Expr) slotRef));
        zeroIfNull.analyze(analyzer);
        subquerySubstitute = zeroIfNull;
      } else if (((FunctionCallExpr)aggFns.get(0)).getReturnType().isStringType()) {
        List<Expr> params = Lists.newArrayList();
        params.add(slotRef);
        params.add(new StringLiteral(""));
        FunctionCallExpr ifnull = new FunctionCallExpr("ifnull", params);
        ifnull.analyze(analyzer);
        subquerySubstitute = ifnull;
      } else {
        throw new AnalysisException("Unsupported aggregate function used in " +
            "a correlated subquery's select list: " + subquery.toSql());
      }
    }
    smap.put(subquery, subquerySubstitute);
    return exprWithSubquery.substitute(smap, analyzer, false);
  }
}