feat(gui): show region layer on workflow editor (#3753)

Author: aglinxinyuan

amber/src/main/scala/org/apache/amber/engine/architecture/controller/Controller.scala

@@ -21,6 +21,7 @@ package org.apache.amber.engine.architecture.controller
 
 import akka.actor.SupervisorStrategy.Stop
 import akka.actor.{AllForOneStrategy, Props, SupervisorStrategy}
+import org.apache.texera.web.model.websocket.response.RegionUpdateEvent
 import org.apache.amber.config.ApplicationConfig
 import org.apache.amber.core.virtualidentity.ChannelIdentity
 import org.apache.amber.core.workflow.{PhysicalPlan, WorkflowContext}
@@ -42,6 +43,7 @@ import org.apache.amber.engine.common.ambermessage.{
 }
 import org.apache.amber.engine.common.virtualidentity.util.{CLIENT, CONTROLLER, SELF}
 import org.apache.amber.engine.common.{CheckpointState, SerializedState}
+import org.apache.texera.web.SessionState
 
 import scala.concurrent.duration.DurationInt
 
@@ -111,6 +113,16 @@ class Controller(
   override def initState(): Unit = {
     attachRuntimeServicesToCPState()
     cp.workflowScheduler.updateSchedule(physicalPlan)
+
+    val regions: List[List[String]] =
+      cp.workflowScheduler.getSchedule.getRegions.map { region =>
+        region.physicalOps.map(_.id.logicalOpId.id).toList
+      }
+
+    SessionState.getAllSessionStates.foreach { state =>
+      state.send(RegionUpdateEvent(regions))
+    }
+
     val controllerRestoreConf = controllerConfig.stateRestoreConfOpt
     if (controllerRestoreConf.isDefined) {
       globalReplayManager.markRecoveryStatus(CONTROLLER, isRecovering = true)

amber/src/main/scala/org/apache/amber/engine/architecture/controller/WorkflowScheduler.scala

@@ -34,6 +34,8 @@ class WorkflowScheduler(
   var physicalPlan: PhysicalPlan = _
   private var schedule: Schedule = _
 
+  def getSchedule: Schedule = schedule
+
   /**
     * Update the schedule to be executed, based on the given physicalPlan.
     */

amber/src/main/scala/org/apache/amber/engine/architecture/scheduling/Schedule.scala

@@ -22,6 +22,8 @@ package org.apache.amber.engine.architecture.scheduling
 case class Schedule(private val levelSets: Map[Int, Set[Region]]) extends Iterator[Set[Region]] {
   private var currentLevel = levelSets.keys.minOption.getOrElse(0)
 
+  def getRegions: List[Region] = levelSets.values.flatten.toList
+
   override def hasNext: Boolean = levelSets.isDefinedAt(currentLevel)
 
   override def next(): Set[Region] = {

amber/src/main/scala/org/apache/texera/web/model/websocket/response/RegionUpdateEvent.scala

@@ -0,0 +1,24 @@
+/*
+ * 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.texera.web.model.websocket.response
+
+import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent
+
+case class RegionUpdateEvent(regions: List[List[String]]) extends TexeraWebSocketEvent

frontend/package.json

@@ -47,7 +47,9 @@
     "@types/plotly.js-basic-dist-min": "2.12.4",
     "ajv": "8.10.0",
     "backbone": "1.4.1",
+    "concaveman": "2.0.0",
     "content-disposition": "0.5.4",
+    "d3": "6.4.0",
     "dagre": "0.8.5",
     "deep-map": "2.0.0",
     "edit-distance": "1.0.4",
@@ -111,7 +113,9 @@
     "@nrwl/nx-cloud": "19.1.0",
     "@nx/angular": "20.0.3",
     "@types/backbone": "1.4.15",
+    "@types/concaveman": "1.1.6",
     "@types/content-disposition": "0",
+    "@types/d3-shape": "2.1.2",
     "@types/dagre": "0.7.47",
     "@types/file-saver": "2.0.5",
     "@types/graphlib": "2.1.8",

frontend/src/app/workspace/component/menu/menu.component.html

@@ -164,8 +164,14 @@
                   >Grid</label
                 >
               </li>
-            </ul>
-            <ul nz-menu>
+              <li nz-menu-item>
+                <label
+                  nz-checkbox
+                  [(ngModel)]="showRegion"
+                  (ngModelChange)="toggleRegion()"
+                  >Region</label
+                >
+              </li>
               <li nz-menu-item>
                 <label
                   nz-checkbox

frontend/src/app/workspace/component/menu/menu.component.ts

@@ -88,6 +88,7 @@ export class MenuComponent implements OnInit, OnDestroy {
   public isWorkflowModifiable: boolean = false;
   public workflowId?: number;
   public isExportDeactivate: boolean = false;
+  public showRegion: boolean = false;
   public showGrid: boolean = false;
   public showNumWorkers: boolean = false;
   protected readonly DASHBOARD_USER_WORKFLOW = DASHBOARD_USER_WORKFLOW;
@@ -469,6 +470,10 @@ export class MenuComponent implements OnInit, OnDestroy {
     this.workflowActionService.getJointGraphWrapper().mainPaper.setGridSize(this.showGrid ? 2 : 1);
   }
 
+  public toggleRegion(): void {
+    this.workflowActionService.getJointGraphWrapper().mainPaper.el.classList.toggle("hide-region", !this.showRegion);
+  }
+
   /**
    * This method will run the autoLayout function
    *

frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.scss

@@ -25,17 +25,8 @@
   height: 100%;
 }
 
-::ng-deep #workflow-editor .connection-wrap {
-  stroke-width: 0;
-}
-
-::ng-deep #workflow-editor .link-tools .tool-remove path {
-  fill: #d8656a;
-  transform: translate(-12px, -12px);
-}
-
-::ng-deep #workflow-editor .link-tools .tool-remove circle {
-  fill-opacity: 0;
+::ng-deep .hide-region .region {
+  display: none;
 }
 
 ::ng-deep .hide-worker-count .operator-worker-count {

frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts

@@ -41,6 +41,8 @@ import * as _ from "lodash";
 import * as joint from "jointjs";
 import { isDefined } from "../../../common/util/predicate";
 import { GuiConfigService } from "../../../common/service/gui-config.service";
+import { line, curveCatmullRomClosed } from "d3-shape";
+import concaveman from "concaveman";
 
 // jointjs interactive options for enabling and disabling interactivity
 // https://resources.jointjs.com/docs/jointjs/v3.2/joint.html#dia.Paper.prototype.options.interactive
@@ -161,6 +163,7 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy
     this.handlePortHighlightEvent();
     this.registerPortDisplayNameChangeHandler();
     this.handleOperatorStatisticsUpdate();
+    this.handleRegionUpdate();
     this.handleOperatorSuggestionHighlightEvent();
     this.handleElementDelete();
     this.handleElementSelectAll();
@@ -330,6 +333,65 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy
       });
   }
 
+  private handleRegionUpdate(): void {
+    this.editor.classList.add("hide-region");
+    const Region = joint.dia.Element.define(
+      "region",
+      {
+        attrs: {
+          body: {
+            fill: "rgba(255,213,79,0.2)",
+            pointerEvents: "none",
+            class: "region",
+          },
+        },
+      },
+      {
+        markup: [{ tagName: "path", selector: "body" }],
+      }
+    );
+
+    let regionMap: { regionElement: joint.dia.Element; operators: joint.dia.Cell[] }[] = [];
+    // update region elements on execution
+    this.executeWorkflowService
+      .getRegionUpdateStream()
+      .pipe(untilDestroyed(this))
+      .subscribe(regions => {
+        this.paper.model
+          .getCells()
+          .filter(element => element instanceof Region)
+          .forEach(element => element.remove());
+
+        regionMap = regions.map(region => {
+          const element = new Region();
+          const ops = region.map(id => this.paper.getModelById(id));
+          this.paper.model.addCell(element);
+          this.updateRegionElement(element, ops);
+          return { regionElement: element, operators: ops };
+        });
+      });
+
+    this.paper.model.on("change:position", operator => {
+      regionMap
+        .filter(region => region.operators.includes(operator))
+        .forEach(region => this.updateRegionElement(region.regionElement, region.operators));
+    });
+  }
+
+  private updateRegionElement(regionElement: joint.dia.Element, operators: joint.dia.Cell[]) {
+    const points = operators.flatMap(op => {
+      const { x, y, width, height } = op.getBBox(),
+        padding = 15;
+      return [
+        [x - padding, y - padding],
+        [x + width + padding, y - padding],
+        [x - padding, y + height + padding + 10],
+        [x + width + padding, y + height + padding + 10],
+      ];
+    });
+    regionElement.attr("body/d", line().curve(curveCatmullRomClosed)(concaveman(points, 2, 0) as [number, number][]));
+  }
+
   /**
    * Handles restore offset default event by translating jointJS paper
    *  back to original position

frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.ts

@@ -85,6 +85,8 @@ export class ExecuteWorkflowService {
     current: ExecutionStateInfo;
   }>();
 
+  private regionUpdateStream = new Subject<readonly string[][]>();
+
   // TODO: move this to another service, or redesign how this
   //   information is stored on the frontend.
   private assignedWorkerIds: Map<string, readonly string[]> = new Map();
@@ -99,6 +101,9 @@ export class ExecuteWorkflowService {
   ) {
     workflowWebsocketService.websocketEvent().subscribe(event => {
       switch (event.type) {
+        case "RegionUpdateEvent":
+          this.regionUpdateStream.next(event.regions);
+          break;
         case "WorkerAssignmentUpdateEvent":
           this.assignedWorkerIds.set(event.operatorId, event.workerIds);
           break;
@@ -329,6 +334,10 @@ export class ExecuteWorkflowService {
     return this.executionStateStream.asObservable();
   }
 
+  public getRegionUpdateStream(): Observable<readonly string[][]> {
+    return this.regionUpdateStream.asObservable();
+  }
+
   public resetExecutionState(): void {
     this.currentState = {
       state: ExecutionState.Uninitialized,