private Vertex createVertexFromMergeWork(JobConf conf, MergeJoinWork mergeJoinWork,
      Path mrScratchDir, VertexType vertexType) throws Exception {
    Utilities.setMergeWork(conf, mergeJoinWork, mrScratchDir, false);
    if (mergeJoinWork.getMainWork() instanceof MapWork) {
      List<BaseWork> mapWorkList = mergeJoinWork.getBaseWorkList();
      MapWork mapWork = (MapWork) (mergeJoinWork.getMainWork());
      Vertex mergeVx = createVertexFromMapWork(
          conf, mapWork, mrScratchDir, vertexType);

      Class<?> inputFormatClass = conf.getClass("mapred.input.format.class",
              HiveInputFormat.class);
      if (inputFormatClass != BucketizedHiveInputFormat.class &&
              inputFormatClass != HiveInputFormat.class) {
        // As of now only these two formats are supported.
        inputFormatClass = HiveInputFormat.class;
      }
      conf.setClass("mapred.input.format.class", inputFormatClass, InputFormat.class);
      // mapreduce.tez.input.initializer.serialize.event.payload should be set
      // to false when using this plug-in to avoid getting a serialized event at run-time.
      conf.setBoolean("mapreduce.tez.input.initializer.serialize.event.payload", false);
      for (int i = 0; i < mapWorkList.size(); i++) {

        mapWork = (MapWork) (mapWorkList.get(i));
        conf.set(TEZ_MERGE_CURRENT_MERGE_FILE_PREFIX, mapWork.getName());
        conf.set(Utilities.INPUT_NAME, mapWork.getName());
        LOG.info("Going through each work and adding MultiMRInput");
        mergeVx.addDataSource(mapWork.getName(),
            MultiMRInput.createConfigBuilder(conf, inputFormatClass).build());
      }

      // To be populated for SMB joins only for all the small tables
      Map<String, Integer> inputToBucketMap = new HashMap<>();
      if (mergeJoinWork.getMergeJoinOperator().getParentOperators().size() == 1
              && mergeJoinWork.getMergeJoinOperator().getOpTraits() != null) {
        // This is an SMB join.
        for (BaseWork work : mapWorkList) {
          MapWork mw = (MapWork) work;
          Map<String, Operator<?>> aliasToWork = mw.getAliasToWork();
          Preconditions.checkState(aliasToWork.size() == 1,
                  "More than 1 alias in SMB mapwork");
          inputToBucketMap.put(mw.getName(), mw.getWorks().get(0).getOpTraits().getNumBuckets());
        }
      }
      VertexManagerPluginDescriptor desc =
        VertexManagerPluginDescriptor.create(CustomPartitionVertex.class.getName());
      // the +1 to the size is because of the main work.
      CustomVertexConfiguration vertexConf =
          new CustomVertexConfiguration(mergeJoinWork.getMergeJoinOperator().getConf()
              .getNumBuckets(), vertexType, mergeJoinWork.getBigTableAlias(),
              mapWorkList.size() + 1, inputToBucketMap);
      DataOutputBuffer dob = new DataOutputBuffer();
      vertexConf.write(dob);
      byte[] userPayload = dob.getData();
      desc.setUserPayload(UserPayload.create(ByteBuffer.wrap(userPayload)));
      mergeVx.setVertexManagerPlugin(desc);
      return mergeVx;
    } else {
      return createVertexFromReduceWork(conf,
          (ReduceWork) mergeJoinWork.getMainWork(), mrScratchDir);
    }
  }
Utility method to create a stripped down configuration for the MR partitioner.

@param partitionerClassName
         the real MR partitioner class name
@param baseConf
         a base configuration to extract relevant properties
@return
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java