@Test public void search_traverse_file() throws Exception { touch("a/a.csv"); touch("b/b.csv"); touch("c/c.csv"); FileSystem fs = getTempFileSystem(); List<FileStatus> results = HadoopDataSourceUtil.search(fs, getBase(), FilePattern.compile("**/*.csv")); assertThat(normalize(results), is(path("a/a.csv", "b/b.csv", "c/c.csv"))); }
public static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern) throws IOException { if (fs == null) { throw new IllegalArgumentException("fs must not be null"); } if (base == null) { throw new IllegalArgumentException("base must not be null"); } if (pattern == null) { throw new IllegalArgumentException("pattern must not be null"); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start searching for files (path={0}, resourcePattern={1})", base, pattern)); } List<FileStatus> current = new ArrayList<>(1); try { FileStatus stat = fs.getFileStatus(base); current.add(stat); } catch (FileNotFoundException e) { return Collections.emptyList(); } int steps = 0; LinkedList<Segment> segments = new LinkedList<>(pattern.getSegments()); while (segments.isEmpty() == false) { if (segments.getFirst().isTraverse()) { segments.removeFirst(); current = recursiveStep(fs, current); } else { List<Path> step = consumeStep(segments); current = globStep(fs, current, step); } steps++; } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish searching for files (path={0}, resourcePattern={1}, results={2}, steps={3})", base, pattern, current.size(), steps)); } return current; }
HadoopDataSourceUtil { public static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern) throws IOException { if (fs == null) { throw new IllegalArgumentException("fs must not be null"); } if (base == null) { throw new IllegalArgumentException("base must not be null"); } if (pattern == null) { throw new IllegalArgumentException("pattern must not be null"); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start searching for files (path={0}, resourcePattern={1})", base, pattern)); } List<FileStatus> current = new ArrayList<>(1); try { FileStatus stat = fs.getFileStatus(base); current.add(stat); } catch (FileNotFoundException e) { return Collections.emptyList(); } int steps = 0; LinkedList<Segment> segments = new LinkedList<>(pattern.getSegments()); while (segments.isEmpty() == false) { if (segments.getFirst().isTraverse()) { segments.removeFirst(); current = recursiveStep(fs, current); } else { List<Path> step = consumeStep(segments); current = globStep(fs, current, step); } steps++; } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish searching for files (path={0}, resourcePattern={1}, results={2}, steps={3})", base, pattern, current.size(), steps)); } return current; } }
HadoopDataSourceUtil { public static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern) throws IOException { if (fs == null) { throw new IllegalArgumentException("fs must not be null"); } if (base == null) { throw new IllegalArgumentException("base must not be null"); } if (pattern == null) { throw new IllegalArgumentException("pattern must not be null"); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start searching for files (path={0}, resourcePattern={1})", base, pattern)); } List<FileStatus> current = new ArrayList<>(1); try { FileStatus stat = fs.getFileStatus(base); current.add(stat); } catch (FileNotFoundException e) { return Collections.emptyList(); } int steps = 0; LinkedList<Segment> segments = new LinkedList<>(pattern.getSegments()); while (segments.isEmpty() == false) { if (segments.getFirst().isTraverse()) { segments.removeFirst(); current = recursiveStep(fs, current); } else { List<Path> step = consumeStep(segments); current = globStep(fs, current, step); } steps++; } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish searching for files (path={0}, resourcePattern={1}, results={2}, steps={3})", base, pattern, current.size(), steps)); } return current; } private  HadoopDataSourceUtil(); }
HadoopDataSourceUtil { public static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern) throws IOException { if (fs == null) { throw new IllegalArgumentException("fs must not be null"); } if (base == null) { throw new IllegalArgumentException("base must not be null"); } if (pattern == null) { throw new IllegalArgumentException("pattern must not be null"); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start searching for files (path={0}, resourcePattern={1})", base, pattern)); } List<FileStatus> current = new ArrayList<>(1); try { FileStatus stat = fs.getFileStatus(base); current.add(stat); } catch (FileNotFoundException e) { return Collections.emptyList(); } int steps = 0; LinkedList<Segment> segments = new LinkedList<>(pattern.getSegments()); while (segments.isEmpty() == false) { if (segments.getFirst().isTraverse()) { segments.removeFirst(); current = recursiveStep(fs, current); } else { List<Path> step = consumeStep(segments); current = globStep(fs, current, step); } steps++; } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish searching for files (path={0}, resourcePattern={1}, results={2}, steps={3})", base, pattern, current.size(), steps)); } return current; } private  HadoopDataSourceUtil(); static List<DirectDataSourceProfile> loadProfiles(Configuration conf); static DirectDataSourceRepository loadRepository(Configuration conf); static boolean isLocalAttemptOutputDefined(LocalFileSystem localFileSystem); static Path getLocalTemporaryDirectory(LocalFileSystem localFileSystem); static OutputTransactionContext createContext(String executionId, String datasourceId); @Deprecated static OutputTransactionContext createContext(JobContext context, String datasourceId); @Deprecated static OutputAttemptContext createContext(TaskAttemptContext context, String datasourceId); static String getTransactionInfoExecutionId(Path transactionInfoPath); static Path getTransactionInfoPath(Configuration conf, String executionId); static Path getCommitMarkPath(Configuration conf, String executionId); static Collection<FileStatus> findAllTransactionInfoFiles(Configuration conf); static Path getSystemDir(Configuration conf); static Path getSystemDir(Configuration conf, boolean resolve); static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern); static List<FileStatus> onlyMinimalCovered(List<FileStatus> statList); static boolean contains(Path parent, Path child); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to); static void moveFromLocal(
            Counter counter,
            LocalFileSystem localFs, FileSystem fs,
            Path from, Path to); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to, int threads); static HadoopFileFormat<T> toHadoopFileFormat(
            Configuration configuration, DataFormat<T> format); }
