@Test public void test() throws Exception { Path partition1Path = new Path("/path/to/partition1"); long modTime = 100000; CopyContext copyContext = new CopyContext(); CopyConfiguration copyConfiguration = Mockito.mock(CopyConfiguration.class); Mockito.doReturn(copyContext).when(copyConfiguration).getCopyContext(); HiveDataset dataset = Mockito.mock(HiveDataset.class); FileSystem fs = Mockito.spy(FileSystem.getLocal(new Configuration())); FileStatus status = new FileStatus(1, false, 1, 1, modTime, partition1Path); Path qualifiedPath = fs.makeQualified(partition1Path); Mockito.doReturn(status).when(fs).getFileStatus(qualifiedPath); Mockito.doReturn(status).when(fs).getFileStatus(partition1Path); Mockito.doReturn(fs).when(dataset).getFs(); HiveCopyEntityHelper helper = Mockito.mock(HiveCopyEntityHelper.class); Mockito.doReturn(copyConfiguration).when(helper).getConfiguration(); Mockito.doReturn(dataset).when(helper).getDataset(); RegistrationTimeSkipPredicate predicate = new RegistrationTimeSkipPredicate(helper); HivePartitionFileSet pc = createPartitionCopy(partition1Path, modTime - 1, true); Assert.assertFalse(predicate.apply(pc)); pc = createPartitionCopy(partition1Path, modTime, true); Assert.assertFalse(predicate.apply(pc)); pc = createPartitionCopy(partition1Path, modTime + 1, true); Assert.assertTrue(predicate.apply(pc)); pc = createPartitionCopy(partition1Path, modTime + 1, false); Assert.assertFalse(predicate.apply(pc)); pc = createPartitionCopy(partition1Path, modTime + 1, true); pc.getExistingTargetPartition().get().getParameters().clear(); Assert.assertFalse(predicate.apply(pc)); }
@Override public boolean apply(@Nullable HivePartitionFileSet input) { if (input == null) { return true; } if (input.getExistingTargetPartition() == null) { throw new RuntimeException("Existing target partition has not been computed! This is an error in the code."); } if (PathUtils.isGlob(input.getPartition().getDataLocation())) { log.error(String.format("%s cannot be applied to globbed location %s. Will not skip.", this.getClass().getSimpleName(), input.getPartition().getDataLocation())); return false; } if (!input.getExistingTargetPartition().isPresent()) { return false; } if (!input.getExistingTargetPartition().get().getParameters().containsKey(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)) { return false; } try { long oldRegistrationTime = Long.parseLong( input.getExistingTargetPartition().get().getParameters().get(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)); Optional<FileStatus> sourceFileStatus = this.helper.getConfiguration().getCopyContext(). getFileStatus(this.helper.getDataset().getFs(), input.getPartition().getDataLocation()); if (!sourceFileStatus.isPresent()) { throw new RuntimeException(String.format("Source path %s does not exist!", input.getPartition().getDataLocation())); } return oldRegistrationTime > sourceFileStatus.get().getModificationTime(); } catch (NumberFormatException nfe) { log.warn(String.format("Cannot parse %s in partition %s. Will not skip.", HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS, input.getPartition().getCompleteName())); return false; } catch (IOException ioe) { throw new RuntimeException(ioe); } }
RegistrationTimeSkipPredicate implements Predicate<HivePartitionFileSet> { @Override public boolean apply(@Nullable HivePartitionFileSet input) { if (input == null) { return true; } if (input.getExistingTargetPartition() == null) { throw new RuntimeException("Existing target partition has not been computed! This is an error in the code."); } if (PathUtils.isGlob(input.getPartition().getDataLocation())) { log.error(String.format("%s cannot be applied to globbed location %s. Will not skip.", this.getClass().getSimpleName(), input.getPartition().getDataLocation())); return false; } if (!input.getExistingTargetPartition().isPresent()) { return false; } if (!input.getExistingTargetPartition().get().getParameters().containsKey(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)) { return false; } try { long oldRegistrationTime = Long.parseLong( input.getExistingTargetPartition().get().getParameters().get(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)); Optional<FileStatus> sourceFileStatus = this.helper.getConfiguration().getCopyContext(). getFileStatus(this.helper.getDataset().getFs(), input.getPartition().getDataLocation()); if (!sourceFileStatus.isPresent()) { throw new RuntimeException(String.format("Source path %s does not exist!", input.getPartition().getDataLocation())); } return oldRegistrationTime > sourceFileStatus.get().getModificationTime(); } catch (NumberFormatException nfe) { log.warn(String.format("Cannot parse %s in partition %s. Will not skip.", HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS, input.getPartition().getCompleteName())); return false; } catch (IOException ioe) { throw new RuntimeException(ioe); } } }
RegistrationTimeSkipPredicate implements Predicate<HivePartitionFileSet> { @Override public boolean apply(@Nullable HivePartitionFileSet input) { if (input == null) { return true; } if (input.getExistingTargetPartition() == null) { throw new RuntimeException("Existing target partition has not been computed! This is an error in the code."); } if (PathUtils.isGlob(input.getPartition().getDataLocation())) { log.error(String.format("%s cannot be applied to globbed location %s. Will not skip.", this.getClass().getSimpleName(), input.getPartition().getDataLocation())); return false; } if (!input.getExistingTargetPartition().isPresent()) { return false; } if (!input.getExistingTargetPartition().get().getParameters().containsKey(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)) { return false; } try { long oldRegistrationTime = Long.parseLong( input.getExistingTargetPartition().get().getParameters().get(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)); Optional<FileStatus> sourceFileStatus = this.helper.getConfiguration().getCopyContext(). getFileStatus(this.helper.getDataset().getFs(), input.getPartition().getDataLocation()); if (!sourceFileStatus.isPresent()) { throw new RuntimeException(String.format("Source path %s does not exist!", input.getPartition().getDataLocation())); } return oldRegistrationTime > sourceFileStatus.get().getModificationTime(); } catch (NumberFormatException nfe) { log.warn(String.format("Cannot parse %s in partition %s. Will not skip.", HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS, input.getPartition().getCompleteName())); return false; } catch (IOException ioe) { throw new RuntimeException(ioe); } }  }
RegistrationTimeSkipPredicate implements Predicate<HivePartitionFileSet> { @Override public boolean apply(@Nullable HivePartitionFileSet input) { if (input == null) { return true; } if (input.getExistingTargetPartition() == null) { throw new RuntimeException("Existing target partition has not been computed! This is an error in the code."); } if (PathUtils.isGlob(input.getPartition().getDataLocation())) { log.error(String.format("%s cannot be applied to globbed location %s. Will not skip.", this.getClass().getSimpleName(), input.getPartition().getDataLocation())); return false; } if (!input.getExistingTargetPartition().isPresent()) { return false; } if (!input.getExistingTargetPartition().get().getParameters().containsKey(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)) { return false; } try { long oldRegistrationTime = Long.parseLong( input.getExistingTargetPartition().get().getParameters().get(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)); Optional<FileStatus> sourceFileStatus = this.helper.getConfiguration().getCopyContext(). getFileStatus(this.helper.getDataset().getFs(), input.getPartition().getDataLocation()); if (!sourceFileStatus.isPresent()) { throw new RuntimeException(String.format("Source path %s does not exist!", input.getPartition().getDataLocation())); } return oldRegistrationTime > sourceFileStatus.get().getModificationTime(); } catch (NumberFormatException nfe) { log.warn(String.format("Cannot parse %s in partition %s. Will not skip.", HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS, input.getPartition().getCompleteName())); return false; } catch (IOException ioe) { throw new RuntimeException(ioe); } }  @Override boolean apply(@Nullable HivePartitionFileSet input); }
RegistrationTimeSkipPredicate implements Predicate<HivePartitionFileSet> { @Override public boolean apply(@Nullable HivePartitionFileSet input) { if (input == null) { return true; } if (input.getExistingTargetPartition() == null) { throw new RuntimeException("Existing target partition has not been computed! This is an error in the code."); } if (PathUtils.isGlob(input.getPartition().getDataLocation())) { log.error(String.format("%s cannot be applied to globbed location %s. Will not skip.", this.getClass().getSimpleName(), input.getPartition().getDataLocation())); return false; } if (!input.getExistingTargetPartition().isPresent()) { return false; } if (!input.getExistingTargetPartition().get().getParameters().containsKey(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)) { return false; } try { long oldRegistrationTime = Long.parseLong( input.getExistingTargetPartition().get().getParameters().get(HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS)); Optional<FileStatus> sourceFileStatus = this.helper.getConfiguration().getCopyContext(). getFileStatus(this.helper.getDataset().getFs(), input.getPartition().getDataLocation()); if (!sourceFileStatus.isPresent()) { throw new RuntimeException(String.format("Source path %s does not exist!", input.getPartition().getDataLocation())); } return oldRegistrationTime > sourceFileStatus.get().getModificationTime(); } catch (NumberFormatException nfe) { log.warn(String.format("Cannot parse %s in partition %s. Will not skip.", HiveDataset.REGISTRATION_GENERATION_TIME_MILLIS, input.getPartition().getCompleteName())); return false; } catch (IOException ioe) { throw new RuntimeException(ioe); } }  @Override boolean apply(@Nullable HivePartitionFileSet input);  }