diff --git a/core/amber/src/main/scala/edu/uci/ics/amber/clustering/ClusterListener.scala b/core/amber/src/main/scala/edu/uci/ics/amber/clustering/ClusterListener.scala
index 9c639946d3c..f60c0c09dc7 100644
--- a/core/amber/src/main/scala/edu/uci/ics/amber/clustering/ClusterListener.scala
+++ b/core/amber/src/main/scala/edu/uci/ics/amber/clustering/ClusterListener.scala
@@ -27,6 +27,8 @@ object ClusterListener {
   final case class GetAvailableNodeAddresses()
 
   var numWorkerNodesInCluster = 0
+
+  var currentAddresses: Set[Address] = Set.empty
 }
 
 class ClusterListener extends Actor with AmberLogging {
@@ -117,9 +119,13 @@ class ClusterListener extends Actor with AmberLogging {
       case other => //skip
     }
 
+    ClusterListener.currentAddresses = getAllAddress.toSet
     numWorkerNodesInCluster = getAllAddress.size
+
+    val addressesStr = ClusterListener.currentAddresses.map(_.toString).toSeq
+
     SessionState.getAllSessionStates.foreach { state =>
-      state.send(ClusterStatusUpdateEvent(numWorkerNodesInCluster))
+      state.send(ClusterStatusUpdateEvent(numWorkerNodesInCluster, addressesStr))
     }
 
     logger.info(
diff --git a/core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/common/ExecutorDeployment.scala b/core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/common/ExecutorDeployment.scala
index e0a9d53eacd..f76c4dcded3 100644
--- a/core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/common/ExecutorDeployment.scala
+++ b/core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/common/ExecutorDeployment.scala
@@ -1,8 +1,13 @@
 package edu.uci.ics.amber.engine.architecture.common
 
-import akka.actor.{Address, Deploy}
+import akka.actor.{Address, AddressFromURIString, Deploy}
 import akka.remote.RemoteScope
-import edu.uci.ics.amber.core.workflow.{PhysicalOp, PreferController, RoundRobinPreference}
+import edu.uci.ics.amber.core.workflow.{
+  GoToSpecificNode,
+  PhysicalOp,
+  PreferController,
+  RoundRobinPreference
+}
 import edu.uci.ics.amber.engine.architecture.controller.execution.OperatorExecution
 import edu.uci.ics.amber.engine.architecture.deploysemantics.AddressInfo
 import edu.uci.ics.amber.engine.architecture.pythonworker.PythonWorkflowWorker
@@ -38,6 +43,16 @@ object ExecutorDeployment {
       val preferredAddress: Address = locationPreference match {
         case PreferController =>
           addressInfo.controllerAddress
+        case node: GoToSpecificNode =>
+          val targetAddress = AddressFromURIString(node.nodeAddr)
+          addressInfo.allAddresses.find(addr => addr == targetAddress) match {
+            case Some(address) => address
+            case None =>
+              throw new IllegalStateException(
+                s"Designated node address '${node.nodeAddr}' not found among available addresses: " +
+                  addressInfo.allAddresses.map(_.host.getOrElse("None")).mkString(", ")
+              )
+          }
         case RoundRobinPreference =>
           assert(
             addressInfo.allAddresses.nonEmpty,
diff --git a/core/amber/src/main/scala/edu/uci/ics/amber/engine/common/AmberRuntime.scala b/core/amber/src/main/scala/edu/uci/ics/amber/engine/common/AmberRuntime.scala
index 8b47e3ead99..671c07ab601 100644
--- a/core/amber/src/main/scala/edu/uci/ics/amber/engine/common/AmberRuntime.scala
+++ b/core/amber/src/main/scala/edu/uci/ics/amber/engine/common/AmberRuntime.scala
@@ -8,7 +8,7 @@ import edu.uci.ics.amber.clustering.ClusterListener
 import edu.uci.ics.amber.engine.architecture.messaginglayer.DeadLetterMonitorActor
 
 import java.io.{BufferedReader, InputStreamReader}
-import java.net.URL
+import java.net.{InetAddress, URL}
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
 
@@ -46,7 +46,9 @@ object AmberRuntime {
     try {
       val query = new URL("http://checkip.amazonaws.com")
       val in = new BufferedReader(new InputStreamReader(query.openStream()))
-      in.readLine()
+      val ip = in.readLine()
+      val localIp = InetAddress.getLocalHost().getHostAddress()
+      ip
     } catch {
       case e: Exception => throw e
     }
@@ -56,18 +58,33 @@ object AmberRuntime {
     var localIpAddress = "localhost"
     if (clusterMode) {
       localIpAddress = getNodeIpAddress
-    }
 
-    val masterConfig = ConfigFactory
-      .parseString(s"""
+      val localPrivateIdAddress = InetAddress.getLocalHost().getHostAddress()
+
+      val masterConfig = ConfigFactory
+        .parseString(s"""
+              akka.remote.artery.canonical.port = 2552
+              akka.remote.artery.canonical.hostname = $localIpAddress
+              akka.remote.artery.bind.hostname = $localPrivateIdAddress
+              akka.remote.artery.bind.port = 2552
+              akka.cluster.seed-nodes = [ "akka://Amber@$localIpAddress:2552" ]
+              """)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
+      createAmberSystem(masterConfig)
+    } else {
+      val masterConfig = ConfigFactory
+        .parseString(s"""
         akka.remote.artery.canonical.port = 2552
         akka.remote.artery.canonical.hostname = $localIpAddress
         akka.cluster.seed-nodes = [ "akka://Amber@$localIpAddress:2552" ]
         """)
-      .withFallback(akkaConfig)
-      .resolve()
-    AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
-    createAmberSystem(masterConfig)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
+      createAmberSystem(masterConfig)
+    }
   }
 
   def akkaConfig: Config =
@@ -79,18 +96,35 @@ object AmberRuntime {
     val addr = mainNodeAddress.getOrElse("localhost")
     var localIpAddress = "localhost"
     if (mainNodeAddress.isDefined) {
+
       localIpAddress = getNodeIpAddress
-    }
-    val workerConfig = ConfigFactory
-      .parseString(s"""
+
+      val localPrivateIdAddress = InetAddress.getLocalHost().getHostAddress()
+
+      val workerConfig = ConfigFactory
+        .parseString(s"""
+              akka.remote.artery.canonical.hostname = $localIpAddress
+              akka.remote.artery.canonical.port = 0
+              akka.remote.artery.bind.hostname = $localPrivateIdAddress
+              akka.remote.artery.bind.port = 0
+              akka.cluster.seed-nodes = [ "akka://Amber@$addr:2552" ]
+              """)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(addr)
+      createAmberSystem(workerConfig)
+    } else {
+      val workerConfig = ConfigFactory
+        .parseString(s"""
         akka.remote.artery.canonical.hostname = $localIpAddress
         akka.remote.artery.canonical.port = 0
         akka.cluster.seed-nodes = [ "akka://Amber@$addr:2552" ]
         """)
-      .withFallback(akkaConfig)
-      .resolve()
-    AmberConfig.masterNodeAddr = createMasterAddress(addr)
-    createAmberSystem(workerConfig)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(addr)
+      createAmberSystem(workerConfig)
+    }
   }
 
   private def createAmberSystem(actorSystemConf: Config): Unit = {
diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ClusterStatusUpdateEvent.scala b/core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ClusterStatusUpdateEvent.scala
index bda2d246f4c..6dc7623058e 100644
--- a/core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ClusterStatusUpdateEvent.scala
+++ b/core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ClusterStatusUpdateEvent.scala
@@ -2,4 +2,5 @@ package edu.uci.ics.texera.web.model.websocket.response
 
 import edu.uci.ics.texera.web.model.websocket.event.TexeraWebSocketEvent
 
-case class ClusterStatusUpdateEvent(numWorkers: Int) extends TexeraWebSocketEvent
+case class ClusterStatusUpdateEvent(numWorkers: Int, addresses: Seq[String] = Seq.empty)
+    extends TexeraWebSocketEvent
diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala
index 864f84784f0..12b308b02c1 100644
--- a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala
+++ b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala
@@ -36,7 +36,10 @@ class WorkflowWebsocketResource extends LazyLogging {
     val workflowState =
       WorkflowService.getOrCreate(WorkflowIdentity(wid))
     sessionState.subscribe(workflowState)
-    sessionState.send(ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster))
+    val addressesStr = ClusterListener.currentAddresses.map(_.toString).toSeq
+    sessionState.send(
+      ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster, addressesStr)
+    )
     logger.info("connection open")
   }
 
diff --git a/core/gui/src/app/app.module.ts b/core/gui/src/app/app.module.ts
index ab3e5c29ddb..64c46191f8e 100644
--- a/core/gui/src/app/app.module.ts
+++ b/core/gui/src/app/app.module.ts
@@ -81,6 +81,7 @@ import { AdminGuardService } from "./dashboard/service/admin/guard/admin-guard.s
 import { ContextMenuComponent } from "./workspace/component/workflow-editor/context-menu/context-menu/context-menu.component";
 import { CoeditorUserIconComponent } from "./workspace/component/menu/coeditor-user-icon/coeditor-user-icon.component";
 import { InputAutoCompleteComponent } from "./workspace/component/input-autocomplete/input-autocomplete.component";
+import { InputNodeAddressComponent } from "./workspace/component/input-node-address/input-node-address.component";
 import { CollabWrapperComponent } from "./common/formly/collab-wrapper/collab-wrapper/collab-wrapper.component";
 import { NzSwitchModule } from "ng-zorro-antd/switch";
 import { AboutComponent } from "./hub/component/about/about.component";
@@ -214,6 +215,7 @@ registerLocaleData(en);
     ContextMenuComponent,
     CoeditorUserIconComponent,
     InputAutoCompleteComponent,
+    InputNodeAddressComponent,
     FileSelectionComponent,
     CollabWrapperComponent,
     AboutComponent,
diff --git a/core/gui/src/app/common/formly/formly-config.ts b/core/gui/src/app/common/formly/formly-config.ts
index bbd4be968eb..27f62c79ef0 100644
--- a/core/gui/src/app/common/formly/formly-config.ts
+++ b/core/gui/src/app/common/formly/formly-config.ts
@@ -7,6 +7,7 @@ import { CodeareaCustomTemplateComponent } from "../../workspace/component/codea
 import { PresetWrapperComponent } from "./preset-wrapper/preset-wrapper.component";
 import { InputAutoCompleteComponent } from "../../workspace/component/input-autocomplete/input-autocomplete.component";
 import { CollabWrapperComponent } from "./collab-wrapper/collab-wrapper/collab-wrapper.component";
+import { InputNodeAddressComponent } from "../../workspace/component/input-node-address/input-node-address.component";
 
 /**
  * Configuration for using Json Schema with Formly.
@@ -57,6 +58,7 @@ export const TEXERA_FORMLY_CONFIG = {
     { name: "multischema", component: MultiSchemaTypeComponent },
     { name: "codearea", component: CodeareaCustomTemplateComponent },
     { name: "inputautocomplete", component: InputAutoCompleteComponent, wrappers: ["form-field"] },
+    { name: "inputnodeaddress", component: InputNodeAddressComponent, wrappers: ["form-field"] },
   ],
   wrappers: [
     { name: "preset-wrapper", component: PresetWrapperComponent },
diff --git a/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.html b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.html
new file mode 100644
index 00000000000..810254385c8
--- /dev/null
+++ b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.html
@@ -0,0 +1,10 @@
+<nz-select
+  [formControl]="formControl"
+  [nzPlaceHolder]="'Select Node Address'"
+  nzAllowClear>
+  <nz-option
+    *ngFor="let addr of nodeAddresses"
+    [nzValue]="addr"
+    [nzLabel]="addr">
+  </nz-option>
+</nz-select>
diff --git a/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.scss b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.scss
new file mode 100644
index 00000000000..003f7e1a20f
--- /dev/null
+++ b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.scss
@@ -0,0 +1,7 @@
+:host {
+  display: block;
+}
+
+nz-select {
+  width: 100%;
+}
diff --git a/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.ts b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.ts
new file mode 100644
index 00000000000..c8f6ec7bde5
--- /dev/null
+++ b/core/gui/src/app/workspace/component/input-node-address/input-node-address.component.ts
@@ -0,0 +1,40 @@
+import { Component, OnInit, OnDestroy } from "@angular/core";
+import { FieldType, FieldTypeConfig } from "@ngx-formly/core";
+import { Subject, Subscription } from "rxjs";
+import { WorkflowWebsocketService } from "../../service/workflow-websocket/workflow-websocket.service";
+import { takeUntil } from "rxjs/operators";
+
+@Component({
+  selector: "texera-input-node-address",
+  templateUrl: "./input-node-address.component.html",
+  styleUrls: ["./input-node-address.component.scss"],
+})
+export class InputNodeAddressComponent extends FieldType<FieldTypeConfig> implements OnInit, OnDestroy {
+  public nodeAddresses: string[] = [];
+
+  private wsSubscription!: Subscription;
+
+  private componentDestroy = new Subject<void>();
+
+  constructor(private workflowWebsocketService: WorkflowWebsocketService) {
+    super();
+  }
+
+  ngOnInit(): void {
+    this.nodeAddresses = this.workflowWebsocketService.workerAddresses;
+
+    this.wsSubscription = this.workflowWebsocketService
+      .websocketEvent()
+      .pipe(takeUntil(this.componentDestroy))
+      .subscribe(event => {
+        if (event.type === "ClusterStatusUpdateEvent") {
+          this.nodeAddresses = event.addresses;
+        }
+      });
+  }
+
+  ngOnDestroy(): void {
+    this.componentDestroy.next();
+    this.componentDestroy.complete();
+  }
+}
diff --git a/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.html b/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.html
index 4ddd0a56809..831a06abf0c 100644
--- a/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.html
+++ b/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.html
@@ -69,23 +69,53 @@
   (focusout)="disconnectQuillFromText()"
   (keyup.enter)="disconnectQuillFromText()"></div>
 
-<form
-  nz-form
-  [nzLayout]="'vertical'"
-  *ngIf="formlyFields && formlyFormGroup"
-  [formGroup]="formlyFormGroup"
-  class="property-editor-form">
-  <formly-form
-    (modelChange)="onFormChanges($event)"
-    [fields]="formlyFields"
-    [form]="formlyFormGroup"
-    [model]="formData"
-    [options]="formlyOptions">
-  </formly-form>
-  <texera-type-casting-display
-    *ngIf="isTypeCasting"
-    currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
-</form>
+<nz-tabset>
+  <nz-tab
+    nzTitle="Basic Settings"
+    *ngIf="basicFields.length > 0">
+    <form
+      nz-form
+      nzLayout="vertical"
+      *ngIf="formlyFields && formlyFormGroup"
+      [formGroup]="formlyFormGroup"
+      class="property-editor-form">
+      <formly-form
+        (modelChange)="onFormChanges($event)"
+        [fields]="basicFields"
+        [form]="formlyFormGroup"
+        [model]="formData"
+        [options]="formlyOptions">
+      </formly-form>
+
+      <texera-type-casting-display
+        *ngIf="isTypeCasting"
+        currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
+    </form>
+  </nz-tab>
+
+  <nz-tab
+    nzTitle="Advanced Settings"
+    *ngIf="advancedFields.length > 0">
+    <form
+      nz-form
+      nzLayout="vertical"
+      *ngIf="formlyFields && formlyFormGroup"
+      [formGroup]="formlyFormGroup"
+      class="property-editor-form">
+      <formly-form
+        (modelChange)="onFormChanges($event)"
+        [fields]="advancedFields"
+        [form]="formlyFormGroup"
+        [model]="formData"
+        [options]="formlyOptions">
+      </formly-form>
+
+      <texera-type-casting-display
+        *ngIf="isTypeCasting"
+        currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
+    </form>
+  </nz-tab>
+</nz-tabset>
 
 <button
   (click)="allowModifyOperatorLogic()"
diff --git a/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts b/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
index 834cad397c2..d1b0db9374e 100644
--- a/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
+++ b/core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
@@ -104,6 +104,9 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
   formlyFields: FormlyFieldConfig[] | undefined;
   formTitle: string | undefined;
 
+  public basicFields: FormlyFieldConfig[] = [];
+  public advancedFields: FormlyFieldConfig[] = [];
+
   // The field name and its css style to be overridden, e.g., for showing the diff between two workflows.
   // example: new Map([
   //     ["attribute", "outline: 3px solid green; transition: 0.3s ease-in-out outline;"],
@@ -379,6 +382,20 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
         },
       };
 
+      if (mappedField.key == "nodeAddr") {
+        mappedField.props = {
+          ...mappedField.props,
+          tab: "AdvancedSettings",
+        };
+      }
+
+      if (mappedField.key == "autoSelectNodeAddress") {
+        mappedField.props = {
+          ...mappedField.props,
+          tab: "AdvancedSettings",
+        };
+      }
+
       // Disable dummy operator for user
       if (mappedField.key === "dummyOperator") {
         mappedField.expressions = {
@@ -432,6 +449,11 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
         mappedField.type = "inputautocomplete";
       }
 
+      // if the title is nodeAddr, then change it to custom inputnodeaddress input template
+      if (mappedField.key == "nodeAddr") {
+        mappedField.type = "inputnodeaddress";
+      }
+
       // if the title is python script (for Python UDF), then make this field a custom template 'codearea'
       if (mapSource?.description?.toLowerCase() === "input your code here") {
         if (mappedField.type) {
@@ -669,6 +691,10 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
     // not return field.fieldGroup directly because
     // doing so the validator in the field will not be triggered
     this.formlyFields = [field];
+
+    const fieldGroup = this.formlyFields?.[0]?.fieldGroup ?? [];
+    this.basicFields = fieldGroup.filter(f => f.props?.tab !== "AdvancedSettings");
+    this.advancedFields = fieldGroup.filter(f => f.props?.tab === "AdvancedSettings");
   }
 
   allowModifyOperatorLogic(): void {
diff --git a/core/gui/src/app/workspace/service/workflow-websocket/workflow-websocket.service.ts b/core/gui/src/app/workspace/service/workflow-websocket/workflow-websocket.service.ts
index c6c5da39f04..0423ba79dac 100644
--- a/core/gui/src/app/workspace/service/workflow-websocket/workflow-websocket.service.ts
+++ b/core/gui/src/app/workspace/service/workflow-websocket/workflow-websocket.service.ts
@@ -26,6 +26,9 @@ export class WorkflowWebsocketService {
 
   public isConnected: boolean = false;
   public numWorkers: number = -1;
+
+  public workerAddresses: string[] = [];
+
   private connectedWid: number = 0;
   private connectedCuid?: number;
 
@@ -109,6 +112,7 @@ export class WorkflowWebsocketService {
     this.websocketEvent().subscribe(evt => {
       if (evt.type === "ClusterStatusUpdateEvent") {
         this.numWorkers = evt.numWorkers;
+        this.workerAddresses = evt.addresses;
       }
       this.isConnected = true;
       this.connectedWid = wId;
diff --git a/core/gui/src/app/workspace/types/workflow-websocket.interface.ts b/core/gui/src/app/workspace/types/workflow-websocket.interface.ts
index 7efabf28557..dae7fc0a535 100644
--- a/core/gui/src/app/workspace/types/workflow-websocket.interface.ts
+++ b/core/gui/src/app/workspace/types/workflow-websocket.interface.ts
@@ -152,6 +152,7 @@ export type ExecutionDurationUpdateEvent = Readonly<{
 
 export type ClusterStatusUpdateEvent = Readonly<{
   numWorkers: number;
+  addresses: string[];
 }>;
 
 export type ModifyLogicResponse = Readonly<{
diff --git a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/workflow/LocationPreference.scala b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/workflow/LocationPreference.scala
index 7a44ea4eb0c..c1bf048dbe9 100644
--- a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/workflow/LocationPreference.scala
+++ b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/workflow/LocationPreference.scala
@@ -12,3 +12,5 @@ object PreferController extends LocationPreference
 // - Operator A: Worker 1 -> Node 1, Worker 2 -> Node 2, Worker 3 -> Node 3
 // - Operator B: Worker 1 -> Node 1, Worker 2 -> Node 2
 object RoundRobinPreference extends LocationPreference
+
+case class GoToSpecificNode(nodeAddr: String) extends LocationPreference
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/DesignatedLocationConfigurable.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/DesignatedLocationConfigurable.scala
new file mode 100644
index 00000000000..57158b05b91
--- /dev/null
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/DesignatedLocationConfigurable.scala
@@ -0,0 +1,51 @@
+package edu.uci.ics.amber.operator
+
+import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.{
+  JsonSchemaInject,
+  JsonSchemaString,
+  JsonSchemaTitle
+}
+import edu.uci.ics.amber.operator.metadata.annotations.HideAnnotation
+import edu.uci.ics.amber.core.workflow.{GoToSpecificNode, PhysicalOp}
+
+/**
+  * Provides configuration for users to designate a specific node location for executing operations.
+  * When `autoSelectNodeAddress` is false and `nodeAddr` is not null, the specified node address will be used.
+  */
+trait DesignatedLocationConfigurable {
+
+  @JsonProperty(defaultValue = "true")
+  @JsonSchemaTitle("Auto Select Node Address")
+  @JsonPropertyDescription(
+    "Set to false to manually specify the node address instead of using the default RoundRobin strategy"
+  )
+  var autoSelectNodeAddress: Boolean = true
+
+  @JsonProperty(required = false)
+  @JsonSchemaTitle("Designated Node Address")
+  @JsonPropertyDescription("The node address to use when autoSelectNodeAddress is disabled")
+  @JsonSchemaInject(
+    strings = Array(
+      // This configuration hides `nodeAddr` when `autoSelectNodeAddress` equals "true".
+      new JsonSchemaString(path = HideAnnotation.hideTarget, value = "autoSelectNodeAddress"),
+      new JsonSchemaString(path = HideAnnotation.hideType, value = HideAnnotation.Type.equals),
+      new JsonSchemaString(path = HideAnnotation.hideExpectedValue, value = "true")
+    )
+  )
+  var nodeAddr: String = _
+
+  /**
+    * Applies the manual location configuration to the given PhysicalOp.
+    *
+    * @param baseOp The base PhysicalOp object
+    * @return The PhysicalOp object with the applied location preference if manual is enabled; otherwise, the original baseOp.
+    */
+  def configureLocationPreference(baseOp: PhysicalOp): PhysicalOp = {
+    if (!autoSelectNodeAddress && nodeAddr != null) {
+      baseOp.withLocationPreference(Some(GoToSpecificNode(nodeAddr)))
+    } else {
+      baseOp
+    }
+  }
+}
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/PythonOperatorDescriptor.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/PythonOperatorDescriptor.scala
index 02c43b4177b..98d86e0bb68 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/PythonOperatorDescriptor.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/PythonOperatorDescriptor.scala
@@ -5,7 +5,9 @@ import edu.uci.ics.amber.core.tuple.Schema
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, PortIdentity, SchemaPropagationFunc}
 
-trait PythonOperatorDescriptor extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+trait PythonOperatorDescriptor extends LogicalOp with DesignatedLocationConfigurable {
   private def generatePythonCodeForRaisingException(ex: Throwable): String = {
     s"#EXCEPTION DURING CODE GENERATION: ${ex.getMessage}"
   }
@@ -44,6 +46,7 @@ trait PythonOperatorDescriptor extends LogicalOp {
       .withOutputPorts(operatorInfo.outputPorts)
       .withParallelizable(parallelizable())
       .withPropagateSchema(SchemaPropagationFunc(inputSchemas => getOutputSchemas(inputSchemas)))
+      .pipe(configureLocationPreference)
   }
 
   def parallelizable(): Boolean = false
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/aggregate/AggregateOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/aggregate/AggregateOpDesc.scala
index 80ad4892782..8899819d21c 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/aggregate/AggregateOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/aggregate/AggregateOpDesc.scala
@@ -10,14 +10,15 @@ import edu.uci.ics.amber.core.virtualidentity.{
   WorkflowIdentity
 }
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeNameList
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
 import javax.validation.constraints.{NotNull, Size}
+import scala.util.chaining.scalaUtilChainingOps
 
-class AggregateOpDesc extends LogicalOp {
+class AggregateOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(value = "aggregations", required = true)
   @JsonPropertyDescription("multiple aggregation functions")
   @NotNull(message = "aggregation cannot be null")
@@ -63,6 +64,7 @@ class AggregateOpDesc extends LogicalOp {
           Map(PortIdentity(internal = true) -> outputSchema)
         })
       )
+      .pipe(configureLocationPreference)
 
     val finalInputPort = InputPort(PortIdentity(0, internal = true))
     val finalOutputPort = OutputPort(PortIdentity(0), blocking = true)
@@ -88,6 +90,7 @@ class AggregateOpDesc extends LogicalOp {
       )
       .withPartitionRequirement(List(Option(HashPartition(groupByKeys))))
       .withDerivePartition(_ => HashPartition(groupByKeys))
+      .pipe(configureLocationPreference)
 
     var plan = PhysicalPlan(
       operators = Set(partialPhysicalOp, finalPhysicalOp),
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/cartesianProduct/CartesianProductOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
index ca6db486b43..99718516f26 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
@@ -4,10 +4,11 @@ import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.tuple.{Attribute, Schema}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{LogicalOp, DesignatedLocationConfigurable}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import scala.util.chaining.scalaUtilChainingOps
 
-class CartesianProductOpDesc extends LogicalOp {
+class CartesianProductOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -68,7 +69,7 @@ class CartesianProductOpDesc extends LogicalOp {
       // TODO : refactor to parallelize this operator for better performance and scalability:
       //  can consider hash partition on larger input, broadcast smaller table to each partition
       .withParallelizable(false)
-
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/dictionary/DictionaryMatcherOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/dictionary/DictionaryMatcherOpDesc.scala
index 2e6d29f4296..afcfde13c5b 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/dictionary/DictionaryMatcherOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/dictionary/DictionaryMatcherOpDesc.scala
@@ -5,17 +5,20 @@ import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.tuple.{Attribute, AttributeType}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, SchemaPropagationFunc}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.map.MapOpDesc
 import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
+import scala.util.chaining.scalaUtilChainingOps
+
 /**
   * Dictionary matcher operator matches a tuple if the specified column is in the given dictionary.
   * It outputs an extra column to label the tuple if it is matched or not
   * This is the description of the operator
   */
-class DictionaryMatcherOpDesc extends MapOpDesc {
+class DictionaryMatcherOpDesc extends MapOpDesc with DesignatedLocationConfigurable {
   @JsonProperty(value = "Dictionary", required = true)
   @JsonPropertyDescription("dictionary values separated by a comma") var dictionary: String = _
 
@@ -54,6 +57,7 @@ class DictionaryMatcherOpDesc extends MapOpDesc {
           )
         })
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/difference/DifferenceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/difference/DifferenceOpDesc.scala
index 8cb81f186c3..ebd87fabf34 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/difference/DifferenceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/difference/DifferenceOpDesc.scala
@@ -4,10 +4,12 @@ import com.google.common.base.Preconditions
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class DifferenceOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class DifferenceOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
@@ -29,6 +31,7 @@ class DifferenceOpDesc extends LogicalOp {
         val outputSchema = inputSchemas.values.head
         operatorInfo.outputPorts.map(port => port.id -> outputSchema).toMap
       }))
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/distinct/DistinctOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/distinct/DistinctOpDesc.scala
index 7f851743b18..c30b0a6fc2c 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/distinct/DistinctOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/distinct/DistinctOpDesc.scala
@@ -3,10 +3,12 @@ package edu.uci.ics.amber.operator.distinct
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{HashPartition, InputPort, OutputPort, PhysicalOp}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class DistinctOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class DistinctOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
@@ -23,7 +25,7 @@ class DistinctOpDesc extends LogicalOp {
       .withOutputPorts(operatorInfo.outputPorts)
       .withPartitionRequirement(List(Option(HashPartition())))
       .withDerivePartition(_ => HashPartition())
-
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/filter/SpecializedFilterOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/filter/SpecializedFilterOpDesc.scala
index 340e2ee8b48..2d5e2166d2f 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/filter/SpecializedFilterOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/filter/SpecializedFilterOpDesc.scala
@@ -2,13 +2,15 @@ package edu.uci.ics.amber.operator.filter
 
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
-import edu.uci.ics.amber.core.workflow.PhysicalOp
+import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PhysicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
-import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class SpecializedFilterOpDesc extends FilterOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class SpecializedFilterOpDesc extends FilterOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(value = "predicates", required = true)
   @JsonPropertyDescription("multiple predicates in OR")
@@ -30,6 +32,7 @@ class SpecializedFilterOpDesc extends FilterOpDesc {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/hashJoin/HashJoinOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/hashJoin/HashJoinOpDesc.scala
index 9b8429e4e18..9c2688e2250 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/hashJoin/HashJoinOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/hashJoin/HashJoinOpDesc.scala
@@ -10,7 +10,7 @@ import edu.uci.ics.amber.core.virtualidentity.{
   WorkflowIdentity
 }
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME
 import edu.uci.ics.amber.operator.metadata.annotations.{
   AutofillAttributeName,
@@ -19,6 +19,8 @@ import edu.uci.ics.amber.operator.metadata.annotations.{
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
+import scala.util.chaining.scalaUtilChainingOps
+
 object HashJoinOpDesc {
   val HASH_JOIN_INTERNAL_KEY_NAME = "__internal__hashtable__key__"
 }
@@ -34,7 +36,7 @@ object HashJoinOpDesc {
   }
 }
 """)
-class HashJoinOpDesc[K] extends LogicalOp {
+class HashJoinOpDesc[K] extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("Left Input Attribute")
@@ -85,6 +87,7 @@ class HashJoinOpDesc[K] extends LogicalOp {
           )
         )
         .withParallelizable(true)
+        .pipe(configureLocationPreference)
 
     val probeBuildInputPort = InputPort(PortIdentity(0, internal = true))
     val probeDataInputPort =
@@ -144,6 +147,7 @@ class HashJoinOpDesc[K] extends LogicalOp {
             Map(PortIdentity() -> outputSchema)
           })
         )
+        .pipe(configureLocationPreference)
 
     PhysicalPlan(
       operators = Set(buildPhysicalOp, probePhysicalOp),
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/ifStatement/IfOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/ifStatement/IfOpDesc.scala
index 65d9b0cd6e0..987d9dcb07d 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/ifStatement/IfOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/ifStatement/IfOpDesc.scala
@@ -11,11 +11,13 @@ import edu.uci.ics.amber.core.workflow.{
   PortIdentity,
   SchemaPropagationFunc
 }
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
-class IfOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class IfOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Condition State")
   @JsonPropertyDescription("name of the state variable to evaluate")
@@ -46,6 +48,7 @@ class IfOpDesc extends LogicalOp {
             .toMap
         )
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intersect/IntersectOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intersect/IntersectOpDesc.scala
index 48cc74ea8a7..d10d39f5df9 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intersect/IntersectOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intersect/IntersectOpDesc.scala
@@ -3,10 +3,12 @@ package edu.uci.ics.amber.operator.intersect
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class IntersectOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class IntersectOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
@@ -23,6 +25,7 @@ class IntersectOpDesc extends LogicalOp {
       .withOutputPorts(operatorInfo.outputPorts)
       .withPartitionRequirement(List(Option(HashPartition()), Option(HashPartition())))
       .withDerivePartition(_ => HashPartition())
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intervalJoin/IntervalJoinOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
index dd61c510a1a..5ce070a077e 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
@@ -6,7 +6,7 @@ import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchema
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.tuple.{Attribute, Schema}
 import edu.uci.ics.amber.core.workflow.{HashPartition, PhysicalOp, SchemaPropagationFunc}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.operator.metadata.annotations.{
   AutofillAttributeName,
@@ -16,6 +16,8 @@ import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
 
+import scala.util.chaining.scalaUtilChainingOps
+
 /** This Operator have two assumptions:
   * 1. The tuples in both inputs come in ascending order
   * 2. The left input join key takes as points, join condition is: left key in the range of (right key, right key + constant)
@@ -34,7 +36,7 @@ import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
   }
 }
 """)
-class IntervalJoinOpDesc extends LogicalOp {
+class IntervalJoinOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("Left Input attr")
@@ -111,6 +113,7 @@ class IntervalJoinOpDesc extends LogicalOp {
         })
       )
       .withPartitionRequirement(partitionRequirement)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/keywordSearch/KeywordSearchOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/keywordSearch/KeywordSearchOpDesc.scala
index b3e41f267e6..f3e1c7f73b7 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/keywordSearch/KeywordSearchOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/keywordSearch/KeywordSearchOpDesc.scala
@@ -10,8 +10,11 @@ import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class KeywordSearchOpDesc extends FilterOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class KeywordSearchOpDesc extends FilterOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("attribute")
@@ -40,6 +43,7 @@ class KeywordSearchOpDesc extends FilterOpDesc {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/limit/LimitOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/limit/LimitOpDesc.scala
index ae0d7768bb1..a31c8cd770d 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/limit/LimitOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/limit/LimitOpDesc.scala
@@ -4,15 +4,16 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.workflow.PhysicalOp
-import edu.uci.ics.amber.operator.{LogicalOp, StateTransferFunc}
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp, StateTransferFunc}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
 
+import scala.util.chaining.scalaUtilChainingOps
 import scala.util.{Success, Try}
 
-class LimitOpDesc extends LogicalOp {
+class LimitOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("Limit")
@@ -36,6 +37,7 @@ class LimitOpDesc extends LogicalOp {
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
       .withParallelizable(false)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/projection/ProjectionOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/projection/ProjectionOpDesc.scala
index 80bc4d4e645..2e27bba1093 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/projection/ProjectionOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/projection/ProjectionOpDesc.scala
@@ -7,11 +7,13 @@ import edu.uci.ics.amber.core.tuple.Schema
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.PhysicalOp.oneToOnePhysicalOp
 import edu.uci.ics.amber.core.workflow._
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.map.MapOpDesc
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
+import scala.util.chaining.scalaUtilChainingOps
 
-class ProjectionOpDesc extends MapOpDesc {
+class ProjectionOpDesc extends MapOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true, defaultValue = "false")
   @JsonSchemaTitle("Drop Option")
@@ -53,6 +55,7 @@ class ProjectionOpDesc extends MapOpDesc {
 
         Map(operatorInfo.outputPorts.head.id -> outputSchema)
       }))
+      .pipe(configureLocationPreference)
   }
 
   def derivePartition()(partition: List[PartitionInfo]): PartitionInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/randomksampling/RandomKSamplingOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/randomksampling/RandomKSamplingOpDesc.scala
index 3fd4849c27c..286141f51e2 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/randomksampling/RandomKSamplingOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/randomksampling/RandomKSamplingOpDesc.scala
@@ -4,11 +4,14 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 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.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.filter.FilterOpDesc
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
-class RandomKSamplingOpDesc extends FilterOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class RandomKSamplingOpDesc extends FilterOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(value = "random k sample percentage", required = true)
   @JsonPropertyDescription("random k sampling with given percentage")
@@ -30,6 +33,7 @@ class RandomKSamplingOpDesc extends FilterOpDesc {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/regex/RegexOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/regex/RegexOpDesc.scala
index 070417540fb..0fe6051136c 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/regex/RegexOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/regex/RegexOpDesc.scala
@@ -10,8 +10,11 @@ import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class RegexOpDesc extends FilterOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class RegexOpDesc extends FilterOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(value = "attribute", required = true)
   @JsonPropertyDescription("column to search regex on")
@@ -43,6 +46,7 @@ class RegexOpDesc extends FilterOpDesc {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/reservoirsampling/ReservoirSamplingOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/reservoirsampling/ReservoirSamplingOpDesc.scala
index cc65876a2f5..1b85f0e03e2 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/reservoirsampling/ReservoirSamplingOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/reservoirsampling/ReservoirSamplingOpDesc.scala
@@ -3,13 +3,15 @@ package edu.uci.ics.amber.operator.reservoirsampling
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.workflow.PhysicalOp
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
-class ReservoirSamplingOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class ReservoirSamplingOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(value = "number of item sampled in reservoir sampling", required = true)
   @JsonPropertyDescription("reservoir sampling with k items being kept randomly")
@@ -31,6 +33,7 @@ class ReservoirSamplingOpDesc extends LogicalOp {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sortPartitions/SortPartitionsOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sortPartitions/SortPartitionsOpDesc.scala
index 3d4809e34f5..e27905c34b7 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sortPartitions/SortPartitionsOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sortPartitions/SortPartitionsOpDesc.scala
@@ -4,13 +4,15 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, RangePartition}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
 
+import scala.util.chaining.scalaUtilChainingOps
+
 @JsonSchemaInject(json = """
 {
   "attributeTypeRules": {
@@ -20,7 +22,7 @@ import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
   }
 }
 """)
-class SortPartitionsOpDesc extends LogicalOp {
+class SortPartitionsOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("Attribute")
@@ -41,7 +43,7 @@ class SortPartitionsOpDesc extends LogicalOp {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     PhysicalOp
       .oneToOnePhysicalOp(
         workflowId,
@@ -57,6 +59,8 @@ class SortPartitionsOpDesc extends LogicalOp {
       .withPartitionRequirement(
         List(Option(RangePartition(List(sortAttributeName), domainMin, domainMax)))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def operatorInfo: OperatorInfo =
     OperatorInfo(
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterFullArchiveSearchSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterFullArchiveSearchSourceOpDesc.scala
index d4050625b1b..8ac390771fc 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterFullArchiveSearchSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterFullArchiveSearchSourceOpDesc.scala
@@ -13,8 +13,13 @@ import edu.uci.ics.amber.operator.metadata.annotations.UIWidget
 import edu.uci.ics.amber.operator.source.apis.twitter.TwitterSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class TwitterFullArchiveSearchSourceOpDesc extends TwitterSourceOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class TwitterFullArchiveSearchSourceOpDesc
+    extends TwitterSourceOpDesc
+    with DesignatedLocationConfigurable {
 
   @JsonIgnore
   override val APIName: Option[String] = Some("Full Archive Search")
@@ -43,7 +48,7 @@ class TwitterFullArchiveSearchSourceOpDesc extends TwitterSourceOpDesc {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     // TODO: use multiple workers
     PhysicalOp
       .sourcePhysicalOp(
@@ -60,6 +65,8 @@ class TwitterFullArchiveSearchSourceOpDesc extends TwitterSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def sourceSchema(): Schema = {
 
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterSearchSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterSearchSourceOpDesc.scala
index 20960c3181d..53af6f1da2b 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterSearchSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/apis/twitter/v2/TwitterSearchSourceOpDesc.scala
@@ -13,8 +13,11 @@ import edu.uci.ics.amber.operator.metadata.annotations.UIWidget
 import edu.uci.ics.amber.operator.source.apis.twitter.TwitterSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class TwitterSearchSourceOpDesc extends TwitterSourceOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class TwitterSearchSourceOpDesc extends TwitterSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonIgnore
   override val APIName: Option[String] = Some("Search")
@@ -33,7 +36,7 @@ class TwitterSearchSourceOpDesc extends TwitterSourceOpDesc {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     // TODO: use multiple workers
     PhysicalOp
       .sourcePhysicalOp(
@@ -50,6 +53,8 @@ class TwitterSearchSourceOpDesc extends TwitterSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def sourceSchema(): Schema = {
 
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/fetcher/URLFetcherOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/fetcher/URLFetcherOpDesc.scala
index 054af768565..b64e8bf7956 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/fetcher/URLFetcherOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/fetcher/URLFetcherOpDesc.scala
@@ -10,8 +10,11 @@ import edu.uci.ics.amber.operator.source.SourceOperatorDescriptor
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class URLFetcherOpDesc extends SourceOperatorDescriptor {
+import scala.util.chaining.scalaUtilChainingOps
+
+class URLFetcherOpDesc extends SourceOperatorDescriptor with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("URL")
@@ -54,6 +57,7 @@ class URLFetcherOpDesc extends SourceOperatorDescriptor {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/FileScanSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/FileScanSourceOpDesc.scala
index 437d2f2bbe3..ae76cc5cb26 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/FileScanSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/FileScanSourceOpDesc.scala
@@ -13,9 +13,15 @@ import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
 import edu.uci.ics.amber.operator.metadata.annotations.HideAnnotation
 import edu.uci.ics.amber.operator.source.scan.text.TextSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
+
+import scala.util.chaining.scalaUtilChainingOps
 
 @JsonIgnoreProperties(value = Array("limit", "offset", "fileEncoding"))
-class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc {
+class FileScanSourceOpDesc
+    extends ScanSourceOpDesc
+    with TextSourceOpDesc
+    with DesignatedLocationConfigurable {
   @JsonProperty(defaultValue = "UTF_8", required = true)
   @JsonSchemaTitle("Encoding")
   @JsonSchemaInject(
@@ -63,6 +69,7 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def sourceSchema(): Schema = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
index 8f63a02af4c..7253c80acd3 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
@@ -7,6 +7,7 @@ import edu.uci.ics.amber.core.tuple.Schema
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
 import edu.uci.ics.amber.operator.source.scan.ScanSourceOpDesc
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.util.ArrowUtils
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
@@ -19,9 +20,10 @@ import org.apache.arrow.vector.ipc.ArrowFileReader
 import org.apache.arrow.vector.types.pojo.{Schema => ArrowSchema}
 
 import scala.util.Using
+import scala.util.chaining.scalaUtilChainingOps
 
 @JsonIgnoreProperties(value = Array("fileEncoding"))
-class ArrowSourceOpDesc extends ScanSourceOpDesc {
+class ArrowSourceOpDesc extends ScanSourceOpDesc with DesignatedLocationConfigurable {
 
   fileTypeName = Option("Arrow")
 
@@ -45,6 +47,7 @@ class ArrowSourceOpDesc extends ScanSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> inferSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   /**
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
index ac9538c25b9..e94883d08c2 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
@@ -11,11 +11,13 @@ import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
 import edu.uci.ics.amber.operator.source.scan.ScanSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
 import java.io.{IOException, InputStreamReader}
 import java.net.URI
+import scala.util.chaining.scalaUtilChainingOps
 
-class CSVScanSourceOpDesc extends ScanSourceOpDesc {
+class CSVScanSourceOpDesc extends ScanSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
@@ -55,6 +57,7 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def sourceSchema(): Schema = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
index 3e362ec7325..f64291fdefd 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
@@ -10,13 +10,15 @@ import edu.uci.ics.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 import edu.uci.ics.amber.core.tuple.{Attribute, AttributeType, Schema}
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.source.scan.ScanSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
 import java.io.IOException
 import java.net.URI
+import scala.util.chaining.scalaUtilChainingOps
 
-class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
+class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
@@ -57,6 +59,7 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def sourceSchema(): Schema = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
index 97be3d95b24..d115cc5e6f9 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
@@ -9,13 +9,15 @@ import edu.uci.ics.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 import edu.uci.ics.amber.core.tuple.{Attribute, AttributeType, Schema}
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.source.scan.ScanSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 
 import java.io.IOException
 import java.net.URI
+import scala.util.chaining.scalaUtilChainingOps
 
-class CSVOldScanSourceOpDesc extends ScanSourceOpDesc {
+class CSVOldScanSourceOpDesc extends ScanSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
@@ -53,6 +55,7 @@ class CSVOldScanSourceOpDesc extends ScanSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def sourceSchema(): Schema = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
index 30d0e58d045..ca9f97eb35d 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
@@ -8,6 +8,7 @@ import edu.uci.ics.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 import edu.uci.ics.amber.core.tuple.{Attribute, Schema}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 import edu.uci.ics.amber.operator.source.scan.ScanSourceOpDesc
 import edu.uci.ics.amber.util.JSONUtils.{JSONToMap, objectMapper}
 
@@ -15,8 +16,9 @@ import java.io._
 import java.net.URI
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters.IteratorHasAsScala
+import scala.util.chaining.scalaUtilChainingOps
 
-class JSONLScanSourceOpDesc extends ScanSourceOpDesc {
+class JSONLScanSourceOpDesc extends ScanSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonPropertyDescription("flatten nested objects and arrays")
@@ -46,6 +48,7 @@ class JSONLScanSourceOpDesc extends ScanSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
   }
 
   override def sourceSchema(): Schema = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/text/TextInputSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
index 597424a068b..553a3ca7e4d 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
@@ -11,8 +11,14 @@ import edu.uci.ics.amber.operator.source.SourceOperatorDescriptor
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class TextInputSourceOpDesc extends SourceOperatorDescriptor with TextSourceOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class TextInputSourceOpDesc
+    extends SourceOperatorDescriptor
+    with TextSourceOpDesc
+    with DesignatedLocationConfigurable {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Text")
   @JsonSchemaInject(json = UIWidget.UIWidgetTextArea)
@@ -21,7 +27,7 @@ class TextInputSourceOpDesc extends SourceOperatorDescriptor with TextSourceOpDe
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     PhysicalOp
       .sourcePhysicalOp(
         workflowId,
@@ -37,6 +43,8 @@ class TextInputSourceOpDesc extends SourceOperatorDescriptor with TextSourceOpDe
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def sourceSchema(): Schema =
     Schema().add(attributeName, attributeType.getType)
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDesc.scala
index ccccaa720a4..e7fbb792050 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDesc.scala
@@ -26,9 +26,12 @@ import edu.uci.ics.amber.operator.source.sql.asterixdb.AsterixDBConnUtil.{
 }
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import kong.unirest.json.JSONObject
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
+
+import scala.util.chaining.scalaUtilChainingOps
 
 @JsonIgnoreProperties(value = Array("username", "password"))
-class AsterixDBSourceOpDesc extends SQLSourceOpDesc {
+class AsterixDBSourceOpDesc extends SQLSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(defaultValue = "false")
   @JsonSchemaTitle("Geo Search?")
@@ -92,7 +95,7 @@ class AsterixDBSourceOpDesc extends SQLSourceOpDesc {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     PhysicalOp
       .sourcePhysicalOp(
         workflowId,
@@ -108,6 +111,8 @@ class AsterixDBSourceOpDesc extends SQLSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def operatorInfo: OperatorInfo =
     OperatorInfo(
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/mysql/MySQLSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/mysql/MySQLSourceOpDesc.scala
index fbc583d8b68..bfe0447169b 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/mysql/MySQLSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/mysql/MySQLSourceOpDesc.scala
@@ -8,15 +8,17 @@ import edu.uci.ics.amber.operator.source.sql.mysql.MySQLConnUtil.connect
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
 import java.sql.{Connection, SQLException}
+import scala.util.chaining.scalaUtilChainingOps
 
-class MySQLSourceOpDesc extends SQLSourceOpDesc {
+class MySQLSourceOpDesc extends SQLSourceOpDesc with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     PhysicalOp
       .sourcePhysicalOp(
         workflowId,
@@ -32,6 +34,8 @@ class MySQLSourceOpDesc extends SQLSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def operatorInfo: OperatorInfo =
     OperatorInfo(
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/postgresql/PostgreSQLSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/postgresql/PostgreSQLSourceOpDesc.scala
index 529ec85d971..9730245955d 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/postgresql/PostgreSQLSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/source/sql/postgresql/PostgreSQLSourceOpDesc.scala
@@ -12,10 +12,12 @@ import edu.uci.ics.amber.operator.source.sql.postgresql.PostgreSQLConnUtil.conne
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
 import java.sql.{Connection, SQLException}
+import scala.util.chaining.scalaUtilChainingOps
 
-class PostgreSQLSourceOpDesc extends SQLSourceOpDesc {
+class PostgreSQLSourceOpDesc extends SQLSourceOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty()
   @JsonSchemaTitle("Keywords to Search")
@@ -29,7 +31,7 @@ class PostgreSQLSourceOpDesc extends SQLSourceOpDesc {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
-  ): PhysicalOp =
+  ): PhysicalOp = {
     PhysicalOp
       .sourcePhysicalOp(
         workflowId,
@@ -45,6 +47,8 @@ class PostgreSQLSourceOpDesc extends SQLSourceOpDesc {
       .withPropagateSchema(
         SchemaPropagationFunc(_ => Map(operatorInfo.outputPorts.head.id -> sourceSchema()))
       )
+      .pipe(configureLocationPreference)
+  }
 
   override def operatorInfo: OperatorInfo =
     OperatorInfo(
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/split/SplitOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/split/SplitOpDesc.scala
index 9003fb54b6c..265b2e6e2e7 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/split/SplitOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/split/SplitOpDesc.scala
@@ -10,12 +10,13 @@ import com.kjetland.jackson.jsonSchema.annotations.{
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow._
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{LogicalOp, DesignatedLocationConfigurable}
 import edu.uci.ics.amber.operator.metadata.annotations.HideAnnotation
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
+import scala.util.chaining.scalaUtilChainingOps
 
-class SplitOpDesc extends LogicalOp {
+class SplitOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonSchemaTitle("Split Percentage")
   @JsonProperty(defaultValue = "80")
@@ -63,6 +64,7 @@ class SplitOpDesc extends LogicalOp {
           operatorInfo.outputPorts.map(port => port.id -> outputSchema).toMap
         })
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
index 3dc311f77d5..dbb9ef335ad 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
@@ -10,11 +10,13 @@ import edu.uci.ics.amber.core.workflow.{
   PortIdentity,
   SchemaPropagationFunc
 }
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 
-class SymmetricDifferenceOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class SymmetricDifferenceOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
@@ -39,6 +41,7 @@ class SymmetricDifferenceOpDesc extends LogicalOp {
         val outputSchema = inputSchemas.values.head
         operatorInfo.outputPorts.map(port => port.id -> outputSchema).toMap
       }))
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/typecasting/TypeCastingOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/typecasting/TypeCastingOpDesc.scala
index f5d502437b4..ff1232d5f45 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/typecasting/TypeCastingOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/typecasting/TypeCastingOpDesc.scala
@@ -10,8 +10,11 @@ import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class TypeCastingOpDesc extends MapOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class TypeCastingOpDesc extends MapOpDesc with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("TypeCasting Units")
@@ -43,6 +46,7 @@ class TypeCastingOpDesc extends MapOpDesc {
           Map(operatorInfo.outputPorts.head.id -> outputSchema)
         }
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/java/JavaUDFOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/java/JavaUDFOpDesc.scala
index f5b3b529589..32138c0d8c9 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/java/JavaUDFOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/java/JavaUDFOpDesc.scala
@@ -12,12 +12,18 @@ import edu.uci.ics.amber.core.workflow.{
   UnknownPartition
 }
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
-import edu.uci.ics.amber.operator.{LogicalOp, PortDescription, StateTransferFunc}
+import edu.uci.ics.amber.operator.{
+  DesignatedLocationConfigurable,
+  LogicalOp,
+  PortDescription,
+  StateTransferFunc
+}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
 
+import scala.util.chaining.scalaUtilChainingOps
 import scala.util.{Success, Try}
-class JavaUDFOpDesc extends LogicalOp {
+class JavaUDFOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(
     required = true,
     defaultValue =
@@ -90,7 +96,7 @@ class JavaUDFOpDesc extends LogicalOp {
       Map(operatorInfo.outputPorts.head.id -> outputSchema)
     }
 
-    if (workers > 1)
+    if (workers > 1) {
       PhysicalOp
         .oneToOnePhysicalOp(
           workflowId,
@@ -106,7 +112,8 @@ class JavaUDFOpDesc extends LogicalOp {
         .withParallelizable(true)
         .withSuggestedWorkerNum(workers)
         .withPropagateSchema(SchemaPropagationFunc(propagateSchema))
-    else
+        .pipe(configureLocationPreference)
+    } else {
       PhysicalOp
         .manyToOnePhysicalOp(
           workflowId,
@@ -121,6 +128,8 @@ class JavaUDFOpDesc extends LogicalOp {
         .withIsOneToManyOp(true)
         .withParallelizable(false)
         .withPropagateSchema(SchemaPropagationFunc(propagateSchema))
+        .pipe(configureLocationPreference)
+    }
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
index 57f85a663c5..823f82add73 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
@@ -6,12 +6,14 @@ import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import edu.uci.ics.amber.core.executor.OpExecWithCode
 import edu.uci.ics.amber.core.tuple.{Attribute, Schema}
 import edu.uci.ics.amber.core.workflow.{PhysicalOp, SchemaPropagationFunc, UnknownPartition}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
 
-class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class DualInputPortsPythonUDFOpDescV2 extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(
     required = true,
     defaultValue =
@@ -111,6 +113,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
           Map(operatorInfo.outputPorts.head.id -> outputSchema)
         })
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/PythonUDFOpDescV2.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/PythonUDFOpDescV2.scala
index 802b8d7d544..70cd2fd7071 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/PythonUDFOpDescV2.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/PythonUDFOpDescV2.scala
@@ -8,11 +8,17 @@ import edu.uci.ics.amber.core.tuple.{Attribute, Schema}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow._
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
-import edu.uci.ics.amber.operator.{LogicalOp, PortDescription, StateTransferFunc}
+import edu.uci.ics.amber.operator.{
+  DesignatedLocationConfigurable,
+  LogicalOp,
+  PortDescription,
+  StateTransferFunc
+}
 
+import scala.util.chaining.scalaUtilChainingOps
 import scala.util.{Success, Try}
 
-class PythonUDFOpDescV2 extends LogicalOp {
+class PythonUDFOpDescV2 extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(
     required = true,
     defaultValue =
@@ -121,6 +127,7 @@ class PythonUDFOpDescV2 extends LogicalOp {
       .withPartitionRequirement(partitionRequirement)
       .withIsOneToManyOp(true)
       .withPropagateSchema(SchemaPropagationFunc(propagateSchema))
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
index ca45e0408d0..f2bf4a06fdb 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
@@ -9,8 +9,11 @@ import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo
 import edu.uci.ics.amber.operator.source.SourceOperatorDescriptor
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor {
+import scala.util.chaining.scalaUtilChainingOps
+
+class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor with DesignatedLocationConfigurable {
 
   @JsonProperty(
     required = true,
@@ -58,6 +61,7 @@ class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor {
     } else {
       physicalOp.withParallelizable(false)
     }
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFOpDesc.scala
index 5e0815f506a..009340d511b 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFOpDesc.scala
@@ -11,14 +11,20 @@ import edu.uci.ics.amber.core.workflow.{
   SchemaPropagationFunc,
   UnknownPartition
 }
-import edu.uci.ics.amber.operator.{LogicalOp, PortDescription, StateTransferFunc}
+import edu.uci.ics.amber.operator.{
+  DesignatedLocationConfigurable,
+  LogicalOp,
+  PortDescription,
+  StateTransferFunc
+}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
 
+import scala.util.chaining.scalaUtilChainingOps
 import scala.util.{Success, Try}
 
-class RUDFOpDesc extends LogicalOp {
+class RUDFOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(
     required = true,
     defaultValue =
@@ -118,6 +124,7 @@ class RUDFOpDesc extends LogicalOp {
       .withPartitionRequirement(partitionRequirement)
       .withIsOneToManyOp(true)
       .withPropagateSchema(SchemaPropagationFunc(propagateSchema))
+      .pipe(configureLocationPreference)
 
   }
 
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFSourceOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFSourceOpDesc.scala
index e84900f8ddf..e6b587abc02 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFSourceOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/r/RUDFSourceOpDesc.scala
@@ -9,8 +9,11 @@ import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo
 import edu.uci.ics.amber.operator.source.SourceOperatorDescriptor
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class RUDFSourceOpDesc extends SourceOperatorDescriptor {
+import scala.util.chaining.scalaUtilChainingOps
+
+class RUDFSourceOpDesc extends SourceOperatorDescriptor with DesignatedLocationConfigurable {
 
   @JsonProperty(
     required = true,
@@ -73,6 +76,7 @@ class RUDFSourceOpDesc extends SourceOperatorDescriptor {
     } else {
       physicalOp.withParallelizable(false)
     }
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo = {
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/union/UnionOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/union/UnionOpDesc.scala
index 6ee4fea20d9..2bafce013ce 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/union/UnionOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/union/UnionOpDesc.scala
@@ -3,10 +3,12 @@ package edu.uci.ics.amber.operator.union
 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, PortIdentity}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class UnionOpDesc extends LogicalOp {
+import scala.util.chaining.scalaUtilChainingOps
+
+class UnionOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
@@ -21,6 +23,7 @@ class UnionOpDesc extends LogicalOp {
       )
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/unneststring/UnnestStringOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/unneststring/UnnestStringOpDesc.scala
index 6db50ce069e..53816a6a2c0 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/unneststring/UnnestStringOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/unneststring/UnnestStringOpDesc.scala
@@ -10,8 +10,11 @@ import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort}
+import edu.uci.ics.amber.operator.DesignatedLocationConfigurable
 
-class UnnestStringOpDesc extends FlatMapOpDesc {
+import scala.util.chaining.scalaUtilChainingOps
+
+class UnnestStringOpDesc extends FlatMapOpDesc with DesignatedLocationConfigurable {
   @JsonProperty(value = "Delimiter", required = true, defaultValue = ",")
   @JsonPropertyDescription("string that separates the data")
   var delimiter: String = _
@@ -60,5 +63,6 @@ class UnnestStringOpDesc extends FlatMapOpDesc {
           Map(operatorInfo.outputPorts.head.id -> outputSchema)
         })
       )
+      .pipe(configureLocationPreference)
   }
 }
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
index 3453f49c6dd..66ea0435a30 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
@@ -5,18 +5,20 @@ import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.tuple.{AttributeType, Schema}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, SchemaPropagationFunc}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.core.workflow.OutputPort.OutputMode
 
+import scala.util.chaining.scalaUtilChainingOps
+
 /**
   * HTML Visualization operator to render any given HTML code
   * This is the description of the operator
   */
-class HtmlVizOpDesc extends LogicalOp {
+class HtmlVizOpDesc extends LogicalOp with DesignatedLocationConfigurable {
   @JsonProperty(required = true)
   @JsonSchemaTitle("HTML content")
   @AutofillAttributeName var htmlContentAttrName: String = _
@@ -43,6 +45,7 @@ class HtmlVizOpDesc extends LogicalOp {
           Map(operatorInfo.outputPorts.head.id -> outputSchema)
         })
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =
diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/urlviz/UrlVizOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
index a11b294f3af..31d622c59ba 100644
--- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
+++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
@@ -5,13 +5,15 @@ import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchema
 import edu.uci.ics.amber.core.executor.OpExecWithClassName
 import edu.uci.ics.amber.core.tuple.{AttributeType, Schema}
 import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, SchemaPropagationFunc}
-import edu.uci.ics.amber.operator.LogicalOp
+import edu.uci.ics.amber.operator.{DesignatedLocationConfigurable, LogicalOp}
 import edu.uci.ics.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import edu.uci.ics.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import edu.uci.ics.amber.operator.metadata.annotations.AutofillAttributeName
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
 import edu.uci.ics.amber.core.workflow.OutputPort.OutputMode
 
+import scala.util.chaining.scalaUtilChainingOps
+
 /**
   * URL Visualization operator to render any content in given URL link
   * This is the description of the operator
@@ -25,7 +27,7 @@ import edu.uci.ics.amber.core.workflow.OutputPort.OutputMode
    }
  }
  """)
-class UrlVizOpDesc extends LogicalOp {
+class UrlVizOpDesc extends LogicalOp with DesignatedLocationConfigurable {
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("URL content")
@@ -54,6 +56,7 @@ class UrlVizOpDesc extends LogicalOp {
           Map(operatorInfo.outputPorts.head.id -> outputSchema)
         })
       )
+      .pipe(configureLocationPreference)
   }
 
   override def operatorInfo: OperatorInfo =