@Override
  public ParseContext transform(ParseContext pctx) throws SemanticException {
    LOG.info("SharedWorkOptimizer start");

    final Map<String, TableScanOperator> topOps = pctx.getTopOps();
    if (topOps.size() < 2) {
      // Nothing to do, bail out
      return pctx;
    }

    // Map of dbName.TblName -> TSOperator
    ListMultimap<String, TableScanOperator> tableNameToOps = splitTableScanOpsByTable(pctx);

    // Check whether all tables in the plan are unique
    boolean tablesReferencedOnlyOnce =
        tableNameToOps.asMap().entrySet().stream().noneMatch(e -> e.getValue().size() > 1);
    if (tablesReferencedOnlyOnce) {
      // Nothing to do, bail out
      return pctx;
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("Before SharedWorkOptimizer:\n" + Operator.toString(pctx.getTopOps().values()));
    }

    // We enforce a certain order when we do the reutilization.
    // In particular, we use size of table x number of reads to
    // rank the tables.
    List<Entry<String, Long>> sortedTables = rankTablesByAccumulatedSize(pctx);
    LOG.debug("Sorted tables by size: {}", sortedTables);

    // Cache to use during optimization
    SharedWorkOptimizerCache optimizerCache = new SharedWorkOptimizerCache();

    // Gather information about the DPP table scans and store it in the cache
    gatherDPPTableScanOps(pctx, optimizerCache);

    final int batchSize = HiveConf.getIntVar(pctx.getConf(), ConfVars.HIVE_SHARED_WORK_MAX_SIBLINGS);
    Preconditions.checkArgument(batchSize == -1 || batchSize > 0, "%s must be -1 or greater than 0",
        ConfVars.HIVE_SHARED_WORK_MAX_SIBLINGS.varname);
    for (List<TableScanOperator> scans : groupTableScanOperators(sortedTables, tableNameToOps, batchSize)) {
      // Execute shared work optimization
      runSharedWorkOptimization(pctx, optimizerCache, scans, Mode.SubtreeMerge);

      if (LOG.isDebugEnabled()) {
        LOG.debug("After SharedWorkOptimizer:\n" + Operator.toString(pctx.getTopOps().values()));
      }

      if (pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_EXTENDED_OPTIMIZATION)) {
        // Execute extended shared work optimization
        sharedWorkExtendedOptimization(pctx, optimizerCache);

        if (LOG.isDebugEnabled()) {
          LOG.debug("After SharedWorkExtendedOptimizer:\n" + Operator.toString(pctx.getTopOps().values()));
        }
      }

      if (pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_SEMIJOIN_OPTIMIZATION)) {

        // Execute shared work optimization with semijoin removal

        boolean optimized = runSharedWorkOptimization(pctx, optimizerCache, scans, Mode.RemoveSemijoin);
        if (optimized && pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_EXTENDED_OPTIMIZATION)) {
          // If it was further optimized, execute a second round of extended shared work optimizer
          sharedWorkExtendedOptimization(pctx, optimizerCache);
        }

        if (LOG.isDebugEnabled()) {
          LOG.debug("After SharedWorkSJOptimizer:\n" + Operator.toString(pctx.getTopOps().values()));
        }
      }

      if (pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_DPPUNION_OPTIMIZATION)) {
        boolean optimized = runSharedWorkOptimization(pctx, optimizerCache, scans, Mode.DPPUnion);

        if (optimized && pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_EXTENDED_OPTIMIZATION)) {
          // If it was further optimized, do a round of extended shared work optimizer
          sharedWorkExtendedOptimization(pctx, optimizerCache);
        }
        if (LOG.isDebugEnabled()) {
          LOG.debug("After DPPUnion:\n" + Operator.toString(pctx.getTopOps().values()));
        }
      }

    }

    if (pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_REUSE_MAPJOIN_CACHE)) {
      runMapJoinCacheReuseOptimization(pctx, optimizerCache);
    }

    // If we are running tests, we are going to verify that the contents of the cache
    // correspond with the contents of the plan, and otherwise we fail.
    // This check always run when we are running in test mode, independently on whether
    // we use the basic or the extended version of the optimizer.
    if (pctx.getConf().getBoolVar(ConfVars.HIVE_IN_TEST)) {
      Set<Operator<?>> visited = new HashSet<>();
      Iterator<Entry<String, TableScanOperator>> it = topOps.entrySet().iterator();
      while (it.hasNext()) {
        Entry<String, TableScanOperator> e = it.next();
        for (Operator<?> op : OperatorUtils.findOperators(e.getValue(), Operator.class)) {
          if (!visited.contains(op)) {
            Set<Operator<?>> workCachedOps = findWorkOperators(optimizerCache, op);
            Set<Operator<?>> workPlanOps = findWorkOperators(op, new HashSet<>());
            if (!workCachedOps.equals(workPlanOps)) {
              throw new SemanticException("Error in shared work optimizer: operator cache contents "
                  + "and actual plan differ\nIn cache: " + workCachedOps + "\nIn plan: " + workPlanOps);
            }
            visited.add(op);
          }
        }
      }
    }

    LOG.info("SharedWorkOptimizer end");
    return pctx;
  }
Shared computation optimizer.

<p>Originally, this rule would find scan operators over the same table
in the query plan and merge them if they met some preconditions.
<pre>
 TS   TS             TS
 |    |     -&gt;   /  \
 Op   Op           Op  Op
</pre>
<p>Now the rule has been extended to find opportunities to other operators
downstream, not only a single table scan.

 TS1   TS2    TS1   TS2            TS1   TS2
  |     |      |     |              |     |
  |    RS      |    RS              |    RS
   \   /        \   /       -&gt;    \   /
  MapJoin      MapJoin              MapJoin
     |            |                  /   \
     Op           Op                Op   Op

<p>If the extended version of the optimizer is enabled, it can go beyond
a work boundary to find reutilization opportunities.

<p>The optimization only works with the Tez execution engine.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java