@Test(enabled = false) public void testGetFilesAtPath() throws IOException { LocalDateTime endDate = LocalDateTime.now(DateTimeZone.forID(TimeAwareRecursiveCopyableDataset.DEFAULT_DATE_PATTERN_TIMEZONE)); for (int i = 0; i < MAX_NUM_DAILY_DIRS; i++) { for (int j = 0; j < NUM_DIRS_PER_DAY; j++) { Path subDirPath = new Path(baseSrcDir, new Path(endDate.toDateTime().plusSeconds(60).getMillis() + "-PT-100000")); fs.mkdirs(subDirPath); for (int k = 0; k < NUM_FILES_PER_DIR; k++) { Path filePath = new Path(subDirPath, k + ".avro"); fs.create(filePath); } endDate = endDate.minusMinutes(10); } endDate = endDate.minusDays(1); } PathFilter ACCEPT_ALL_PATH_FILTER = new PathFilter() { @Override public boolean accept(Path path) { return true; } }; Properties properties = new Properties(); properties.setProperty("gobblin.dataset.pattern", sourceDir); properties.setProperty(TimeAwareRecursiveCopyableDataset.DATE_PATTERN_TIMEZONE_KEY, TimeAwareRecursiveCopyableDataset.DEFAULT_DATE_PATTERN_TIMEZONE); properties.setProperty(TimeAwareRecursiveCopyableDataset.LOOKBACK_TIME_KEY, NUM_LOOKBACK_DAYS_STR); properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "ALL"); properties.setProperty(UnixTimestampRecursiveCopyableDataset.TIMESTAMP_REGEEX, ".*/([0-9]{13})-PT-.*/.*"); UnixTimestampCopyableDatasetFinder finder = new UnixTimestampCopyableDatasetFinder(fs, properties); String datasetRoot = rootPath + "/" + databaseName; UnixTimestampRecursiveCopyableDataset dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); List<FileStatus> fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 30); properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "EARLIEST"); finder = new UnixTimestampCopyableDatasetFinder(fs, properties); dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 6); properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "latest"); finder = new UnixTimestampCopyableDatasetFinder(fs, properties); dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 6); properties.setProperty(UnixTimestampRecursiveCopyableDataset.TIMESTAMP_REGEEX, "([0-9]{13})-PT-.*/.*"); finder = new UnixTimestampCopyableDatasetFinder(fs, properties); datasetRoot = rootPath + "/" + databaseName + "/" + tableName; properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "ALL"); dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 30); properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "EARLIEST"); finder = new UnixTimestampCopyableDatasetFinder(fs, properties); dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 6); properties.setProperty(UnixTimestampRecursiveCopyableDataset.VERSION_SELECTION_POLICY, "latest"); finder = new UnixTimestampCopyableDatasetFinder(fs, properties); dataset = (UnixTimestampRecursiveCopyableDataset) finder.datasetAtPath(new Path(datasetRoot)); fileStatusList = dataset.getFilesAtPath(fs, baseSrcDir, ACCEPT_ALL_PATH_FILTER); Assert.assertTrue(fileStatusList.size() == 6); }
@Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { PathFilter andPathFilter = new AndPathFilter(fileFilter, new TimestampPathFilter()); List<FileStatus> files = super.getFilesAtPath(fs, path, andPathFilter); if (VersionSelectionPolicy.ALL == versionSelectionPolicy) { return files; } Map<Pair<String, LocalDate>, TreeMap<Long, List<FileStatus>>> pathTimestampFilesMap = new HashMap<>(); for (FileStatus fileStatus : files) { String relativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()), datasetRoot()).toString(); Matcher matcher = timestampPattern.matcher(relativePath); if (!matcher.matches()) { continue; } String timestampStr = matcher.group(1); String rootPath = relativePath.substring(0, relativePath.indexOf(timestampStr)); Long unixTimestamp = Long.parseLong(timestampStr); LocalDate localDate = new LocalDateTime(unixTimestamp,dateTimeZone).toLocalDate(); Pair<String, LocalDate> key = new ImmutablePair<>(rootPath, localDate); if (!pathTimestampFilesMap.containsKey(key)) { pathTimestampFilesMap.put(key, new TreeMap<Long, List<FileStatus>>()); } Map<Long, List<FileStatus>> timestampFilesMap = pathTimestampFilesMap.get(key); if (!timestampFilesMap.containsKey(unixTimestamp)) { timestampFilesMap.put(unixTimestamp, Lists.newArrayList()); } List<FileStatus> filesStatuses = timestampFilesMap.get(unixTimestamp); filesStatuses.add(fileStatus); } List<FileStatus> result = new ArrayList<>(); for(TreeMap<Long, List<FileStatus>> timestampFileStatus : pathTimestampFilesMap.values()) { if(timestampFileStatus.size() <=0 ) { continue; } switch (versionSelectionPolicy) { case EARLIEST: result.addAll(timestampFileStatus.firstEntry().getValue()); break; case LATEST: result.addAll(timestampFileStatus.lastEntry().getValue()); break; default: throw new RuntimeException("Unsupported version selection policy"); } } return result; }
UnixTimestampRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { PathFilter andPathFilter = new AndPathFilter(fileFilter, new TimestampPathFilter()); List<FileStatus> files = super.getFilesAtPath(fs, path, andPathFilter); if (VersionSelectionPolicy.ALL == versionSelectionPolicy) { return files; } Map<Pair<String, LocalDate>, TreeMap<Long, List<FileStatus>>> pathTimestampFilesMap = new HashMap<>(); for (FileStatus fileStatus : files) { String relativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()), datasetRoot()).toString(); Matcher matcher = timestampPattern.matcher(relativePath); if (!matcher.matches()) { continue; } String timestampStr = matcher.group(1); String rootPath = relativePath.substring(0, relativePath.indexOf(timestampStr)); Long unixTimestamp = Long.parseLong(timestampStr); LocalDate localDate = new LocalDateTime(unixTimestamp,dateTimeZone).toLocalDate(); Pair<String, LocalDate> key = new ImmutablePair<>(rootPath, localDate); if (!pathTimestampFilesMap.containsKey(key)) { pathTimestampFilesMap.put(key, new TreeMap<Long, List<FileStatus>>()); } Map<Long, List<FileStatus>> timestampFilesMap = pathTimestampFilesMap.get(key); if (!timestampFilesMap.containsKey(unixTimestamp)) { timestampFilesMap.put(unixTimestamp, Lists.newArrayList()); } List<FileStatus> filesStatuses = timestampFilesMap.get(unixTimestamp); filesStatuses.add(fileStatus); } List<FileStatus> result = new ArrayList<>(); for(TreeMap<Long, List<FileStatus>> timestampFileStatus : pathTimestampFilesMap.values()) { if(timestampFileStatus.size() <=0 ) { continue; } switch (versionSelectionPolicy) { case EARLIEST: result.addAll(timestampFileStatus.firstEntry().getValue()); break; case LATEST: result.addAll(timestampFileStatus.lastEntry().getValue()); break; default: throw new RuntimeException("Unsupported version selection policy"); } } return result; } }
UnixTimestampRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { PathFilter andPathFilter = new AndPathFilter(fileFilter, new TimestampPathFilter()); List<FileStatus> files = super.getFilesAtPath(fs, path, andPathFilter); if (VersionSelectionPolicy.ALL == versionSelectionPolicy) { return files; } Map<Pair<String, LocalDate>, TreeMap<Long, List<FileStatus>>> pathTimestampFilesMap = new HashMap<>(); for (FileStatus fileStatus : files) { String relativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()), datasetRoot()).toString(); Matcher matcher = timestampPattern.matcher(relativePath); if (!matcher.matches()) { continue; } String timestampStr = matcher.group(1); String rootPath = relativePath.substring(0, relativePath.indexOf(timestampStr)); Long unixTimestamp = Long.parseLong(timestampStr); LocalDate localDate = new LocalDateTime(unixTimestamp,dateTimeZone).toLocalDate(); Pair<String, LocalDate> key = new ImmutablePair<>(rootPath, localDate); if (!pathTimestampFilesMap.containsKey(key)) { pathTimestampFilesMap.put(key, new TreeMap<Long, List<FileStatus>>()); } Map<Long, List<FileStatus>> timestampFilesMap = pathTimestampFilesMap.get(key); if (!timestampFilesMap.containsKey(unixTimestamp)) { timestampFilesMap.put(unixTimestamp, Lists.newArrayList()); } List<FileStatus> filesStatuses = timestampFilesMap.get(unixTimestamp); filesStatuses.add(fileStatus); } List<FileStatus> result = new ArrayList<>(); for(TreeMap<Long, List<FileStatus>> timestampFileStatus : pathTimestampFilesMap.values()) { if(timestampFileStatus.size() <=0 ) { continue; } switch (versionSelectionPolicy) { case EARLIEST: result.addAll(timestampFileStatus.firstEntry().getValue()); break; case LATEST: result.addAll(timestampFileStatus.lastEntry().getValue()); break; default: throw new RuntimeException("Unsupported version selection policy"); } } return result; } UnixTimestampRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob); }
UnixTimestampRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { PathFilter andPathFilter = new AndPathFilter(fileFilter, new TimestampPathFilter()); List<FileStatus> files = super.getFilesAtPath(fs, path, andPathFilter); if (VersionSelectionPolicy.ALL == versionSelectionPolicy) { return files; } Map<Pair<String, LocalDate>, TreeMap<Long, List<FileStatus>>> pathTimestampFilesMap = new HashMap<>(); for (FileStatus fileStatus : files) { String relativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()), datasetRoot()).toString(); Matcher matcher = timestampPattern.matcher(relativePath); if (!matcher.matches()) { continue; } String timestampStr = matcher.group(1); String rootPath = relativePath.substring(0, relativePath.indexOf(timestampStr)); Long unixTimestamp = Long.parseLong(timestampStr); LocalDate localDate = new LocalDateTime(unixTimestamp,dateTimeZone).toLocalDate(); Pair<String, LocalDate> key = new ImmutablePair<>(rootPath, localDate); if (!pathTimestampFilesMap.containsKey(key)) { pathTimestampFilesMap.put(key, new TreeMap<Long, List<FileStatus>>()); } Map<Long, List<FileStatus>> timestampFilesMap = pathTimestampFilesMap.get(key); if (!timestampFilesMap.containsKey(unixTimestamp)) { timestampFilesMap.put(unixTimestamp, Lists.newArrayList()); } List<FileStatus> filesStatuses = timestampFilesMap.get(unixTimestamp); filesStatuses.add(fileStatus); } List<FileStatus> result = new ArrayList<>(); for(TreeMap<Long, List<FileStatus>> timestampFileStatus : pathTimestampFilesMap.values()) { if(timestampFileStatus.size() <=0 ) { continue; } switch (versionSelectionPolicy) { case EARLIEST: result.addAll(timestampFileStatus.firstEntry().getValue()); break; case LATEST: result.addAll(timestampFileStatus.lastEntry().getValue()); break; default: throw new RuntimeException("Unsupported version selection policy"); } } return result; } UnixTimestampRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob);  }
UnixTimestampRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { PathFilter andPathFilter = new AndPathFilter(fileFilter, new TimestampPathFilter()); List<FileStatus> files = super.getFilesAtPath(fs, path, andPathFilter); if (VersionSelectionPolicy.ALL == versionSelectionPolicy) { return files; } Map<Pair<String, LocalDate>, TreeMap<Long, List<FileStatus>>> pathTimestampFilesMap = new HashMap<>(); for (FileStatus fileStatus : files) { String relativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()), datasetRoot()).toString(); Matcher matcher = timestampPattern.matcher(relativePath); if (!matcher.matches()) { continue; } String timestampStr = matcher.group(1); String rootPath = relativePath.substring(0, relativePath.indexOf(timestampStr)); Long unixTimestamp = Long.parseLong(timestampStr); LocalDate localDate = new LocalDateTime(unixTimestamp,dateTimeZone).toLocalDate(); Pair<String, LocalDate> key = new ImmutablePair<>(rootPath, localDate); if (!pathTimestampFilesMap.containsKey(key)) { pathTimestampFilesMap.put(key, new TreeMap<Long, List<FileStatus>>()); } Map<Long, List<FileStatus>> timestampFilesMap = pathTimestampFilesMap.get(key); if (!timestampFilesMap.containsKey(unixTimestamp)) { timestampFilesMap.put(unixTimestamp, Lists.newArrayList()); } List<FileStatus> filesStatuses = timestampFilesMap.get(unixTimestamp); filesStatuses.add(fileStatus); } List<FileStatus> result = new ArrayList<>(); for(TreeMap<Long, List<FileStatus>> timestampFileStatus : pathTimestampFilesMap.values()) { if(timestampFileStatus.size() <=0 ) { continue; } switch (versionSelectionPolicy) { case EARLIEST: result.addAll(timestampFileStatus.firstEntry().getValue()); break; case LATEST: result.addAll(timestampFileStatus.lastEntry().getValue()); break; default: throw new RuntimeException("Unsupported version selection policy"); } } return result; } UnixTimestampRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob);  static final String VERSION_SELECTION_POLICY; static final String TIMESTAMP_REGEEX; static final String DEFAULT_TIMESTAMP_REGEX; }