Copilot commented on code in PR #8858:
URL: https://github.com/apache/hadoop/pull/8858#discussion_r4358034303
##########
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java:
##########
@@ -654,6 +654,13 @@ private void concatFileChunks(Configuration conf, Path
sourceFile,
DistCpUtils.compareFileLengthsAndChecksums(srcFileStatus.getLen(),
srcfs, sourceFile, null, dstfs,
targetFile, skipCrc, srcFileStatus.getLen());
+ // Concat changes the modification time preserved on the first chunk.
+ // Restore times after validation, which may also update the access time.
+ if (DistCpUtils.unpackAttributes(conf.get(CONF_LABEL_PRESERVE_STATUS))
+ .contains(FileAttribute.TIMES)) {
+ dstfs.setTimes(targetFile, srcFileStatus.getModificationTime(),
+ srcFileStatus.getAccessTime());
Review Comment:
`setTimes(Path, mtime, atime)` typically uses `-1` to mean 'do not change'.
If `srcFileStatus.getAccessTime()` is `0` (common on filesystems that don’t
expose atime), this will actively set the destination atime to epoch rather
than preserving/ignoring it. Consider passing `-1` when the source atime is
non-positive (e.g., compute `long atime = srcFileStatus.getAccessTime() > 0 ?
srcFileStatus.getAccessTime() : -1`) so enabling `TIMES` doesn’t inadvertently
clobber atime on FSes without atime support.
##########
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java:
##########
@@ -252,6 +271,243 @@ protected Path getRemoteDir() {
return remoteDir;
}
+ @ParameterizedTest(name = "preserveTimes={0}, direct={1}, atomic={2}")
+ @CsvSource({
+ "true, false, false",
+ "false, false, false",
+ "true, true, false",
+ "false, true, false",
+ "true, false, true",
+ "false, false, true"
+ })
+ public void testChunkCopyPreserveTimes(boolean preserveTimes,
+ boolean directWrite, boolean atomicCommit) throws Exception {
+ requireChunkCopySupport();
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+
+ DistCpOptions.Builder builder = chunkCopyOptions(source, target,
+ preserveTimes).withDirectWrite(directWrite || shouldUseDirectWrite())
+ .withAtomicCommit(atomicCommit);
+ runDistCp(builder.build(), true);
+
+ // The source directory itself is in the copy listing, so its times and
+ // those of nested directories must be restored after all concatenations.
+ verifyChunkCopyTree(target, sourceStatuses, preserveTimes, true);
+ }
+
+ @ParameterizedTest(name = "preserveTimes={0}, renameTarget={1}")
+ @CsvSource({"true, false", "false, false", "true, true", "false, true"})
+ public void testChunkCopyOutsideParentScope(boolean preserveTimes,
+ boolean renameTarget)
+ throws Exception {
+ requireChunkCopySupport();
+ Path sourceParent = new Path(remoteDir, "source-parent");
+ Path source = new Path(sourceParent, "file");
+ Path targetParent = new Path(remoteDir, "target-parent");
+ Path target = new Path(targetParent,
+ renameTarget ? "renamed-file" : source.getName());
+ createChunkCopyFile(source, TEST_FILE_LEN * 3 + 1);
+ FileStatus sourceStatus = setChunkCopyTimes(source, 0);
+ FileStatus sourceParentStatus = setChunkCopyTimes(sourceParent, 1);
+ mkdirs(remoteFS, targetParent);
+ FileStatus targetParentStatus = setChunkCopyTimes(targetParent, 2);
+
+ DistCpOptions.Builder builder = chunkCopyOptions(source,
+ renameTarget ? target : targetParent, preserveTimes);
+ if (renameTarget) {
+ builder.withDirectWrite(true);
+ }
+ runDistCp(builder.build(), true);
+
+ assertChunkCopyTimes(sourceStatus, remoteFS.getFileStatus(target),
+ preserveTimes);
+ // Creating/concatenating children may change the destination parent's
+ // mtime naturally. It must not inherit the source parent's timestamps.
+ FileStatus parentAfterCopy = remoteFS.getFileStatus(targetParent);
+ Assertions.assertThat(parentAfterCopy.getModificationTime())
+ .describedAs("mtime of parent outside copy scope")
+ .isNotEqualTo(sourceParentStatus.getModificationTime());
+ assertEquals(targetParentStatus.getAccessTime(),
+ parentAfterCopy.getAccessTime(), "atime of parent outside copy scope");
+ verifyFileContents(remoteFS, target,
+ dataset((int) sourceStatus.getLen(), 'a', 26));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testChunkCopyUpdateRootTimes(boolean updateRoot)
+ throws Exception {
+ requireChunkCopySupport();
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+ // Force an existing destination file to be replaced by a chunked copy.
+ createFile(remoteFS, new Path(target, "large"), true, new byte[1]);
+ FileStatus targetRootStatus = setChunkCopyTimes(target, 20);
+ DistCpOptions options = chunkCopyOptions(source, target, true)
+ .withSyncFolder(true).withUpdateRoot(updateRoot).build();
+
+ runDistCp(options, true);
+ verifyChunkCopyTree(target, sourceStatuses, true, updateRoot);
+ if (!updateRoot) {
+ assertEquals(targetRootStatus.getAccessTime(),
+ remoteFS.getFileStatus(target).getAccessTime(),
+ "-update must not preserve root times without -updateRoot");
+ }
+
+ // A fully skipped update must still preserve file times without
+ // attempting to concatenate nonexistent chunk files.
+ setChunkCopyTimes(new Path(target, "large"), 30);
+ Job skipped = runDistCp(options, true);
+ assertCounterInRange(skipped, CopyMapper.Counter.COPY, 0, 0);
+ assertCounterInRange(skipped, CopyMapper.Counter.SKIP, 1, -1);
+ assertCounterInRange(skipped, CopyMapper.Counter.BYTESCOPIED, 0, 0);
+ verifyChunkCopyTree(target, sourceStatuses, true, updateRoot);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testChunkCopyPreserveOtherAttributes(boolean preserveTimes)
+ throws Exception {
+ requireChunkCopySupport();
+ skipIfUnsupported(SUPPORTS_UNIX_PERMISSIONS);
+ if (!remoteFS.hasPathCapability(remoteDir, FS_ACLS)
+ || !remoteFS.hasPathCapability(remoteDir, FS_XATTRS)) {
+ skip("ACL and XAttr support required");
+ }
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+ Path sourceFile = new Path(source, "large");
+ remoteFS.setPermission(sourceFile, new FsPermission((short) 0640));
+ remoteFS.modifyAclEntries(sourceFile, Collections.singletonList(
+ new AclEntry.Builder().setScope(AclEntryScope.ACCESS)
+ .setType(AclEntryType.USER).setName("distcp-reader")
+ .setPermission(FsAction.READ).build()));
+ String xattrName = "user.distcp";
+ byte[] xattrValue = {1, 2, 3};
+ remoteFS.setXAttr(sourceFile, xattrName, xattrValue);
+ sourceStatuses.put("large", setChunkCopyTimes(sourceFile, 0));
+ sourceStatuses.put("", setChunkCopyTimes(source, 11));
+
+ runDistCp(chunkCopyOptions(source, target, preserveTimes)
+ .preserve("rbugpcax").build(), true);
+
+ // File listing entries omit extended ACLs: restoring all attributes from
+ // them after concat would erase ACLs already preserved by the mapper.
+ Path targetFile = new Path(target, "large");
+ assertEquals(remoteFS.getAclStatus(sourceFile).getEntries(),
+ remoteFS.getAclStatus(targetFile).getEntries(), "file ACL entries");
+ assertArrayEquals(xattrValue, remoteFS.getXAttr(targetFile, xattrName));
+ FileStatus expected = sourceStatuses.get("large");
+ FileStatus actual = remoteFS.getFileStatus(targetFile);
+ assertEquals(expected.getPermission(), actual.getPermission());
+ assertEquals(expected.getOwner(), actual.getOwner());
+ assertEquals(expected.getGroup(), actual.getGroup());
+ assertEquals(expected.getReplication(), actual.getReplication());
+ assertEquals(expected.getBlockSize(), actual.getBlockSize());
+ verifyChunkCopyTree(target, sourceStatuses, preserveTimes, true);
+ }
+
+ private void requireChunkCopySupport() throws IOException {
+ skipIfUnsupported(SUPPORTS_CONCAT);
+ skipIfUnsupported(SUPPORTS_SETTIMES);
+ skipIfUnsupported(SUPPORTS_BLOCK_LOCALITY);
+ conf.setInt("mapreduce.local.map.tasks.maximum", 2);
+ }
+
+ private DistCpOptions.Builder chunkCopyOptions(Path source, Path target,
+ boolean preserveTimes) {
+ DistCpOptions.Builder builder = new DistCpOptions.Builder(
+ Collections.singletonList(source), target)
+ .withBlocksPerChunk(1).withDirectWrite(shouldUseDirectWrite());
+ if (preserveTimes) {
+ builder.preserve(FileAttribute.TIMES);
+ }
+ return builder;
+ }
+
+ private Map<String, FileStatus> createChunkCopyTree(Path source)
+ throws IOException {
+ Map<String, FileStatus> statuses = new LinkedHashMap<>();
+ String[] names = {"large", "large2", "nested/large", "block", "small",
+ "empty"};
+ int[] lengths = {TEST_FILE_LEN * 3 + 1, TEST_FILE_LEN * 4,
+ TEST_FILE_LEN * 2, TEST_FILE_LEN, TEST_FILE_LEN / 2, 0};
+ for (int i = 0; i < names.length; i++) {
+ Path file = new Path(source, names[i]);
+ createChunkCopyFile(file, lengths[i]);
+ statuses.put(names[i], setChunkCopyTimes(file, i));
+ }
+ // Set directory times only after all children have been created.
+ statuses.put("nested", setChunkCopyTimes(new Path(source, "nested"), 10));
+ statuses.put("", setChunkCopyTimes(source, 11));
+ return statuses;
+ }
+
+ private void createChunkCopyFile(Path path, int length) throws IOException {
+ writeDataset(remoteFS, path, dataset(length, 'a', 26), length,
+ TEST_FILE_LEN, true);
+ if (length > TEST_FILE_LEN) {
+ FileStatus status = remoteFS.getFileStatus(path);
+ assertTrue(status.getLen() > status.getBlockSize(),
+ "File must exceed its block size to exercise chunked copying");
+ assertTrue(remoteFS.getFileBlockLocations(status, 0, length).length > 1,
+ "File must have multiple blocks to exercise concatenation");
+ }
+ }
+
+ private FileStatus setChunkCopyTimes(Path path, int offset)
+ throws IOException {
+ long mtime = CHUNK_COPY_TIME + offset * MODIFICATION_TIME_OFFSET;
+ remoteFS.setTimes(path, mtime,
+ CHUNK_COPY_ACCESS_TIME + offset * MODIFICATION_TIME_OFFSET);
+ FileStatus status = remoteFS.getFileStatus(path);
+ assertEquals(mtime, status.getModificationTime(), "source fixture mtime");
Review Comment:
The fixture validates only mtime, but later tests assert both mtime and
atime preservation. If `setTimes()` silently ignores atime on a given FS (or
normalizes it), failures later can be harder to diagnose. Consider also
asserting the expected atime when available (e.g., if `status.getAccessTime() >
0`, assert it matches the requested value), or documenting that atime may not
round-trip even when `SUPPORTS_SETTIMES` is true.
##########
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java:
##########
@@ -654,6 +654,13 @@ private void concatFileChunks(Configuration conf, Path
sourceFile,
DistCpUtils.compareFileLengthsAndChecksums(srcFileStatus.getLen(),
srcfs, sourceFile, null, dstfs,
targetFile, skipCrc, srcFileStatus.getLen());
+ // Concat changes the modification time preserved on the first chunk.
+ // Restore times after validation, which may also update the access time.
+ if (DistCpUtils.unpackAttributes(conf.get(CONF_LABEL_PRESERVE_STATUS))
+ .contains(FileAttribute.TIMES)) {
Review Comment:
`DistCpUtils.unpackAttributes(...)` appears to parse/unpack the
preserved-attributes set on every `concatFileChunks()` call. Since concat may
run once per copied file, this can add avoidable overhead at scale. Consider
unpacking once (e.g., in the committer initialization / caller) and passing a
boolean `preserveTimes`, or caching the unpacked set for reuse.
##########
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java:
##########
@@ -76,6 +89,12 @@ public abstract class AbstractContractDistCpTest
/** Using offset to change modification time in tests. */
private static final long MODIFICATION_TIME_OFFSET = 10000;
+ private static final long CHUNK_COPY_TIME = 1600000000000L;
+
+ // Keep reads and checksum verification from advancing the source atime.
+ private static final long CHUNK_COPY_ACCESS_TIME =
+ (System.currentTimeMillis() / 1000 + 86400) * 1000;
Review Comment:
This makes test behavior time-dependent by setting atime relative to the
wall clock (and into the future). On some filesystems/implementations,
read-atime updates may set atime to 'now' even if that is earlier than the
current atime, which could create non-deterministic failures depending on FS
semantics. A more deterministic approach is to use a fixed constant (like
`CHUNK_COPY_TIME +/- offset`) and, if needed, re-stat the source after the copy
to derive the expected atime (or conditionally assert atime only when it
remains stable on the contract FS).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
