// 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.List;
import java.util.stream.Collectors;

import org.apache.impala.analysis.AnalysisContext;
import org.apache.impala.analysis.AnalysisContext.AnalysisResult;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.ColumnLineageGraph;
import org.apache.impala.analysis.ColumnLineageGraph.ColumnLabel;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ExprSubstitutionMap;
import org.apache.impala.analysis.InsertStmt;
import org.apache.impala.analysis.JoinOperator;
import org.apache.impala.analysis.QueryStmt;
import org.apache.impala.analysis.SortInfo;
import org.apache.impala.analysis.TupleId;
import org.apache.impala.catalog.FeHBaseTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.common.PrintUtils;
import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryExecRequest;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TRuntimeFilterMode;
import org.apache.impala.thrift.TTableName;
import org.apache.impala.util.EventSequence;
import org.apache.impala.util.KuduUtil;
import org.apache.impala.util.MathUtil;
import org.apache.impala.util.MaxRowsProcessedVisitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;

import static org.apache.impala.analysis.ToSqlOptions.SHOW_IMPLICIT_CASTS;

/**
 * Creates an executable plan from an analyzed parse tree and query options.
 */
public class Planner {
  private final static Logger LOG = LoggerFactory.getLogger(Planner.class);

  // Minimum per-host resource requirements to ensure that no plan node set can have
  // estimates of zero, even if the contained PlanNodes have estimates of zero.
  public static final long MIN_PER_HOST_MEM_ESTIMATE_BYTES = 10 * 1024 * 1024;

  // The amount of memory added to a dedicated coordinator's memory estimate to
  // compensate for underestimation. In the general case estimates for exec
  // nodes tend to overestimate and should work fine but for estimates in the
  // 100-500 MB space, underestimates can be a problem. We pick a value of 100MB
  // because it is trivial for large estimates and small enough to not make a
  // huge impact on the coordinator's process memory (which ideally would be
  // large).
  public static final long DEDICATED_COORD_SAFETY_BUFFER_BYTES = 100 * 1024 * 1024;

  public static final ResourceProfile MIN_PER_HOST_RESOURCES =
      new ResourceProfileBuilder()
          .setMemEstimateBytes(MIN_PER_HOST_MEM_ESTIMATE_BYTES)
          .setMinMemReservationBytes(0)
          .build();

  private final PlannerContext ctx_;

  public Planner(AnalysisResult analysisResult, TQueryCtx queryCtx,
      EventSequence timeline) {
    ctx_ = new PlannerContext(analysisResult, queryCtx, timeline);
  }

  public TQueryCtx getQueryCtx() { return ctx_.getQueryCtx(); }
  public AnalysisContext.AnalysisResult getAnalysisResult() {
    return ctx_.getAnalysisResult();
  }

