HADOOP-12161. Add getStoragePolicy API to the FileSystem interface. (Contributed by Brahma Reddy Battula)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -716,6 +716,9 @@ Release 2.8.0 - UNRELEASED
     HADOOP-12184. Remove unused Linux-specific constants in NativeIO (Martin
     Walsh via Colin P. McCabe)
 
+    HADOOP-12161. Add getStoragePolicy API to the FileSystem interface.
+    (Brahma Reddy Battula via Arpit Agarwal)
+
   OPTIMIZATIONS
 
     HADOOP-11785. Reduce the number of listStatus operation in distcp

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/AbstractFileSystem.java

@@ -1236,6 +1236,19 @@ public abstract class AbstractFileSystem {
         + " doesn't support setStoragePolicy");
   }
 
+  /**
+   * Retrieve the storage policy for a given file or directory.
+   *
+   * @param src file or directory path.
+   * @return storage policy for give file.
+   * @throws IOException
+   */
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    throw new UnsupportedOperationException(getClass().getSimpleName()
+        + " doesn't support getStoragePolicy");
+  }
+
   /**
    * Retrieve all the storage policies supported by this file system.
    *

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileContext.java

@@ -49,6 +49,7 @@ import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
+
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.ipc.RpcClientException;
 import org.apache.hadoop.ipc.RpcServerException;
@@ -2691,6 +2692,25 @@ public class FileContext {
     }.resolve(this, absF);
   }
 
+  /**
+   * Query the effective storage policy ID for the given file or directory.
+   *
+   * @param src file or directory path.
+   * @return storage policy for give file.
+   * @throws IOException
+   */
+  public BlockStoragePolicySpi getStoragePolicy(Path path) throws IOException {
+    final Path absF = fixRelativePart(path);
+    return new FSLinkResolver<BlockStoragePolicySpi>() {
+      @Override
+      public BlockStoragePolicySpi next(final AbstractFileSystem fs,
+          final Path p)
+          throws IOException {
+        return fs.getStoragePolicy(p);
+      }
+    }.resolve(this, absF);
+  }
+
   /**
    * Retrieve all the storage policies supported by this file system.
    *

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -2628,6 +2628,19 @@ public abstract class FileSystem extends Configured implements Closeable {
         + " doesn't support setStoragePolicy");
   }
 
+  /**
+   * Query the effective storage policy ID for the given file or directory.
+   *
+   * @param src file or directory path.
+   * @return storage policy for give file.
+   * @throws IOException
+   */
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    throw new UnsupportedOperationException(getClass().getSimpleName()
+        + " doesn't support getStoragePolicy");
+  }
+
   /**
    * Retrieve all the storage policies supported by this file system.
    *

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java

@@ -627,6 +627,12 @@ public class FilterFileSystem extends FileSystem {
     fs.setStoragePolicy(src, policyName);
   }
 
+  @Override
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    return fs.getStoragePolicy(src);
+  }
+
   @Override
   public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
       throws IOException {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFs.java

@@ -405,6 +405,12 @@ public abstract class FilterFs extends AbstractFileSystem {
     myFs.setStoragePolicy(path, policyName);
   }
 
+  @Override
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    return myFs.getStoragePolicy(src);
+  }
+
   @Override
   public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
       throws IOException {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ChRootedFs.java

@@ -385,6 +385,12 @@ class ChRootedFs extends AbstractFileSystem {
     myFs.setStoragePolicy(fullPath(path), policyName);
   }
 
+  @Override
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    return myFs.getStoragePolicy(src);
+  }
+
   @Override
   public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
       throws IOException {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFs.java

@@ -34,6 +34,7 @@ import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.AbstractFileSystem;
 import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.BlockStoragePolicySpi;
 import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -748,6 +749,20 @@ public class ViewFs extends AbstractFileSystem {
     res.targetFileSystem.setStoragePolicy(res.remainingPath, policyName);
   }
 
+  /**
+   * Retrieve the storage policy for a given file or directory.
+   *
+   * @param src file or directory path.
+   * @return storage policy for give file.
+   * @throws IOException
+   */
+  public BlockStoragePolicySpi getStoragePolicy(final Path src)
+      throws IOException {
+    InodeTree.ResolveResult<AbstractFileSystem> res =
+        fsState.resolve(getUriPath(src), true);
+    return res.targetFileSystem.getStoragePolicy(res.remainingPath);
+  }
+
   /*
    * An instance of this class represents an internal dir of the viewFs 
    * ie internal dir of the mount table.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java

@@ -210,6 +210,9 @@ public class TestHarFileSystem {
     public void setStoragePolicy(Path src, String policyName)
         throws IOException;
 
+    public BlockStoragePolicySpi getStoragePolicy(final Path src)
+        throws IOException;
+
     public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
         throws IOException;
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/fs/Hdfs.java

@@ -472,6 +472,11 @@ public class Hdfs extends AbstractFileSystem {
     dfs.setStoragePolicy(getUriPath(path), policyName);
   }
 
+  @Override
+  public BlockStoragePolicySpi getStoragePolicy(Path src) throws IOException {
+    return dfs.getStoragePolicy(getUriPath(src));
+  }
+
   @Override
   public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
       throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -1573,6 +1573,24 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
     }
   }
 
+  /**
+   * @return Get the storage policy for specified path
+   */
+  public BlockStoragePolicy getStoragePolicy(String path) throws IOException {
+    HdfsFileStatus status = getFileInfo(path);
+    if (status == null) {
+      throw new FileNotFoundException("File does not exist: " + path);
+    }
+    byte storagePolicyId = status.getStoragePolicy();
+    BlockStoragePolicy[] policies = getStoragePolicies();
+    for (BlockStoragePolicy policy : policies) {
+      if (policy.getId() == storagePolicyId) {
+        return policy;
+      }
+    }
+    return null;
+  }
+
   /**
    * @return All the existing storage policies
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -556,6 +556,25 @@ public class DistributedFileSystem extends FileSystem {
     }.resolve(this, absF);
   }
 
+  @Override
+  public BlockStoragePolicySpi getStoragePolicy(Path path) throws IOException {
+    statistics.incrementReadOps(1);
+    Path absF = fixRelativePart(path);
+
+    return new FileSystemLinkResolver<BlockStoragePolicySpi>() {
+      @Override
+      public BlockStoragePolicySpi doCall(final Path p) throws IOException {
+        return getClient().getStoragePolicy(getPathName(p));
+      }
+
+      @Override
+      public BlockStoragePolicySpi next(final FileSystem fs, final Path p)
+          throws IOException, UnresolvedLinkException {
+        return fs.getStoragePolicy(p);
+      }
+    }.resolve(this, absF);
+  }
+
   @Override
   public Collection<BlockStoragePolicy> getAllStoragePolicies()
       throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java

@@ -871,9 +871,25 @@ public class TestBlockStoragePolicy {
         GenericTestUtils.assertExceptionContains(invalidPath.toString(), e);
       }
 
+      try {
+        fs.getStoragePolicy(invalidPath);
+        Assert.fail("Should throw a FileNotFoundException");
+      } catch (FileNotFoundException e) {
+        GenericTestUtils.assertExceptionContains(invalidPath.toString(), e);
+      }
+
       fs.setStoragePolicy(fooFile, HdfsServerConstants.COLD_STORAGE_POLICY_NAME);
       fs.setStoragePolicy(barDir, HdfsServerConstants.WARM_STORAGE_POLICY_NAME);
       fs.setStoragePolicy(barFile2, HdfsServerConstants.HOT_STORAGE_POLICY_NAME);
+      Assert.assertEquals("File storage policy should be COLD",
+          HdfsServerConstants.COLD_STORAGE_POLICY_NAME,
+          fs.getStoragePolicy(fooFile).getName());
+      Assert.assertEquals("File storage policy should be WARM",
+          HdfsServerConstants.WARM_STORAGE_POLICY_NAME,
+          fs.getStoragePolicy(barDir).getName());
+      Assert.assertEquals("File storage policy should be HOT",
+          HdfsServerConstants.HOT_STORAGE_POLICY_NAME,
+          fs.getStoragePolicy(barFile2).getName());
 
       dirList = fs.getClient().listPaths(dir.toString(),
           HdfsFileStatus.EMPTY_NAME).getPartialListing();
@@ -1306,4 +1322,5 @@ public class TestBlockStoragePolicy {
       Assert.assertEquals(StorageType.ARCHIVE, i.next().getKey());
     }
   }
+
 }