diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ConfigUtil.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ConfigUtil.java
index 4c3dae9a9f99b..c5cb6b772abcd 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ConfigUtil.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ConfigUtil.java
@@ -161,6 +161,41 @@ public static void addLinkNfly(final Configuration conf, final String src,
         targets);
   }
 
+
+  /**
+   * Add a LinkRegex to the config for the specified mount table.
+   * @param conf
+   * @param mountTableName
+   * @param srcRegex
+   * @param targetStr
+   */
+  public static void addLinkRegex(
+      Configuration conf, final String mountTableName,
+      final String srcRegex, final String targetStr) {
+    addLinkRegex(conf, mountTableName, srcRegex, targetStr, null);
+  }
+
+  /**
+   * Add a LinkRegex to the config for the specified mount table.
+   * @param conf
+   * @param mountTableName
+   * @param srcRegex
+   * @param targetStr
+   * @param interceptorSettings
+   */
+  public static void addLinkRegex(
+      Configuration conf, final String mountTableName, final String srcRegex,
+      final String targetStr, final String interceptorSettings) {
+    String prefix = getConfigViewFsPrefix(mountTableName) + "."
+        + Constants.CONFIG_VIEWFS_LINK_REGEX + ".";
+    if ((interceptorSettings != null) && (!interceptorSettings.isEmpty())) {
+      prefix = prefix + interceptorSettings
+          + RegexMountPoint.SETTING_SRCREGEX_SEP;
+    }
+    String key = prefix + srcRegex;
+    conf.set(key, targetStr);
+  }
+
   /**
    * Add config variable for homedir for default mount table
    * @param conf - add to this conf
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
index aa1bc7e63771a..ff651dceadfa8 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
@@ -75,7 +75,23 @@ public interface Constants {
    */
   String CONFIG_VIEWFS_LINK_MERGE_SLASH = "linkMergeSlash";
 
+  /**
+   * Config variable for specifying a regex link which uses regular expressions
+   * as source and target could use group captured in src.
+   * E.g. (^/(?<firstDir>\\w+), /prefix-${firstDir}) =>
+   *   (/path1/file1 => /prefix-path1/file1)
+   */
+  String CONFIG_VIEWFS_LINK_REGEX = "linkRegex";
+
   FsPermission PERMISSION_555 = new FsPermission((short) 0555);
 
   String CONFIG_VIEWFS_RENAME_STRATEGY = "fs.viewfs.rename.strategy";