  /**
   * Returns a list of plan fragments for executing an analyzed parse tree.
   * May return a single-node or distributed executable plan. If enabled (through a
   * query option), computes runtime filters for dynamic partition pruning.
   *
   * Plan generation may fail and throw for the following reasons:
   * 1. Expr evaluation failed, e.g., during partition pruning.
   * 2. A certain feature is not yet implemented, e.g., physical join implementation for
   *    outer/semi joins without equi conjuncts.
   * 3. Expr substitution failed, e.g., because an expr was substituted with a type that
   *    render the containing expr semantically invalid. Analysis should have ensured
   *    that such an expr substitution during plan generation never fails. If it does,
   *    that typically means there is a bug in analysis, or a broken/missing smap.
   */
  public List<PlanFragment> createPlan() throws ImpalaException {
    SingleNodePlanner singleNodePlanner = new SingleNodePlanner(ctx_);
    DistributedPlanner distributedPlanner = new DistributedPlanner(ctx_);
    PlanNode singleNodePlan = singleNodePlanner.createSingleNodePlan();
    ctx_.getTimeline().markEvent("Single node plan created");
    List<PlanFragment> fragments = null;

    checkForSmallQueryOptimization(singleNodePlan);

    // Join rewrites.
    invertJoins(singleNodePlan, ctx_.isSingleNodeExec());
    singleNodePlan = useNljForSingularRowBuilds(singleNodePlan, ctx_.getRootAnalyzer());

    // MT_DOP > 0 is not supported by default for plans with base table joins or table
    // sinks: we only allow MT_DOP > 0 with such plans if --unlock_mt_dop=true is
    // specified. We allow single node plans with mt_dop since there is no actual
    // parallelism.
    if (!ctx_.isSingleNodeExec() && ctx_.getQueryOptions().mt_dop > 0
        && (!RuntimeEnv.INSTANCE.isTestEnv()
               || RuntimeEnv.INSTANCE.isMtDopValidationEnabled())
        && !BackendConfig.INSTANCE.isMtDopUnlocked()
        && (ctx_.hasTableSink()
               || singleNodePlanner.hasUnsupportedMtDopJoin(singleNodePlan))) {
      if (BackendConfig.INSTANCE.mtDopAutoFallback()) {
        // Fall back to non-dop mode. This assumes that the mt_dop value is only used
        // in the distributed planning process, which should be generally true as long
        // as the value isn't cached in any plan nodes.
        ctx_.getQueryOptions().setMt_dop(0);
      } else {
        throw new NotImplementedException(
            "MT_DOP not supported for plans with base table joins or table sinks.");
      }
    }

    singleNodePlanner.validatePlan(singleNodePlan);

    if (ctx_.isSingleNodeExec()) {
      // create one fragment containing the entire single-node plan tree
      fragments = Lists.newArrayList(new PlanFragment(
          ctx_.getNextFragmentId(), singleNodePlan, DataPartition.UNPARTITIONED));
    } else {
      // create distributed plan
      fragments = distributedPlanner.createPlanFragments(singleNodePlan);
    }

    // Create runtime filters.
    PlanFragment rootFragment = fragments.get(fragments.size() - 1);
    if (ctx_.getQueryOptions().getRuntime_filter_mode() != TRuntimeFilterMode.OFF) {
      RuntimeFilterGenerator.generateRuntimeFilters(ctx_, rootFragment.getPlanRoot());
      ctx_.getTimeline().markEvent("Runtime filters computed");
    }

    rootFragment.verifyTree();
    ExprSubstitutionMap rootNodeSmap = rootFragment.getPlanRoot().getOutputSmap();
    if (ctx_.isInsertOrCtas()) {
      InsertStmt insertStmt = ctx_.getAnalysisResult().getInsertStmt();
      insertStmt.substituteResultExprs(rootNodeSmap, ctx_.getRootAnalyzer());
      if (!ctx_.isSingleNodeExec()) {
        // repartition on partition keys
        rootFragment = distributedPlanner.createInsertFragment(
            rootFragment, insertStmt, ctx_.getRootAnalyzer(), fragments);
      }
      // Add optional sort node to the plan, based on clustered/noclustered plan hint.
      createPreInsertSort(insertStmt, rootFragment, ctx_.getRootAnalyzer());
      // set up table sink for root fragment
      rootFragment.setSink(insertStmt.createDataSink());
    } else {
      QueryStmt queryStmt = ctx_.getQueryStmt();
      queryStmt.substituteResultExprs(rootNodeSmap, ctx_.getRootAnalyzer());
      List<Expr> resultExprs = queryStmt.getResultExprs();
      if (ctx_.isUpdate()) {
        // Set up update sink for root fragment
        rootFragment.setSink(
            ctx_.getAnalysisResult().getUpdateStmt().createDataSink(resultExprs));
      } else if (ctx_.isDelete()) {
        // Set up delete sink for root fragment
        rootFragment.setSink(
            ctx_.getAnalysisResult().getDeleteStmt().createDataSink(resultExprs));
      } else if (ctx_.isQuery()) {
        rootFragment.setSink(
            ctx_.getAnalysisResult().getQueryStmt().createDataSink(resultExprs));
      }
    }

    // The check for disabling codegen uses estimates of rows per node so must be done
    // on the distributed plan.
    checkForDisableCodegen(rootFragment.getPlanRoot());

    if (LOG.isTraceEnabled()) {
      LOG.trace("desctbl: " + ctx_.getRootAnalyzer().getDescTbl().debugString());
      LOG.trace("root sink: " + rootFragment.getSink().getExplainString(
            "", "", ctx_.getQueryOptions(), TExplainLevel.VERBOSE));
      LOG.trace("finalize plan fragments");
    }
    for (PlanFragment fragment: fragments) {
      fragment.finalizeExchanges(ctx_.getRootAnalyzer());
    }

    Collections.reverse(fragments);
    ctx_.getTimeline().markEvent("Distributed plan created");

    ColumnLineageGraph graph = ctx_.getRootAnalyzer().getColumnLineageGraph();
    if (BackendConfig.INSTANCE.getComputeLineage() || RuntimeEnv.INSTANCE.isTestEnv()) {
      // Lineage is disabled for UPDATE AND DELETE statements
      if (ctx_.isUpdateOrDelete()) return fragments;
      // Compute the column lineage graph
      if (ctx_.isInsertOrCtas()) {
        InsertStmt insertStmt = ctx_.getAnalysisResult().getInsertStmt();
        FeTable targetTable = insertStmt.getTargetTable();
        Preconditions.checkNotNull(targetTable);
        if (targetTable instanceof FeKuduTable) {
          if (ctx_.isInsert()) {
            // For insert statements on Kudu tables, we only need to consider
            // the labels of columns mentioned in the column list.
            List<String> mentionedColumns = insertStmt.getMentionedColumns();
            Preconditions.checkState(!mentionedColumns.isEmpty());
            List<ColumnLabel> targetColLabels = new ArrayList<>();
            String tblFullName = targetTable.getFullName();
            for (String column: mentionedColumns) {
              targetColLabels.add(new ColumnLabel(column, targetTable.getTableName()));
            }
            graph.addTargetColumnLabels(targetColLabels);
          } else {
            graph.addTargetColumnLabels(targetTable);
          }
        } else if (targetTable instanceof FeHBaseTable) {
          graph.addTargetColumnLabels(targetTable);
        } else {
          graph.addTargetColumnLabels(targetTable);
        }
      } else {
        graph.addTargetColumnLabels(ctx_.getQueryStmt().getColLabels().stream()
            .map(col -> new ColumnLabel(col))
            .collect(Collectors.toList()));
      }
      List<Expr> outputExprs = new ArrayList<>();
      rootFragment.getSink().collectExprs(outputExprs);
      graph.computeLineageGraph(outputExprs, ctx_.getRootAnalyzer());
      if (LOG.isTraceEnabled()) LOG.trace("lineage: " + graph.debugString());
      ctx_.getTimeline().markEvent("Lineage info computed");
    }

    return fragments;
  }

