@Test public void testFullPathDiff() throws Exception { Map<Path, FileStatus> sourceMap = Maps.newHashMap(); Map<Path, FileStatus> targetDesiredMap = Maps.newHashMap(); List<Path> expectedFilesToCopy = Lists.newArrayList(); List<Path> expectedFilesToSkipCopy = Lists.newArrayList(); List<Path> expectedFilesToDelete = Lists.newArrayList(); List<Path> expectedFilesToSkipDelete = Lists.newArrayList(); populateSourceAndTargetEntities(sourceMap, targetDesiredMap, expectedFilesToCopy, expectedFilesToSkipCopy, expectedFilesToDelete, expectedFilesToSkipDelete); TestLocationDescriptor sourceLocation = new TestLocationDescriptor(sourceMap); TestLocationDescriptor targetDesiredLocation = new TestLocationDescriptor(targetDesiredMap); TestLocationDescriptor existingTargetLocation = new TestLocationDescriptor(Maps.newHashMap(targetDesiredMap)); MultiTimingEvent timer = Mockito.mock(MultiTimingEvent.class); HiveCopyEntityHelper helper = Mockito.mock(HiveCopyEntityHelper.class); Mockito.when(helper.isEnforceFileSizeMatch()).thenReturn(true); HiveTargetPathHelper targetPathHelper = Mockito.mock(HiveTargetPathHelper.class); Mockito.when(targetPathHelper .getTargetPath(Mockito.any(Path.class), Mockito.any(FileSystem.class), Mockito.any(Optional.class), Mockito.anyBoolean())).then(new Answer<Path>() { @Override public Path answer(InvocationOnMock invocation) throws Throwable { Path path = (Path) invocation.getArguments()[0]; return new Path(path.toString().replace(sourceRoot.toString(), targetRoot.toString())); } }); Mockito.when(helper.getTargetPathHelper()).thenReturn(targetPathHelper); HiveCopyEntityHelper.DiffPathSet diff = HiveCopyEntityHelper.fullPathDiff(sourceLocation, targetDesiredLocation, Optional.<HiveLocationDescriptor>of(existingTargetLocation), Optional.<Partition>absent(), timer, helper); Assert.assertEquals(diff.filesToCopy.size(), expectedFilesToCopy.size()); for (Path expectedFileToCopy : expectedFilesToCopy) { Assert.assertTrue(containsPath(diff.filesToCopy, expectedFileToCopy)); } for (Path expectedFileToSkipCopy : expectedFilesToSkipCopy) { Assert.assertFalse(containsPath(diff.filesToCopy, expectedFileToSkipCopy)); } Assert.assertEquals(diff.pathsToDelete.size(), expectedFilesToDelete.size()); for (Path expectedFileToDelete : expectedFilesToDelete) { Assert.assertTrue(diff.pathsToDelete.contains(expectedFileToDelete)); } for (Path expectedFileToSkipDelete : expectedFilesToSkipDelete) { Assert.assertFalse(diff.pathsToDelete.contains(expectedFileToSkipDelete)); } }
@VisibleForTesting protected static DiffPathSet fullPathDiff(HiveLocationDescriptor sourceLocation, HiveLocationDescriptor desiredTargetLocation, Optional<HiveLocationDescriptor> currentTargetLocation, Optional<Partition> partition, MultiTimingEvent multiTimer, HiveCopyEntityHelper helper) throws IOException { sourceLocation.populateDataFileVersionStrategy(); desiredTargetLocation.populateDataFileVersionStrategy(); DiffPathSet.DiffPathSetBuilder builder = DiffPathSet.builder(); if (!sourceLocation.versionStrategy.isPresent() || !desiredTargetLocation.versionStrategy.isPresent()) { log.warn("Version strategy doesn't exist ({},{}), cannot handle copy.", sourceLocation.versionStrategy.isPresent(), desiredTargetLocation.versionStrategy.isPresent()); return builder.build(); } if (!sourceLocation.versionStrategy.get().getClass().getName() .equals(desiredTargetLocation.versionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", sourceLocation.versionStrategy.get().getClass().getName(), desiredTargetLocation.versionStrategy.get().getClass().getName()); return builder.build(); } multiTimer.nextStage(Stages.SOURCE_PATH_LISTING); Map<Path, FileStatus> sourcePaths = sourceLocation.getPaths(); multiTimer.nextStage(Stages.TARGET_EXISTING_PATH_LISTING); Map<Path, FileStatus> targetExistingPaths = currentTargetLocation.isPresent() ? currentTargetLocation.get().getPaths() : Maps.<Path, FileStatus> newHashMap(); multiTimer.nextStage(Stages.DESIRED_PATHS_LISTING); Map<Path, FileStatus> desiredTargetExistingPaths; try { desiredTargetExistingPaths = desiredTargetLocation.getPaths(); } catch (InvalidInputException ioe) { desiredTargetExistingPaths = Maps.newHashMap(); } multiTimer.nextStage(Stages.PATH_DIFF); for (FileStatus sourcePath : sourcePaths.values()) { Path newPath = helper.getTargetPathHelper().getTargetPath(sourcePath.getPath(), desiredTargetLocation.getFileSystem(), partition, true); boolean shouldCopy = true; boolean useDirectGetModTime = sourceLocation.versionStrategy.isPresent() && sourceLocation.versionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (desiredTargetExistingPaths.containsKey(newPath)) { FileStatus existingTargetStatus = desiredTargetExistingPaths.get(newPath); Comparable srcVer = useDirectGetModTime ? sourcePath.getModificationTime() : sourceLocation.versionStrategy.get().getVersion(sourcePath.getPath()); Comparable dstVer = useDirectGetModTime ? existingTargetStatus.getModificationTime() : desiredTargetLocation.versionStrategy.get().getVersion(existingTargetStatus.getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!helper.isEnforceFileSizeMatch() || existingTargetStatus.getLen() == sourcePath.getLen()) { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can be skipped since file size ({} bytes) is matching", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, sourcePath.getLen()); shouldCopy = false; } else { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can not be skipped because the file size is not matching or it is enforced by this config: {}", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, CopyConfiguration.ENFORCE_FILE_LENGTH_MATCH); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer); } } if (shouldCopy) { builder.copyFile(sourcePath); } else { targetExistingPaths.remove(newPath); desiredTargetExistingPaths.remove(newPath); } } multiTimer.nextStage(Stages.COMPUTE_DELETE_PATHS); for (Path delete : targetExistingPaths.keySet()) { builder.deleteFile(delete); desiredTargetExistingPaths.remove(delete); } if (desiredTargetExistingPaths.size() > 0 && helper.getUnmanagedDataPolicy() != UnmanagedDataPolicy.DELETE_UNMANAGED_DATA) { throw new IOException(String.format( "New table / partition would pick up existing, undesired files in target file system. " + "%s, files %s.", partition.isPresent() ? partition.get().getCompleteName() : helper.getDataset().getTable().getCompleteName(), Arrays.toString(desiredTargetExistingPaths.keySet().toArray()))); } else if (desiredTargetExistingPaths.size() > 0) { for (Path delete : desiredTargetExistingPaths.keySet()) { builder.deleteFile(delete); } log.warn(String.format("Un-managed files detected in target file system, however deleting them " + "because of the policy: %s Files to be deleted are: %s", UnmanagedDataPolicy.DELETE_UNMANAGED_DATA, StringUtils.join(desiredTargetExistingPaths.keySet(), ","))); } return builder.build(); }
HiveCopyEntityHelper { @VisibleForTesting protected static DiffPathSet fullPathDiff(HiveLocationDescriptor sourceLocation, HiveLocationDescriptor desiredTargetLocation, Optional<HiveLocationDescriptor> currentTargetLocation, Optional<Partition> partition, MultiTimingEvent multiTimer, HiveCopyEntityHelper helper) throws IOException { sourceLocation.populateDataFileVersionStrategy(); desiredTargetLocation.populateDataFileVersionStrategy(); DiffPathSet.DiffPathSetBuilder builder = DiffPathSet.builder(); if (!sourceLocation.versionStrategy.isPresent() || !desiredTargetLocation.versionStrategy.isPresent()) { log.warn("Version strategy doesn't exist ({},{}), cannot handle copy.", sourceLocation.versionStrategy.isPresent(), desiredTargetLocation.versionStrategy.isPresent()); return builder.build(); } if (!sourceLocation.versionStrategy.get().getClass().getName() .equals(desiredTargetLocation.versionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", sourceLocation.versionStrategy.get().getClass().getName(), desiredTargetLocation.versionStrategy.get().getClass().getName()); return builder.build(); } multiTimer.nextStage(Stages.SOURCE_PATH_LISTING); Map<Path, FileStatus> sourcePaths = sourceLocation.getPaths(); multiTimer.nextStage(Stages.TARGET_EXISTING_PATH_LISTING); Map<Path, FileStatus> targetExistingPaths = currentTargetLocation.isPresent() ? currentTargetLocation.get().getPaths() : Maps.<Path, FileStatus> newHashMap(); multiTimer.nextStage(Stages.DESIRED_PATHS_LISTING); Map<Path, FileStatus> desiredTargetExistingPaths; try { desiredTargetExistingPaths = desiredTargetLocation.getPaths(); } catch (InvalidInputException ioe) { desiredTargetExistingPaths = Maps.newHashMap(); } multiTimer.nextStage(Stages.PATH_DIFF); for (FileStatus sourcePath : sourcePaths.values()) { Path newPath = helper.getTargetPathHelper().getTargetPath(sourcePath.getPath(), desiredTargetLocation.getFileSystem(), partition, true); boolean shouldCopy = true; boolean useDirectGetModTime = sourceLocation.versionStrategy.isPresent() && sourceLocation.versionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (desiredTargetExistingPaths.containsKey(newPath)) { FileStatus existingTargetStatus = desiredTargetExistingPaths.get(newPath); Comparable srcVer = useDirectGetModTime ? sourcePath.getModificationTime() : sourceLocation.versionStrategy.get().getVersion(sourcePath.getPath()); Comparable dstVer = useDirectGetModTime ? existingTargetStatus.getModificationTime() : desiredTargetLocation.versionStrategy.get().getVersion(existingTargetStatus.getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!helper.isEnforceFileSizeMatch() || existingTargetStatus.getLen() == sourcePath.getLen()) { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can be skipped since file size ({} bytes) is matching", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, sourcePath.getLen()); shouldCopy = false; } else { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can not be skipped because the file size is not matching or it is enforced by this config: {}", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, CopyConfiguration.ENFORCE_FILE_LENGTH_MATCH); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer); } } if (shouldCopy) { builder.copyFile(sourcePath); } else { targetExistingPaths.remove(newPath); desiredTargetExistingPaths.remove(newPath); } } multiTimer.nextStage(Stages.COMPUTE_DELETE_PATHS); for (Path delete : targetExistingPaths.keySet()) { builder.deleteFile(delete); desiredTargetExistingPaths.remove(delete); } if (desiredTargetExistingPaths.size() > 0 && helper.getUnmanagedDataPolicy() != UnmanagedDataPolicy.DELETE_UNMANAGED_DATA) { throw new IOException(String.format( "New table / partition would pick up existing, undesired files in target file system. " + "%s, files %s.", partition.isPresent() ? partition.get().getCompleteName() : helper.getDataset().getTable().getCompleteName(), Arrays.toString(desiredTargetExistingPaths.keySet().toArray()))); } else if (desiredTargetExistingPaths.size() > 0) { for (Path delete : desiredTargetExistingPaths.keySet()) { builder.deleteFile(delete); } log.warn(String.format("Un-managed files detected in target file system, however deleting them " + "because of the policy: %s Files to be deleted are: %s", UnmanagedDataPolicy.DELETE_UNMANAGED_DATA, StringUtils.join(desiredTargetExistingPaths.keySet(), ","))); } return builder.build(); } }
HiveCopyEntityHelper { @VisibleForTesting protected static DiffPathSet fullPathDiff(HiveLocationDescriptor sourceLocation, HiveLocationDescriptor desiredTargetLocation, Optional<HiveLocationDescriptor> currentTargetLocation, Optional<Partition> partition, MultiTimingEvent multiTimer, HiveCopyEntityHelper helper) throws IOException { sourceLocation.populateDataFileVersionStrategy(); desiredTargetLocation.populateDataFileVersionStrategy(); DiffPathSet.DiffPathSetBuilder builder = DiffPathSet.builder(); if (!sourceLocation.versionStrategy.isPresent() || !desiredTargetLocation.versionStrategy.isPresent()) { log.warn("Version strategy doesn't exist ({},{}), cannot handle copy.", sourceLocation.versionStrategy.isPresent(), desiredTargetLocation.versionStrategy.isPresent()); return builder.build(); } if (!sourceLocation.versionStrategy.get().getClass().getName() .equals(desiredTargetLocation.versionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", sourceLocation.versionStrategy.get().getClass().getName(), desiredTargetLocation.versionStrategy.get().getClass().getName()); return builder.build(); } multiTimer.nextStage(Stages.SOURCE_PATH_LISTING); Map<Path, FileStatus> sourcePaths = sourceLocation.getPaths(); multiTimer.nextStage(Stages.TARGET_EXISTING_PATH_LISTING); Map<Path, FileStatus> targetExistingPaths = currentTargetLocation.isPresent() ? currentTargetLocation.get().getPaths() : Maps.<Path, FileStatus> newHashMap(); multiTimer.nextStage(Stages.DESIRED_PATHS_LISTING); Map<Path, FileStatus> desiredTargetExistingPaths; try { desiredTargetExistingPaths = desiredTargetLocation.getPaths(); } catch (InvalidInputException ioe) { desiredTargetExistingPaths = Maps.newHashMap(); } multiTimer.nextStage(Stages.PATH_DIFF); for (FileStatus sourcePath : sourcePaths.values()) { Path newPath = helper.getTargetPathHelper().getTargetPath(sourcePath.getPath(), desiredTargetLocation.getFileSystem(), partition, true); boolean shouldCopy = true; boolean useDirectGetModTime = sourceLocation.versionStrategy.isPresent() && sourceLocation.versionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (desiredTargetExistingPaths.containsKey(newPath)) { FileStatus existingTargetStatus = desiredTargetExistingPaths.get(newPath); Comparable srcVer = useDirectGetModTime ? sourcePath.getModificationTime() : sourceLocation.versionStrategy.get().getVersion(sourcePath.getPath()); Comparable dstVer = useDirectGetModTime ? existingTargetStatus.getModificationTime() : desiredTargetLocation.versionStrategy.get().getVersion(existingTargetStatus.getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!helper.isEnforceFileSizeMatch() || existingTargetStatus.getLen() == sourcePath.getLen()) { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can be skipped since file size ({} bytes) is matching", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, sourcePath.getLen()); shouldCopy = false; } else { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can not be skipped because the file size is not matching or it is enforced by this config: {}", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, CopyConfiguration.ENFORCE_FILE_LENGTH_MATCH); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer); } } if (shouldCopy) { builder.copyFile(sourcePath); } else { targetExistingPaths.remove(newPath); desiredTargetExistingPaths.remove(newPath); } } multiTimer.nextStage(Stages.COMPUTE_DELETE_PATHS); for (Path delete : targetExistingPaths.keySet()) { builder.deleteFile(delete); desiredTargetExistingPaths.remove(delete); } if (desiredTargetExistingPaths.size() > 0 && helper.getUnmanagedDataPolicy() != UnmanagedDataPolicy.DELETE_UNMANAGED_DATA) { throw new IOException(String.format( "New table / partition would pick up existing, undesired files in target file system. " + "%s, files %s.", partition.isPresent() ? partition.get().getCompleteName() : helper.getDataset().getTable().getCompleteName(), Arrays.toString(desiredTargetExistingPaths.keySet().toArray()))); } else if (desiredTargetExistingPaths.size() > 0) { for (Path delete : desiredTargetExistingPaths.keySet()) { builder.deleteFile(delete); } log.warn(String.format("Un-managed files detected in target file system, however deleting them " + "because of the policy: %s Files to be deleted are: %s", UnmanagedDataPolicy.DELETE_UNMANAGED_DATA, StringUtils.join(desiredTargetExistingPaths.keySet(), ","))); } return builder.build(); } HiveCopyEntityHelper(HiveDataset dataset, CopyConfiguration configuration, FileSystem targetFs); }
HiveCopyEntityHelper { @VisibleForTesting protected static DiffPathSet fullPathDiff(HiveLocationDescriptor sourceLocation, HiveLocationDescriptor desiredTargetLocation, Optional<HiveLocationDescriptor> currentTargetLocation, Optional<Partition> partition, MultiTimingEvent multiTimer, HiveCopyEntityHelper helper) throws IOException { sourceLocation.populateDataFileVersionStrategy(); desiredTargetLocation.populateDataFileVersionStrategy(); DiffPathSet.DiffPathSetBuilder builder = DiffPathSet.builder(); if (!sourceLocation.versionStrategy.isPresent() || !desiredTargetLocation.versionStrategy.isPresent()) { log.warn("Version strategy doesn't exist ({},{}), cannot handle copy.", sourceLocation.versionStrategy.isPresent(), desiredTargetLocation.versionStrategy.isPresent()); return builder.build(); } if (!sourceLocation.versionStrategy.get().getClass().getName() .equals(desiredTargetLocation.versionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", sourceLocation.versionStrategy.get().getClass().getName(), desiredTargetLocation.versionStrategy.get().getClass().getName()); return builder.build(); } multiTimer.nextStage(Stages.SOURCE_PATH_LISTING); Map<Path, FileStatus> sourcePaths = sourceLocation.getPaths(); multiTimer.nextStage(Stages.TARGET_EXISTING_PATH_LISTING); Map<Path, FileStatus> targetExistingPaths = currentTargetLocation.isPresent() ? currentTargetLocation.get().getPaths() : Maps.<Path, FileStatus> newHashMap(); multiTimer.nextStage(Stages.DESIRED_PATHS_LISTING); Map<Path, FileStatus> desiredTargetExistingPaths; try { desiredTargetExistingPaths = desiredTargetLocation.getPaths(); } catch (InvalidInputException ioe) { desiredTargetExistingPaths = Maps.newHashMap(); } multiTimer.nextStage(Stages.PATH_DIFF); for (FileStatus sourcePath : sourcePaths.values()) { Path newPath = helper.getTargetPathHelper().getTargetPath(sourcePath.getPath(), desiredTargetLocation.getFileSystem(), partition, true); boolean shouldCopy = true; boolean useDirectGetModTime = sourceLocation.versionStrategy.isPresent() && sourceLocation.versionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (desiredTargetExistingPaths.containsKey(newPath)) { FileStatus existingTargetStatus = desiredTargetExistingPaths.get(newPath); Comparable srcVer = useDirectGetModTime ? sourcePath.getModificationTime() : sourceLocation.versionStrategy.get().getVersion(sourcePath.getPath()); Comparable dstVer = useDirectGetModTime ? existingTargetStatus.getModificationTime() : desiredTargetLocation.versionStrategy.get().getVersion(existingTargetStatus.getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!helper.isEnforceFileSizeMatch() || existingTargetStatus.getLen() == sourcePath.getLen()) { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can be skipped since file size ({} bytes) is matching", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, sourcePath.getLen()); shouldCopy = false; } else { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can not be skipped because the file size is not matching or it is enforced by this config: {}", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, CopyConfiguration.ENFORCE_FILE_LENGTH_MATCH); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer); } } if (shouldCopy) { builder.copyFile(sourcePath); } else { targetExistingPaths.remove(newPath); desiredTargetExistingPaths.remove(newPath); } } multiTimer.nextStage(Stages.COMPUTE_DELETE_PATHS); for (Path delete : targetExistingPaths.keySet()) { builder.deleteFile(delete); desiredTargetExistingPaths.remove(delete); } if (desiredTargetExistingPaths.size() > 0 && helper.getUnmanagedDataPolicy() != UnmanagedDataPolicy.DELETE_UNMANAGED_DATA) { throw new IOException(String.format( "New table / partition would pick up existing, undesired files in target file system. " + "%s, files %s.", partition.isPresent() ? partition.get().getCompleteName() : helper.getDataset().getTable().getCompleteName(), Arrays.toString(desiredTargetExistingPaths.keySet().toArray()))); } else if (desiredTargetExistingPaths.size() > 0) { for (Path delete : desiredTargetExistingPaths.keySet()) { builder.deleteFile(delete); } log.warn(String.format("Un-managed files detected in target file system, however deleting them " + "because of the policy: %s Files to be deleted are: %s", UnmanagedDataPolicy.DELETE_UNMANAGED_DATA, StringUtils.join(desiredTargetExistingPaths.keySet(), ","))); } return builder.build(); } HiveCopyEntityHelper(HiveDataset dataset, CopyConfiguration configuration, FileSystem targetFs); FileSystem getTargetFileSystem(); }
HiveCopyEntityHelper { @VisibleForTesting protected static DiffPathSet fullPathDiff(HiveLocationDescriptor sourceLocation, HiveLocationDescriptor desiredTargetLocation, Optional<HiveLocationDescriptor> currentTargetLocation, Optional<Partition> partition, MultiTimingEvent multiTimer, HiveCopyEntityHelper helper) throws IOException { sourceLocation.populateDataFileVersionStrategy(); desiredTargetLocation.populateDataFileVersionStrategy(); DiffPathSet.DiffPathSetBuilder builder = DiffPathSet.builder(); if (!sourceLocation.versionStrategy.isPresent() || !desiredTargetLocation.versionStrategy.isPresent()) { log.warn("Version strategy doesn't exist ({},{}), cannot handle copy.", sourceLocation.versionStrategy.isPresent(), desiredTargetLocation.versionStrategy.isPresent()); return builder.build(); } if (!sourceLocation.versionStrategy.get().getClass().getName() .equals(desiredTargetLocation.versionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", sourceLocation.versionStrategy.get().getClass().getName(), desiredTargetLocation.versionStrategy.get().getClass().getName()); return builder.build(); } multiTimer.nextStage(Stages.SOURCE_PATH_LISTING); Map<Path, FileStatus> sourcePaths = sourceLocation.getPaths(); multiTimer.nextStage(Stages.TARGET_EXISTING_PATH_LISTING); Map<Path, FileStatus> targetExistingPaths = currentTargetLocation.isPresent() ? currentTargetLocation.get().getPaths() : Maps.<Path, FileStatus> newHashMap(); multiTimer.nextStage(Stages.DESIRED_PATHS_LISTING); Map<Path, FileStatus> desiredTargetExistingPaths; try { desiredTargetExistingPaths = desiredTargetLocation.getPaths(); } catch (InvalidInputException ioe) { desiredTargetExistingPaths = Maps.newHashMap(); } multiTimer.nextStage(Stages.PATH_DIFF); for (FileStatus sourcePath : sourcePaths.values()) { Path newPath = helper.getTargetPathHelper().getTargetPath(sourcePath.getPath(), desiredTargetLocation.getFileSystem(), partition, true); boolean shouldCopy = true; boolean useDirectGetModTime = sourceLocation.versionStrategy.isPresent() && sourceLocation.versionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (desiredTargetExistingPaths.containsKey(newPath)) { FileStatus existingTargetStatus = desiredTargetExistingPaths.get(newPath); Comparable srcVer = useDirectGetModTime ? sourcePath.getModificationTime() : sourceLocation.versionStrategy.get().getVersion(sourcePath.getPath()); Comparable dstVer = useDirectGetModTime ? existingTargetStatus.getModificationTime() : desiredTargetLocation.versionStrategy.get().getVersion(existingTargetStatus.getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!helper.isEnforceFileSizeMatch() || existingTargetStatus.getLen() == sourcePath.getLen()) { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can be skipped since file size ({} bytes) is matching", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, sourcePath.getLen()); shouldCopy = false; } else { log.debug("Copy from src {} (version:{}) to dst {} (version:{}) can not be skipped because the file size is not matching or it is enforced by this config: {}", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer, CopyConfiguration.ENFORCE_FILE_LENGTH_MATCH); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", sourcePath.getPath(), srcVer, existingTargetStatus.getPath(), dstVer); } } if (shouldCopy) { builder.copyFile(sourcePath); } else { targetExistingPaths.remove(newPath); desiredTargetExistingPaths.remove(newPath); } } multiTimer.nextStage(Stages.COMPUTE_DELETE_PATHS); for (Path delete : targetExistingPaths.keySet()) { builder.deleteFile(delete); desiredTargetExistingPaths.remove(delete); } if (desiredTargetExistingPaths.size() > 0 && helper.getUnmanagedDataPolicy() != UnmanagedDataPolicy.DELETE_UNMANAGED_DATA) { throw new IOException(String.format( "New table / partition would pick up existing, undesired files in target file system. " + "%s, files %s.", partition.isPresent() ? partition.get().getCompleteName() : helper.getDataset().getTable().getCompleteName(), Arrays.toString(desiredTargetExistingPaths.keySet().toArray()))); } else if (desiredTargetExistingPaths.size() > 0) { for (Path delete : desiredTargetExistingPaths.keySet()) { builder.deleteFile(delete); } log.warn(String.format("Un-managed files detected in target file system, however deleting them " + "because of the policy: %s Files to be deleted are: %s", UnmanagedDataPolicy.DELETE_UNMANAGED_DATA, StringUtils.join(desiredTargetExistingPaths.keySet(), ","))); } return builder.build(); } HiveCopyEntityHelper(HiveDataset dataset, CopyConfiguration configuration, FileSystem targetFs); FileSystem getTargetFileSystem(); static final String EXISTING_ENTITY_POLICY_KEY; static final String DEFAULT_EXISTING_ENTITY_POLICY; static final String UNMANAGED_DATA_POLICY_KEY; static final String DEFAULT_UNMANAGED_DATA_POLICY; static final String TARGET_METASTORE_URI_KEY; static final String TARGET_DATABASE_KEY; static final String COPY_PARTITIONS_FILTER_CONSTANT; static final String COPY_PARTITION_FILTER_GENERATOR; static final String FAST_PARTITION_SKIP_PREDICATE; static final String FAST_TABLE_SKIP_PREDICATE; static final String DELETE_FILES_ON_DEREGISTER; static final DeregisterFileDeleteMethod DEFAULT_DEREGISTER_DELETE_METHOD; static final String HIVE_PARTITION_EXTENDED_FILTER_TYPE; static final String GOBBLIN_DISTCP; }