Using TachyonURI and creating directories on mount.

clients/unshaded/src/main/java/tachyon/client/TachyonFS.java

@@ -342,8 +342,7 @@ public synchronized long createFile(TachyonURI path, TachyonURI ufsPath, long bl
       if (blockSizeByte > 0) {
         return mFSMasterClient.createFile(path.getPath(), blockSizeByte, recursive);
       } else {
-        return mFSMasterClient.loadFileInfoFromUfs(path.getPath(), ufsPath.toString(),
-                                                   blockSizeByte, recursive);
+        return mFSMasterClient.loadFileFromUfs(path.getPath(), recursive);
       }
     } catch (TException e) {
       throw new IOException(e);

clients/unshaded/src/main/java/tachyon/client/file/TachyonFileSystem.java

@@ -202,21 +202,19 @@ public List<FileInfo> listStatus(TachyonFile file) throws IOException, FileDoesN
    * updated and no data will be transferred. The data can be added to Tachyon space by doing an
    * operation with the cache option specified, for example reading.
    *
-   * @param path the path to create the file in Tachyon
-   * @param ufsPath the under storage system path of the file that will back the Tachyon file
+   * @param path the Tachyon path of the file to load
    * @param recursive if true, the parent directories to the file in Tachyon will be created
    * @return the file id of the resulting file in Tachyon
    * @throws FileDoesNotExistException if there is no file at the given path
    * @throws IOException if the Tachyon path is invalid or the ufsPath does not exist
    */
-  public long loadFileInfoFromUfs(TachyonURI path, TachyonURI ufsPath, boolean recursive)
+  public long loadFileFromUfs(TachyonURI path, boolean recursive)
       throws IOException, FileDoesNotExistException {
     FileSystemMasterClient masterClient = mContext.acquireMasterClient();
     try {
       long fileId =
-          masterClient.loadFileInfoFromUfs(path.getPath(), ufsPath.toString(), -1L, recursive);
-      LOG.info(
-          "Loaded file " + path.getPath() + " from " + ufsPath + (recursive ? " recursively" : ""));
+          masterClient.loadFileFromUfs(path.getPath(), recursive);
+      LOG.info("Loaded file " + path.getPath() + (recursive ? " recursively" : ""));
       return fileId;
     } finally {
       mContext.releaseMasterClient(masterClient);

common/src/main/java/tachyon/client/FileSystemMasterClient.java

@@ -270,20 +270,19 @@ public synchronized long createFile(String path, long blockSizeBytes, boolean re
   /**
    * Loads a file from the under file system.
    *
-   * @param path the file path
-   * @param ufsPath the under file system path
+   * @param path the Tachyon path of the file
    * @param recursive whether parent directories should be loaded if not present yet
    * @return the file id
    * @throws FileDoesNotExistException if the file does not exist
    * @throws IOException if an I/O error occurs
    */
-  public synchronized long loadFileInfoFromUfs(String path, String ufsPath, boolean recursive)
+  public synchronized long loadFileFromUfs(String path, boolean recursive)
       throws IOException, FileDoesNotExistException {
     int retry = 0;
     while (!mClosed && (retry ++) <= RPC_MAX_NUM_RETRY) {
       connect();
       try {
-        return mClient.loadFileFromUfs(ufsPath, recursive);
+        return mClient.loadFileFromUfs(path, recursive);
       } catch (FileDoesNotExistException e) {
         throw e;
       } catch (TException e) {

integration-tests/src/test/java/tachyon/master/JournalIntegrationTest.java

@@ -114,7 +114,7 @@ public void AddCheckpointTest() throws Exception {
     FileInfo fInfo = mTfs.getInfo(mTfs.open(new TachyonURI("/xyz")));
     TachyonURI ckPath = new TachyonURI("/xyz_ck");
     // TODO(cc): What's the counterpart in the new client API for this?
-    mTfs.loadFileInfoFromUfs(new TachyonURI("/xyz_ck"), new TachyonURI(fInfo.getUfsPath()), true);
+    mTfs.loadFileFromUfs(new TachyonURI("/xyz_ck"), true);
     FileInfo ckFileInfo = mTfs.getInfo(mTfs.open(ckPath));
     mLocalTachyonCluster.stopTFS();
     AddCheckpointTestUtil(fInfo, ckFileInfo);

servers/src/main/java/tachyon/master/file/FileSystemMaster.java

@@ -1091,16 +1091,13 @@ public List<Integer> getPriorityDependencyList() {
     }
   }
 
-  public long loadFileFromUfs(String ufsPath, boolean recursive) throws TachyonException {
-    if (ufsPath == null || ufsPath.isEmpty()) {
-      throw new IllegalArgumentException("the underFS path is not provided");
-    }
+  public long loadFileFromUfs(TachyonURI tachyonPath, boolean recursive) throws TachyonException {
+    String ufsPath = mMountTable.lookup(tachyonPath).toString();
     UnderFileSystem underfs = UnderFileSystem.get(ufsPath, MasterContext.getConf());
     try {
       long ufsBlockSizeByte = underfs.getBlockSizeByte(ufsPath);
       long fileSizeByte = underfs.getFileSize(ufsPath);
-      String path = mMountTable.reverseLookup(ufsPath);
-      long fileId = createFile(new TachyonURI(path), ufsBlockSizeByte, recursive);
+      long fileId = createFile(tachyonPath, ufsBlockSizeByte, recursive);
       if (fileId != -1) {
         completeFileCheckpoint(-1, fileId, fileSizeByte, new TachyonURI(ufsPath));
       }
@@ -1115,22 +1112,25 @@ public long loadFileFromUfs(String ufsPath, boolean recursive) throws TachyonExc
       throw new TachyonException(ipe.getMessage());
     } catch (IOException ioe) {
       throw new TachyonException(ioe.getMessage());
-    } catch (NotFoundException nfe) {
-      throw new TachyonException(nfe.getMessage());
     } catch (SuspectedFileSizeException sfse) {
       throw new TachyonException(sfse.getMessage());
     }
   }
 
-  public void mount(String tachyonPath, String ufsPath) throws AlreadyExistsException {
+  public void mount(TachyonURI tachyonPath, TachyonURI ufsPath) throws AlreadyExistsException,
+      FileAlreadyExistException, InvalidPathException {
+    mkdirs(tachyonPath, true);
     mMountTable.add(tachyonPath, ufsPath);
     writeJournalEntry(new AddMountPointEntry(tachyonPath, ufsPath));
     flushJournal();
   }
 
-  public void unmount(String tachyonPath) throws NotFoundException {
-    // TODO(jiri): Persist files nested under tachyonPath and then void its namespace.
+  // TODO(jiri): Account for asynchronously persisted files once lineage is implemented.
+  public void unmount(TachyonURI tachyonPath) throws FileDoesNotExistException,
+      InvalidPathException, NotFoundException {
     mMountTable.delete(tachyonPath);
+    long fileId = getFileId(tachyonPath);
+    // TODO(jiri): Delete the files here.
     writeJournalEntry(new DeleteMountPointEntry(tachyonPath));
     flushJournal();
   }

servers/src/main/java/tachyon/master/file/FileSystemMasterServiceHandler.java

@@ -15,7 +15,6 @@
 
 package tachyon.master.file;
 
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Set;
@@ -34,7 +33,6 @@
 import tachyon.thrift.InvalidPathException;
 import tachyon.thrift.SuspectedFileSizeException;
 import tachyon.thrift.TachyonException;
-import tachyon.underfs.UnderFileSystem;
 
 public final class FileSystemMasterServiceHandler implements FileSystemMasterService.Iface {
   private final FileSystemMaster mFileSystemMaster;
@@ -170,22 +168,30 @@ public void requestFilesInDependency(int depId) throws DependencyDoesNotExistExc
 
   @Override
   public long loadFileFromUfs(String ufsPath, boolean recursive) throws TachyonException {
-    return mFileSystemMaster.loadFileFromUfs(ufsPath, recursive);
+    return mFileSystemMaster.loadFileFromUfs(new TachyonURI(ufsPath), recursive);
   }
 
   @Override
   public void mount(String tachyonPath, String ufsPath) throws TachyonException {
     try {
-      mFileSystemMaster.mount(tachyonPath, ufsPath);
+      mFileSystemMaster.mount(new TachyonURI(tachyonPath), new TachyonURI(ufsPath));
     } catch (AlreadyExistsException aee) {
       throw new TachyonException(aee.getMessage());
+    } catch (FileAlreadyExistException faee) {
+      throw new TachyonException(faee.getMessage());
+    } catch (InvalidPathException ipe) {
+      throw new TachyonException(ipe.getMessage());
     }
   }
 
   @Override
   public void unmount(String tachyonPath) throws TachyonException {
     try {
-      mFileSystemMaster.unmount(tachyonPath);
+      mFileSystemMaster.unmount(new TachyonURI(tachyonPath));
+    } catch (FileDoesNotExistException fdnee) {
+      throw new TachyonException(fdnee.getMessage());
+    } catch (InvalidPathException ipe) {
+      throw new TachyonException(ipe.getMessage());
     } catch (NotFoundException nfe) {
       throw new TachyonException(nfe.getMessage());
     }

servers/src/main/java/tachyon/master/file/journal/AddMountPointEntry.java

@@ -19,24 +19,25 @@
 
 import com.google.common.collect.Maps;
 
+import tachyon.TachyonURI;
 import tachyon.master.journal.JournalEntry;
 import tachyon.master.journal.JournalEntryType;
 
 public class AddMountPointEntry implements JournalEntry {
   private final String mTachyonPath;
   private final String mUfsPath;
 
-  public AddMountPointEntry(String tachyonPath, String ufsPath) {
-    mTachyonPath = tachyonPath;
-    mUfsPath = ufsPath;
+  public AddMountPointEntry(TachyonURI tachyonPath, TachyonURI ufsPath) {
+    mTachyonPath = tachyonPath.toString();
+    mUfsPath = ufsPath.toString();
   }
 
-  public String getTachyonPath() {
-    return mTachyonPath;
+  public TachyonURI getTachyonPath() {
+    return new TachyonURI(mTachyonPath);
   }
 
-  public String getUfsPath() {
-    return mUfsPath;
+  public TachyonURI getUfsPath() {
+    return new TachyonURI(mUfsPath);
   }
 
   @Override

servers/src/main/java/tachyon/master/file/journal/DeleteMountPointEntry.java

@@ -19,18 +19,19 @@
 
 import com.google.common.collect.Maps;
 
+import tachyon.TachyonURI;
 import tachyon.master.journal.JournalEntry;
 import tachyon.master.journal.JournalEntryType;
 
 public class DeleteMountPointEntry implements JournalEntry {
   private final String mTachyonPath;
 
-  public DeleteMountPointEntry(String tachyonPath) {
-    mTachyonPath = tachyonPath;
+  public DeleteMountPointEntry(TachyonURI tachyonPath) {
+    mTachyonPath = tachyonPath.toString();
   }
 
-  public String getTachyonPath() {
-    return mTachyonPath;
+  public TachyonURI getTachyonPath() {
+    return new TachyonURI(mTachyonPath);
   }
 
   @Override

servers/src/main/java/tachyon/master/file/meta/MountTable.java

@@ -18,20 +18,22 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import tachyon.TachyonURI;
 import tachyon.exception.AlreadyExistsException;
 import tachyon.exception.NotFoundException;
 
 /** This class is used for keeping track of Tachyon mount points. It is thread safe. */
 public class MountTable {
-  private Map<String, String> mMountTable;
+  private Map<TachyonURI, TachyonURI> mMountTable;
 
   public MountTable() {
     final int INITIAL_CAPACITY = 10;
-    mMountTable = new HashMap<String, String>(INITIAL_CAPACITY);
+    mMountTable = new HashMap<TachyonURI, TachyonURI>(INITIAL_CAPACITY);
   }
 
-  public synchronized void add(String tachyonPath, String ufsPath) throws AlreadyExistsException {
-    for (Map.Entry<String, String> entry : mMountTable.entrySet()) {
+  public synchronized void add(TachyonURI tachyonPath, TachyonURI ufsPath)
+      throws AlreadyExistsException {
+    for (Map.Entry<TachyonURI, TachyonURI> entry : mMountTable.entrySet()) {
       if (hasPrefix(tachyonPath, entry.getKey())) {
         // Cannot mount a path under an existing mount point.
         throw new AlreadyExistsException("Tachyon path " + tachyonPath
@@ -41,36 +43,26 @@ public synchronized void add(String tachyonPath, String ufsPath) throws AlreadyE
     mMountTable.put(tachyonPath, ufsPath);
   }
 
-  public synchronized void delete(String tachyonPath) throws NotFoundException {
+  public synchronized void delete(TachyonURI tachyonPath) throws NotFoundException {
     if (mMountTable.containsKey(tachyonPath)) {
       mMountTable.remove(tachyonPath);
     }
     // Cannot mount a path under an existing mount point.
     throw new NotFoundException("Tachyon path " + tachyonPath + " is not a valid mount point.");
   }
 
-  public synchronized String lookup(String tachyonPath) {
-    for (Map.Entry<String, String> entry : mMountTable.entrySet()) {
+  public synchronized TachyonURI lookup(TachyonURI tachyonPath) {
+    for (Map.Entry<TachyonURI, TachyonURI> entry : mMountTable.entrySet()) {
       if (hasPrefix(tachyonPath, entry.getKey())) {
-        return entry.getValue() + tachyonPath.substring(entry.getKey().length());
+        return new TachyonURI(entry.getValue()
+            + tachyonPath.toString().substring(entry.getKey().toString().length()));
       }
     }
     // If the given path is not found in the mount table, the lookup is an identity.
     return tachyonPath;
   }
 
-  public synchronized String reverseLookup(String ufsPath) throws NotFoundException {
-    for (Map.Entry<String, String> entry : mMountTable.entrySet()) {
-      if (hasPrefix(ufsPath, entry.getValue())) {
-        return entry.getKey() + ufsPath.substring(entry.getValue().length());
-      }
-    }
-    // If the given path is not found in the mount table, the reverse lookup fails.
-    throw new NotFoundException("UFS path " + ufsPath + " is not mounted in Tachyon namespace.");
-  }
-
-  private boolean hasPrefix(String path, String prefix) {
-    // TODO(jiri): Use canonical representation for UFS scheme and authority.
-    return path.startsWith(prefix);
+  private boolean hasPrefix(TachyonURI path, TachyonURI prefix) {
+    return path.toString().startsWith(prefix.toString());
   }
 }

servers/src/main/java/tachyon/worker/block/BlockDataManager.java

@@ -81,7 +81,7 @@ public final class BlockDataManager {
   public BlockDataManager(WorkerSource workerSource,
       WorkerBlockMasterClient workerBlockMasterClient,
       WorkerFileSystemMasterClient workerFileSystemMasterClient) throws IOException {
-    // TODO(jiri): We may not need to assign the conf to a variable
+    // TODO(jiri): We may not need to assign the conf to a variable.
     mTachyonConf = WorkerContext.getConf();
     mHeartbeatReporter = new BlockHeartbeatReporter();
     mBlockStore = new TieredBlockStore();
@@ -136,8 +136,8 @@ public void accessBlock(long sessionId, long blockId) throws NotFoundException {
    * Add the checkpoint information of a file. The information is from the session
    * <code>sessionId</code>.
    *
-   * This method is normally triggered from {@link tachyon.client.FileOutStream#close()} if and only
-   * if {@link tachyon.client.WriteType#isThrough()} is true. The current implementation of
+   * This method is normally triggered from {@link tachyon.client.file.FileOutStream#close()} if and
+   * only if {@link tachyon.client.WriteType#isThrough()} is true. The current implementation of
    * checkpointing is that through {@link tachyon.client.WriteType} operations write to
    * {@link tachyon.underfs.UnderFileSystem} on the client's write path, but under a session temp
    * directory (temp directory is defined in the worker as {@link #getSessionUfsTmpFolder(long)}).