package org.edx.hadoop.input;

import java.io.*;
import java.util.List;
import java.util.ArrayList;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;


public class ManifestTextInputFormat extends KeyValueTextInputFormat {
    public static int RETRY_LIMIT = 10;
    public static long REPORTING_INTERVAL = 1000l * 60 * 5;

    protected FileStatus[] listStatus(JobConf job) throws IOException {
        FileStatus[] manifests = super.listStatus(job);
        List<FileStatus> paths = new ArrayList<FileStatus>();
        long lastReportTime = System.currentTimeMillis();

        for (FileStatus manifest : manifests) {
            List<Path> globPaths = this.readManifest(manifest.getPath(), job);

            for(int i = 0; i < globPaths.size(); i++) {
                Path globPath = globPaths.get(i);

                paths.addAll(this.expandPath(globPath, job));

                if ((System.currentTimeMillis() - lastReportTime) >= REPORTING_INTERVAL) {
                    lastReportTime = System.currentTimeMillis();

                    if (globPath != null)
                        LOG.info("Fetched " + i + " of " + globPaths.size() + " paths, most recently: " + globPath.toUri());
                    else
                        LOG.info("Fetched " + i + " of " + globPaths.size() + " paths.");
                }
            }
        }

        return paths.toArray(new FileStatus[1]);
    }

    private List<Path> readManifest(Path manifestPath, JobConf job) throws IOException {
        FileSystem fs = manifestPath.getFileSystem(job);
        List<Path> paths = new ArrayList<Path>();
        DataInputStream dataStream = fs.open(manifestPath);
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader(dataStream));
            String line = reader.readLine();
            while(line != null) {
                paths.add(new Path(line));
                line = reader.readLine();
            }
        } finally {
            dataStream.close();
        }

        return paths;
    }

    private List<FileStatus> expandPath(Path globPath, JobConf conf) throws IOException {
        FileSystem fs = globPath.getFileSystem(conf);

        int attempts = 0;
        boolean success = false;
        FileStatus[] matches = null;
        while (!success && attempts < RETRY_LIMIT) {
            attempts += 1;
            try {
                matches = fs.globStatus(globPath);
                success = true;
            } catch (Exception exc) {
                LOG.info("Exception while calling globStatus, retrying...", exc);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException("This should never happen", e);
                }
            }
        }

        if (!success) {
            throw new RuntimeException("Unable to fetch glob status '" + globPath.toUri() + "' after " + RETRY_LIMIT + " attempts");
        }

        List<FileStatus> paths = new ArrayList<FileStatus>();

        // fs.globStatus returns null when the path looks like a file and it does not exist.
        if (matches == null) {
            return paths;
        }

        for (int i = 0; i < matches.length; i++) {
            FileStatus match = matches[i];
            if (match.isDirectory()) {
                FileStatus[] childStatuses = fs.listStatus(match.getPath());
                for (int j = 0; j < childStatuses.length; j++) {
                    paths.add(childStatuses[j]);
                }
            } else {
                paths.add(match);
            }
        }

        return paths;
    }
}

@@ -1022,16 +1022,23 @@
 ath, job
+, false
 ));%0A%0A   
@@ -2270,16 +2270,41 @@
 onf conf
+, boolean use_list_status
 ) throws
@@ -2559,32 +2559,273 @@
           try %7B%0A
+                // Expand using globStatus at the top level, but recurse below that using listStatus to get children.%0A                if (use_list_status) %7B%0A                    matches = fs.listStatus(globPath);%0A                %7D else %7B%0A    
                 
@@ -2851,32 +2851,50 @@
 atus(globPath);%0A
+                %7D%0A
                 
@@ -2993,20 +2993,16 @@
 ile 
-calling glob
+getting 
 Stat
@@ -3190,32 +3190,53 @@
 on(%22
-This should never happen
+Fatal unexpected interruption while retrying 
 %22, e
@@ -3365,20 +3365,19 @@
 tch 
-glob 
 status 
+for 
 '%22 +
@@ -3825,200 +3825,137 @@
     
-FileStatus%5B%5D childStatuses = fs.listStatus(match.getPath());%0A                for (int j = 0; j %3C childStatuses.length; j++) %7B%0A                    paths.add(childStatuses%5Bj%5D);%0A                %7D
+// Recursively expand directories, using listStatus.%0A                paths.addAll(this.expandPath(match.getPath(), conf, true));%0A
 %0A   

Call expandPath recursively to expand directories.