@Override
  public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
    init(job);

    Path[] dirs = getInputPaths(job);

    JobConf newjob = new JobConf(job);
    ArrayList<InputSplit> result = new ArrayList<InputSplit>();

    int numOrigSplits = 0;
    // for each dir, get all files under the dir, do getSplits to each
    // individual file,
    // and then create a BucketizedHiveInputSplit on it

    ArrayList<Path> currentDir = null;
    for (Path dir : dirs) {
      PartitionDesc part = getPartitionDescFromPath(pathToPartitionInfo, dir);
      // create a new InputFormat instance if this is the first time to see this class
      Class inputFormatClass = part.getInputFileFormatClass();
      InputFormat inputFormat = getInputFormatFromCache(inputFormatClass, job);
      newjob.setInputFormat(inputFormat.getClass());

      ValidWriteIdList mmIds = null;
      if (part.getTableDesc() != null) {
        // This can happen for truncate table case for non-MM tables.
        mmIds = getMmValidWriteIds(newjob, part.getTableDesc(), null);
      }
      // TODO: should this also handle ACID operation, etc.? seems to miss a lot of stuff from HIF.
      List<Path> finalDirs = null, dirsWithMmOriginals = null;
      if (mmIds == null) {
        finalDirs = Lists.newArrayList(dir);
      } else {
        finalDirs = new ArrayList<>();
        dirsWithMmOriginals = new ArrayList<>();
        processPathsForMmRead(
            Lists.newArrayList(dir), newjob, mmIds, finalDirs, dirsWithMmOriginals);
      }
      if (finalDirs.isEmpty() && (dirsWithMmOriginals == null || dirsWithMmOriginals.isEmpty())) {
        continue; // No valid inputs - possible in MM case.
      }

      for (Path finalDir : finalDirs) {
        FileStatus[] listStatus = listStatus(newjob, finalDir);
        for (FileStatus status : listStatus) {
          numOrigSplits = addBHISplit(
              status, inputFormat, inputFormatClass, numOrigSplits, newjob, result);
        }
      }
      if (dirsWithMmOriginals != null) {
        for (Path originalsDir : dirsWithMmOriginals) {
          FileSystem fs = originalsDir.getFileSystem(job);
          FileStatus[] listStatus = fs.listStatus(dir, FileUtils.HIDDEN_FILES_PATH_FILTER);
          for (FileStatus status : listStatus) {
            if (status.isDirectory()) continue;
            numOrigSplits = addBHISplit(
                status, inputFormat, inputFormatClass, numOrigSplits, newjob, result);
          }
        }
      }
    }

    LOG.info(result.size() + " bucketized splits generated from "
        + numOrigSplits + " original splits.");
    return result.toArray(new BucketizedHiveInputSplit[result.size()]);
  }
Recursively lists status for all files starting from the directory dir
@param job
@param dir
@return
@throws IOException
ql/src/java/org/apache/hadoop/hive/ql/io/BucketizedHiveInputFormat.java