HADOOP-901. Add support for recursive renaming to the S3 filesystem. Contributed by Tom White.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@498829 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -54,6 +54,9 @@ Trunk (unreleased changes)
 16. HADOOP-908.  Add a new contrib package, Abacus, that simplifies
     counting and aggregation, built on MapReduce.  (Runping Qi via cutting)
 
+17. HADOOP-901.  Add support for recursive renaming to the S3 filesystem.
+    (Tom White via cutting)
+
 
 Release 0.10.1 - 2007-01-10
 

src/java/org/apache/hadoop/fs/s3/FileSystemStore.java

@@ -25,6 +25,7 @@ interface FileSystemStore {
   void deleteBlock(Block block) throws IOException;
 
   Set<Path> listSubPaths(Path path) throws IOException;
+  Set<Path> listDeepSubPaths(Path path) throws IOException;
 
   /**
    * Delete everything. Used for testing.

src/java/org/apache/hadoop/fs/s3/Jets3tFileSystemStore.java

@@ -188,6 +188,27 @@ class Jets3tFileSystemStore implements FileSystemStore {
       throw new S3Exception(e);
     }
   }
+  
+  public Set<Path> listDeepSubPaths(Path path) throws IOException {
+    try {
+      String prefix = pathToKey(path);
+      if (!prefix.endsWith(PATH_DELIMITER)) {
+        prefix += PATH_DELIMITER;
+      }
+      S3Object[] objects = s3Service.listObjects(bucket, prefix, null);
+      Set<Path> prefixes = new TreeSet<Path>();
+      for (int i = 0; i < objects.length; i++) {
+        prefixes.add(keyToPath(objects[i].getKey()));
+      }
+      prefixes.remove(path);
+      return prefixes;
+    } catch (S3ServiceException e) {
+      if (e.getCause() instanceof IOException) {
+        throw (IOException) e.getCause();
+      }
+      throw new S3Exception(e);
+    }    
+  }
 
   private void put(String key, InputStream in, long length) throws IOException {
     try {

src/java/org/apache/hadoop/fs/s3/S3FileSystem.java

@@ -174,15 +174,48 @@ public class S3FileSystem extends FileSystem {
 
   @Override
   public boolean renameRaw(Path src, Path dst) throws IOException {
-    // TODO: Check corner cases: dst already exists,
-    // or if path is directory with children
     Path absoluteSrc = makeAbsolute(src);
-    INode inode = store.getINode(absoluteSrc);
-    if (inode == null) {
-      throw new IOException("No such file.");
+    INode srcINode = store.getINode(absoluteSrc);
+    if (srcINode == null) {
+      // src path doesn't exist
+      return false; 
+    }
+    Path absoluteDst = makeAbsolute(dst);
+    INode dstINode = store.getINode(absoluteDst);
+    if (dstINode != null && dstINode.isDirectory()) {
+      absoluteDst = new Path(absoluteDst, absoluteSrc.getName());
+      dstINode = store.getINode(absoluteDst);
+    }
+    if (dstINode != null) {
+      // dst path already exists - can't overwrite
+      return false;
+    }
+    Path dstParent = absoluteDst.getParent();
+    if (dstParent != null) {
+      INode dstParentINode = store.getINode(dstParent);
+      if (dstParentINode == null || dstParentINode.isFile()) {
+        // dst parent doesn't exist or is a file
+        return false;
+      }
+    }
+    return renameRawRecursive(absoluteSrc, absoluteDst);
+  }
+  
+  private boolean renameRawRecursive(Path src, Path dst) throws IOException {
+    INode srcINode = store.getINode(src);
+    store.storeINode(dst, srcINode);
+    store.deleteINode(src);
+    if (srcINode.isDirectory()) {
+      for (Path oldSrc : store.listDeepSubPaths(src)) {
+        INode inode = store.getINode(oldSrc);
+        if (inode == null) {
+          return false;
+        }
+        Path newDst = new Path(oldSrc.toString().replaceFirst(src.toString(), dst.toString()));
+        store.storeINode(newDst, inode);
+        store.deleteINode(oldSrc);
+      }
     }
-    store.storeINode(makeAbsolute(dst), inode);
-    store.deleteINode(absoluteSrc);
     return true;
   }
 

src/test/org/apache/hadoop/fs/s3/InMemoryFileSystemStore.java

@@ -66,6 +66,21 @@ class InMemoryFileSystemStore implements FileSystemStore {
     return subPaths;
   }
 
+  public Set<Path> listDeepSubPaths(Path path) throws IOException {
+    String pathString = path.toUri().getPath();
+    if (!pathString.endsWith("/")) {
+      pathString += "/";
+    }
+    // This is inefficient but more than adequate for testing purposes.
+    Set<Path> subPaths = new LinkedHashSet<Path>();
+    for (Path p : inodes.tailMap(path).keySet()) {
+      if (p.toUri().getPath().startsWith(pathString)) {
+        subPaths.add(p);
+      }
+    }
+    return subPaths;
+  }
+
   public void storeINode(Path path, INode inode) throws IOException {
     inodes.put(path, inode);
   }

src/test/org/apache/hadoop/fs/s3/S3FileSystemBaseTest.java

@@ -228,32 +228,96 @@ public abstract class S3FileSystemBaseTest extends TestCase {
     assertFalse("file2 exists", s3FileSystem.exists(file2));
     
   }
+  
+  public void testRenameNonExistentPath() throws Exception {
+    Path src = new Path("/test/hadoop/path");
+    Path dst = new Path("/test/new/newpath");
+    rename(src, dst, false, false, false);
+  }
 
-  public void testRename() throws Exception {
-    int len = BLOCK_SIZE;
-    
-    Path path = new Path("/test/hadoop/file");
-    
-    s3FileSystem.mkdirs(path.getParent());
-
-    createEmptyFile(path);
+  public void testRenameFileMoveToNonExistentDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/file");
+    createEmptyFile(src);
+    Path dst = new Path("/test/new/newfile");
+    rename(src, dst, false, true, false);
+  }
 
-    assertTrue("Exists", s3FileSystem.exists(path));
+  public void testRenameFileMoveToExistingDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/file");
+    createEmptyFile(src);
+    Path dst = new Path("/test/new/newfile");
+    s3FileSystem.mkdirs(dst.getParent());
+    rename(src, dst, true, false, true);
+  }
 
-    Path newPath = new Path("/test/hadoop/newfile");
-    s3FileSystem.rename(path, newPath);
-    assertFalse("No longer exists", s3FileSystem.exists(path));
-    assertTrue("Moved", s3FileSystem.exists(newPath));
+  public void testRenameFileAsExistingFile() throws Exception {
+    Path src = new Path("/test/hadoop/file");
+    createEmptyFile(src);
+    Path dst = new Path("/test/new/newfile");
+    createEmptyFile(dst);
+    rename(src, dst, false, true, true);
+  }
 
-    FSInputStream in = s3FileSystem.openRaw(newPath);
-    byte[] buf = new byte[len];
+  public void testRenameFileAsExistingDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/file");
+    createEmptyFile(src);
+    Path dst = new Path("/test/new/newdir");
+    s3FileSystem.mkdirs(dst);
+    rename(src, dst, true, false, true);
+    assertTrue("Destination changed", s3FileSystem.exists(new Path("/test/new/newdir/file")));    
+  }
+  
+  public void testRenameDirectoryMoveToNonExistentDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/dir");
+    s3FileSystem.mkdirs(src);
+    Path dst = new Path("/test/new/newdir");
+    rename(src, dst, false, true, false);
+  }
+  
+  public void testRenameDirectoryMoveToExistingDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/dir");
+    s3FileSystem.mkdirs(src);
+    createEmptyFile(new Path("/test/hadoop/dir/file1"));
+    createEmptyFile(new Path("/test/hadoop/dir/subdir/file2"));
     
-    in.readFully(0, buf);
-
-    assertEquals(len, buf.length);
-    for (int i = 0; i < buf.length; i++) {
-      assertEquals("Position " + i, data[i], buf[i]);
-    }
+    Path dst = new Path("/test/new/newdir");
+    s3FileSystem.mkdirs(dst.getParent());
+    rename(src, dst, true, false, true);
+    
+    assertFalse("Nested file1 exists", s3FileSystem.exists(new Path("/test/hadoop/dir/file1")));
+    assertFalse("Nested file2 exists", s3FileSystem.exists(new Path("/test/hadoop/dir/subdir/file2")));
+    assertTrue("Renamed nested file1 exists", s3FileSystem.exists(new Path("/test/new/newdir/file1")));
+    assertTrue("Renamed nested exists", s3FileSystem.exists(new Path("/test/new/newdir/subdir/file2")));
+  }
+  
+  public void testRenameDirectoryAsExistingFile() throws Exception {
+    Path src = new Path("/test/hadoop/dir");
+    s3FileSystem.mkdirs(src);
+    Path dst = new Path("/test/new/newfile");
+    createEmptyFile(dst);
+    rename(src, dst, false, true, true);
+  }
+  
+  public void testRenameDirectoryAsExistingDirectory() throws Exception {
+    Path src = new Path("/test/hadoop/dir");
+    s3FileSystem.mkdirs(src);
+    createEmptyFile(new Path("/test/hadoop/dir/file1"));
+    createEmptyFile(new Path("/test/hadoop/dir/subdir/file2"));
+    
+    Path dst = new Path("/test/new/newdir");
+    s3FileSystem.mkdirs(dst);
+    rename(src, dst, true, false, true);
+    assertTrue("Destination changed", s3FileSystem.exists(new Path("/test/new/newdir/dir")));    
+    assertFalse("Nested file1 exists", s3FileSystem.exists(new Path("/test/hadoop/dir/file1")));
+    assertFalse("Nested file2 exists", s3FileSystem.exists(new Path("/test/hadoop/dir/subdir/file2")));
+    assertTrue("Renamed nested file1 exists", s3FileSystem.exists(new Path("/test/new/newdir/dir/file1")));
+    assertTrue("Renamed nested exists", s3FileSystem.exists(new Path("/test/new/newdir/dir/subdir/file2")));
+  }
+  
+  private void rename(Path src, Path dst, boolean renameSucceeded, boolean srcExists, boolean dstExists) throws IOException {
+    assertEquals("Rename result", renameSucceeded, s3FileSystem.rename(src, dst));
+    assertEquals("Source exists", srcExists, s3FileSystem.exists(src));
+    assertEquals("Destination exists", dstExists, s3FileSystem.exists(dst));
   }
 
   private void createEmptyFile(Path path) throws IOException {