  /**
   * Return a list of plans, each represented by the root of their fragment trees.
   * TODO: roll into createPlan()
   */
  public List<PlanFragment> createParallelPlans() throws ImpalaException {
    Preconditions.checkState(ctx_.getQueryOptions().mt_dop > 0);
    List<PlanFragment> distrPlan = createPlan();
    Preconditions.checkNotNull(distrPlan);
    List<PlanFragment> parallelPlans;
    // TODO: IMPALA-4224: Parallel plans are not executable
    if (RuntimeEnv.INSTANCE.isTestEnv()) {
      ParallelPlanner planner = new ParallelPlanner(ctx_);
      parallelPlans = planner.createPlans(distrPlan.get(0));
    } else {
      parallelPlans = Collections.singletonList(distrPlan.get(0));
    }
    ctx_.getTimeline().markEvent("Parallel plans created");
    return parallelPlans;
  }

  /**
   * Return combined explain string for all plan fragments.
   * Includes the estimated resource requirements from the request if set.
   * Uses a default level of EXTENDED, unless overriden by the
   * 'explain_level' query option.
   */
  public String getExplainString(List<PlanFragment> fragments,
      TQueryExecRequest request) {
    // use EXTENDED by default for all non-explain statements
    TExplainLevel explainLevel = TExplainLevel.EXTENDED;
    // use the query option for explain stmts and tests (e.g., planner tests)
    if (ctx_.getAnalysisResult().isExplainStmt() || RuntimeEnv.INSTANCE.isTestEnv()) {
      explainLevel = ctx_.getQueryOptions().getExplain_level();
    }
    return getExplainString(fragments, request, explainLevel);
  }

