private static void sharedWorkExtendedOptimization(ParseContext pctx, SharedWorkOptimizerCache optimizerCache)
      throws SemanticException {
    // Gather RS operators that 1) belong to root works, i.e., works containing TS operators,
    // and 2) share the same input operator.
    // These will be the first target for extended shared work optimization
    Multimap<Operator<?>, ReduceSinkOperator> parentToRsOps = ArrayListMultimap.create();
    Set<Operator<?>> visited = new HashSet<>();
    for (Entry<String, TableScanOperator> e : pctx.getTopOps().entrySet()) {
      gatherReduceSinkOpsByInput(parentToRsOps, visited,
          findWorkOperators(optimizerCache, e.getValue()));
    }

    Set<Operator<?>> removedOps = new HashSet<>();
    while (!parentToRsOps.isEmpty()) {
      // As above, we enforce a certain order when we do the reutilization.
      // In particular, we use size of data in RS x number of uses.
      List<Entry<Operator<?>, Long>> sortedRSGroups =
          rankOpsByAccumulatedSize(parentToRsOps.keySet());
      LOG.debug("Sorted operators by size: {}", sortedRSGroups);

      // Execute extended optimization
      // For each RS, check whether other RS in same work could be merge into this one.
      // If they are merged, RS operators in the resulting work will be considered
      // mergeable in next loop iteration.
      Multimap<Operator<?>, ReduceSinkOperator> existingRsOps = ArrayListMultimap.create();
      for (Entry<Operator<?>, Long> rsGroupInfo : sortedRSGroups) {
        Operator<?> rsParent = rsGroupInfo.getKey();
        for (ReduceSinkOperator discardableRsOp : parentToRsOps.get(rsParent)) {
          if (removedOps.contains(discardableRsOp)) {
            LOG.debug("Skip {} as it has already been removed", discardableRsOp);
            continue;
          }
          Collection<ReduceSinkOperator> otherRsOps = existingRsOps.get(rsParent);
          for (ReduceSinkOperator retainableRsOp : otherRsOps) {
            if (retainableRsOp.getChildOperators().size() == 0) {
              // just skip this RS - its a semijoin/bloomfilter related RS
              continue;
            }
            if (removedOps.contains(retainableRsOp)) {
              LOG.debug("Skip {} as it has already been removed", retainableRsOp);
              continue;
            }

            // First we quickly check if the two RS operators can actually be merged.
            // We already know that these two RS operators have the same parent, but
            // we need to check whether both RS are actually equal. Further, we check
            // whether their child is also equal. If any of these conditions are not
            // met, we are not going to try to merge.
            boolean mergeable = compareOperator(pctx, retainableRsOp, discardableRsOp) &&
                compareOperator(pctx, retainableRsOp.getChildOperators().get(0),
                    discardableRsOp.getChildOperators().get(0));
            if (!mergeable) {
              // Skip
              LOG.debug("{} and {} cannot be merged", retainableRsOp, discardableRsOp);
              continue;
            }

            LOG.debug("Checking additional conditions for merging subtree starting at {}"
                + " into subtree starting at {}", discardableRsOp, retainableRsOp);

            // Secondly, we extract information about the part of the tree that can be merged
            // as well as some structural information (memory consumption) that needs to be
            // used to determined whether the merge can happen
            Operator<?> retainableRsOpChild = retainableRsOp.getChildOperators().get(0);
            Operator<?> discardableRsOpChild = discardableRsOp.getChildOperators().get(0);
            SharedResult sr = extractSharedOptimizationInfo(
                pctx, optimizerCache, retainableRsOp, discardableRsOp,
                retainableRsOpChild, discardableRsOpChild);

            // It seems these two operators can be merged.
            // Check that plan meets some preconditions before doing it.
            // In particular, in the presence of map joins in the upstream plan:
            // - we cannot exceed the noconditional task size, and
            // - if we already merged the big table, we cannot merge the broadcast
            // tables.
            if (sr.retainableOps.isEmpty() || !validPreConditions(pctx, optimizerCache, sr)) {
              // Skip
              LOG.debug("{} and {} do not meet preconditions", retainableRsOp, discardableRsOp);
              continue;
            }

            deduplicateReduceTraits(retainableRsOp.getConf(), discardableRsOp.getConf());

            // We can merge
            Operator<?> lastRetainableOp = sr.retainableOps.get(sr.retainableOps.size() - 1);
            Operator<?> lastDiscardableOp = sr.discardableOps.get(sr.discardableOps.size() - 1);
            if (lastDiscardableOp.getNumChild() != 0) {
              List<Operator<? extends OperatorDesc>> allChildren =
                  Lists.newArrayList(lastDiscardableOp.getChildOperators());
              for (Operator<? extends OperatorDesc> op : allChildren) {
                lastDiscardableOp.getChildOperators().remove(op);
                op.replaceParent(lastDiscardableOp, lastRetainableOp);
                lastRetainableOp.getChildOperators().add(op);
              }
            }

            LOG.debug("Merging subtree starting at {} into subtree starting at {}",
                discardableRsOp, retainableRsOp);

            // First we remove the input operators of the expression that
            // we are going to eliminate
            for (Operator<?> op : sr.discardableInputOps) {
              OperatorUtils.removeOperator(op);
              optimizerCache.removeOp(op);
              removedOps.add(op);
              // Remove DPP predicates
              if (op instanceof ReduceSinkOperator) {
                SemiJoinBranchInfo sjbi = pctx.getRsToSemiJoinBranchInfo().get(op);
                if (sjbi != null && !sr.discardableOps.contains(sjbi.getTsOp()) &&
                    !sr.discardableInputOps.contains(sjbi.getTsOp())) {
                  GenTezUtils.removeSemiJoinOperator(
                      pctx, (ReduceSinkOperator) op, sjbi.getTsOp());
                  optimizerCache.tableScanToDPPSource.remove(sjbi.getTsOp(), op);
                }
              } else if (op instanceof AppMasterEventOperator) {
                DynamicPruningEventDesc dped = (DynamicPruningEventDesc) op.getConf();
                if (!sr.discardableOps.contains(dped.getTableScan()) &&
                    !sr.discardableInputOps.contains(dped.getTableScan())) {
                  GenTezUtils.removeSemiJoinOperator(
                      pctx, (AppMasterEventOperator) op, dped.getTableScan());
                  optimizerCache.tableScanToDPPSource.remove(dped.getTableScan(), op);
                }
              }
              LOG.debug("Input operator removed: {}", op);
            }
            // We remove the discardable RS operator
            OperatorUtils.removeOperator(discardableRsOp);
            optimizerCache.removeOp(discardableRsOp);
            removedOps.add(discardableRsOp);
            LOG.debug("Operator removed: {}", discardableRsOp);
            // Then we merge the operators of the works we are going to merge
            optimizerCache.removeOpAndCombineWork(discardableRsOpChild, retainableRsOpChild);
            // Finally we remove the rest of the expression from the tree
            for (Operator<?> op : sr.discardableOps) {
              OperatorUtils.removeOperator(op);
              optimizerCache.removeOp(op);
              removedOps.add(op);
              LOG.debug("Operator removed: {}", op);
            }

            if (pctx.getConf().getBoolVar(ConfVars.HIVE_SHARED_WORK_DOWNSTREAM_MERGE)) {
              if (sr.discardableOps.size() == 1) {
                downStreamMerge(retainableRsOp, optimizerCache, pctx);
              }
            }

            break;
          }

          if (removedOps.contains(discardableRsOp)) {
            // This operator has been removed, remove it from the list of existing operators
            existingRsOps.remove(rsParent, discardableRsOp);
          } else {
            // This operator has not been removed, include it in the list of existing operators
            existingRsOps.put(rsParent, discardableRsOp);
          }
        }
      }

      // We gather the operators that will be used for next iteration of extended optimization
      // (if any)
      parentToRsOps = ArrayListMultimap.create();
      visited = new HashSet<>();
      for (Entry<Operator<?>, ReduceSinkOperator> e : existingRsOps.entries()) {
        if (removedOps.contains(e.getValue()) || e.getValue().getNumChild() < 1) {
          // If 1) RS has been removed, or 2) it does not have a child (for instance, it is a
          // semijoin RS), we can quickly skip this one
          continue;
        }
        gatherReduceSinkOpsByInput(parentToRsOps, visited,
            findWorkOperators(optimizerCache, e.getValue().getChildOperators().get(0)));
      }
    }

    // Remove unused table scan operators
    pctx.getTopOps().entrySet().removeIf(
        (Entry<String, TableScanOperator> e) -> e.getValue().getNumChild() == 0);
  }
When we call this method, we have already verified that the SJ expressions targeting
two TS operators are the same.
Since we already had a method to push the filter expressions on top of the discardable
TS (pushFilterToTopOfTableScan), here we remove the old SJ expressions from the
discardable TS (and follow-up Filters if present) and we add the SJ expressions
from the retainable TS. That way the SJ expressions will be pushed on top of the
discardable TS by pushFilterToTopOfTableScan.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java