public List<Task<?>> getTasks(HiveConf conf, Object objCtx) {
    ConditionalResolverMergeFilesCtx ctx = (ConditionalResolverMergeFilesCtx) objCtx;
    String dirName = ctx.getDir();

    List<Task<?>> resTsks = new ArrayList<Task<?>>();
    // check if a map-reduce job is needed to merge the files
    // If the current size is smaller than the target, merge
    long trgtSize = conf.getLongVar(HiveConf.ConfVars.HIVE_MERGE_MAP_FILES_SIZE);
    long avgConditionSize = conf
        .getLongVar(HiveConf.ConfVars.HIVE_MERGE_MAP_FILES_AVG_SIZE);
    trgtSize = Math.max(trgtSize, avgConditionSize);

    Task<?> mvTask = ctx.getListTasks().get(0);
    Task<?> mrTask = ctx.getListTasks().get(1);
    Task<?> mrAndMvTask = ctx.getListTasks().get(2);

    try {
      Path dirPath = new Path(dirName);
      FileSystem inpFs = dirPath.getFileSystem(conf);
      DynamicPartitionCtx dpCtx = ctx.getDPCtx();
      HiveStorageHandler storageHandler = HiveUtils.getStorageHandler(conf, ctx.getStorageHandlerClass());
      boolean dirExists = inpFs.exists(dirPath);
      boolean useCustomStorageHandler = storageHandler != null && storageHandler.supportsMergeFiles();

      MapWork work = null;
      // For each dynamic partition, check if it needs to be merged.
      if (mrTask.getWork() instanceof MapredWork) {
        work = ((MapredWork) mrTask.getWork()).getMapWork();
      } else if (mrTask.getWork() instanceof TezWork){
        work = (MapWork) ((TezWork) mrTask.getWork()).getAllWork().get(0);
      } else {
        work = (MapWork) mrTask.getWork();
      }

      if (dirExists) {
        int lbLevel = (ctx.getLbCtx() == null) ? 0 : ctx.getLbCtx().calculateListBucketingLevel();
        boolean manifestFilePresent = false;
        FileSystem manifestFs = dirPath.getFileSystem(conf);
        if (manifestFs.exists(new Path(dirPath, Utilities.BLOB_MANIFEST_FILE))) {
          manifestFilePresent = true;
        }

        /**
         * In order to make code easier to read, we write the following in the way:
         * 1. the first if clause to differ dynamic partition and static partition
         * 2. with static partition, we differ list bucketing from non-list bucketing.
         * Another way to write it is to merge static partition w/ LB wit DP. In that way,
         * we still need to further differ them, since one uses lbLevel and
         * another lbLevel+numDPCols.
         * The first one is selected mainly for easy to read.
         */
        // Dynamic partition: replace input path (root to dp paths) with dynamic partition
        // input paths.
        if (dpCtx != null &&  dpCtx.getNumDPCols() > 0) {
          int numDPCols = dpCtx.getNumDPCols();
          int dpLbLevel = numDPCols + lbLevel;

          generateActualTasks(conf, resTsks, trgtSize, avgConditionSize, mvTask, mrTask,
              mrAndMvTask, dirPath, inpFs, ctx, work, dpLbLevel, manifestFilePresent, storageHandler);
        } else { // no dynamic partitions
          if(lbLevel == 0) {
            // static partition without list bucketing
            List<FileStatus> manifestFilePaths = Lists.newArrayList();
            long totalSize;
            if (manifestFilePresent) {
              manifestFilePaths = getManifestFilePaths(conf, dirPath);
              totalSize = getMergeSize(manifestFilePaths, avgConditionSize);
            } else {
              totalSize = getMergeSize(inpFs, dirPath, avgConditionSize);
              Utilities.FILE_OP_LOGGER.debug("merge resolve simple case - totalSize " + totalSize + " from " + dirPath);
            }

            if (totalSize >= 0) { // add the merge job
              if (manifestFilePresent) {
                setupWorkWhenUsingManifestFile(work, manifestFilePaths, dirPath, true);
              }
              setupMapRedWork(conf, work, trgtSize, totalSize);
              resTsks.add(mrTask);
            } else { // don't need to merge, add the move job
              resTsks.add(mvTask);
            }
          } else {
            // static partition and list bucketing
            generateActualTasks(conf, resTsks, trgtSize, avgConditionSize, mvTask, mrTask,
                mrAndMvTask, dirPath, inpFs, ctx, work, lbLevel, manifestFilePresent, storageHandler);
          }
        }
      } else if (useCustomStorageHandler) {
        generateActualTasks(conf, resTsks, trgtSize, avgConditionSize, mvTask, mrTask,
                mrAndMvTask, dirPath, inpFs, ctx, work, 0, false, storageHandler);
      } else {
        Utilities.FILE_OP_LOGGER.info("Resolver returning movetask for " + dirPath);
        resTsks.add(mvTask);
      }
    } catch (IOException e) {
      LOG.warn("Exception while getting tasks", e);
    } catch (ClassNotFoundException | HiveException e) {
      throw new RuntimeException("Failed to load storage handler: {}" + e.getMessage());
    }

    // Only one of the tasks should ever be added to resTsks
    assert(resTsks.size() == 1);

    return resTsks;
  }
@param lbCtx the lbCtx to set
ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java