  /**
   * Return combined explain string for all plan fragments and with an
   * explicit explain level.
   * Includes the estimated resource requirements from the request if set.
   */
  public String getExplainString(List<PlanFragment> fragments,
      TQueryExecRequest request, TExplainLevel explainLevel) {
    StringBuilder str = new StringBuilder();
    boolean hasHeader = false;

    // Only some requests (queries, DML, etc) have a resource profile.
    if (request.isSetMax_per_host_min_mem_reservation()) {
      Preconditions.checkState(request.isSetMax_per_host_thread_reservation());
      Preconditions.checkState(request.isSetPer_host_mem_estimate());
      str.append(String.format(
          "Max Per-Host Resource Reservation: Memory=%s Threads=%d\n",
          PrintUtils.printBytes(request.getMax_per_host_min_mem_reservation()),
          request.getMax_per_host_thread_reservation()));
      str.append(String.format("Per-Host Resource Estimates: Memory=%s\n",
          PrintUtils.printBytesRoundedToMb(request.getPer_host_mem_estimate())));
      if (BackendConfig.INSTANCE.useDedicatedCoordinatorEstimates()) {
        str.append(String.format("Dedicated Coordinator Resource Estimate: Memory=%s\n",
            PrintUtils.printBytesRoundedToMb(request.getDedicated_coord_mem_estimate())));
      }
      hasHeader = true;
    }
    // Warn if the planner is running in DEBUG mode.
    if (request.query_ctx.client_request.query_options.planner_testcase_mode) {
      str.append("WARNING: The planner is running in TESTCASE mode. This should only be "
          + "used by developers for debugging.\nTo disable it, do SET " +
          "PLANNER_TESTCASE_MODE=false.\n");
    }
    if (request.query_ctx.disable_codegen_hint) {
      str.append("Codegen disabled by planner\n");
    }

    // IMPALA-1983 In the case of corrupt stats, issue a warning for all queries except
    // child queries of 'compute stats'.
    if (!request.query_ctx.isSetParent_query_id() &&
        request.query_ctx.isSetTables_with_corrupt_stats() &&
        !request.query_ctx.getTables_with_corrupt_stats().isEmpty()) {
      List<String> tableNames = new ArrayList<>();
      for (TTableName tableName: request.query_ctx.getTables_with_corrupt_stats()) {
        tableNames.add(tableName.db_name + "." + tableName.table_name);
      }
      str.append(
          "WARNING: The following tables have potentially corrupt table statistics.\n" +
          "Drop and re-compute statistics to resolve this problem.\n" +
          Joiner.on(", ").join(tableNames) + "\n");
      hasHeader = true;
    }

    // Append warning about tables missing stats except for child queries of
    // 'compute stats'. The parent_query_id is only set for compute stats child queries.
    if (!request.query_ctx.isSetParent_query_id() &&
        request.query_ctx.isSetTables_missing_stats() &&
        !request.query_ctx.getTables_missing_stats().isEmpty()) {
      List<String> tableNames = new ArrayList<>();
      for (TTableName tableName: request.query_ctx.getTables_missing_stats()) {
        tableNames.add(tableName.db_name + "." + tableName.table_name);
      }
      str.append("WARNING: The following tables are missing relevant table " +
          "and/or column statistics.\n" + Joiner.on(", ").join(tableNames) + "\n");
      hasHeader = true;
    }

    if (request.query_ctx.isSetTables_missing_diskids()) {
      List<String> tableNames = new ArrayList<>();
      for (TTableName tableName: request.query_ctx.getTables_missing_diskids()) {
        tableNames.add(tableName.db_name + "." + tableName.table_name);
      }
      str.append("WARNING: The following tables have scan ranges with missing " +
          "disk id information.\n" + Joiner.on(", ").join(tableNames) + "\n");
      hasHeader = true;
    }

    if (request.query_ctx.isDisable_spilling()) {
      str.append("WARNING: Spilling is disabled for this query as a safety guard.\n" +
          "Reason: Query option disable_unsafe_spills is set, at least one table\n" +
          "is missing relevant stats, and no plan hints were given.\n");
      hasHeader = true;
    }

    if (explainLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
      // In extended explain include the analyzed query text showing implicit casts
      String queryText = ctx_.getQueryStmt().toSql(SHOW_IMPLICIT_CASTS);
      String wrappedText = PrintUtils.wrapString("Analyzed query: " + queryText, 80);
      str.append(wrappedText).append("\n");
      hasHeader = true;
    }
    // Note that the analyzed query text must be the last thing in the header.
    // This is to help tests that parse the header.

    // Add the blank line that indicates the end of the header
    if (hasHeader) str.append("\n");

    if (explainLevel.ordinal() < TExplainLevel.VERBOSE.ordinal()) {
      // Print the non-fragmented parallel plan.
      str.append(fragments.get(0).getExplainString(ctx_.getQueryOptions(), explainLevel));
    } else {
      // Print the fragmented parallel plan.
      for (int i = 0; i < fragments.size(); ++i) {
        PlanFragment fragment = fragments.get(i);
        str.append(fragment.getExplainString(ctx_.getQueryOptions(), explainLevel));
        if (i < fragments.size() - 1) str.append("\n");
      }
    }
    return str.toString();
  }

