Added addLineage in lineage master

Alluxio · Sep 17, 2015 · 7663fc3 · 7663fc3
1 parent 184ce13
commit 7663fc3
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 4 deletions.
diff --git a/servers/src/main/java/tachyon/master/lineage/LineageMaster.java b/servers/src/main/java/tachyon/master/lineage/LineageMaster.java
@@ -113,8 +113,7 @@ public void streamToJournalCheckpoint(JournalOutputStream outputStream) throws I
   }
 
   public long createLineage(List<TachyonFile> inputFiles, List<TachyonFile> outputFiles, Job job) {
-    // TODO create lineage
+    return mLineageStore.addLineage(inputFiles, outputFiles, job);
-    return -1;
   }
 
   public boolean deleteLineage(long lineageId) {

diff --git a/servers/src/main/java/tachyon/master/lineage/meta/Lineage.java b/servers/src/main/java/tachyon/master/lineage/meta/Lineage.java
@@ -65,4 +65,8 @@ public LineageState getState() {
   public void setState(LineageState newState) {
     mState = newState;
   }
+
+  public long getId() {
+    return mId;
+  }
 }
diff --git a/servers/src/main/java/tachyon/master/lineage/meta/LineageStore.java b/servers/src/main/java/tachyon/master/lineage/meta/LineageStore.java
@@ -14,12 +14,15 @@
  */
 package tachyon.master.lineage.meta;
 
+import java.util.List;
 import java.util.Map;
 
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
 import tachyon.client.file.TachyonFile;
 import tachyon.dag.DAG;
+import tachyon.job.Job;
 
 /**
  * A store of lineages.
@@ -28,17 +31,42 @@ public final class LineageStore {
   private DAG<Lineage> mLineageDAG;
 
   /** Indices for lineages */
-  private Map<TachyonFile, Lineage> mInputFileIndex;
+  /** Index of the output files of lineage to lineage */
   private Map<TachyonFile, Lineage> mOutputFileIndex;
   private Map<LineageState, Lineage> mStateIndex;
+  private Map<Long, Lineage> mIdIndex;
 
   public LineageStore() {
     mLineageDAG = new DAG<Lineage>();
-    mInputFileIndex = Maps.newHashMap();
     mOutputFileIndex = Maps.newHashMap();
     mStateIndex = Maps.newHashMap();
+    mIdIndex = Maps.newHashMap();
   }
 
+  public long addLineage(List<TachyonFile> inputFiles, List<TachyonFile> outputFiles, Job job) {
+    Lineage lineage  = new Lineage(inputFiles, outputFiles, job);
 
+    List<Lineage> parentLineages = Lists.newArrayList();
+    for(TachyonFile inputFile:inputFiles) {
+      if(mOutputFileIndex.containsKey(inputFile)) {
+        parentLineages.add(mOutputFileIndex.get(inputFile));
+      }
+    }
 
+    mLineageDAG.add(lineage, parentLineages);
+
+    // update index
+    for (TachyonFile outputFile : outputFiles) {
+      mOutputFileIndex.put(outputFile, lineage);
+    }
+    mStateIndex.put(lineage.getState(), lineage);
+    mIdIndex.put(lineage.getId(), lineage);
+
+    return lineage.getId();
+  }
+
+  private void updateState(Lineage lineage, LineageState newState) {
+    lineage.setState(newState);
+    mStateIndex.put(newState, lineage);
+  }
 }