HadoopDataSourceUtil { public static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern) throws IOException { if (fs == null) { throw new IllegalArgumentException("fs must not be null"); } if (base == null) { throw new IllegalArgumentException("base must not be null"); } if (pattern == null) { throw new IllegalArgumentException("pattern must not be null"); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start searching for files (path={0}, resourcePattern={1})", base, pattern)); } List<FileStatus> current = new ArrayList<>(1); try { FileStatus stat = fs.getFileStatus(base); current.add(stat); } catch (FileNotFoundException e) { return Collections.emptyList(); } int steps = 0; LinkedList<Segment> segments = new LinkedList<>(pattern.getSegments()); while (segments.isEmpty() == false) { if (segments.getFirst().isTraverse()) { segments.removeFirst(); current = recursiveStep(fs, current); } else { List<Path> step = consumeStep(segments); current = globStep(fs, current, step); } steps++; } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish searching for files (path={0}, resourcePattern={1}, results={2}, steps={3})", base, pattern, current.size(), steps)); } return current; } private  HadoopDataSourceUtil(); static List<DirectDataSourceProfile> loadProfiles(Configuration conf); static DirectDataSourceRepository loadRepository(Configuration conf); static boolean isLocalAttemptOutputDefined(LocalFileSystem localFileSystem); static Path getLocalTemporaryDirectory(LocalFileSystem localFileSystem); static OutputTransactionContext createContext(String executionId, String datasourceId); @Deprecated static OutputTransactionContext createContext(JobContext context, String datasourceId); @Deprecated static OutputAttemptContext createContext(TaskAttemptContext context, String datasourceId); static String getTransactionInfoExecutionId(Path transactionInfoPath); static Path getTransactionInfoPath(Configuration conf, String executionId); static Path getCommitMarkPath(Configuration conf, String executionId); static Collection<FileStatus> findAllTransactionInfoFiles(Configuration conf); static Path getSystemDir(Configuration conf); static Path getSystemDir(Configuration conf, boolean resolve); static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern); static List<FileStatus> onlyMinimalCovered(List<FileStatus> statList); static boolean contains(Path parent, Path child); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to); static void moveFromLocal(
            Counter counter,
            LocalFileSystem localFs, FileSystem fs,
            Path from, Path to); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to, int threads); static HadoopFileFormat<T> toHadoopFileFormat(
            Configuration configuration, DataFormat<T> format); static final String PREFIX; static final String KEY_PATH; static final String KEY_SYSTEM_DIR; static final String KEY_LOCAL_TEMPDIR; static final String PREFIX_BEGIN_MARK; static final String PREFIX_COMMIT_MARK; static final String DEFAULT_SYSTEM_DIR; static final String TRANSACTION_INFO_DIR; static final Charset COMMENT_CHARSET; }