  /**
   * Computes the per-host resource profile for the given plans, i.e. the peak resources
   * consumed by all fragment instances belonging to the query per host. Sets the
   * per-host resource values in 'request'.
   */
  public void computeResourceReqs(List<PlanFragment> planRoots,
      TQueryCtx queryCtx, TQueryExecRequest request) {
    Preconditions.checkState(!planRoots.isEmpty());
    Preconditions.checkNotNull(request);
    TQueryOptions queryOptions = ctx_.getRootAnalyzer().getQueryOptions();
    int mtDop = queryOptions.getMt_dop();

    // Peak per-host peak resources for all plan fragments, assuming that all fragments
    // are scheduled on all nodes. The actual per-host resource requirements are computed
    // after scheduling.
    ResourceProfile maxPerHostPeakResources = ResourceProfile.invalid();
    long totalRuntimeFilterMemBytes = 0;

    // Do a pass over all the fragments to compute resource profiles. Compute the
    // profiles bottom-up since a fragment's profile may depend on its descendants.
    PlanFragment rootFragment = planRoots.get(0);
    List<PlanFragment> allFragments = rootFragment.getNodesPostOrder();
    for (PlanFragment fragment: allFragments) {
      // Compute the per-node, per-sink and aggregate profiles for the fragment.
      fragment.computeResourceProfile(ctx_.getRootAnalyzer());

      // Different fragments do not synchronize their Open() and Close(), so the backend
      // does not provide strong guarantees about whether one fragment instance releases
      // resources before another acquires them. Conservatively assume that all fragment
      // instances run on all backends with max DOP, and can consume their peak resources
      // at the same time, i.e. that the query-wide peak resources is the sum of the
      // per-fragment-instance peak resources.
      maxPerHostPeakResources = maxPerHostPeakResources.sum(
          fragment.getResourceProfile().multiply(fragment.getNumInstancesPerHost(mtDop)));
      // Coordinator has to have a copy of each of the runtime filters to perform filter
      // aggregation.
      totalRuntimeFilterMemBytes += fragment.getRuntimeFiltersMemReservationBytes();
    }
    rootFragment.computePipelineMembership();

    Preconditions.checkState(maxPerHostPeakResources.getMemEstimateBytes() >= 0,
        maxPerHostPeakResources.getMemEstimateBytes());
    Preconditions.checkState(maxPerHostPeakResources.getMinMemReservationBytes() >= 0,
        maxPerHostPeakResources.getMinMemReservationBytes());

    maxPerHostPeakResources = MIN_PER_HOST_RESOURCES.max(maxPerHostPeakResources);

    request.setPer_host_mem_estimate(maxPerHostPeakResources.getMemEstimateBytes());
    request.setMax_per_host_min_mem_reservation(
        maxPerHostPeakResources.getMinMemReservationBytes());
    request.setMax_per_host_thread_reservation(
        maxPerHostPeakResources.getThreadReservation());
    if (getAnalysisResult().isQueryStmt()) {
      request.setDedicated_coord_mem_estimate(MathUtil.saturatingAdd(rootFragment
          .getResourceProfile().getMemEstimateBytes(), totalRuntimeFilterMemBytes +
          DEDICATED_COORD_SAFETY_BUFFER_BYTES));
    } else {
      // For queries that don't have a coordinator fragment, estimate a small
      // amount of memory that the query state spwaned on the coordinator can use.
      request.setDedicated_coord_mem_estimate(totalRuntimeFilterMemBytes +
          DEDICATED_COORD_SAFETY_BUFFER_BYTES);
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Max per-host min reservation: " +
          maxPerHostPeakResources.getMinMemReservationBytes());
      LOG.trace("Max estimated per-host memory: " +
          maxPerHostPeakResources.getMemEstimateBytes());
      LOG.trace("Max estimated per-host thread reservation: " +
          maxPerHostPeakResources.getThreadReservation());
    }
  }


  /**
   * Traverses the plan tree rooted at 'root' and inverts joins in the following
   * situations:
   * 1. If the left-hand side is a SingularRowSrcNode then we invert the join because
   *    then the build side is guaranteed to have only a single row.
   * 2. There is no backend support for distributed non-equi right outer/semi joins,
   *    so we invert them (any distributed left semi/outer join is ok).
   * 3. If we estimate that the inverted join is cheaper (see isInvertedJoinCheaper()).
   *    Do not invert if relevant stats are missing.
   * The first two inversion rules are independent of the presence/absence of stats.
   * Left Null Aware Anti Joins are never inverted due to lack of backend support.
   * Joins that originate from query blocks with a straight join hint are not inverted.
   * The 'isLocalPlan' parameter indicates whether the plan tree rooted at 'root'
   * will be executed locally within one machine, i.e., without any data exchanges.
   */
  private void invertJoins(PlanNode root, boolean isLocalPlan) {
    if (root instanceof SubplanNode) {
      invertJoins(root.getChild(0), isLocalPlan);
      invertJoins(root.getChild(1), true);
    } else {
      for (PlanNode child: root.getChildren()) invertJoins(child, isLocalPlan);
    }

    if (root instanceof JoinNode) {
      JoinNode joinNode = (JoinNode) root;
      JoinOperator joinOp = joinNode.getJoinOp();

      if (!joinNode.isInvertible(isLocalPlan)) {
        // Re-compute tuple ids since their order must correspond to the order
        // of children.
        root.computeTupleIds();
        return;
      }

      if (joinNode.getChild(0) instanceof SingularRowSrcNode) {
        // Always place a singular row src on the build side because it
        // only produces a single row.
        joinNode.invertJoin();
      } else if (!isLocalPlan && joinNode instanceof NestedLoopJoinNode &&
          (joinOp.isRightSemiJoin() || joinOp.isRightOuterJoin())) {
        // The current join is a distributed non-equi right outer or semi join
        // which has no backend support. Invert the join to make it executable.
        joinNode.invertJoin();
      } else if (isInvertedJoinCheaper(joinNode, isLocalPlan)) {
        joinNode.invertJoin();
      }
    }

    // Re-compute tuple ids because the backend assumes that their order corresponds to
    // the order of children.
    root.computeTupleIds();
  }

  /**
   * Return true if we estimate that 'joinNode' will be cheaper to execute after
   * inversion. Returns false if any join input is missing relevant stats.
   *
   * For nested loop joins, we simply assume that the cost is determined by the size of
   * the build side.
   *
   * For hash joins, the cost model is more nuanced and depends on:
   * - est. number of rows in the build and probe: lhsCard and rhsCard
   * - est. size of the rows in the build and probe: lhsAvgRowSize and rhsAvgRowSize
   * - est. parallelism with which the lhs and rhs trees execute: lhsNumNodes
   *   and rhsNumNodes. The parallelism of the join is determined by the lhs.
   *
   * The assumptions are:
   * - the join strategy is PARTITIONED and rows are distributed evenly. We don't know
   *   what join strategy will be chosen until later in planning so this assumption
   *   simplifies the analysis. Generally if one input is small enough that broadcast
   *   join is viable then this formula will prefer to put that input on the right side
   *   anyway.
   * - processing a build row is twice as expensive as processing a probe row of the
   *   same size.
   * - the cost of processing each byte of a row has a fixed component (C) (e.g.
   *   hashing and comparing the row) and a variable component (e.g. looking up the
   *   hash table).
   * - The variable component grows proportionally to the log of the build side, to
   *   approximate the effect of accesses to the the hash table hitting slower levels
   *   of the memory hierarchy.
   *
   * The estimated per-host cost of a hash join before and after inversion, measured in
   * an arbitrary unit of time, is then:
   *
   *    (log_b(rhsBytes) + C) * (lhsBytes + 2 * rhsBytes) / lhsNumNodes
   *    vs.
   *    (log_b(lhsBytes) + C) * (rhsBytes + 2 * lhsBytes) / rhsNumNodes
   *
   * where lhsBytes = lhsCard * lhsAvgRowSize and rhsBytes = rhsCard * rhsAvgRowSize
   *
   * We choose b = 10 and C = 5 empirically because it seems to give reasonable
   * results for a range of inputs. The model is not particularly sensitive to the
   * parameters.
   *
   * If the parallelism of both sides is the same then this reduces to comparing
   * the size of input on both sides. Otherwise, if inverting a hash join reduces
   * parallelism significantly, then a significant difference between lhs and rhs
   * bytes is needed to justify inversion.
   */
  private boolean isInvertedJoinCheaper(JoinNode joinNode, boolean isLocalPlan) {
    long lhsCard = joinNode.getChild(0).getCardinality();
    long rhsCard = joinNode.getChild(1).getCardinality();
    // Need cardinality estimates to make a decision.
    if (lhsCard == -1 || rhsCard == -1) return false;
    double lhsBytes = lhsCard * joinNode.getChild(0).getAvgRowSize();
    double rhsBytes = rhsCard * joinNode.getChild(1).getAvgRowSize();
    if (joinNode instanceof NestedLoopJoinNode) {
      // For NLJ, simply try to minimize the size of the build side, since it needs to
      // be broadcast to all participating nodes.
      return lhsBytes < rhsBytes;
    }
    Preconditions.checkState(joinNode instanceof HashJoinNode);
    int lhsNumNodes = isLocalPlan ? 1 : joinNode.getChild(0).getNumNodes();
    int rhsNumNodes = isLocalPlan ? 1 : joinNode.getChild(1).getNumNodes();
    // Need parallelism to determine whether inverting a hash join is profitable.
    if (lhsNumNodes <= 0 || rhsNumNodes <= 0) return false;

    final long CONSTANT_COST_PER_BYTE = 5;
    // Add 1 to the log argument to avoid taking log of 0.
    double totalCost =
        (Math.log10(rhsBytes + 1) + CONSTANT_COST_PER_BYTE) * (lhsBytes + 2 * rhsBytes);
    double invertedTotalCost =
        (Math.log10(lhsBytes + 1) + CONSTANT_COST_PER_BYTE) * (rhsBytes + 2 * lhsBytes);
    double perNodeCost = totalCost / lhsNumNodes;
    double invertedPerNodeCost = invertedTotalCost / rhsNumNodes;
    if (LOG.isTraceEnabled()) {
      LOG.trace("isInvertedJoinCheaper() " + TupleId.printIds(joinNode.getTupleIds()));
      LOG.trace("lhsCard " + lhsCard + " lhsBytes " + lhsBytes +
          " lhsNumNodes " + lhsNumNodes);
      LOG.trace("rhsCard " + rhsCard + " rhsBytes " + rhsBytes +
          " rhsNumNodes " + rhsNumNodes);
      LOG.trace("cost " + perNodeCost + " invCost " + invertedPerNodeCost);
      LOG.trace("INVERT? " + (invertedPerNodeCost < perNodeCost));
    }
    return invertedPerNodeCost < perNodeCost;
  }

  /**
   * Converts hash joins to nested-loop joins if the right-side is a SingularRowSrcNode.
   * Does not convert Null Aware Anti Joins because we only support that join op with
   * a hash join.
   * Throws if JoinNode.init() fails on the new nested-loop join node.
   */
  private PlanNode useNljForSingularRowBuilds(PlanNode root, Analyzer analyzer)
      throws ImpalaException {
    for (int i = 0; i < root.getChildren().size(); ++i) {
      root.setChild(i, useNljForSingularRowBuilds(root.getChild(i), analyzer));
    }
    if (!(root instanceof JoinNode)) return root;
    if (root instanceof NestedLoopJoinNode) return root;
    if (!(root.getChild(1) instanceof SingularRowSrcNode)) return root;
    JoinNode joinNode = (JoinNode) root;
    if (joinNode.getJoinOp().isNullAwareLeftAntiJoin()) {
      Preconditions.checkState(joinNode instanceof HashJoinNode);
      return root;
    }
    List<Expr> otherJoinConjuncts = Lists.newArrayList(joinNode.getOtherJoinConjuncts());
    otherJoinConjuncts.addAll(joinNode.getEqJoinConjuncts());
    JoinNode newJoinNode = new NestedLoopJoinNode(joinNode.getChild(0),
        joinNode.getChild(1), joinNode.isStraightJoin(),
        joinNode.getDistributionModeHint(), joinNode.getJoinOp(), otherJoinConjuncts);
    newJoinNode.getConjuncts().addAll(joinNode.getConjuncts());
    newJoinNode.setId(joinNode.getId());
    newJoinNode.init(analyzer);
    return newJoinNode;
  }

  private void checkForSmallQueryOptimization(PlanNode singleNodePlan) {
    MaxRowsProcessedVisitor visitor = new MaxRowsProcessedVisitor();
    singleNodePlan.accept(visitor);
    // TODO: IMPALA-3335: support the optimization for plans with joins.
    if (!visitor.valid() || visitor.foundJoinNode()) return;
    // This optimization executes the plan on a single node so the threshold must
    // be based on the total number of rows processed.
    long maxRowsProcessed = visitor.getMaxRowsProcessed();
    int threshold = ctx_.getQueryOptions().exec_single_node_rows_threshold;
    if (maxRowsProcessed < threshold) {
      // Execute on a single node and disable codegen for small results
      LOG.trace("Query is small enough to execute on a single node: maxRowsProcessed = "
          + maxRowsProcessed);
      ctx_.getQueryOptions().setNum_nodes(1);
      ctx_.getQueryCtx().disable_codegen_hint = true;
      if (maxRowsProcessed < ctx_.getQueryOptions().batch_size ||
          maxRowsProcessed < 1024 && ctx_.getQueryOptions().batch_size == 0) {
        // Only one scanner thread for small queries
        ctx_.getQueryOptions().setNum_scanner_threads(1);
      }
      // disable runtime filters
      ctx_.getQueryOptions().setRuntime_filter_mode(TRuntimeFilterMode.OFF);
    }
  }

  private void checkForDisableCodegen(PlanNode distributedPlan) {
    MaxRowsProcessedVisitor visitor = new MaxRowsProcessedVisitor();
    distributedPlan.accept(visitor);
    if (!visitor.valid()) return;
    // This heuristic threshold tries to determine if the per-node codegen time will
    // reduce per-node execution time enough to justify the cost of codegen. Per-node
    // execution time is correlated with the number of rows flowing through the plan.
    if (visitor.getMaxRowsProcessedPerNode()
        < ctx_.getQueryOptions().getDisable_codegen_rows_threshold()) {
      ctx_.getQueryCtx().disable_codegen_hint = true;
    }
  }

  /**
   * Insert a sort node on top of the plan, depending on the clustered/noclustered
   * plan hint and on the 'sort.columns' table property. If clustering is enabled in
   * insertStmt or additional columns are specified in the 'sort.columns' table property,
   * then the ordering columns will start with the clustering columns (key columns for
   * Kudu tables), so that partitions can be written sequentially in the table sink. Any
   * additional non-clustering columns specified by the 'sort.columns' property will be
   * added to the ordering columns and after any clustering columns. If no clustering is
   * requested and the table does not contain columns in the 'sort.columns' property, then
   * no sort node will be added to the plan.
   */
  public void createPreInsertSort(InsertStmt insertStmt, PlanFragment inputFragment,
       Analyzer analyzer) throws ImpalaException {
    List<Expr> orderingExprs = new ArrayList<>();

    boolean partialSort = false;
    if (insertStmt.getTargetTable() instanceof FeKuduTable) {
      // Always sort if the 'clustered' hint is present. Otherwise, don't sort if either
      // the 'noclustered' hint is present, or this is a single node exec, or if the
      // target table is unpartitioned.
      if (insertStmt.hasClusteredHint() || (!insertStmt.hasNoClusteredHint()
          && !ctx_.isSingleNodeExec() && !insertStmt.getPartitionKeyExprs().isEmpty())) {
        orderingExprs.add(
            KuduUtil.createPartitionExpr(insertStmt, ctx_.getRootAnalyzer()));
        orderingExprs.addAll(insertStmt.getPrimaryKeyExprs());
        partialSort = true;
      }
    } else if (insertStmt.requiresClustering()) {
      orderingExprs.addAll(insertStmt.getPartitionKeyExprs());
    }
    orderingExprs.addAll(insertStmt.getSortExprs());
    // Ignore constants for the sake of clustering.
    Expr.removeConstants(orderingExprs);

    if (orderingExprs.isEmpty()) return;

    // Build sortinfo to sort by the ordering exprs.
    List<Boolean> isAscOrder = Collections.nCopies(orderingExprs.size(), true);
    List<Boolean> nullsFirstParams = Collections.nCopies(orderingExprs.size(), false);
    SortInfo sortInfo = new SortInfo(orderingExprs, isAscOrder, nullsFirstParams,
        insertStmt.getSortingOrder());
    sortInfo.createSortTupleInfo(insertStmt.getResultExprs(), analyzer);
    sortInfo.getSortTupleDescriptor().materializeSlots();
    insertStmt.substituteResultExprs(sortInfo.getOutputSmap(), analyzer);

    PlanNode node = null;
    if (partialSort) {
      node = SortNode.createPartialSortNode(
          ctx_.getNextNodeId(), inputFragment.getPlanRoot(), sortInfo);
    } else {
      node = SortNode.createTotalSortNode(
          ctx_.getNextNodeId(), inputFragment.getPlanRoot(), sortInfo, 0);
    }
    node.init(analyzer);

    inputFragment.setPlanRoot(node);
  }
}