private void removeSemijoinOptimizationByBenefit(OptimizeTezProcContext procCtx)
      throws SemanticException {

    Map<ReduceSinkOperator, SemiJoinBranchInfo> map = procCtx.parseContext.getRsToSemiJoinBranchInfo();
    if (map.isEmpty()) {
      // Nothing to do
      return;
    }

    // Scale down stats for tables with DPP
    Map<FilterOperator, Statistics> adjustedStatsMap = new HashMap<>();
    List<ReduceSinkOperator> semijoinRsToRemove = new ArrayList<>();
    double semijoinReductionThreshold = procCtx.conf.getFloatVar(
        HiveConf.ConfVars.TEZ_DYNAMIC_SEMIJOIN_REDUCTION_THRESHOLD);
    // Using SortedSet to make iteration order deterministic
    final Comparator<ReduceSinkOperator> rsOpComp =
        (ReduceSinkOperator o1, ReduceSinkOperator o2) -> (o1.toString().compareTo(o2.toString()));
    SortedSet<ReduceSinkOperator> semiJoinRsOps = new TreeSet<>(rsOpComp);
    semiJoinRsOps.addAll(map.keySet());
    ListMultimap<FilterOperator, SemijoinOperatorInfo> globalReductionFactorMap = ArrayListMultimap.create();
    while (!semiJoinRsOps.isEmpty()) {
      // We will gather the SJs to keep in the plan in the following map
      Map<FilterOperator, SemijoinOperatorInfo> reductionFactorMap = new HashMap<>();
      SortedSet<ReduceSinkOperator> semiJoinRsOpsNewIter = new TreeSet<>(rsOpComp);
      for (ReduceSinkOperator rs : semiJoinRsOps) {
        SemiJoinBranchInfo sjInfo = map.get(rs);
        if (sjInfo.getIsHint() || !sjInfo.getShouldRemove()) {
          // Semijoin created using hint or marked useful, skip it
          continue;
        }
        // rs is semijoin optimization branch, which should look like <Parent>-SEL-GB1-RS1-GB2-RS2
        SelectOperator sel = OperatorUtils.ancestor(rs, SelectOperator.class, 0, 0, 0, 0);

        // Check the ndv/rows from the SEL vs the destination tablescan the semijoin opt is going to.
        TableScanOperator ts = sjInfo.getTsOp();
        RuntimeValuesInfo rti = procCtx.parseContext.getRsToRuntimeValuesInfoMap().get(rs);
        List<ExprNodeDesc> targetColumns = rti.getTargetColumns();
        // In semijoin branches the SEL operator has the following forms:
        // SEL[c1] - single column semijoin reduction
        // SEL[c1, c2,..., ck, hash(hash(hash(c1, c2),...),ck)] - multi column semijoin reduction
        // The source columns in the above cases are c1, c2,...,ck.
        // We need to exclude the hash(...) expression, if it is present.
        List<ExprNodeDesc> sourceColumns = sel.getConf().getColList().subList(0, targetColumns.size());

        if (LOG.isDebugEnabled()) {
          LOG.debug("Computing BloomFilter cost/benefit for " + OperatorUtils.getOpNamePretty(rs)
              + " - " + OperatorUtils.getOpNamePretty(ts) + " " + targetColumns + " ");
        }

        FilterOperator filterOperator = (FilterOperator) ts.getChildOperators().get(0);
        Statistics filterStats = adjustedStatsMap.get(filterOperator);
        if (filterStats == null && filterOperator.getStatistics() != null) {
          filterStats = filterOperator.getStatistics().clone();
          adjustedStatsMap.put(filterOperator, filterStats);
        }
        double reductionFactor = computeBloomFilterNetBenefit(sel, sourceColumns, filterStats, targetColumns);
        if (reductionFactor < semijoinReductionThreshold) {
          // This semijoin optimization should be removed. Do it after we're done iterating
          semijoinRsToRemove.add(rs);
        } else {
          // This semijoin qualifies, add it to the result set
          if (filterStats != null) {
            ImmutableSet.Builder<String> colNames = ImmutableSet.builder();
            for (ExprNodeDesc tsExpr : targetColumns) {
              Set<ExprNodeColumnDesc> allReferencedColumns = ExprNodeDescUtils.findAllColumnDescs(tsExpr);
              for (ExprNodeColumnDesc col : allReferencedColumns) {
                colNames.add(col.getColumn());
              }
            }
            // We check whether there was already another SJ over this TS that was selected
            // in previous iteration
            SemijoinOperatorInfo prevResult = reductionFactorMap.get(filterOperator);
            if (prevResult != null) {
              if (prevResult.reductionFactor < reductionFactor) {
                // We should pick up new SJ as its reduction factor is greater than the previous one
                // that we found. We add the previous RS where SJ was originating to RS ops for new
                // iteration
                reductionFactorMap.put(filterOperator, new SemijoinOperatorInfo(rs, filterOperator,
                    filterStats, colNames.build(), reductionFactor));
                semiJoinRsOpsNewIter.add(prevResult.rsOperator);
                if (LOG.isDebugEnabled()) {
                  LOG.debug("Adding " + OperatorUtils.getOpNamePretty(prevResult.rsOperator)
                      + " for re-iteration");
                }
              } else {
                // We should pick up old SJ. We just need to add new RS where SJ was originating
                // to RS ops for new iteration
                semiJoinRsOpsNewIter.add(rs);
                if (LOG.isDebugEnabled()) {
                  LOG.debug("Adding " + OperatorUtils.getOpNamePretty(rs) + " for re-iteration");
                }
              }
            } else {
              // Another SJ did not exist for this TS, hence just add it to SJs to keep
              reductionFactorMap.put(filterOperator, new SemijoinOperatorInfo(rs, filterOperator,
                  filterStats, colNames.build(), reductionFactor));
            }
          }
        }
      }

      for (SemijoinOperatorInfo roi : reductionFactorMap.values()) {
        // This semijoin will be kept
        // We are going to adjust the filter statistics
        long newNumRows = (long) (1.0 - roi.reductionFactor) * roi.filterStats.getNumRows();
        if (LOG.isDebugEnabled()) {
          LOG.debug("Old stats for {}: {}", roi.filterOperator, roi.filterStats);
          LOG.debug("Number of rows reduction: {}/{}", newNumRows, roi.filterStats.getNumRows());
        }
        StatsUtils.updateStats(roi.filterStats, newNumRows,
            true, roi.filterOperator, roi.colNames);
        if (LOG.isDebugEnabled()) {
          LOG.debug("New stats for {}: {}", roi.filterOperator, roi.filterStats);
        }
        adjustedStatsMap.put(roi.filterOperator, roi.filterStats);
        globalReductionFactorMap.put(roi.filterOperator, roi);
      }

      semiJoinRsOps = semiJoinRsOpsNewIter;
    }

    for (ReduceSinkOperator rs : semijoinRsToRemove) {
      TableScanOperator ts = map.get(rs).getTsOp();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Reduction factor not satisfied for " + OperatorUtils.getOpNamePretty(rs)
            + "-" + OperatorUtils.getOpNamePretty(ts) + ". Removing semijoin optimization.");
      }
      GenTezUtils.removeBranch(rs);
      GenTezUtils.removeSemiJoinOperator(procCtx.parseContext, rs, ts);
    }

    if (!globalReductionFactorMap.isEmpty()) {
      sortSemijoinFilters(procCtx, globalReductionFactorMap);
    }
  }
Sort semijoin filters depending on the benefit (computed depending on selectivity and cost)
that they provide. We create three blocks: first all normal predicates, second between clauses
for the min/max dynamic values, and finally the in bloom filter predicates. The intuition is
that evaluating the between clause will be cheaper than evaluating the bloom filter predicates.
Hence, after this method runs, normal predicates come first (possibly sorted by Calcite),
then we will have sorted between clauses, and finally sorted in bloom filter clauses.
ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java