Removed lineage file state from lineage store

servers/src/main/java/tachyon/master/lineage/LineageMaster.java

@@ -34,7 +34,6 @@
 
 import tachyon.Constants;
 import tachyon.TachyonURI;
-import tachyon.client.file.TachyonFile;
 import tachyon.conf.TachyonConf;
 import tachyon.exception.BlockInfoException;
 import tachyon.exception.ExceptionMessage;
@@ -59,7 +58,6 @@
 import tachyon.master.lineage.checkpoint.CheckpointPlan;
 import tachyon.master.lineage.checkpoint.CheckpointSchedulingExcecutor;
 import tachyon.master.lineage.meta.Lineage;
-import tachyon.master.lineage.meta.LineageFile;
 import tachyon.master.lineage.meta.LineageFileState;
 import tachyon.master.lineage.meta.LineageIdGenerator;
 import tachyon.master.lineage.meta.LineageStore;
@@ -219,18 +217,18 @@ public LineageStoreView getLineageStoreView() {
   public synchronized long createLineage(List<TachyonURI> inputFiles, List<TachyonURI> outputFiles,
       Job job) throws InvalidPathException, FileAlreadyExistsException, BlockInfoException,
       IOException {
-    List<TachyonFile> inputTachyonFiles = Lists.newArrayList();
+    List<Long> inputTachyonFiles = Lists.newArrayList();
     for (TachyonURI inputFile : inputFiles) {
       long fileId;
       fileId = mFileSystemMaster.getFileId(inputFile);
       if (fileId == IdUtils.INVALID_FILE_ID) {
         throw new InvalidPathException(
             ExceptionMessage.LINEAGE_INPUT_FILE_NOT_EXIST.getMessage(inputFile));
       }
-      inputTachyonFiles.add(new TachyonFile(fileId));
+      inputTachyonFiles.add(fileId);
     }
     // create output files
-    List<LineageFile> outputTachyonFiles = Lists.newArrayList();
+    List<Long> outputTachyonFiles = Lists.newArrayList();
     for (TachyonURI outputFile : outputFiles) {
       long fileId;
       // TODO(yupeng): delete the placeholder files if the creation fails.
@@ -239,7 +237,7 @@ public synchronized long createLineage(List<TachyonURI> inputFiles, List<Tachyon
           new CreateOptions.Builder(MasterContext.getConf()).setRecursive(true)
               .setBlockSizeBytes(Constants.KB).build();
       fileId = mFileSystemMaster.create(outputFile, options);
-      outputTachyonFiles.add(new LineageFile(fileId));
+      outputTachyonFiles.add(fileId);
     }
 
     LOG.info("Create lineage of input:{}, output:{}, job:{}", inputTachyonFiles,

servers/src/main/java/tachyon/master/lineage/meta/LineageStore.java

@@ -20,12 +20,10 @@
 import java.util.Map;
 import java.util.Set;
 
-import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
-import tachyon.client.file.TachyonFile;
 import tachyon.collections.DirectedAcyclicGraph;
 import tachyon.exception.ExceptionMessage;
 import tachyon.exception.LineageDoesNotExistException;
@@ -78,8 +76,8 @@ public synchronized void addLineageFromJournal(LineageEntry entry) {
    * @param job the job
    * @return the id of the created lineage
    */
-  public synchronized long createLineage(List<TachyonFile> inputFiles,
-      List<LineageFile> outputFiles, Job job) {
+  public synchronized long createLineage(List<Long> inputFiles,
+      List<Long> outputFiles, Job job) {
     long lineageId = mLineageIdGenerator.generateId();
     Lineage lineage = new Lineage(lineageId, inputFiles, outputFiles, job);
     addLineageInternal(lineage);
@@ -88,31 +86,20 @@ public synchronized long createLineage(List<TachyonFile> inputFiles,
 
   private void addLineageInternal(Lineage lineage) {
     List<Lineage> parentLineages = Lists.newArrayList();
-    for (TachyonFile inputFile : lineage.getInputFiles()) {
-      if (mOutputFileIndex.containsKey(inputFile.getFileId())) {
-        parentLineages.add(mOutputFileIndex.get(inputFile.getFileId()));
+    for (long inputFile : lineage.getInputFiles()) {
+      if (mOutputFileIndex.containsKey(inputFile)) {
+        parentLineages.add(mOutputFileIndex.get(inputFile));
       }
     }
     mLineageDAG.add(lineage, parentLineages);
 
     // update index
-    for (TachyonFile outputFile : lineage.getOutputFiles()) {
-      mOutputFileIndex.put(outputFile.getFileId(), lineage);
+    for (long outputFile : lineage.getOutputFiles()) {
+      mOutputFileIndex.put(outputFile, lineage);
     }
     mIdIndex.put(lineage.getId(), lineage);
   }
 
-  /**
-   * Completes an output file.
-   *
-   * @param fileId the file id
-   */
-  public synchronized void completeFile(long fileId) {
-    Preconditions.checkState(mOutputFileIndex.containsKey(fileId));
-    Lineage lineage = mOutputFileIndex.get(fileId);
-    lineage.updateOutputFileState(fileId, LineageFileState.COMPLETED);
-  }
-
   /**
    * Deletes a lineage.
    *
@@ -142,24 +129,11 @@ private void deleteLineage(long lineageId, Set<Long> deleted)
     mLineageDAG.deleteLeaf(toDelete);
     mIdIndex.remove(lineageId);
     deleted.add(lineageId);
-    for (TachyonFile outputFile : toDelete.getOutputFiles()) {
-      mOutputFileIndex.remove(outputFile.getFileId());
+    for (long outputFile : toDelete.getOutputFiles()) {
+      mOutputFileIndex.remove(outputFile);
     }
   }
 
-  /**
-   * Requests an output file as being persisted.
-   *
-   * @param fileId the file id
-   * @throws LineageDoesNotExistException if the lineage does not exist
-   */
-  public synchronized void requestFilePersistence(long fileId) throws LineageDoesNotExistException {
-    LineageDoesNotExistException.check(mOutputFileIndex.containsKey(fileId),
-        ExceptionMessage.LINEAGE_OUTPUT_FILE_NOT_EXIST, fileId);
-    Lineage lineage = mOutputFileIndex.get(fileId);
-    lineage.updateOutputFileState(fileId, LineageFileState.PERSISENCE_REQUESTED);
-  }
-
   /**
    * Gets the lineage.
    *
@@ -200,45 +174,13 @@ public synchronized List<Lineage> getParents(Lineage lineage)
     return mLineageDAG.getParents(lineage);
   }
 
-  /**
-   * Reports an output file as lost.
-   *
-   * @param fileId the file id
-   * @return the lineage containing the output file, null if no lineage outputs the given file
-   * @throws LineageDoesNotExistException
-   */
-  public synchronized Lineage reportLostFile(long fileId) throws LineageDoesNotExistException {
-    Lineage lineage = mOutputFileIndex.get(fileId);
-    LineageDoesNotExistException.check(lineage != null, ExceptionMessage.LINEAGE_DOES_NOT_EXIST,
-        fileId);
-    // TODO(yupeng) push the persisted info to FS master
-    if (lineage.getOutputFileState(fileId) != LineageFileState.PERSISTED) {
-      lineage.updateOutputFileState(fileId, LineageFileState.LOST);
-    }
-    return lineage;
-  }
-
   /**
    * @return the list of all root lineages
    */
   public synchronized List<Lineage> getRootLineages() {
     return mLineageDAG.getRoots();
   }
 
-  /**
-   * Commits a file as persisted
-   *
-   * @param fileId the file id
-   * @throws LineageDoesNotExistException if the lineage does not exist
-   */
-  public synchronized void commitFilePersistence(Long fileId) throws LineageDoesNotExistException {
-    LineageDoesNotExistException.check(mOutputFileIndex.containsKey(fileId),
-        ExceptionMessage.LINEAGE_OUTPUT_FILE_NOT_EXIST, fileId);
-
-    Lineage lineage = mOutputFileIndex.get(fileId);
-    lineage.updateOutputFileState(fileId, LineageFileState.PERSISTED);
-  }
-
   /**
    * Sorts a given set of lineages topologically.
    *
@@ -275,17 +217,4 @@ public synchronized void streamToJournalCheckpoint(JournalOutputStream outputStr
   public boolean hasOutputFile(long fileId) {
     return mOutputFileIndex.containsKey(fileId);
   }
-
-  /**
-   * @param fileId the fild id
-   * @return the lineage state of the given file
-   * @throws LineageDoesNotExistException if the file does not belong to any lineage
-   */
-  public synchronized LineageFileState getLineageFileState(long fileId)
-      throws LineageDoesNotExistException {
-    if (!mOutputFileIndex.containsKey(fileId)) {
-      throw new LineageDoesNotExistException("No lineage has output file " + fileId);
-    }
-    return mOutputFileIndex.get(fileId).getOutputFileState(fileId);
-  }
 }