public MapJoinConversion getMapJoinConversion(JoinOperator joinOp, OptimizeTezProcContext context,
      int buckets, boolean skipJoinTypeChecks, long maxSize, boolean checkMapJoinThresholds)
              throws SemanticException {
    JoinDesc joinDesc = joinOp.getConf();
    JoinCondDesc[] conds = joinDesc.getConds();
    if (!skipJoinTypeChecks) {
      /*
       * HIVE-9038: Join tests fail in tez when we have more than 1 join on the same key and there is
       * an outer join down the join tree that requires filterTag. We disable this conversion to map
       * join here now. We need to emulate the behavior of HashTableSinkOperator as in MR or create a
       * new operation to be able to support this. This seems like a corner case enough to special
       * case this for now.
       */
      if (conds.length > 1) {
        if (hasOuterJoin(joinOp)) {
          return null;
        }
      }
    }

    // Assume.
    boolean isFullOuterEnabledForDynamicPartitionHashJoin = false;
    boolean isFullOuterEnabledForMapJoin = false;

    boolean isFullOuterJoin =
        MapJoinProcessor.precheckFullOuter(context.conf, joinOp);
    if (isFullOuterJoin) {

      boolean isFullOuterEnabled = MapJoinProcessor.isFullOuterMapEnabled(context.conf, joinOp);
      if (isFullOuterEnabled) {

        // FUTURE: Currently, we only support DPHJ.
        isFullOuterEnabledForDynamicPartitionHashJoin =
            MapJoinProcessor.isFullOuterEnabledForDynamicPartitionHashJoin(context.conf, joinOp);
      }
    }

    Set<Integer> bigTableCandidateSet =
        MapJoinProcessor.getBigTableCandidates(conds, /* isSupportFullOuter */ true);

    if (bigTableCandidateSet.isEmpty()) {
      return null;
    }

    int bigTablePosition = -1;
    // big input cumulative row count
    long bigInputCumulativeCardinality = -1L;
    // stats of the big input
    Statistics bigInputStat = null;

    // bigTableFound means we've encountered a table that's bigger than the
    // max. This table is either the the big table or we cannot convert.
    boolean foundInputNotFittingInMemory = false;

    // total size of the inputs
    long totalSize = 0;

    // convert to DPHJ
    boolean convertDPHJ = false;

    for (int pos = 0; pos < joinOp.getParentOperators().size(); pos++) {
      Operator<? extends OperatorDesc> parentOp = joinOp.getParentOperators().get(pos);

      Statistics currInputStat = parentOp.getStatistics();
      if (currInputStat == null) {
        LOG.warn("Couldn't get statistics from: " + parentOp);
        return null;
      }

      long inputSize = computeOnlineDataSize(currInputStat);
      LOG.info("Join input#{}; onlineDataSize: {}; Statistics: {}", pos, inputSize, currInputStat);

      boolean currentInputNotFittingInMemory = false;
      if ((bigInputStat == null)
          || (inputSize > computeOnlineDataSize(bigInputStat))) {

        if (foundInputNotFittingInMemory) {
          // cannot convert to map join; we've already chosen a big table
          // on size and there's another one that's bigger.
          return null;
        }

        if (inputSize/buckets > maxSize) {
          if (!bigTableCandidateSet.contains(pos)) {
            // can't use the current table as the big table, but it's too
            // big for the map side.
            return null;
          }

          currentInputNotFittingInMemory = true;
          foundInputNotFittingInMemory = true;
        }
      }

      long currentInputCumulativeCardinality;
      if (foundInputNotFittingInMemory) {
        currentInputCumulativeCardinality = -1L;
      } else {
        Long cardinality = computeCumulativeCardinality(parentOp);
        if (cardinality == null) {
          // We could not get stats, we cannot convert
          return null;
        }
        currentInputCumulativeCardinality = cardinality;
      }

      // This input is the big table if it is contained in the big candidates set, and either:
      // 1) we have not chosen a big table yet, or
      // 2) it has been chosen as the big table above, or
      // 3) the cumulative cardinality for this input is higher, or
      // 4) the cumulative cardinality is equal, but the size is bigger,
      boolean selectedBigTable = bigTableCandidateSet.contains(pos) &&
              (bigInputStat == null || currentInputNotFittingInMemory ||
                      (!foundInputNotFittingInMemory && (currentInputCumulativeCardinality > bigInputCumulativeCardinality ||
                  (currentInputCumulativeCardinality == bigInputCumulativeCardinality
                      && inputSize > computeOnlineDataSize(bigInputStat)))));

      if (bigInputStat != null && selectedBigTable) {
        // We are replacing the current big table with a new one, thus
        // we need to count the current one as a map table then.
        totalSize += computeOnlineDataSize(bigInputStat);
        // Check if number of distinct keys is greater than given max number of entries
        // for HashMap
        if (checkMapJoinThresholds && !checkNumberOfEntriesForHashTable(joinOp, bigTablePosition, context)) {
          convertDPHJ = true;
        }
      } else if (!selectedBigTable) {
        // This is not the first table and we are not using it as big table,
        // in fact, we're adding this table as a map table
        totalSize += inputSize;
        // Check if number of distinct keys is greater than given max number of entries
        // for HashMap
        if (checkMapJoinThresholds && !checkNumberOfEntriesForHashTable(joinOp, pos, context)) {
          convertDPHJ = true;
        }
      }

      if (totalSize/buckets > maxSize) {
        // sum of small tables size in this join exceeds configured limit
        // hence cannot convert.
        return null;
      }

      if (selectedBigTable) {
        bigTablePosition = pos;
        bigInputCumulativeCardinality = currentInputCumulativeCardinality;
        bigInputStat = currInputStat;
      }

    }

    if (bigTablePosition == -1) {
      LOG.debug("No big table selected, no MapJoin");
      return null;
    }

    // Check if size of data to shuffle (larger table) is less than given max size
    if (checkMapJoinThresholds && convertDPHJ
            && checkShuffleSizeForLargeTable(joinOp, bigTablePosition, context)) {
      LOG.debug("Conditions to convert to MapJoin are not met");
      return null;
    }

    // only allow cross product in map joins if build side is 'small'
    boolean cartesianProductEdgeEnabled =
      HiveConf.getBoolVar(context.conf, HiveConf.ConfVars.TEZ_CARTESIAN_PRODUCT_EDGE_ENABLED);
    if (cartesianProductEdgeEnabled && !hasOuterJoin(joinOp) && isCrossProduct(joinOp)) {
      for (int i = 0 ; i < joinOp.getParentOperators().size(); i ++) {
        if (i != bigTablePosition) {
          Statistics parentStats = joinOp.getParentOperators().get(i).getStatistics();
          if (parentStats.getNumRows() >
            HiveConf.getIntVar(context.conf, HiveConf.ConfVars.XPROD_SMALL_TABLE_ROWS_THRESHOLD)) {
            // if any of smaller side is estimated to generate more than
            // threshold rows we would disable mapjoin
            return null;
          }
        }
      }
    }

    // We store the total memory that this MapJoin is going to use,
    // which is calculated as totalSize/buckets, with totalSize
    // equal to sum of small tables size.
    joinOp.getConf().setInMemoryDataSize(totalSize / buckets);

    return
        new MapJoinConversion(
            bigTablePosition,
            isFullOuterJoin,
            isFullOuterEnabledForDynamicPartitionHashJoin,
            isFullOuterEnabledForMapJoin);
  }
Obtain big table position for join.

@param joinOp join operator
@param context optimization context
@param buckets bucket count for Bucket Map Join conversion consideration or reduce count
for Dynamic Hash Join conversion consideration
@param skipJoinTypeChecks whether to skip join type checking
@param maxSize size threshold for Map Join conversion
@param checkMapJoinThresholds whether to check thresholds to convert to Map Join
@return returns big table position or -1 if it cannot be determined
@throws SemanticException
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java