feat(operator): sleep operator

core/gui/src/app/workspace/component/left-panel/operator-menu/operator-menu.component.ts

@@ -78,10 +78,12 @@ export class OperatorMenuComponent {
         );
         this.groupNames = operatorMetadata.groups;
         ops.forEach(x => {
-          const group = x.additionalMetadata.operatorGroupName;
-          const list = this.opList.get(group) || [];
-          list.push(x);
-          this.opList.set(group, list);
+          if (x.operatorType !== "Sleep") {
+            const group = x.additionalMetadata.operatorGroupName;
+            const list = this.opList.get(group) || [];
+            list.push(x);
+            this.opList.set(group, list);
+          }
         });
         this.opList.forEach(value => {
           value.sort((a, b) => a.operatorType.localeCompare(b.operatorType));

core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/LogicalOp.scala

@@ -63,6 +63,7 @@ import edu.uci.ics.amber.operator.randomksampling.RandomKSamplingOpDesc
 import edu.uci.ics.amber.operator.regex.RegexOpDesc
 import edu.uci.ics.amber.operator.reservoirsampling.ReservoirSamplingOpDesc
 import edu.uci.ics.amber.operator.sklearn._
+import edu.uci.ics.amber.operator.sleep.SleepOpDesc
 import edu.uci.ics.amber.operator.sklearn.training.{
   SklearnTrainingAdaptiveBoostingOpDesc,
   SklearnTrainingBaggingOpDesc,
@@ -215,6 +216,7 @@ trait StateTransferFunc
     new Type(value = classOf[AsterixDBSourceOpDesc], name = "AsterixDBSource"),
     new Type(value = classOf[TypeCastingOpDesc], name = "TypeCasting"),
     new Type(value = classOf[LimitOpDesc], name = "Limit"),
+    new Type(value = classOf[SleepOpDesc], name = "Sleep"),
     new Type(value = classOf[RandomKSamplingOpDesc], name = "RandomKSampling"),
     new Type(value = classOf[ReservoirSamplingOpDesc], name = "ReservoirSampling"),
     new Type(value = classOf[HashJoinOpDesc[String]], name = "HashJoin"),

core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sleep/SleepOpDesc.scala

@@ -0,0 +1,66 @@
+/*
+ * 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 edu.uci.ics.amber.operator.sleep
+
+import com.fasterxml.jackson.annotation.JsonProperty
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import edu.uci.ics.amber.core.executor.OpExecWithClassName
+import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PhysicalOp}
+import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import edu.uci.ics.amber.util.JSONUtils.objectMapper
+
+class SleepOpDesc extends LogicalOp {
+
+  @JsonProperty(required = true)
+  @JsonSchemaTitle("Sleep Time (seconds)")
+  var sleepTime: Int = _
+
+  override def getPhysicalOp(
+      workflowId: WorkflowIdentity,
+      executionId: ExecutionIdentity
+  ): PhysicalOp = {
+    PhysicalOp
+      .oneToOnePhysicalOp(
+        workflowId,
+        executionId,
+        operatorIdentifier,
+        OpExecWithClassName(
+          "edu.uci.ics.amber.operator.sleep.SleepOpExec",
+          objectMapper.writeValueAsString(this)
+        )
+      )
+      .withInputPorts(operatorInfo.inputPorts)
+      .withOutputPorts(operatorInfo.outputPorts)
+      .withParallelizable(false)
+      .withSuggestedWorkerNum(1)
+
+  }
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      "Sleep",
+      "Sleep n seconds between each tuple",
+      OperatorGroupConstants.CONTROL_GROUP,
+      inputPorts = List(InputPort()),
+      outputPorts = List(OutputPort())
+    )
+}

core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sleep/SleepOpExec.scala

@@ -0,0 +1,33 @@
+/*
+ * 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 edu.uci.ics.amber.operator.sleep
+
+import edu.uci.ics.amber.core.executor.OperatorExecutor
+import edu.uci.ics.amber.core.tuple.{Tuple, TupleLike}
+import edu.uci.ics.amber.util.JSONUtils.objectMapper
+
+class SleepOpExec(descString: String) extends OperatorExecutor {
+  private val desc: SleepOpDesc = objectMapper.readValue(descString, classOf[SleepOpDesc])
+
+  override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = {
+    Thread.sleep(1000 * desc.sleepTime)
+    Iterator(tuple)
+  }
+}