[NEMO-332] Refactor RunTimePass

client/src/main/java/org/apache/nemo/client/JobLauncher.java

@@ -20,13 +20,15 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
-import org.apache.nemo.common.dag.DAG;
+import org.apache.nemo.common.ir.IRDAG;
+import org.apache.nemo.compiler.backend.nemo.NemoPlanRewriter;
 import org.apache.nemo.conf.JobConf;
 import org.apache.nemo.driver.NemoDriver;
 import org.apache.nemo.runtime.common.comm.ControlMessage;
 import org.apache.nemo.runtime.common.message.MessageEnvironment;
 import org.apache.nemo.runtime.common.message.MessageParameters;
 import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
 import org.apache.nemo.runtime.master.scheduler.Scheduler;
 import org.apache.reef.client.DriverConfiguration;
 import org.apache.reef.client.DriverLauncher;
@@ -206,15 +208,15 @@ public static void shutdown() {
    * @param dag the application DAG.
    */
   // When modifying the signature of this method, see CompilerTestUtil#compileDAG and make corresponding changes
-  public static void launchDAG(final DAG dag) {
+  public static void launchDAG(final IRDAG dag) {
     launchDAG(dag, Collections.emptyMap(), "");
   }
 
   /**
    * @param dag the application DAG.
    * @param jobId job ID.
    */
-  public static void launchDAG(final DAG dag, final String jobId) {
+  public static void launchDAG(final IRDAG dag, final String jobId) {
     launchDAG(dag, Collections.emptyMap(), jobId);
   }
 
@@ -223,7 +225,9 @@ public static void launchDAG(final DAG dag, final String jobId) {
    * @param broadcastVariables broadcast variables (can be empty).
    * @param jobId job ID.
    */
-  public static void launchDAG(final DAG dag, final Map<Serializable, Object> broadcastVariables, final String jobId) {
+  public static void launchDAG(final IRDAG dag,
+                               final Map<Serializable, Object> broadcastVariables,
+                               final String jobId) {
     // launch driver if it hasn't been already
     if (driverReadyLatch == null) {
       try {
@@ -318,6 +322,7 @@ private static Configuration getSchedulerConf(final Configuration jobConf)
     final JavaConfigurationBuilder jcb = Tang.Factory.getTang().newConfigurationBuilder();
     final Class schedulerImpl = ((Class<Scheduler>) Class.forName(classImplName));
     jcb.bindImplementation(Scheduler.class, schedulerImpl);
+    jcb.bindImplementation(PlanRewriter.class, NemoPlanRewriter.class);
     return jcb.build();
   }
 

common/src/main/java/org/apache/nemo/common/HashRange.java

@@ -24,24 +24,21 @@
  * Descriptor for hash range.
  */
 public final class HashRange implements KeyRange<Integer> {
-  private static final HashRange ALL = new HashRange(0, Integer.MAX_VALUE, false);
+  private static final HashRange ALL = new HashRange(0, Integer.MAX_VALUE);
   private final int rangeBeginInclusive;
   private final int rangeEndExclusive;
-  private boolean isSkewed;
 
   /**
    * Private constructor.
    * @param rangeBeginInclusive point at which the hash range starts (inclusive).
    * @param rangeEndExclusive point at which the hash range ends (exclusive).
-   * @param isSkewed whether or not the range is skewed
    */
-  private HashRange(final int rangeBeginInclusive, final int rangeEndExclusive, final boolean isSkewed) {
+  private HashRange(final int rangeBeginInclusive, final int rangeEndExclusive) {
     if (rangeBeginInclusive < 0 || rangeEndExclusive < 0) {
       throw new RuntimeException("Each boundary value of the range have to be non-negative.");
     }
     this.rangeBeginInclusive = rangeBeginInclusive;
     this.rangeEndExclusive = rangeEndExclusive;
-    this.isSkewed = isSkewed;
   }
 
   /**
@@ -54,11 +51,10 @@ public static HashRange all() {
   /**
    * @param rangeStartInclusive the start of the range (inclusive)
    * @param rangeEndExclusive   the end of the range (exclusive)
-   * @param isSkewed  whether or not the range is skewed
    * @return A hash range descriptor representing [{@code rangeBeginInclusive}, {@code rangeEndExclusive})
    */
-  public static HashRange of(final int rangeStartInclusive, final int rangeEndExclusive, final boolean isSkewed) {
-    return new HashRange(rangeStartInclusive, rangeEndExclusive, isSkewed);
+  public static HashRange of(final int rangeStartInclusive, final int rangeEndExclusive) {
+    return new HashRange(rangeStartInclusive, rangeEndExclusive);
   }
 
   /**
@@ -114,8 +110,7 @@ public boolean equals(final Object o) {
     }
     final HashRange hashRange = (HashRange) o;
     if (rangeBeginInclusive != hashRange.rangeBeginInclusive
-        || rangeEndExclusive != hashRange.rangeEndExclusive
-        || isSkewed != hashRange.isSkewed) {
+      || rangeEndExclusive != hashRange.rangeEndExclusive) {
       return false;
     }
     return true;
@@ -127,16 +122,8 @@ public boolean equals(final Object o) {
   @Override
   public int hashCode() {
     return Arrays.hashCode(new Object[] {
-        rangeBeginInclusive,
-        rangeEndExclusive,
-        isSkewed,
+      rangeBeginInclusive,
+      rangeEndExclusive,
     });
   }
-
-  /**
-   * @return whether or not the range is skewed.
-   */
-  public boolean isSkewed() {
-    return isSkewed;
-  }
 }

common/src/main/java/org/apache/nemo/common/KeyRange.java

@@ -26,7 +26,7 @@
  */
 public interface KeyRange<K extends Serializable> extends Serializable {
 
- /**
+  /**
    * @return whether this instance represents the entire range or not.
    */
   boolean isAll();

common/src/main/java/org/apache/nemo/common/coder/PairDecoderFactory.java

@@ -63,6 +63,17 @@ public Decoder<Pair<A, B>> create(final InputStream inputStream) throws IOExcept
     return new PairDecoder<>(inputStream, leftDecoderFactory, rightDecoderFactory);
   }
 
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder();
+    sb.append("Pair(");
+    sb.append(leftDecoderFactory.toString());
+    sb.append(", ");
+    sb.append(rightDecoderFactory.toString());
+    sb.append(")");
+    return sb.toString();
+  }
+
   /**
    * PairDecoder.
    * @param <T1> type for the left coder.

common/src/main/java/org/apache/nemo/common/coder/PairEncoderFactory.java

@@ -62,6 +62,17 @@ public Encoder<Pair<A, B>> create(final OutputStream outputStream) throws IOExce
     return new PairEncoder<>(outputStream, leftEncoderFactory, rightEncoderFactory);
   }
 
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder();
+    sb.append("Pair(");
+    sb.append(leftEncoderFactory.toString());
+    sb.append(", ");
+    sb.append(rightEncoderFactory.toString());
+    sb.append(")");
+    return sb.toString();
+  }
+
   /**
    * PairEncoder.
    * @param <T1> type for the left coder.