Fix pseudo-directory renaming operation

Change-Id: I977603b60d9ae4949b8dbced5df8dd5d20843412
Closes-Bug: 1429055

hadoop-swiftfs/src/main/java/org/apache/hadoop/fs/swift/snative/SwiftFileStatus.java

@@ -31,6 +31,8 @@ public class SwiftFileStatus extends FileStatus {
 
   private SwiftObjectPath dloPrefix = null;
 
+  private boolean isPseudoDirFlag = false;
+
   public SwiftFileStatus() {
   }
 
@@ -63,6 +65,14 @@ public class SwiftFileStatus extends FileStatus {
             access_time, permission, owner, group, path);
   }
 
+  public static SwiftFileStatus createPseudoDirStatus(Path path) {
+    SwiftFileStatus status = new SwiftFileStatus(0, true, 1, 0,
+                                                 System.currentTimeMillis(),
+                                                 path);
+    status.isPseudoDirFlag = true;
+    return status;
+  }
+
   /**
    * A entry is a file if it is not a directory.
    * By implementing it <i>and not marking as an override</i> this
@@ -89,6 +99,10 @@ public class SwiftFileStatus extends FileStatus {
     return dloPrefix;
   }
 
+  public boolean isPseudoDir() {
+    return isPseudoDirFlag;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();

hadoop-swiftfs/src/main/java/org/apache/hadoop/fs/swift/snative/SwiftNativeFileSystemStore.java

@@ -223,17 +223,19 @@ public class SwiftNativeFileSystemStore {
     try {
       headers = stat(objectPath, newest);
     } catch (FileNotFoundException e) {
-      // if path is pseudo-directory, ignore FileNotFoundException.
+      try {
+        // retry stat for direcotry file
+        objectPath = toDirPath(path);
+        headers = stat(objectPath, newest);
+      } catch (FileNotFoundException ex) {
+        // if path is pseudo-directory, ignore FileNotFoundException.
+      }
     }
     //no headers is treated as a missing file or pseudo-directory
     if (headers == null || headers.length == 0) {
       if (existsPseudoDirectory(objectPath)) {
-        return new SwiftFileStatus(0,
-                                   true,
-                                   1,
-                                   getBlocksize(),
-                                   System.currentTimeMillis(),
-                                   getCorrectSwiftPath(statusPath));
+        Path pseudoDirPath = getCorrectSwiftPath(statusPath);
+        return SwiftFileStatus.createPseudoDirStatus(pseudoDirPath);
       }
       throw new FileNotFoundException("Not Found " + path.toUri());
     }
@@ -758,8 +760,7 @@ public class SwiftNativeFileSystemStore {
     //calculate the destination
     SwiftObjectPath destPath;
 
-    boolean srcIsFile = !srcMetadata.isDirectory();
-    if (srcIsFile) {
+    if (srcMetadata.isFile()) {
 
       //source is a simple file OR a partitioned file
       // outcomes:
@@ -870,7 +871,9 @@ public class SwiftNativeFileSystemStore {
       //now rename self. If missing, create the dest directory and warn
       if (!SwiftUtils.isRootDir(srcObject)) {
         try {
-          copyThenDeleteObject(srcObject, srcMetadata, targetObjectPath);
+          if (!srcMetadata.isPseudoDir()) {
+            copyThenDeleteObject(srcObject, srcMetadata, targetObjectPath);
+          }
         } catch (FileNotFoundException e) {
           //create the destination directory
           LOG.warn("Source directory deleted during rename", e);

hadoop-swiftfs/src/test/java/org/apache/hadoop/fs/swift/TestSwiftFileSystemRename.java

@@ -21,9 +21,12 @@ package org.apache.hadoop.fs.swift;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.swift.http.SwiftRestClient;
+import org.apache.hadoop.fs.swift.util.SwiftObjectPath;
 import org.apache.hadoop.fs.swift.util.SwiftTestUtils;
 import org.junit.Test;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 
 import static org.apache.hadoop.fs.swift.util.SwiftTestUtils.compareByteArrays;
@@ -267,4 +270,18 @@ public class TestSwiftFileSystemRename extends SwiftFileSystemBaseTest {
     rename(path, path2, false, false, false);
   }
 
+  @Test(timeout = SWIFT_TEST_TIMEOUT)
+  public void testRenamePseudoDir() throws Throwable {
+    assumeRenameSupported();
+
+    // create file directory (don't create directory file)
+    SwiftRestClient client;
+    client = SwiftRestClient.getInstance(fs.getUri(), fs.getConf());
+    SwiftObjectPath path = SwiftObjectPath.fromPath(fs.getUri(), new Path("/test/olddir/file"));
+    client.upload(path, new ByteArrayInputStream(new byte[0]), 0);
+
+    rename(path("/test/olddir"), path("/test/newdir"), true, false, true);
+    SwiftTestUtils.assertIsDirectory(fs, path("/test/newdir"));
+    assertIsFile(path("/test/newdir/file"));
+  }
 }