Implement Lambda Pass, LambdaPolicy and ResourceLambdaProperty

common/src/main/java/org/apache/nemo/common/ir/vertex/executionproperty/ResourceLambdaProperty.java

@@ -0,0 +1,55 @@
+/*
+ * 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
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.nemo.common.ir.vertex.executionproperty;
+
+import org.apache.nemo.common.ir.executionproperty.VertexExecutionProperty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public final class ResourceLambdaProperty extends VertexExecutionProperty<ResourceLambdaProperty.Value>{
+  private static final Logger LOG = LoggerFactory.getLogger(ResourceLambdaProperty.class.getName());
+  /**
+   * Constructor.
+   *
+   * @param value value of the execution property.
+   */
+  private ResourceLambdaProperty(final ResourceLambdaProperty.Value value) {
+    super(value);
+    LOG.info("LambdaProperty Instantiated");
+  }
+
+  /**
+   * Static method exposing the constructor.
+   *
+   * @param value value of the new execution property.
+   * @return the newly created execution property.
+   */
+  public static ResourceLambdaProperty of(final ResourceLambdaProperty.Value value) {
+    return new ResourceLambdaProperty(value);
+  }
+
+  /**
+   * Possible values of DataStore ExecutionProperty.
+   */
+  public enum Value {
+    On,
+    Off,
+  }
+}

compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/pass/compiletime/annotating/LambdaPass.java

@@ -0,0 +1,54 @@
+/*
+ * 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
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.nemo.compiler.optimizer.pass.compiletime.annotating;
+
+import org.apache.nemo.common.ir.IRDAG;
+import org.apache.nemo.common.ir.vertex.executionproperty.ResourceLambdaProperty;
+import org.apache.nemo.common.ir.executionproperty.ExecutionProperty;
+import org.apache.nemo.compiler.optimizer.pass.compiletime.CompileTimePass;
+import org.apache.nemo.compiler.optimizer.pass.compiletime.annotating.AnnotatingPass;
+import org.apache.nemo.common.ir.edge.executionproperty.DataStoreProperty;
+
+import org.apache.nemo.compiler.optimizer.policy.LambdaPolicy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/*
+ * Lambda Policy
+ * Maintainer: Gao Zhiyuan
+ * Description: A part of lambda executor, assigning LambdaResourceProperty
+ */
+
+@Annotates(ResourceLambdaProperty.class)
+public final class LambdaPass extends AnnotatingPass {
+  private static final Logger LOG = LoggerFactory.getLogger(LambdaPolicy.class.getName());
+
+  public LambdaPass() {
+    super(LambdaPass.class);
+    LOG.info("LambdaPass initialized");
+  }
+
+  @Override
+  public IRDAG apply(final IRDAG dag) {
+    dag.getVertices().forEach(vertex -> {
+      vertex.setPropertyPermanently(ResourceLambdaProperty.of(ResourceLambdaProperty.Value.On));
+    });
+    return dag;
+  }
+}

compiler/optimizer/src/main/java/org/apache/nemo/compiler/optimizer/policy/LambdaPolicy.java

@@ -1,7 +1,25 @@
+/*
+ * 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
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.nemo.compiler.optimizer.policy;
 
 import org.apache.nemo.common.ir.IRDAG;
-import org.apache.nemo.compiler.optimizer.pass.compiletime.annotating.PipeTransferForAllEdgesPass;
+import org.apache.nemo.compiler.optimizer.pass.compiletime.annotating.LambdaPass;
 import org.apache.nemo.compiler.optimizer.pass.compiletime.composite.DefaultCompositePass;
 import org.apache.nemo.compiler.optimizer.pass.runtime.Message;
 
@@ -24,7 +42,7 @@ public LambdaPolicy() {
     LOG.info("LambdaPolicy!!!!!");
     final PolicyBuilder builder = new PolicyBuilder();
     builder.registerCompileTimePass(new DefaultCompositePass());
-    builder.registerCompileTimePass(new PipeTransferForAllEdgesPass());
+    builder.registerCompileTimePass(new LambdaPass());
     this.policy = builder.build();
   }