+
+  /**
+   * Config capacity of mount point resolution cache.
+   * Value <= 0 means disable cache.
+   */
+  String CONFIG_VIEWFS_PATH_RESOLUTION_CACHE_CAPACITY
+      = "fs.viewfs.path.resolution.cache.capacity";
+  int CONFIG_VIEWFS_PATH_RESOLUTION_CACHE_CAPACITY_DEFAULT = 0;
 }
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
index 69923438ecc20..7245023a46340 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
@@ -17,7 +17,9 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
@@ -29,7 +31,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import org.apache.commons.collections.map.LRUMap;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -39,25 +43,31 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * InodeTree implements a mount-table as a tree of inodes.
  * It is used to implement ViewFs and ViewFileSystem.
  * In order to use it the caller must subclass it and implement
  * the abstract methods {@link #getTargetFileSystem(INodeDir)}, etc.
- *
- * The mountable is initialized from the config variables as 
+ * <p>
+ * The mountable is initialized from the config variables as
  * specified in {@link ViewFs}
  *
  * @param <T> is AbstractFileSystem or FileSystem
- *
- * The two main methods are
- * {@link #InodeTree(Configuration, String)} // constructor
- * {@link #resolve(String, boolean)} 
+ *            <p>
+ *            The two main methods are
+ *            {@link #InodeTree(Configuration, String)} // constructor
+ *            {@link #resolve(String, boolean)}
  */
 
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 abstract class InodeTree<T> {
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(InodeTree.class.getName());
+
   enum ResultKind {
     INTERNAL_DIR,
     EXTERNAL_DIR
@@ -71,6 +81,11 @@ enum ResultKind {
   // the homedir for this mount table
   private final String homedirPrefix;
   private List<MountPoint<T>> mountPoints = new ArrayList<MountPoint<T>>();
+  private List<RegexMountPoint<T>> regexMountPointList =
+      new ArrayList<RegexMountPoint<T>>();
+  private LRUMap pathResolutionCache;
+  private ReentrantReadWriteLock cacheRWLock;
+  private int pathResolutionCacheCapacity;
 
   static class MountPoint<T> {
     String src;
@@ -84,6 +99,7 @@ static class MountPoint<T> {
 
   /**
    * Breaks file path into component names.
+   *
    * @param path
    * @return array of names component names
    */
@@ -93,6 +109,7 @@ static String[] breakIntoPathComponents(final String path) {
 
   /**
    * Internal class for INode tree.
+   *
    * @param <T>
    */
   abstract static class INode<T> {
@@ -117,11 +134,12 @@ boolean isLink() {
 
   /**
    * Internal class to represent an internal dir of the mount table.
+   *
    * @param <T>
    */
   static class INodeDir<T> extends INode<T> {
     private final Map<String, INode<T>> children = new HashMap<>();
-    private T internalDirFs =  null; //filesystem of this internal directory
+    private T internalDirFs = null; //filesystem of this internal directory
     private boolean isRoot = false;
 
     INodeDir(final String pathToNode, final UserGroupInformation aUgi) {
@@ -213,18 +231,25 @@ enum LinkType {
      * Config prefix: fs.viewfs.mounttable.<mnt_tbl_name>.linkNfly
      * Refer: {@link Constants#CONFIG_VIEWFS_LINK_NFLY}
      */
-    NFLY;
+    NFLY,
+    /**
+     * Link entry which source are regex exrepssions and target refer matched
+     * group from source
+     * Config prefix: fs.viewfs.mounttable.<mnt_tbl_name>.linkMerge
+     * Refer: {@link Constants#CONFIG_VIEWFS_LINK_REGEX}
+     */
+    REGEX;
   }
 
   /**
    * An internal class to represent a mount link.
    * A mount link can be single dir link or a merge dir link.
-
+   * <p>
    * A merge dir link is  a merge (junction) of links to dirs:
    * example : merge of 2 dirs
-   *     /users -> hdfs:nn1//users
-   *     /users -> hdfs:nn2//users
-   *
+   * /users -> hdfs:nn1//users
+   * /users -> hdfs:nn2//users
+   * <p>
    * For a merge, each target is checked to be dir when created but if target
    * is changed later it is then ignored (a dir with null entries)
    */
@@ -315,20 +340,20 @@ private void createLink(final String src, final String target,
 
     // Now process the last component
     // Add the link in 2 cases: does not exist or a link exists
-    String iPath = srcPaths[i];// last component
+    String iPath = srcPaths[i]; // last component
     if (curInode.resolveInternal(iPath) != null) {
       //  directory/link already exists
       StringBuilder strB = new StringBuilder(srcPaths[0]);
       for (int j = 1; j <= i; ++j) {
         strB.append('/').append(srcPaths[j]);
       }
-      throw new FileAlreadyExistsException("Path " + strB +
-          " already exists as dir; cannot create link here");
+      throw new FileAlreadyExistsException(
+          "Path " + strB + " already exists as dir; cannot create link here");
     }
 
     final INodeLink<T> newLink;
-    final String fullPath = curInode.fullPath + (curInode == root ? "" : "/")
-        + iPath;
+    final String fullPath =
+        curInode.fullPath + (curInode == root ? "" : "/") + iPath;
     switch (linkType) {
     case SINGLE:
       newLink = new INodeLink<T>(fullPath, aUgi,
@@ -341,10 +366,10 @@ private void createLink(final String src, final String target,
       throw new IllegalArgumentException("Unexpected linkType: " + linkType);
     case MERGE:
     case NFLY:
-      final URI[] targetUris = StringUtils.stringToURI(
-          StringUtils.getStrings(target));
+      final URI[] targetUris =
+          StringUtils.stringToURI(StringUtils.getStrings(target));
       newLink = new INodeLink<T>(fullPath, aUgi,
-            getTargetFileSystem(settings, targetUris), targetUris);
+          getTargetFileSystem(settings, targetUris), targetUris);
       break;
     default:
       throw new IllegalArgumentException(linkType + ": Infeasible linkType");
@@ -356,6 +381,7 @@ private void createLink(final String src, final String target,
   /**
    * The user of this class must subclass and implement the following
    * 3 abstract methods.
+   *
    * @throws IOException
    */
   protected abstract T getTargetFileSystem(URI uri)
@@ -389,6 +415,7 @@ private INodeLink<T> getRootFallbackLink() {
   /**
    * An internal class representing the ViewFileSystem mount table
    * link entries and their attributes.
+   *
    * @see LinkType
    */
   private static class LinkEntry {
@@ -439,9 +466,10 @@ Configuration getConfig() {
   }
 
   /**
-   * Create Inode Tree from the specified mount-table specified in Config
-   * @param config - the mount table keys are prefixed with 
-   *       FsConstants.CONFIG_VIEWFS_PREFIX
+   * Create Inode Tree from the specified mount-table specified in Config.
+   *
+   * @param config   - the mount table keys are prefixed with
+   *                 FsConstants.CONFIG_VIEWFS_PREFIX
    * @param viewName - the name of the mount table - if null use defaultMT name
    * @throws UnsupportedFileSystemException
    * @throws URISyntaxException
@@ -463,94 +491,116 @@ protected InodeTree(final Configuration config, final String viewName)
 
     final String mountTablePrefix =
         Constants.CONFIG_VIEWFS_PREFIX + "." + mountTableName + ".";
-    final String linkPrefix = Constants.CONFIG_VIEWFS_LINK + ".";
-    final String linkFallbackPrefix = Constants.CONFIG_VIEWFS_LINK_FALLBACK;
-    final String linkMergePrefix = Constants.CONFIG_VIEWFS_LINK_MERGE + ".";
-    final String linkMergeSlashPrefix =
-        Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH;
     boolean gotMountTableEntry = false;
     final UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
     for (Entry<String, String> si : config) {
       final String key = si.getKey();
-      if (key.startsWith(mountTablePrefix)) {
-        gotMountTableEntry = true;
-        LinkType linkType;
-        String src = key.substring(mountTablePrefix.length());
-        String settings = null;
-        if (src.startsWith(linkPrefix)) {
-          src = src.substring(linkPrefix.length());
-          if (src.equals(SlashPath.toString())) {
-            throw new UnsupportedFileSystemException("Unexpected mount table "
-                + "link entry '" + key + "'. Use "
-                + Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH  + " instead!");
-          }
-          linkType = LinkType.SINGLE;
-        } else if (src.startsWith(linkFallbackPrefix)) {
-          if (src.length() != linkFallbackPrefix.length()) {
-            throw new IOException("ViewFs: Mount points initialization error." +
-                " Invalid " + Constants.CONFIG_VIEWFS_LINK_FALLBACK +
-                " entry in config: " + src);
-          }
-          linkType = LinkType.SINGLE_FALLBACK;
-        } else if (src.startsWith(linkMergePrefix)) { // A merge link
-          src = src.substring(linkMergePrefix.length());
-          linkType = LinkType.MERGE;
-        } else if (src.startsWith(linkMergeSlashPrefix)) {
-          // This is a LinkMergeSlash entry. This entry should
-          // not have any additional source path.
-          if (src.length() != linkMergeSlashPrefix.length()) {
-            throw new IOException("ViewFs: Mount points initialization error." +
-                " Invalid " + Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH +
-                " entry in config: " + src);
-          }
-          linkType = LinkType.MERGE_SLASH;
-        } else if (src.startsWith(Constants.CONFIG_VIEWFS_LINK_NFLY)) {
-          // prefix.settings.src
-          src = src.substring(Constants.CONFIG_VIEWFS_LINK_NFLY.length() + 1);
-          // settings.src
-          settings = src.substring(0, src.indexOf('.'));
-          // settings
-
-          // settings.src
-          src = src.substring(settings.length() + 1);
-          // src
-
-          linkType = LinkType.NFLY;
-        } else if (src.startsWith(Constants.CONFIG_VIEWFS_HOMEDIR)) {
-          // ignore - we set home dir from config
-          continue;
+      if (!key.startsWith(mountTablePrefix)) {
+        continue;
+      }
+      gotMountTableEntry = true;
+      String src = key.substring(mountTablePrefix.length());
+      if (src.startsWith(Constants.CONFIG_VIEWFS_HOMEDIR)) {
+        // ignore - we set home dir from config
+        continue;
+      }
+      LinkType linkType = checkAndParseLinkType(src);
+      final String target = si.getValue();
+      String linkKeyPath = null;
+      String settings = null;
+      switch (linkType) {
+      case SINGLE:
+        final String linkPrefix = Constants.CONFIG_VIEWFS_LINK + ".";
+        linkKeyPath = src.substring(linkPrefix.length());
+        linkEntries.add(
+            new LinkEntry(linkKeyPath, target, linkType, settings, ugi,
+                config));
+        break;
+      case SINGLE_FALLBACK:
+        final String linkFallbackPrefix = Constants.CONFIG_VIEWFS_LINK_FALLBACK;
+        linkKeyPath = src.substring(linkFallbackPrefix.length());
+        linkEntries.add(
+            new LinkEntry(linkKeyPath, target, linkType, settings, ugi,
+                config));
+        break;
+      case MERGE:
+        final String linkMergePrefix = Constants.CONFIG_VIEWFS_LINK_MERGE + ".";
+        linkKeyPath = src.substring(linkMergePrefix.length());
+        linkEntries.add(
+            new LinkEntry(linkKeyPath, target, linkType, settings, ugi,
+                config));
+        break;
+      case MERGE_SLASH:
+        if (isMergeSlashConfigured) {
+          throw new IOException("Mount table " + mountTableName
+              + " has already been configured with a merge slash link. "
+              + "Multiple merge slash links for the same mount table is "
+              + "not allowed.");
+        }
+        isMergeSlashConfigured = true;
+        mergeSlashTarget = target;
+        break;
+      case NFLY:
+        final String linkNflyPrefix = Constants.CONFIG_VIEWFS_LINK_NFLY + ".";
+        String settingAndKeyPathStr = src.substring(linkNflyPrefix.length());
+        // settings.src
+        settings = settingAndKeyPathStr
+            .substring(0, settingAndKeyPathStr.indexOf('.'));
+        // settings
+        // settings.src
+        linkKeyPath = settingAndKeyPathStr.substring(settings.length() + 1);
+        linkEntries.add(
+            new LinkEntry(linkKeyPath, target, linkType, settings, ugi,
+                config));
+        break;
+      case REGEX:
+        final String linkRegexPrefix = Constants.CONFIG_VIEWFS_LINK_REGEX + ".";
+        // settings#.linkKey
+        String settingsAndLinkKeyPath = src.substring(linkRegexPrefix.length());
+        int settingLinkKeySepIndex = settingsAndLinkKeyPath
+            .indexOf(RegexMountPoint.SETTING_SRCREGEX_SEP);
+        if (settingLinkKeySepIndex == -1) {
+          // There's no settings
+          linkKeyPath = settingsAndLinkKeyPath;
+          settings = null;
         } else {
-          throw new IOException("ViewFs: Cannot initialize: Invalid entry in " +
-              "Mount table in config: " + src);
+          // settings#.linkKey style configuration
+          // settings from settings#.linkKey
+          settings =
+              settingsAndLinkKeyPath.substring(0, settingLinkKeySepIndex);
+          // linkKeyPath
+          linkKeyPath = settingsAndLinkKeyPath.substring(
+              settings.length() + RegexMountPoint.SETTING_SRCREGEX_SEP
+                  .length());
         }
-
-        final String target = si.getValue();
+        linkEntries.add(
+            new LinkEntry(linkKeyPath, target, linkType, settings, ugi,
+                config));
+        break;
+      default:
+        throw new IOException("ViewFs: Cannot initialize: Invalid entry in "
+            + "Mount table in config: " + src);
+      }
+      if (isMergeSlashConfigured) {
         if (linkType != LinkType.MERGE_SLASH) {
-          if (isMergeSlashConfigured) {
-            throw new IOException("Mount table " + mountTableName
-                + " has already been configured with a merge slash link. "
-                + "A regular link should not be added.");
-          }
-          linkEntries.add(
-              new LinkEntry(src, target, linkType, settings, ugi, config));
-        } else {
-          if (!linkEntries.isEmpty()) {
-            throw new IOException("Mount table " + mountTableName
-                + " has already been configured with regular links. "
-                + "A merge slash link should not be configured.");
-          }
-          if (isMergeSlashConfigured) {
-            throw new IOException("Mount table " + mountTableName
-                + " has already been configured with a merge slash link. "
-                + "Multiple merge slash links for the same mount table is "
-                + "not allowed.");
-          }
-          isMergeSlashConfigured = true;
-          mergeSlashTarget = target;
+          throw new IOException("Mount table " + mountTableName
+              + " has already been configured with a merge slash link. "
+              + "A regular link should not be added.");
+        }
+        if (!linkEntries.isEmpty()) {
+          throw new IOException("Mount table " + mountTableName
+              + " has already been configured with regular links. "
+              + "A merge slash link should not be configured.");
         }
       }
     }
 
+    if (!gotMountTableEntry) {
+      throw new IOException(
+          "ViewFs: Cannot initialize: Empty Mount table in config for "
+              + "viewfs://" + mountTableName + "/");
+    }
+
     if (isMergeSlashConfigured) {
       Preconditions.checkNotNull(mergeSlashTarget);
       root = new INodeLink<T>(mountTableName, ugi,
@@ -558,34 +608,101 @@ protected InodeTree(final Configuration config, final String viewName)
           new URI(mergeSlashTarget));
       mountPoints.add(new MountPoint<T>("/", (INodeLink<T>) root));
       rootFallbackLink = null;
-    } else {
-      root = new INodeDir<T>("/", UserGroupInformation.getCurrentUser());
-      getRootDir().setInternalDirFs(getTargetFileSystem(getRootDir()));
-      getRootDir().setRoot(true);
-      INodeLink<T> fallbackLink = null;
-      for (LinkEntry le : linkEntries) {
-        if (le.isLinkType(LinkType.SINGLE_FALLBACK)) {
-          if (fallbackLink != null) {
-            throw new IOException("Mount table " + mountTableName
-                + " has already been configured with a link fallback. "
-                + "Multiple fallback links for the same mount table is "
-                + "not allowed.");
-          }
-          fallbackLink = new INodeLink<T>(mountTableName, ugi,
-              getTargetFileSystem(new URI(le.getTarget())),
-              new URI(le.getTarget()));
-        } else {
-          createLink(le.getSrc(), le.getTarget(), le.getLinkType(),
-              le.getSettings(), le.getUgi(), le.getConfig());
+      return;
+    }
+
+    root = new INodeDir<T>("/", UserGroupInformation.getCurrentUser());
+    getRootDir().setInternalDirFs(getTargetFileSystem(getRootDir()));
+    getRootDir().setRoot(true);
+    INodeLink<T> fallbackLink = null;
+    for (LinkEntry le : linkEntries) {
+      switch (le.getLinkType()) {
+      case SINGLE_FALLBACK:
+        if (fallbackLink != null) {
+          throw new IOException("Mount table " + mountTableName
+              + " has already been configured with a link fallback. "
+              + "Multiple fallback links for the same mount table is "
+              + "not allowed.");
         }
+        fallbackLink = new INodeLink<T>(mountTableName, ugi,
+            getTargetFileSystem(new URI(le.getTarget())),
+            new URI(le.getTarget()));
+        break;
+      case REGEX:
+        LOGGER.info("Add regex mount point:" + le.getSrc() + ", target:" + le
+            .getTarget() + ", interceptor settings:" + le.getSettings());
+        RegexMountPoint regexMountPoint =
+            new RegexMountPoint<T>(this, le.getSrc(), le.getTarget(),
+                le.getSettings());
+        regexMountPoint.initialize();
+        regexMountPointList.add(regexMountPoint);
+        break;
+      default:
+        createLink(le.getSrc(), le.getTarget(), le.getLinkType(),
+            le.getSettings(), le.getUgi(), le.getConfig());
       }
-      rootFallbackLink = fallbackLink;
     }
+    rootFallbackLink = fallbackLink;
 
-    if (!gotMountTableEntry) {
-      throw new IOException(
-          "ViewFs: Cannot initialize: Empty Mount table in config for " +
-              "viewfs://" + mountTableName + "/");
+    pathResolutionCacheCapacity = config
+        .getInt(Constants.CONFIG_VIEWFS_PATH_RESOLUTION_CACHE_CAPACITY,
+            Constants.CONFIG_VIEWFS_PATH_RESOLUTION_CACHE_CAPACITY_DEFAULT);
+    if (pathResolutionCacheCapacity > 0) {
+      pathResolutionCache = new LRUMap(pathResolutionCacheCapacity);
+      cacheRWLock = new ReentrantReadWriteLock();
+    }
+  }
+
+  /**
+   * Get link type of mount point.
+   *
+   * @param src
+   * @return link type of current mount point
+   * @throws UnsupportedFileSystemException
+   * @throws IOException
+   */
+  protected LinkType checkAndParseLinkType(String src)
+      throws UnsupportedFileSystemException, IOException {
+    String linkStr = src;
+    int dotIndex = src.indexOf(".");
+    if (dotIndex != -1) {
+      linkStr = linkStr.substring(0, dotIndex);
+    }
+    switch (linkStr) {
+    case Constants.CONFIG_VIEWFS_LINK:
+      String linkPrefixStr = Constants.CONFIG_VIEWFS_LINK + ".";
+      String linkKey = src.substring(linkPrefixStr.length());
+      if (linkKey.equals(SlashPath.toString())) {
+        throw new UnsupportedFileSystemException(
+            "Unexpected mount table " + "link entry '" + src + "'. Use "
+                + Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH + " instead!");
+      }
+      return LinkType.SINGLE;
+    case Constants.CONFIG_VIEWFS_LINK_FALLBACK:
+      if (src.length() != Constants.CONFIG_VIEWFS_LINK_FALLBACK.length()) {
+        throw new IOException(
+            "ViewFs: Mount points initialization error." + " Invalid "
+                + Constants.CONFIG_VIEWFS_LINK_FALLBACK + " entry in config: "
+                + src);
+      }
+      return LinkType.SINGLE_FALLBACK;
+    case Constants.CONFIG_VIEWFS_LINK_MERGE:
+      return LinkType.MERGE;
+    case Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH:
+      if (src.length() != Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH.length()) {
+        throw new IOException(
+            "ViewFs: Mount points initialization error." + " Invalid "
+                + Constants.CONFIG_VIEWFS_LINK_MERGE_SLASH
+                + " entry in config: " + src);
+      }
+      return LinkType.MERGE_SLASH;
+    case Constants.CONFIG_VIEWFS_LINK_NFLY:
+      return LinkType.NFLY;
+    case Constants.CONFIG_VIEWFS_LINK_REGEX:
+      return LinkType.REGEX;
+    default:
+      throw new IOException("ViewFs: Cannot initialize: Invalid entry in "
+          + "Mount table in config: " + src);
     }
   }
 
@@ -593,7 +710,7 @@ protected InodeTree(final Configuration config, final String viewName)
    * Resolve returns ResolveResult.
    * The caller can continue the resolution of the remainingPath
    * in the targetFileSystem.
-   *
+   * <p>
    * If the input pathname leads to link to another file system then
    * the targetFileSystem is the one denoted by the link (except it is
    * file system chrooted to link target.
@@ -621,7 +738,7 @@ boolean isInternalDir() {
   }
 
   /**
-   * Resolve the pathname p relative to root InodeDir
+   * Resolve the pathname p relative to root InodeDir.
    * @param p - input path
    * @param resolveLastComponent
    * @return ResolveResult which allows further resolution of the remaining path
@@ -629,94 +746,200 @@ boolean isInternalDir() {
    */
   ResolveResult<T> resolve(final String p, final boolean resolveLastComponent)
       throws FileNotFoundException {
-    String[] path = breakIntoPathComponents(p);
-    if (path.length <= 1) { // special case for when path is "/"
-      T targetFs = root.isInternalDir() ?
-          getRootDir().getInternalDirFs() : getRootLink().getTargetFileSystem();
-      ResolveResult<T> res = new ResolveResult<T>(ResultKind.INTERNAL_DIR,
-          targetFs, root.fullPath, SlashPath);
-      return res;
-    }
+    ResolveResult<T> resolveResult = null;
+    resolveResult = getResolveResultFromCache(p, resolveLastComponent);
+    if (resolveResult != null) {
+      return resolveResult;
+    }
+
+    try {
+      String[] path = breakIntoPathComponents(p);
+      if (path.length <= 1) { // special case for when path is "/"
+        T targetFs = root.isInternalDir() ?
+            getRootDir().getInternalDirFs() :
+            getRootLink().getTargetFileSystem();
+        resolveResult = new ResolveResult<T>(ResultKind.INTERNAL_DIR, targetFs,
+            root.fullPath, SlashPath);
+        return resolveResult;
+      }
 
-    /**
-     * linkMergeSlash has been configured. The root of this mount table has
-     * been linked to the root directory of a file system.
-     * The first non-slash path component should be name of the mount table.
-     */
-    if (root.isLink()) {
-      Path remainingPath;
-      StringBuilder remainingPathStr = new StringBuilder();
-      // ignore first slash
-      for (int i = 1; i < path.length; i++) {
-        remainingPathStr.append("/").append(path[i]);
+      /**
+       * linkMergeSlash has been configured. The root of this mount table has
+       * been linked to the root directory of a file system.
+       * The first non-slash path component should be name of the mount table.
+       */
+      if (root.isLink()) {
+        Path remainingPath;
+        StringBuilder remainingPathStr = new StringBuilder();
+        // ignore first slash
+        for (int i = 1; i < path.length; i++) {
+          remainingPathStr.append("/").append(path[i]);
+        }
+        remainingPath = new Path(remainingPathStr.toString());
+        resolveResult = new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
+            getRootLink().getTargetFileSystem(), root.fullPath, remainingPath);
+        return resolveResult;
       }
-      remainingPath = new Path(remainingPathStr.toString());
-      ResolveResult<T> res = new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
-          getRootLink().getTargetFileSystem(), root.fullPath, remainingPath);
-      return res;
-    }
-    Preconditions.checkState(root.isInternalDir());
-    INodeDir<T> curInode = getRootDir();
+      Preconditions.checkState(root.isInternalDir());
+      INodeDir<T> curInode = getRootDir();
 
-    int i;
-    // ignore first slash
-    for (i = 1; i < path.length - (resolveLastComponent ? 0 : 1); i++) {
-      INode<T> nextInode = curInode.resolveInternal(path[i]);
-      if (nextInode == null) {
-        if (hasFallbackLink()) {
-          return new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
-              getRootFallbackLink().getTargetFileSystem(),
-              root.fullPath, new Path(p));
-        } else {
-          StringBuilder failedAt = new StringBuilder(path[0]);
-          for (int j = 1; j <= i; ++j) {
-            failedAt.append('/').append(path[j]);
+      // Try to resolve path in the regex mount point
+      resolveResult = tryResolveInRegexMountpoint(p, resolveLastComponent);
+      if (resolveResult != null) {
+        return resolveResult;
+      }
+
+      int i;
+      // ignore first slash
+      for (i = 1; i < path.length - (resolveLastComponent ? 0 : 1); i++) {
+        INode<T> nextInode = curInode.resolveInternal(path[i]);
+        if (nextInode == null) {
+          if (hasFallbackLink()) {
+            resolveResult = new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
+                getRootFallbackLink().getTargetFileSystem(), root.fullPath,
+                new Path(p));
+            return resolveResult;
+          } else {
+            StringBuilder failedAt = new StringBuilder(path[0]);
+            for (int j = 1; j <= i; ++j) {
+              failedAt.append('/').append(path[j]);
+            }
+            throw (new FileNotFoundException(
+                "File/Directory does not exist: " + failedAt.toString()));
           }
-          throw (new FileNotFoundException(
-              "File/Directory does not exist: " + failedAt.toString()));
         }
-      }
 
-      if (nextInode.isLink()) {
-        final INodeLink<T> link = (INodeLink<T>) nextInode;
-        final Path remainingPath;
-        if (i >= path.length - 1) {
-          remainingPath = SlashPath;
-        } else {
-          StringBuilder remainingPathStr = new StringBuilder("/" + path[i + 1]);
-          for (int j = i + 2; j < path.length; ++j) {
-            remainingPathStr.append('/').append(path[j]);
+        if (nextInode.isLink()) {
+          final INodeLink<T> link = (INodeLink<T>) nextInode;
+          final Path remainingPath;
+          if (i >= path.length - 1) {
+            remainingPath = SlashPath;
+          } else {
+            StringBuilder remainingPathStr =
+                new StringBuilder("/" + path[i + 1]);
+            for (int j = i + 2; j < path.length; ++j) {
+              remainingPathStr.append('/').append(path[j]);
+            }
+            remainingPath = new Path(remainingPathStr.toString());
           }
-          remainingPath = new Path(remainingPathStr.toString());
+          resolveResult = new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
+              link.getTargetFileSystem(), nextInode.fullPath, remainingPath);
+          return resolveResult;
+        } else if (nextInode.isInternalDir()) {
+          curInode = (INodeDir<T>) nextInode;
         }
-        final ResolveResult<T> res =
-            new ResolveResult<T>(ResultKind.EXTERNAL_DIR,
-                link.getTargetFileSystem(), nextInode.fullPath, remainingPath);
-        return res;
-      } else if (nextInode.isInternalDir()) {
-        curInode = (INodeDir<T>) nextInode;
+      }
+
+      // We have resolved to an internal dir in mount table.
+      Path remainingPath;
+      if (resolveLastComponent) {
+        remainingPath = SlashPath;
+      } else {
+        // note we have taken care of when path is "/" above
+        // for internal dirs rem-path does not start with / since the lookup
+        // that follows will do a children.get(remaningPath) and will have to
+        // strip-out the initial /
+        StringBuilder remainingPathStr = new StringBuilder("/" + path[i]);
+        for (int j = i + 1; j < path.length; ++j) {
+          remainingPathStr.append('/').append(path[j]);
+        }
+        remainingPath = new Path(remainingPathStr.toString());
+      }
+      resolveResult = new ResolveResult<T>(ResultKind.INTERNAL_DIR,
+          curInode.getInternalDirFs(), curInode.fullPath, remainingPath);
+      return resolveResult;
+    } finally {
+      if (pathResolutionCacheCapacity > 0 && resolveResult != null) {
+        addResolveResultToCache(p, resolveLastComponent, resolveResult);
       }
     }
+  }
 
-    // We have resolved to an internal dir in mount table.
-    Path remainingPath;
-    if (resolveLastComponent) {
-      remainingPath = SlashPath;
-    } else {
-      // note we have taken care of when path is "/" above
-      // for internal dirs rem-path does not start with / since the lookup
-      // that follows will do a children.get(remaningPath) and will have to
-      // strip-out the initial /
-      StringBuilder remainingPathStr = new StringBuilder("/" + path[i]);
-      for (int j = i + 1; j < path.length; ++j) {
-        remainingPathStr.append('/').append(path[j]);
+  /**
+   * Walk through all regex mount points to see
+   * whether the path match any regex expressions.
+   *
+   * @param srcPath
+   * @param resolveLastComponent
+   * @return
+   */
+  protected ResolveResult<T> tryResolveInRegexMountpoint(final String srcPath,
+      final boolean resolveLastComponent) {
+    for (RegexMountPoint regexMountPoint : regexMountPointList) {
+      ResolveResult resolveResult =
+          regexMountPoint.resolve(srcPath, resolveLastComponent);
+      if (resolveResult != null) {
+        return resolveResult;
       }
-      remainingPath = new Path(remainingPathStr.toString());
     }
-    final ResolveResult<T> res =
-        new ResolveResult<T>(ResultKind.INTERNAL_DIR,
-            curInode.getInternalDirFs(), curInode.fullPath, remainingPath);
-    return res;
+    return null;
+  }
+
+  /**
+   * Build resolve result return to caller.
+   *
+   * @param resultKind
+   * @param resolvedPathStr
+   * @param targetOfResolvedPathStr
+   * @param remainingPath
+   * @return
+   */
+  protected ResolveResult<T> buildResolveResultForRegexMountPoint(
+      ResultKind resultKind, String resolvedPathStr,
+      String targetOfResolvedPathStr, Path remainingPath) {
+    try {
+      T targetFs = getTargetFileSystem(new URI(targetOfResolvedPathStr));
+      return new ResolveResult<T>(resultKind, targetFs, resolvedPathStr,
+          remainingPath);
+    } catch (IOException ex) {
+      return null;
+    } catch (URISyntaxException uex) {
+      return null;
+    }
+  }
+
+  /**
+   * Return resolution cache capacity.
+   *
+   * @return
+   */
+  public int getPathResolutionCacheCapacity() {
+    return pathResolutionCacheCapacity;
+  }
+
+  private void addResolveResultToCache(final String pathStr,
+      final Boolean resolveLastComponent,
+      final ResolveResult<T> resolveResult) {
+    try {
+      cacheRWLock.writeLock().lock();
+      String key = getResolveCacheKeyStr(pathStr, resolveLastComponent);
+      pathResolutionCache.put(key, resolveResult);
+    } finally {
+      cacheRWLock.writeLock().unlock();
+    }
+  }
+
+  private ResolveResult<T> getResolveResultFromCache(final String pathStr,
+      final Boolean resolveLastComponent) {
+    if (pathResolutionCacheCapacity <= 0) {
+      return null;
+    }
+    try {
+      cacheRWLock.readLock().lock();
+      String key = getResolveCacheKeyStr(pathStr, resolveLastComponent);
+      return (ResolveResult<T>) pathResolutionCache.get(key);
+    } finally {
+      cacheRWLock.readLock().unlock();
+    }
+  }
+
+  public static String getResolveCacheKeyStr(final String path,
+      Boolean resolveLastComp) {
+    return path + ",resolveLastComp" + resolveLastComp;
+  }
+
+  @VisibleForTesting public LRUMap getPathResolutionCache() {
+    return pathResolutionCache;
   }
 
   List<MountPoint<T>> getMountPoints() {
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPoint.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPoint.java
new file mode 100644
index 0000000000000..a2849f4a4087e
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPoint.java
@@ -0,0 +1,245 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.hadoop.fs.viewfs.InodeTree.SlashPath;
+
+/**
+ * Regex mount point is build to implement regex based mount point.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+class RegexMountPoint<T> {
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(RegexMountPoint.class.getName());
+
+  private InodeTree inodeTree;
+  private String srcPathRegex;
+  private Pattern srcPattern;
+  private String dstPath;
+  private String interceptorSettingsString;
+  private List<RegexMountPointInterceptor> interceptorList;
+
+  public static final String SETTING_SRCREGEX_SEP = "#.";
+  public static final char INTERCEPTOR_SEP = ';';
+  public static final char INTERCEPTOR_INTERNAL_SEP = ':';
+  // ${var},$var
+  public static final Pattern VAR_PATTERN_IN_DEST =
+      Pattern.compile("\\$((\\{\\w+\\})|(\\w+))");
+
+  // key => $key or key = > ${key}
+  private Map<String, Set<String>> varInDestPathMap;
+
+  public Map<String, Set<String>> getVarInDestPathMap() {
+    return varInDestPathMap;
+  }
+
+  RegexMountPoint(InodeTree inodeTree, String sourcePathRegex,
+      String destPath, String settingsStr) {
+    this.inodeTree = inodeTree;
+    this.srcPathRegex = sourcePathRegex;
+    this.dstPath = destPath;
+    this.interceptorSettingsString = settingsStr;
+    this.interceptorList = new ArrayList<>();
+  }
+
+  /**
+   * Initialize regex mount point.
+   *
+   * @throws IOException
+   */
+  public void initialize() throws IOException {
+    try {
+      srcPattern = Pattern.compile(srcPathRegex);
+    } catch (PatternSyntaxException ex) {
+      throw new IOException(
+          "Failed to initialized mount point due to bad src path regex:"
+              + srcPathRegex + ", dstPath:" + dstPath, ex);
+    }
+    varInDestPathMap = getVarListInString(dstPath);
+    initializeInterceptors();
+  }
+
+  private void initializeInterceptors() throws IOException {
+    if (interceptorSettingsString == null
+        || interceptorSettingsString.isEmpty()) {
+      return;
+    }
+    String[] interceptorStrArray =
+        StringUtils.split(interceptorSettingsString, INTERCEPTOR_SEP);
+    for (String interceptorStr : interceptorStrArray) {
+      RegexMountPointInterceptor interceptor =
+          RegexMountPointInterceptorFactory.create(interceptorStr);
+      if (interceptor == null) {
+        throw new IOException(
+            "Illegal settings String " + interceptorSettingsString);
+      }
+      interceptor.initialize();
+      interceptorList.add(interceptor);
+    }
+  }
+
+  /**
+   * Get $var1 and $var2 style variables in string.
+   *
+   * @param input
+   * @return
+   */
+  public static Map<String, Set<String>> getVarListInString(String input) {
+    Map<String, Set<String>> varMap = new HashMap<>();
+    Matcher matcher = VAR_PATTERN_IN_DEST.matcher(input);
+    while (matcher.find()) {
+      // $var or ${var}
+      String varName = matcher.group(0);
+      // var or {var}
+      String strippedVarName = matcher.group(1);
+      if (strippedVarName.startsWith("{")) {
+        // {varName} = > varName
+        strippedVarName =
+            strippedVarName.substring(1, strippedVarName.length() - 1);
+      }
+      varMap.putIfAbsent(strippedVarName, new HashSet<>());
+      varMap.get(strippedVarName).add(varName);
+    }
+    return varMap;
+  }
+
+  public String getSrcPathRegex() {
+    return srcPathRegex;
+  }
+
+  public Pattern getSrcPattern() {
+    return srcPattern;
+  }
+
+  public String getDstPath() {
+    return dstPath;
+  }
+
+  public static Pattern getVarPatternInDest() {
+    return VAR_PATTERN_IN_DEST;
+  }
+
+  /**
+   * Get resolved path from regex mount points.
+   * @param srcPath
+   * @param resolveLastComponent
+   * @return
+   */
+  public InodeTree.ResolveResult<T> resolve(final String srcPath,
+      final boolean resolveLastComponent) {
+    String pathStrToResolve = srcPath;
+    if (!resolveLastComponent) {
+      int lastSlashIndex = srcPath.lastIndexOf(SlashPath.toString());
+      if (lastSlashIndex == -1) {
+        return null;
+      }
+      pathStrToResolve = srcPath.substring(0, lastSlashIndex);
+    }
+    for (RegexMountPointInterceptor interceptor : interceptorList) {
+      // TODO: if the source interceptor change the pathStrToResolve
+      pathStrToResolve = interceptor.interceptSource(pathStrToResolve);
+    }
+    LOGGER.debug("Path to resolve:" + pathStrToResolve + ",srcPattern:"
+        + getSrcPathRegex());
+    Matcher srcMatcher = getSrcPattern().matcher(pathStrToResolve);
+    String parsedDestPath = getDstPath();
+    int mappedCount = 0;
+    String resolvedPathStr = "";
+    while (srcMatcher.find()) {
+      resolvedPathStr = pathStrToResolve.substring(0, srcMatcher.end());
+      Map<String, Set<String>> varMap = getVarInDestPathMap();
+      for (Map.Entry<String, Set<String>> entry : varMap.entrySet()) {
+        String groupNameOrIndexStr = entry.getKey();
+        String groupValue = null;
+        if (groupNameOrIndexStr.matches("\\d+")) {
+          // group index
+          int groupIndex = Integer.parseUnsignedInt(groupNameOrIndexStr);
+          if (groupIndex >= 0 && groupIndex <= srcMatcher.groupCount()) {
+            groupValue = srcMatcher.group(groupIndex);
+          }
+        } else {
+          // named group in regex
+          groupValue = srcMatcher.group(groupNameOrIndexStr);
+        }
+        if (groupValue == null) {
+          continue;
+        }
+        Set<String> varNameListToReplace = entry.getValue();
+        for (String varName : varNameListToReplace) {
+          parsedDestPath = parsedDestPath.replace(varName, groupValue);
+          LOGGER.debug("parsedDestPath value is:" + parsedDestPath);
+        }
+      }
+      ++mappedCount;
+    }
+    if (0 == mappedCount) {
+      return null;
+    }
+    String remainingPathStr = srcPath.substring(resolvedPathStr.length());
+    if (!remainingPathStr.startsWith("/")) {
+      remainingPathStr = "/" + remainingPathStr;
+    }
+    Path remainingPath = new Path(remainingPathStr);
+    for (RegexMountPointInterceptor interceptor : interceptorList) {
+      parsedDestPath = interceptor.interceptResolvedDestPathStr(parsedDestPath);
+      remainingPath =
+          interceptor.interceptRemainingPath(remainingPath);
+    }
+    InodeTree.ResolveResult resolveResult = inodeTree
+        .buildResolveResultForRegexMountPoint(InodeTree.ResultKind.EXTERNAL_DIR,
+            resolvedPathStr, parsedDestPath, remainingPath);
+    return resolveResult;
+  }
+
+  /**
+   * Convert interceptor to string.
+   *
+   * @param interceptorList
+   * @return
+   */
+  public static String convertInterceptorsToString(
+      List<RegexMountPointInterceptor> interceptorList) {
+    StringBuffer stringBuffer = new StringBuffer();
+    for (int index = 0; index < interceptorList.size(); ++index) {
+      stringBuffer.append(interceptorList.get(index).toString());
+      if (index < interceptorList.size() - 1) {
+        stringBuffer.append(INTERCEPTOR_SEP);
+      }
+    }
+    return stringBuffer.toString();
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptor.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptor.java
new file mode 100644
index 0000000000000..37f44b0a51579
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptor.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * RegexMountPointInterceptor is a mechanism provided to intercept
+ * src and resolved path before/after resolution.
+ */
+@InterfaceAudience.LimitedPrivate("Common")
+@InterfaceStability.Unstable
+interface RegexMountPointInterceptor {
+
+  /**
+   * Initialize interceptor and throws IOException if needed.
+   * @throws IOException
+   */
+  void initialize() throws IOException;
+
+  /**
+   * Intercept source before resolution.
+   * @param source
+   * @return
+   */
+  String interceptSource(String source);
+
+  /**
+   * Intercept parsed dest path and return a new one.
+   * @return intercepted string
+   */
+  String interceptResolvedDestPathStr(String parsedDestPathStr);
+
+  /**
+   * Intercept remaining path.
+   * @return intercepted string
+   */
+  Path interceptRemainingPath(Path remainingPath);
+
+  /**
+   * Get interceptor type.
+   * @return
+   */
+  RegexMountPointInterceptorType getType();
+
+  /**
+   * Serialize the interceptor to a string.
+   * @return
+   */
+  String serializeToString();
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorFactory.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorFactory.java
new file mode 100644
index 0000000000000..d9b3f54021333
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorFactory.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * The interceptor factory used to create RegexMountPoint interceptors.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+final class RegexMountPointInterceptorFactory {
+
+  private RegexMountPointInterceptorFactory() {
+
+  }
+
+  /**
+   * interceptorSettingsString string should be like ${type}:${string},
+   * e.g. replaceresolveddstpath:word1,word2.
+   *
+   * @param interceptorSettingsString
+   * @return
+   */
+  public static RegexMountPointInterceptor create(
+      String interceptorSettingsString) {
+    int typeTagIndex = interceptorSettingsString
+        .indexOf(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP);
+    if (typeTagIndex == -1 || (typeTagIndex == (
+        interceptorSettingsString.length() - 1))) {
+      return null;
+    }
+    String typeTag = interceptorSettingsString.substring(0, typeTagIndex).trim()
+        .toLowerCase();
+    RegexMountPointInterceptorType interceptorType =
+        RegexMountPointInterceptorType.get(typeTag);
+    if (interceptorType == null) {
+      return null;
+    }
+    switch (interceptorType) {
+    case REPLACE_RESOLVED_DST_PATH:
+      RegexMountPointInterceptor interceptor =
+          RegexMountPointResolvedDstPathReplaceInterceptor
+              .deserializeFromString(interceptorSettingsString);
+      return interceptor;
+    default:
+      // impossible now
+      return null;
+    }
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorType.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorType.java
new file mode 100644
index 0000000000000..ad953eba24ad9
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointInterceptorType.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * RegexMountPointInterceptorType.
+ */
+public enum RegexMountPointInterceptorType {
+  REPLACE_RESOLVED_DST_PATH("replaceresolveddstpath");
+
+  private final String configName;
+  private static final Map<String, RegexMountPointInterceptorType>
+      INTERCEPTOR_TYPE_MAP
+      = new HashMap<String, RegexMountPointInterceptorType>();
+
+  static {
+    for (RegexMountPointInterceptorType interceptorType
+        : RegexMountPointInterceptorType.values()) {
+      INTERCEPTOR_TYPE_MAP.put(
+          interceptorType.getConfigName(), interceptorType);
+    }
+  }
+
+  RegexMountPointInterceptorType(String configName) {
+    this.configName = configName;
+  }
+
+  public String getConfigName() {
+    return configName;
+  }
+
+  public static RegexMountPointInterceptorType get(String configName) {
+    return INTERCEPTOR_TYPE_MAP.get(configName);
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointResolvedDstPathReplaceInterceptor.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointResolvedDstPathReplaceInterceptor.java
new file mode 100644
index 0000000000000..d89b93df2b973
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/RegexMountPointResolvedDstPathReplaceInterceptor.java
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
+
+import static org.apache.hadoop.fs.viewfs.RegexMountPointInterceptorType.REPLACE_RESOLVED_DST_PATH;
+
+/**
+ * Implementation of RegexMountPointResolvedDstPathReplaceInterceptor.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+class RegexMountPointResolvedDstPathReplaceInterceptor
+    implements RegexMountPointInterceptor {
+
+  private String srcRegexString;
+  private String replaceString;
+  private Pattern srcRegexPattern;
+
+  RegexMountPointResolvedDstPathReplaceInterceptor(String srcRegex,
+      String replaceString) {
+    this.srcRegexString = srcRegex;
+    this.replaceString = replaceString;
+    this.srcRegexPattern = null;
+  }
+
+  public String getSrcRegexString() {
+    return srcRegexString;
+  }
+
+  public String getReplaceString() {
+    return replaceString;
+  }
+
+  public Pattern getSrcRegexPattern() {
+    return srcRegexPattern;
+  }
+
+  @Override public void initialize() throws IOException {
+    try {
+      srcRegexPattern = Pattern.compile(srcRegexString);
+    } catch (PatternSyntaxException ex) {
+      throw new IOException(
+          "Initialize interceptor failed, srcRegx:" + srcRegexString, ex);
+    }
+  }
+
+  /**
+   * Intercept source before resolution.
+   *
+   * @param source
+   * @return
+   */
+  @Override public String interceptSource(String source) {
+    return source;
+  }
+
+  /**
+   * Intercept resolved path, e.g.
+   * Mount point /^(\\w+)/, ${1}.hadoop.net
+   * If incoming path is /user1/home/tmp/job1,
+   * then the resolved path str will be user1.
+   *
+   * @return intercepted string
+   */
+  @Override public String interceptResolvedDestPathStr(
+      String parsedDestPathStr) {
+    Matcher matcher = srcRegexPattern.matcher(parsedDestPathStr);
+    return matcher.replaceAll(replaceString);
+  }
+
+  /**
+   * Intercept remaining path.
+   *
+   * @return intercepted path
+   */
+  @Override public Path interceptRemainingPath(Path remainingPath) {
+    return remainingPath;
+  }
+
+  @Override public RegexMountPointInterceptorType getType() {
+    return REPLACE_RESOLVED_DST_PATH;
+  }
+
+  @Override public String serializeToString() {
+    return REPLACE_RESOLVED_DST_PATH.getConfigName()
+        + RegexMountPoint.INTERCEPTOR_INTERNAL_SEP + srcRegexString
+        + RegexMountPoint.INTERCEPTOR_INTERNAL_SEP + replaceString;
+  }
+
+  /**
+   * Create interceptor from config string. The string should be in
+   * replaceresolvedpath:wordToReplace:replaceString
+   * Note that we'll assume there's no ':' in the regex for the moment.
+   *
+   * @param serializedString
+   * @return
+   */
+  public static RegexMountPointResolvedDstPathReplaceInterceptor
+      deserializeFromString(String serializedString) {
+    String[] strings = serializedString
+        .split(Character.toString(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP));
+    // We'll assume there's no ':' in the regex for the moment.
+    if (strings.length != 3) {
+      return null;
+    }
+    //The format should be like replaceresolvedpath:wordToReplace:replaceString
+    return new RegexMountPointResolvedDstPathReplaceInterceptor(strings[1],
+        strings[2]);
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPoint.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPoint.java
new file mode 100644
index 0000000000000..fb334ec6be9f4
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPoint.java
@@ -0,0 +1,156 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test Regex Mount Point.
+ */
+public class TestRegexMountPoint {
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(TestRegexMountPoint.class.getName());
+
+  private InodeTree inodeTree;
+  private Configuration conf;
+
+  class TestRegexMountPointFileSystem {
+    public URI getUri() {
+      return uri;
+    }
+
+    private URI uri;
+
+    TestRegexMountPointFileSystem(URI uri) {
+      String uriStr = uri == null ? "null" : uri.toString();
+      LOGGER.info("Create TestRegexMountPointFileSystem Via URI:" + uriStr);
+      this.uri = uri;
+    }
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    conf = new Configuration();
+    ConfigUtil.addLink(conf, TestRegexMountPoint.class.getName(), "/mnt",
+        URI.create("file:///"));
+
+    inodeTree = new InodeTree<TestRegexMountPointFileSystem>(conf,
+        TestRegexMountPoint.class.getName()) {
+      @Override
+      protected TestRegexMountPointFileSystem getTargetFileSystem(
+          final URI uri) {
+        return new TestRegexMountPointFileSystem(uri);
+      }
+
+      @Override
+      protected TestRegexMountPointFileSystem getTargetFileSystem(
+          final INodeDir<TestRegexMountPointFileSystem> dir) {
+        return new TestRegexMountPointFileSystem(null);
+      }
+
+      @Override
+      protected TestRegexMountPointFileSystem getTargetFileSystem(
+          final String settings, final URI[] mergeFsURIList) {
+        return new TestRegexMountPointFileSystem(null);
+      }
+    };
+  }
+
+  @After public void tearDown() throws Exception {
+    inodeTree = null;
+  }
+
+  @Test public void testGetVarListInString() throws IOException {
+    String srcRegex = "/(\\w+)";
+    String target = "/$0/${1}/$1/${2}/${2}";
+    RegexMountPoint regexMountPoint =
+        new RegexMountPoint(inodeTree, srcRegex, target, null);
+    regexMountPoint.initialize();
+    Map<String, Set<String>> varMap = regexMountPoint.getVarInDestPathMap();
+    Assert.assertEquals(varMap.size(), 3);
+    Assert.assertEquals(varMap.get("0").size(), 1);
+    Assert.assertTrue(varMap.get("0").contains("$0"));
+    Assert.assertEquals(varMap.get("1").size(), 2);
+    Assert.assertTrue(varMap.get("1").contains("${1}"));
+    Assert.assertTrue(varMap.get("1").contains("$1"));
+    Assert.assertEquals(varMap.get("2").size(), 1);
+    Assert.assertTrue(varMap.get("2").contains("${2}"));
+  }
+
+  @Test public void testResolve() throws IOException {
+    String regexStr = "^/user/(?<username>\\w+)";
+    String dstPathStr = "/namenode1/testResolve/$username";
+    String settingsStr = null;
+    RegexMountPoint regexMountPoint =
+        new RegexMountPoint(inodeTree, regexStr, dstPathStr, settingsStr);
+    regexMountPoint.initialize();
+    InodeTree.ResolveResult resolveResult =
+        regexMountPoint.resolve("/user/hadoop/file1", true);
+    Assert.assertEquals(resolveResult.kind, InodeTree.ResultKind.EXTERNAL_DIR);
+    Assert.assertTrue(
+        resolveResult.targetFileSystem
+            instanceof TestRegexMountPointFileSystem);
+    Assert.assertTrue(resolveResult.resolvedPath.equals("/user/hadoop"));
+    Assert.assertTrue(
+        resolveResult.targetFileSystem
+            instanceof TestRegexMountPointFileSystem);
+    Assert.assertTrue(
+        ((TestRegexMountPointFileSystem) resolveResult.targetFileSystem)
+            .getUri().toString().equals("/namenode1/testResolve/hadoop"));
+    Assert.assertTrue(resolveResult.remainingPath.toString().equals("/file1"));
+  }
+
+  @Test public void testResolveWithInterceptor() throws IOException {
+    String regexStr = "^/user/(?<username>\\w+)";
+    String dstPathStr = "/namenode1/testResolve/$username";
+    // Replace "_" with "-"
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        new RegexMountPointResolvedDstPathReplaceInterceptor("_", "-");
+    // replaceresolvedpath:_:-
+    String settingsStr = interceptor.serializeToString();
+    RegexMountPoint regexMountPoint =
+        new RegexMountPoint(inodeTree, regexStr, dstPathStr, settingsStr);
+    regexMountPoint.initialize();
+    InodeTree.ResolveResult resolveResult =
+        regexMountPoint.resolve("/user/hadoop_user1/file_index", true);
+    Assert.assertEquals(resolveResult.kind, InodeTree.ResultKind.EXTERNAL_DIR);
+    Assert.assertTrue(
+        resolveResult.targetFileSystem
+            instanceof TestRegexMountPointFileSystem);
+    Assert.assertTrue(resolveResult.resolvedPath.equals("/user/hadoop_user1"));
+    Assert.assertTrue(
+        resolveResult.targetFileSystem
+            instanceof TestRegexMountPointFileSystem);
+    Assert.assertTrue(
+        ((TestRegexMountPointFileSystem) resolveResult.targetFileSystem)
+            .getUri().toString().equals("/namenode1/testResolve/hadoop-user1"));
+    Assert.assertTrue(
+        resolveResult.remainingPath.toString().equals("/file_index"));
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointInterceptorFactory.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointInterceptorFactory.java
new file mode 100644
index 0000000000000..0b12332353067
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointInterceptorFactory.java
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test Regex Mount Point Interceptor Factory.
+ */
+public class TestRegexMountPointInterceptorFactory {
+
+  @Test public void testCreateNormalCase() {
+    String replaceInterceptorStr =
+        RegexMountPointInterceptorType.REPLACE_RESOLVED_DST_PATH.getConfigName()
+            + Character.toString(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP)
+            + "src" + Character
+            .toString(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP) + "replace";
+    RegexMountPointInterceptor interceptor =
+        RegexMountPointInterceptorFactory.create(replaceInterceptorStr);
+    Assert.assertTrue(
+        interceptor
+            instanceof RegexMountPointResolvedDstPathReplaceInterceptor);
+  }
+
+  @Test public void testCreateBadCase() {
+    String replaceInterceptorStr =
+        RegexMountPointInterceptorType.REPLACE_RESOLVED_DST_PATH.getConfigName()
+            + "___" + Character
+            .toString(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP) + "src"
+            + Character.toString(RegexMountPoint.INTERCEPTOR_INTERNAL_SEP)
+            + "replace";
+    RegexMountPointInterceptor interceptor =
+        RegexMountPointInterceptorFactory.create(replaceInterceptorStr);
+    Assert.assertTrue(interceptor == null);
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointResolvedDstPathReplaceInterceptor.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointResolvedDstPathReplaceInterceptor.java
new file mode 100644
index 0000000000000..5c44fbb25c089
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestRegexMountPointResolvedDstPathReplaceInterceptor.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.IOException;
+
+import org.apache.hadoop.fs.Path;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.hadoop.fs.viewfs.RegexMountPointInterceptorType.REPLACE_RESOLVED_DST_PATH;
+
+/**
+ * Test RegexMountPointResolvedDstPathReplaceInterceptor.
+ */
+public class TestRegexMountPointResolvedDstPathReplaceInterceptor {
+
+  public String createSerializedString(String regex, String replaceString) {
+    return REPLACE_RESOLVED_DST_PATH.getConfigName()
+        + RegexMountPoint.INTERCEPTOR_INTERNAL_SEP + regex
+        + RegexMountPoint.INTERCEPTOR_INTERNAL_SEP + replaceString;
+  }
+
+  @Test public void testDeserializeFromStringNormalCase() throws IOException {
+    String srcRegex = "-";
+    String replaceString = "_";
+    String serializedString = createSerializedString(srcRegex, replaceString);
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        RegexMountPointResolvedDstPathReplaceInterceptor
+            .deserializeFromString(serializedString);
+    Assert.assertTrue(interceptor.getSrcRegexString().equals(srcRegex));
+    Assert.assertTrue(interceptor.getReplaceString().equals(replaceString));
+    Assert.assertTrue(interceptor.getSrcRegexPattern() == null);
+    interceptor.initialize();
+    Assert.assertTrue(
+        interceptor.getSrcRegexPattern().toString().equals(srcRegex));
+  }
+
+  @Test public void testDeserializeFromStringBadCase() throws IOException {
+    String srcRegex = "-";
+    String replaceString = "_";
+    String serializedString = createSerializedString(srcRegex, replaceString);
+    serializedString = serializedString + ":ddd";
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        RegexMountPointResolvedDstPathReplaceInterceptor
+            .deserializeFromString(serializedString);
+    Assert.assertEquals(interceptor, null);
+  }
+
+  @Test public void testSerialization() {
+    String srcRegex = "word1";
+    String replaceString = "word2";
+    String serializedString = createSerializedString(srcRegex, replaceString);
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        new RegexMountPointResolvedDstPathReplaceInterceptor(srcRegex,
+            replaceString);
+    Assert.assertEquals(interceptor.serializeToString(), serializedString);
+  }
+
+  @Test public void testInterceptSource() {
+    String srcRegex = "word1";
+    String replaceString = "word2";
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        new RegexMountPointResolvedDstPathReplaceInterceptor(srcRegex,
+            replaceString);
+    String sourcePath = "/a/b/l3/dd";
+    sourcePath = interceptor.interceptSource(sourcePath);
+  }
+
+  @Test public void testInterceptResolve() throws IOException {
+    String pathAfterResolution = "/user-hadoop";
+    Path remainingPath = new Path("/ad-data");
+
+    String srcRegex = "hadoop";
+    String replaceString = "hdfs";
+    RegexMountPointResolvedDstPathReplaceInterceptor interceptor =
+        new RegexMountPointResolvedDstPathReplaceInterceptor(srcRegex,
+            replaceString);
+    interceptor.initialize();
+    Assert.assertTrue(
+        interceptor.interceptResolvedDestPathStr(pathAfterResolution)
+            .equals("/user-hdfs"));
+  }
+}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
index b9453df5699f2..ef855b506e304 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
@@ -28,6 +28,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 
+import org.apache.commons.collections.map.LRUMap;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.BlockStoragePolicySpi;
@@ -1272,4 +1273,49 @@ public void testLinkTarget() throws Exception {
           containsString("File does not exist:"));
     }
   }
+
+  @Test
+  public void testMountPointCache() throws Exception {
+    conf.setInt(Constants.CONFIG_VIEWFS_PATH_RESOLUTION_CACHE_CAPACITY, 1);
+    conf.setBoolean("fs.viewfs.impl.disable.cache", true);
+    FileSystem fileSystem = FileSystem.get(FsConstants.VIEWFS_URI, conf);
+    ViewFileSystem viewfs = (ViewFileSystem) fileSystem;
+    Path resolvedPath1 = new Path(targetTestRoot, "dir3/file1");
+    Path srcPath1 = new Path("/internalDir/internalDir2/linkToDir3/file1");
+    LRUMap cacheMap = viewfs.fsState.getPathResolutionCache();
+    FileSystemTestHelper.createFile(fsTarget, resolvedPath1);
+    Assert.assertTrue(
+        resolvedPath1.toString().equals(
+            fileSystem.resolvePath(srcPath1).toString()));
+    Assert.assertEquals(viewfs.fsState.getPathResolutionCacheCapacity(), 1);
+    Assert.assertEquals(viewfs.fsState.getPathResolutionCache().size(), 1);
+    Assert.assertTrue(viewfs.fsState.getPathResolutionCache().isFull());
+    InodeTree.ResolveResult resolveResult1 =
+        viewfs.fsState.resolve(viewfs.getUriPath(srcPath1), true);
+    LOG.info("Resolve result resolve path:" + resolveResult1.resolvedPath +
+        ", remaining path:" + resolveResult1.remainingPath
+        + ", file kind:" + resolveResult1.kind);
+    Assert.assertTrue(resolveResult1.resolvedPath.toString().equals(
+        new Path("/internalDir/internalDir2/linkToDir3").toString()));
+    Assert.assertTrue(resolveResult1.remainingPath.toString().equals(
+        "/file1"));
+    Assert.assertEquals(resolveResult1,
+        cacheMap.get(
+            InodeTree.getResolveCacheKeyStr(
+                viewfs.getUriPath(srcPath1), true)));
+
+    Path srcPath2 = new Path("/internalDir/internalDir2/linkToDir3/file2");
+    InodeTree.ResolveResult resolveResult2 =
+        viewfs.fsState.resolve(viewfs.getUriPath(srcPath2), true);
+    Assert.assertTrue(resolveResult2.resolvedPath.toString().equals(
+        new Path("/internalDir/internalDir2/linkToDir3").toString()));
+    Assert.assertTrue(resolveResult2.remainingPath.toString().equals(
+        "/file2"));
+    Assert.assertEquals(viewfs.fsState.getPathResolutionCache().size(), 1);
+    Assert.assertTrue(viewfs.fsState.getPathResolutionCache().isFull());
+    Assert.assertEquals(resolveResult2,
+        cacheMap.get(
+            InodeTree.getResolveCacheKeyStr(
+                viewfs.getUriPath(srcPath2), true)));
+  }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemLinkRegex.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemLinkRegex.java
new file mode 100644
index 0000000000000..a91086b79f606
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemLinkRegex.java
@@ -0,0 +1,304 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystemTestHelper;
+import org.apache.hadoop.fs.FsConstants;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.MiniDFSNNTopology;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.hadoop.fs.viewfs.RegexMountPoint.INTERCEPTOR_INTERNAL_SEP;
+
+/**
+ * Test linkRegex node type for view file system.
+ */
+public class TestViewFileSystemLinkRegex extends ViewFileSystemBaseTest {
+  public static final Logger LOGGER =
+      LoggerFactory.getLogger(TestViewFileSystemLinkRegex.class);
+
+  private static FileSystem fsDefault;
+  private static MiniDFSCluster cluster;
+  private static final int NAME_SPACES_COUNT = 3;
+  private static final int DATA_NODES_COUNT = 3;
+  private static final int FS_INDEX_DEFAULT = 0;
+  private static final FileSystem[] FS_HDFS = new FileSystem[NAME_SPACES_COUNT];
+  private static final Configuration CONF = new Configuration();
+  private static final String CLUSTER_NAME =
+      "TestViewFileSystemLinkRegexCluster";
+  private static final File TEST_DIR = GenericTestUtils
+      .getTestDir(TestViewFileSystemLinkRegex.class.getSimpleName());
+  private static final String TEST_BASE_PATH =
+      "/tmp/TestViewFileSystemLinkRegex";
+
+  @Override protected FileSystemTestHelper createFileSystemHelper() {
+    return new FileSystemTestHelper(TEST_BASE_PATH);
+  }
+
+  @BeforeClass public static void clusterSetupAtBeginning() throws IOException {
+    SupportsBlocks = true;
+    CONF.setBoolean(DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY,
+        true);
+    cluster = new MiniDFSCluster.Builder(CONF).nnTopology(
+        MiniDFSNNTopology.simpleFederatedTopology(NAME_SPACES_COUNT))
+        .numDataNodes(DATA_NODES_COUNT).build();
+    cluster.waitClusterUp();
+
+    for (int i = 0; i < NAME_SPACES_COUNT; i++) {
+      FS_HDFS[i] = cluster.getFileSystem(i);
+    }
+    fsDefault = FS_HDFS[FS_INDEX_DEFAULT];
+  }
+
+  @AfterClass public static void clusterShutdownAtEnd() throws Exception {
+    if (cluster != null) {
+      cluster.shutdown();
+    }
+  }
+
+  @Override @Before public void setUp() throws Exception {
+    fsTarget = fsDefault;
+    super.setUp();
+  }
+
+  /**
+   * Override this so that we don't set the targetTestRoot to any path under the
+   * root of the FS, and so that we don't try to delete the test dir, but rather
+   * only its contents.
+   */
+  @Override void initializeTargetTestRoot() throws IOException {
+    targetTestRoot = fsDefault.makeQualified(new Path("/"));
+    for (FileStatus status : fsDefault.listStatus(targetTestRoot)) {
+      fsDefault.delete(status.getPath(), true);
+    }
+  }
+
+  @Override void setupMountPoints() {
+    super.setupMountPoints();
+  }
+
+  @Override int getExpectedDelegationTokenCount() {
+    return 1; // all point to the same fs so 1 unique token
+  }
+
+  @Override
+  int getExpectedDelegationTokenCountWithCredentials() {
+    return 1;
+  }
+
+  public String buildReplaceInterceptorSettingString(String srcRegex,
+      String replaceString) {
+    return
+        RegexMountPointInterceptorType.REPLACE_RESOLVED_DST_PATH.getConfigName()
+            + INTERCEPTOR_INTERNAL_SEP + srcRegex + INTERCEPTOR_INTERNAL_SEP
+            + replaceString;
+  }
+
+  public String linkInterceptorSettings(
+      List<String> interceptorSettingStrList) {
+    StringBuilder stringBuilder = new StringBuilder();
+    int listSize = interceptorSettingStrList.size();
+    for (int i = 0; i < listSize; ++i) {
+      stringBuilder.append(interceptorSettingStrList.get(i));
+      if (i < listSize - 1) {
+        stringBuilder.append(RegexMountPoint.INTERCEPTOR_SEP);
+      }
+    }
+    return stringBuilder.toString();
+  }
+
+  @Test
+  public void testConfLinkRegexIndexMapping() throws Exception {
+    // (^/(\w+),/targetTestRoot/$1)
+    // => /targetTestRoot/testConfLinkRegexIndexMapping1
+    URI viewFsUri =
+        new URI(FsConstants.VIEWFS_SCHEME, CLUSTER_NAME, "/", null, null);
+    String regexStr = "^/(\\w+)";
+    String dstPathStr = targetTestRoot + "$1";
+    Path srcPath = new Path("/testConfLinkRegexIndexMapping1");
+    Path expectedResolveResult =
+        new Path(dstPathStr.replace("$1", "testConfLinkRegexIndexMapping1"));
+    FSDataOutputStream outputStream = fsTarget.create((expectedResolveResult));
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    FileSystem vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+
+    // Test ${1} format
+    // ^/(\w+, /targetTestRoot/${1})
+    // => /targetTestRoot/testConfLinkRegexIndexMapping2
+    dstPathStr = targetTestRoot + "${1}";
+    srcPath = new Path("/testConfLinkRegexIndexMapping2");
+    expectedResolveResult =
+        new Path(dstPathStr.replace("${1}", "testConfLinkRegexIndexMapping2"));
+    outputStream = fsTarget.create(expectedResolveResult);
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+
+    //(^/(\w+)/file1, /targetTestRoot/$1)
+    // = > /targetTestRoot/testConfLinkRegexIndexMapping3/file1
+    dstPathStr = targetTestRoot + "$1";
+    srcPath = new Path("/testConfLinkRegexIndexMapping3/file1");
+    expectedResolveResult = new Path(
+        dstPathStr.replace("$1", "testConfLinkRegexIndexMapping3/file1"));
+    outputStream = fsTarget.create(expectedResolveResult);
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+
+    //(^/(\w+)/file1, /targetTestRoot/$1/)
+    // = > /targetTestRoot/testConfLinkRegexIndexMapping4/file1
+    dstPathStr = targetTestRoot + "$1/";
+    srcPath = new Path("/testConfLinkRegexIndexMapping4/file1");
+    expectedResolveResult = new Path(
+        dstPathStr.replace("$1", "testConfLinkRegexIndexMapping4/file1"));
+    outputStream = fsTarget.create(expectedResolveResult);
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+  }
+
+  @Test
+  public void testConfLinkRegexNamedGroupMapping() throws Exception {
+    // ^/(?<firstDir>\\w+) = > /targetTestRoot/$firstDir
+    URI viewFsUri =
+        new URI(FsConstants.VIEWFS_SCHEME, CLUSTER_NAME, "/", null, null);
+    String regexStr = "^/(?<firstDir>\\w+)";
+    String dstPathStr = targetTestRoot + "$firstDir";
+    Path srcPath = new Path("/testConfLinkRegexNamedGroupMapping1");
+    Path expectedResolveResult = new Path(
+        dstPathStr.replace("$firstDir", "testConfLinkRegexNamedGroupMapping1"));
+    FSDataOutputStream outputStream = fsTarget.create((expectedResolveResult));
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    FileSystem vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+
+    // Test ${1} format
+    dstPathStr = targetTestRoot + "${firstDir}";
+    srcPath = new Path("/testConfLinkRegexNamedGroupMapping2");
+    expectedResolveResult = new Path(dstPathStr
+        .replace("${firstDir}", "testConfLinkRegexNamedGroupMapping2"));
+    outputStream = fsTarget.create(expectedResolveResult);
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+  }
+
+  @Test
+  public void testConfLinkRegexFixedDestMapping() throws Exception {
+    URI viewFsUri =
+        new URI(FsConstants.VIEWFS_SCHEME, CLUSTER_NAME, "/", null, null);
+    String regexStr = "^/\\w+";
+    String dstPathStr =
+        targetTestRoot + "testConfLinkRegexFixedDestMappingFile";
+    Path expectedResolveResult = new Path(dstPathStr);
+    FSDataOutputStream outputStream = fsTarget.create((expectedResolveResult));
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil.addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, null);
+    FileSystem vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(
+        expectedResolveResult.equals(vfs.resolvePath(new Path("/misc1"))));
+    Assert.assertTrue(
+        expectedResolveResult.equals(vfs.resolvePath(new Path("/misc2"))));
+  }
+
+  @Test
+  public void testConfLinkRegexWithSingleInterceptor() throws Exception {
+    URI viewFsUri =
+        new URI(FsConstants.VIEWFS_SCHEME, CLUSTER_NAME, "/", null, null);
+    String regexStr = "^/user/(?<username>\\w+)";
+    String dstPathStr = targetTestRoot + "$username";
+    // Replace "_" with "-"
+    String settingString = buildReplaceInterceptorSettingString("_", "-");
+    Path srcPath = new Path("/user/hadoop_user1/hadoop_file1");
+    Path expectedResolveResult =
+        new Path(targetTestRoot, "hadoop-user1/hadoop_file1");
+    FSDataOutputStream outputStream = fsTarget.create((expectedResolveResult));
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil
+        .addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, settingString);
+    FileSystem vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+  }
+
+  @Test
+  public void testConfLinkRegexWithInterceptors() throws Exception {
+    URI viewFsUri =
+        new URI(FsConstants.VIEWFS_SCHEME, CLUSTER_NAME, "/", null, null);
+    String regexStr = "^/user/(?<username>\\w+)/";
+    String dstPathStr = targetTestRoot + "$username";
+    // Replace "_" with "-"
+    String interceptor1 = buildReplaceInterceptorSettingString("_", "-");
+    // Replace "hadoop" with "hdfs"
+    String interceptor2 =
+        buildReplaceInterceptorSettingString("hadoop", "hdfs");
+    String interceptors =
+        linkInterceptorSettings(Arrays.asList(interceptor1, interceptor2));
+    Path srcPath = new Path("/user/hadoop_user1/hadoop_file1");
+    Path expectedResolveResult =
+        new Path(targetTestRoot, "hdfs-user1/hadoop_file1");
+    FSDataOutputStream outputStream = fsTarget.create((expectedResolveResult));
+    fsTarget.listStatus(expectedResolveResult);
+    outputStream.close();
+    ConfigUtil
+        .addLinkRegex(conf, CLUSTER_NAME, regexStr, dstPathStr, interceptors);
+    FileSystem vfs = FileSystem.get(viewFsUri, conf);
+    Assert.assertTrue(expectedResolveResult.equals(vfs.resolvePath(srcPath)));
+    Assert.assertEquals(0L, vfs.getFileStatus(srcPath).getLen());
+  }
+}