private void findPossibleAutoConvertedJoinOperators() throws SemanticException {
    // Guess if CommonJoinResolver will work. If CommonJoinResolver may
    // convert a join operation, correlation optimizer will not merge that join.
    // TODO: If hive.auto.convert.join.noconditionaltask=true, for a JoinOperator
    // that has both intermediate tables and query input tables as input tables,
    // we should be able to guess if this JoinOperator will be converted to a MapJoin
    // based on hive.auto.convert.join.noconditionaltask.size.
    for (JoinOperator joinOp: pCtx.getJoinOps()) {
      boolean isAbleToGuess = true;
      boolean mayConvert = false;
      // Get total size and individual alias's size
      long aliasTotalKnownInputSize = 0;
      Map<String, Long> aliasToSize = new HashMap<String, Long>();
      Map<Integer, Set<String>> posToAliases = new HashMap<Integer, Set<String>>();
      for (int pos = 0; pos < joinOp.getNumParent(); pos++) {
        Operator<? extends OperatorDesc> op = joinOp.getParentOperators().get(pos);
        Set<TableScanOperator> topOps = CorrelationUtilities.findTableScanOperators(op);
        if (topOps.isEmpty()) {
          isAbleToGuess = false;
          break;
        }

        Set<String> aliases = new LinkedHashSet<String>();
        for (TableScanOperator tsop : topOps) {
          Table table = tsop.getConf().getTableMetadata();
          if (table == null) {
            // table should not be null.
            throw new SemanticException("The table of " +
                tsop.getName() + " " + tsop.getIdentifier() +
                " is null, which is not expected.");
          }
          String alias = tsop.getConf().getAlias();
          aliases.add(alias);

          Path p = table.getPath();
          ContentSummary resultCs = null;
          try {
            FileSystem fs = table.getPath().getFileSystem(pCtx.getConf());
            resultCs = fs.getContentSummary(p);
          } catch (IOException e) {
            LOG.warn("Encounter a error while querying content summary of table " +
                table.getCompleteName() + " from FileSystem. " +
                "Cannot guess if CommonJoinOperator will optimize " +
                joinOp.getName() + " " + joinOp.getIdentifier());
          }
          if (resultCs == null) {
            isAbleToGuess = false;
            break;
          }

          long size = resultCs.getLength();
          aliasTotalKnownInputSize += size;
          Long es = aliasToSize.get(alias);
          if(es == null) {
            es = Long.valueOf(0);
          }
          es += size;
          aliasToSize.put(alias, es);
        }
        posToAliases.put(pos, aliases);
      }

      if (!isAbleToGuess) {
        LOG.info("Cannot guess if CommonJoinOperator will optimize " +
            joinOp.getName() + " " + joinOp.getIdentifier());
        continue;
      }

      JoinDesc joinDesc = joinOp.getConf();
      Byte[] order = joinDesc.getTagOrder();
      int numAliases = order.length;
      Set<Integer> bigTableCandidates =
          MapJoinProcessor.getBigTableCandidates(joinDesc.getConds());
      if (bigTableCandidates.isEmpty()) {
        continue;
      }

      long ThresholdOfSmallTblSizeSum = HiveConf.getLongVar(pCtx.getConf(),
          HiveConf.ConfVars.HIVE_SMALL_TABLES_FILESIZE);
      for (int i = 0; i < numAliases; i++) {
        // this table cannot be big table
        if (!bigTableCandidates.contains(i)) {
          continue;
        }
        Set<String> aliases = posToAliases.get(i);
        long aliasKnownSize = Utilities.sumOf(aliasToSize, aliases);
        if (!CommonJoinTaskDispatcher.cannotConvert(aliasKnownSize,
            aliasTotalKnownInputSize, ThresholdOfSmallTblSizeSum)) {
          mayConvert = true;
        }
      }

      if (mayConvert) {
        LOG.info(joinOp.getName() + " " + joinOp.getIdentifier() +
            " may be converted to MapJoin by CommonJoinResolver");
        skipedJoinOperators.add(joinOp);
      }
    }
  }
Implementation of  Correlation Optimizer. This optimizer is based on
the paper "YSmart: Yet Another SQL-to-MapReduce Translator"
(Rubao Lee, Tian Luo, Yin Huai, Fusheng Wang, Yongqiang He, and Xiaodong Zhang)
(http://www.cse.ohio-state.edu/hpcs/WWW/HTML/publications/papers/TR-11-7.pdf).
Correlation Optimizer detects if ReduceSinkOperators share same keys.
Then, it will transform the query plan tree (operator tree) by exploiting
detected correlations. For details, see the original paper of YSmart.

Test queries associated with this optimizer are correlationoptimizer1.q to
correlationoptimizer14.q
ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java