private boolean convertJoinDynamicPartitionedHashJoin(JoinOperator joinOp, OptimizeTezProcContext context)
    throws SemanticException {
    // Attempt dynamic partitioned hash join
    // Since we don't have big table index yet, must start with estimate of numReducers
    int numReducers = estimateNumBuckets(joinOp, false);
    LOG.info("Try dynamic partitioned hash join with estimated " + numReducers + " reducers");
    MapJoinConversion mapJoinConversion =
        getMapJoinConversion(joinOp, context, numReducers, false, maxJoinMemory, false);
    if (mapJoinConversion != null) {
      if (mapJoinConversion.getIsFullOuterJoin() &&
          !mapJoinConversion.getIsFullOuterEnabledForDynamicPartitionHashJoin()) {
        return false;
      }
      final int bigTablePos = mapJoinConversion.getBigTablePos();

      // Now that we have the big table index, get real numReducers value based on big table RS
      ReduceSinkOperator bigTableParentRS =
          (ReduceSinkOperator) (joinOp.getParentOperators().get(bigTablePos));
      numReducers = bigTableParentRS.getConf().getNumReducers();
      LOG.debug("Real big table reducers = " + numReducers);

      MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, mapJoinConversion, false);
      if (mapJoinOp != null) {
        LOG.info("Selected dynamic partitioned hash join");
        MapJoinDesc mapJoinDesc = mapJoinOp.getConf();
        mapJoinDesc.setDynamicPartitionHashJoin(true);

        // Set OpTraits for dynamically partitioned hash join:
        // bucketColNames: Re-use previous joinOp's bucketColNames. Parent operators should be
        //   reduce sink, which should have bucket columns based on the join keys.
        // numBuckets: set to number of reducers
        // sortCols: This is an unsorted join - no sort cols
        OpTraits opTraits = new OpTraits(
            joinOp.getOpTraits().getBucketColNames(),
            joinOp.getOpTraits().getCustomBucketFunctions(),
            numReducers,
            null,
            joinOp.getOpTraits().getNumReduceSinks());
        mapJoinOp.setOpTraits(opTraits);
        preserveOperatorInfos(mapJoinOp, joinOp, context);
        // propagate this change till the next RS
        for (Operator<? extends OperatorDesc> childOp : mapJoinOp.getChildOperators()) {
          setAllChildrenTraits(childOp, mapJoinOp.getOpTraits());
        }
        return true;
      }
    }

    return false;
  }
Estimate the number of buckets in the join, using the parent operators' OpTraits and/or
parent operators' number of reducers
@param joinOp
@param useOpTraits  Whether OpTraits should be used for the estimate.
@return
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java