private boolean selectJoinForLlap(OptimizeTezProcContext context, JoinOperator joinOp,
                          TezBucketJoinProcCtx tezBucketJoinProcCtx,
                          LlapClusterStateForCompile llapInfo,
                          MapJoinConversion mapJoinConversion, int numBuckets) throws SemanticException {
    if (!context.conf.getBoolVar(HiveConf.ConfVars.HIVE_DYNAMIC_PARTITION_HASHJOIN)
            && numBuckets > 1) {
      // DPHJ is disabled, only attempt BMJ or mapjoin
      return convertJoinBucketMapJoin(joinOp, context, mapJoinConversion, tezBucketJoinProcCtx);
    }

    int numExecutorsPerNode = -1;
    if (llapInfo.hasClusterInfo()) {
      numExecutorsPerNode = llapInfo.getNumExecutorsPerNode();
    }
    if (numExecutorsPerNode == -1) {
      numExecutorsPerNode = context.conf.getIntVar(ConfVars.LLAP_DAEMON_NUM_EXECUTORS);
    }

    int numNodes = llapInfo.getKnownExecutorCount()/numExecutorsPerNode;

    LOG.debug("Number of nodes = " + numNodes + ". Number of Executors per node = " + numExecutorsPerNode);

    // Determine the size of small table inputs
    final int mapJoinConversionPos = mapJoinConversion.getBigTablePos();
    long totalSize = 0;
    for (int pos = 0; pos < joinOp.getParentOperators().size(); pos++) {
      if (pos == mapJoinConversionPos) {
        continue;
      }
      Operator<? extends OperatorDesc> parentOp = joinOp.getParentOperators().get(pos);
      totalSize += computeOnlineDataSize(parentOp.getStatistics());
    }

    // Size of bigtable
    long bigTableSize = computeOnlineDataSize(joinOp.getParentOperators().get(mapJoinConversionPos).getStatistics());

    // Network cost of DPHJ
    long networkCostDPHJ = totalSize + bigTableSize;

    LOG.info("Cost of dynamically partitioned hash join : total small table size = " + totalSize
    + " bigTableSize = " + bigTableSize + "networkCostDPHJ = " + networkCostDPHJ);

    // Network cost of map side join
    long networkCostMJ = numNodes * totalSize;
    LOG.info("Cost of Bucket Map Join : numNodes = " + numNodes + " total small table size = "
    + totalSize + " networkCostMJ = " + networkCostMJ);

    if (totalSize <= maxJoinMemory) {
      // mapjoin is applicable; don't try the below algos..
      return false;
    }

    if (networkCostDPHJ < networkCostMJ) {
      LOG.info("Dynamically partitioned Hash Join chosen");
      return convertJoinDynamicPartitionedHashJoin(joinOp, context);
    } else if (numBuckets > 1) {
      LOG.info("Bucket Map Join chosen");
      return convertJoinBucketMapJoin(joinOp, context, mapJoinConversion, tezBucketJoinProcCtx);
    }
    // fallback to mapjoin no bucket scaling
    LOG.info("Falling back to mapjoin no bucket scaling");
    return false;
  }
ConvertJoinMapJoin is an optimization that replaces a common join
(aka shuffle join) with a map join (aka broadcast or fragment replicate
join when possible. Map joins have restrictions on which joins can be
converted (e.g.: full outer joins cannot be handled as map joins) as well
as memory restrictions (one side of the join has to fit into memory).
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java