feat: introduce batch execution mode

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

@@ -45,7 +45,7 @@ class WorkflowScheduler(
       // CostBasedRegionPlanGenerator considers costs to try to find an optimal plan.
       new CostBasedScheduleGenerator(
         workflowContext,
-        physicalPlan,
+        physicalPlan.copy(executionMode = workflowContext.workflowSettings.executionMode),
         actorId
       ).generate()
     this.schedule = generatedSchedule

amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala

@@ -151,8 +151,11 @@ class WorkflowCompiler(
     val (physicalPlan, outputPortsNeedingStorage) =
       expandLogicalPlan(logicalPlan, logicalPlanPojo.opsToViewResult, None)
 
-    context.workflowSettings =
-      WorkflowSettings(context.workflowSettings.dataTransferBatchSize, outputPortsNeedingStorage)
+    context.workflowSettings = WorkflowSettings(
+      context.workflowSettings.dataTransferBatchSize,
+      outputPortsNeedingStorage,
+      context.workflowSettings.executionMode
+    )
 
     Workflow(context, logicalPlan, physicalPlan)
   }

amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala

@@ -23,7 +23,12 @@ import org.apache.pekko.actor.{ActorSystem, Props}
 import org.apache.pekko.testkit.{ImplicitSender, TestKit}
 import org.apache.pekko.util.Timeout
 import org.apache.texera.amber.clustering.SingleNodeListener
-import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext, WorkflowSettings}
+import org.apache.texera.amber.core.workflow.{
+  ExecutionMode,
+  PortIdentity,
+  WorkflowContext,
+  WorkflowSettings
+}
 import org.apache.texera.amber.engine.architecture.controller._
 import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings._
 import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
@@ -119,7 +124,10 @@ class BatchSizePropagationSpec
   "Engine" should "propagate the correct batch size for headerlessCsv workflow" in {
     val expectedBatchSize = 1
 
-    val customWorkflowSettings = WorkflowSettings(dataTransferBatchSize = expectedBatchSize)
+    val customWorkflowSettings = WorkflowSettings(
+      dataTransferBatchSize = expectedBatchSize,
+      executionMode = ExecutionMode.STREAMING
+    )
 
     val context =
       new WorkflowContext(workflowSettings = customWorkflowSettings)
@@ -141,7 +149,10 @@ class BatchSizePropagationSpec
   "Engine" should "propagate the correct batch size for headerlessCsv->keyword workflow" in {
     val expectedBatchSize = 500
 
-    val customWorkflowSettings = WorkflowSettings(dataTransferBatchSize = expectedBatchSize)
+    val customWorkflowSettings = WorkflowSettings(
+      dataTransferBatchSize = expectedBatchSize,
+      executionMode = ExecutionMode.STREAMING
+    )
 
     val context =
       new WorkflowContext(workflowSettings = customWorkflowSettings)
@@ -171,7 +182,10 @@ class BatchSizePropagationSpec
   "Engine" should "propagate the correct batch size for csv->keyword->count workflow" in {
     val expectedBatchSize = 100
 
-    val customWorkflowSettings = WorkflowSettings(dataTransferBatchSize = expectedBatchSize)
+    val customWorkflowSettings = WorkflowSettings(
+      dataTransferBatchSize = expectedBatchSize,
+      executionMode = ExecutionMode.STREAMING
+    )
 
     val context =
       new WorkflowContext(workflowSettings = customWorkflowSettings)
@@ -209,7 +223,10 @@ class BatchSizePropagationSpec
   "Engine" should "propagate the correct batch size for csv->keyword->averageAndGroupBy workflow" in {
     val expectedBatchSize = 300
 
-    val customWorkflowSettings = WorkflowSettings(dataTransferBatchSize = expectedBatchSize)
+    val customWorkflowSettings = WorkflowSettings(
+      dataTransferBatchSize = expectedBatchSize,
+      executionMode = ExecutionMode.STREAMING
+    )
 
     val context =
       new WorkflowContext(workflowSettings = customWorkflowSettings)
@@ -250,7 +267,10 @@ class BatchSizePropagationSpec
   "Engine" should "propagate the correct batch size for csv->(csv->)->join workflow" in {
     val expectedBatchSize = 1
 
-    val customWorkflowSettings = WorkflowSettings(dataTransferBatchSize = expectedBatchSize)
+    val customWorkflowSettings = WorkflowSettings(
+      dataTransferBatchSize = expectedBatchSize,
+      executionMode = ExecutionMode.STREAMING
+    )
 
     val context =
       new WorkflowContext(workflowSettings = customWorkflowSettings)

common/config/src/main/resources/gui.conf

@@ -63,8 +63,11 @@ gui {
     default-data-transfer-batch-size = 400
     default-data-transfer-batch-size = ${?GUI_WORKFLOW_WORKSPACE_DEFAULT_DATA_TRANSFER_BATCH_SIZE}
 
+    # default execution mode for workflows, can be either BATCH or STREAMING
+    default-execution-mode = STREAMING
+    default-execution-mode = ${?GUI_WORKFLOW_WORKSPACE_DEFAULT_EXECUTION_MODE}
+
     # whether selecting files from datasets instead of the local file system.
-    # The user system must be enabled to make this flag work!
     selecting-files-from-datasets-enabled = true
     selecting-files-from-datasets-enabled = ${?GUI_WORKFLOW_WORKSPACE_SELECTING_FILES_FROM_DATASETS_ENABLED}
 

common/config/src/main/scala/org/apache/texera/config/GuiConfig.scala

@@ -47,6 +47,8 @@ object GuiConfig {
     conf.getBoolean("gui.workflow-workspace.auto-attribute-correction-enabled")
   val guiWorkflowWorkspaceDefaultDataTransferBatchSize: Int =
     conf.getInt("gui.workflow-workspace.default-data-transfer-batch-size")
+  val guiWorkflowWorkspaceDefaultExecutionMode: String =
+    conf.getString("gui.workflow-workspace.default-execution-mode")
   val guiWorkflowWorkspaceSelectingFilesFromDatasetsEnabled: Boolean =
     conf.getBoolean("gui.workflow-workspace.selecting-files-from-datasets-enabled")
   val guiWorkflowWorkspaceWorkflowExecutionsTrackingEnabled: Boolean =

common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/ExecutionMode.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.core.workflow;
+
+public enum ExecutionMode {
+    STREAMING,
+    BATCH;
+
+    public static ExecutionMode fromString(String value) { return valueOf(value); }
+}

common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalPlan.scala

@@ -38,7 +38,8 @@ import scala.jdk.CollectionConverters.{CollectionHasAsScala, IteratorHasAsScala}
 
 case class PhysicalPlan(
     operators: Set[PhysicalOp],
-    links: Set[PhysicalLink]
+    links: Set[PhysicalLink],
+    executionMode: ExecutionMode = ExecutionMode.STREAMING
 ) extends LazyLogging {
 
   @transient private lazy val operatorMap: Map[PhysicalOpIdentity, PhysicalOp] =
@@ -245,6 +246,7 @@ case class PhysicalPlan(
                   getOperator(physicalOp.id).isInputLinkDependee(
                     link
                   ) || getOperator(upstreamPhysicalOpId).isOutputLinkBlocking(link)
+                    || executionMode == ExecutionMode.BATCH
                 )
             }
         }

common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala

@@ -30,7 +30,8 @@ object WorkflowContext {
   val DEFAULT_EXECUTION_ID: ExecutionIdentity = ExecutionIdentity(1L)
   val DEFAULT_WORKFLOW_ID: WorkflowIdentity = WorkflowIdentity(1L)
   val DEFAULT_WORKFLOW_SETTINGS: WorkflowSettings = WorkflowSettings(
-    dataTransferBatchSize = 400 // TODO: make this configurable
+    dataTransferBatchSize = 400,
+    executionMode = ExecutionMode.STREAMING
   )
 }
 class WorkflowContext(

common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowSettings.scala

@@ -21,5 +21,6 @@ package org.apache.texera.amber.core.workflow
 
 case class WorkflowSettings(
     dataTransferBatchSize: Int,
-    outputPortsNeedingStorage: Set[GlobalPortIdentity] = Set.empty
+    outputPortsNeedingStorage: Set[GlobalPortIdentity] = Set.empty,
+    executionMode: ExecutionMode
 )

config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala

@@ -46,6 +46,7 @@ class ConfigResource {
       "timetravelEnabled" -> GuiConfig.guiWorkflowWorkspaceTimetravelEnabled,
       "productionSharedEditingServer" -> GuiConfig.guiWorkflowWorkspaceProductionSharedEditingServer,
       "defaultDataTransferBatchSize" -> GuiConfig.guiWorkflowWorkspaceDefaultDataTransferBatchSize,
+      "defaultExecutionMode" -> GuiConfig.guiWorkflowWorkspaceDefaultExecutionMode,
       "workflowEmailNotificationEnabled" -> GuiConfig.guiWorkflowWorkspaceWorkflowEmailNotificationEnabled,
       "sharingComputingUnitEnabled" -> ComputingUnitConfig.sharingComputingUnitEnabled,
       "operatorConsoleMessageBufferSize" -> GuiConfig.guiWorkflowWorkspaceOperatorConsoleMessageBufferSize,

frontend/src/app/app.module.ts

@@ -181,6 +181,7 @@ import { AdminSettingsComponent } from "./dashboard/component/admin/settings/adm
 import { FormlyRepeatDndComponent } from "./common/formly/repeat-dnd/repeat-dnd.component";
 import { NzInputNumberModule } from "ng-zorro-antd/input-number";
 import { NzCheckboxModule } from "ng-zorro-antd/checkbox";
+import { NzRadioModule } from "ng-zorro-antd/radio";
 
 registerLocaleData(en);
 
@@ -344,6 +345,7 @@ registerLocaleData(en);
     NzProgressModule,
     NzInputNumberModule,
     NzCheckboxModule,
+    NzRadioModule,
   ],
   providers: [
     provideNzI18n(en_US),

frontend/src/app/common/service/gui-config.service.mock.ts

@@ -20,6 +20,7 @@
 import { Injectable } from "@angular/core";
 import { Observable, of } from "rxjs";
 import { GuiConfig } from "../type/gui-config";
+import { ExecutionMode } from "../type/workflow";
 
 /**
  * Mock GuiConfigService for testing purposes.
@@ -42,6 +43,7 @@ export class MockGuiConfigService {
     productionSharedEditingServer: false,
     pythonLanguageServerPort: "3000",
     defaultDataTransferBatchSize: 100,
+    defaultExecutionMode: ExecutionMode.STREAMING,
     workflowEmailNotificationEnabled: false,
     sharingComputingUnitEnabled: false,
     operatorConsoleMessageBufferSize: 1000,

frontend/src/app/common/type/gui-config.ts

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+import { ExecutionMode } from "./workflow";
 
 // Please refer to core/config/src/main/resources/gui.conf for the definition of each config item
 export interface GuiConfig {
@@ -33,6 +34,7 @@ export interface GuiConfig {
   productionSharedEditingServer: boolean;
   pythonLanguageServerPort: string;
   defaultDataTransferBatchSize: number;
+  defaultExecutionMode: ExecutionMode;
   workflowEmailNotificationEnabled: boolean;
   sharingComputingUnitEnabled: boolean;
   operatorConsoleMessageBufferSize: number;

frontend/src/app/common/type/workflow.ts

@@ -20,8 +20,14 @@
 import { WorkflowMetadata } from "../../dashboard/type/workflow-metadata.interface";
 import { CommentBox, OperatorLink, OperatorPredicate, Point } from "../../workspace/types/workflow-common.interface";
 
+export enum ExecutionMode {
+  STREAMING = "STREAMING",
+  BATCH = "BATCH",
+}
+
 export interface WorkflowSettings {
   dataTransferBatchSize: number;
+  executionMode: ExecutionMode;
 }
 
 /**

frontend/src/app/dashboard/component/user-dashboard-test-fixtures.ts

@@ -19,7 +19,7 @@
 
 //All times in test Workflows are in PST because our local machine's timezone is PST
 
-import { Workflow, WorkflowContent } from "../../common/type/workflow";
+import { ExecutionMode, Workflow, WorkflowContent } from "../../common/type/workflow";
 import { DashboardEntry } from "../type/dashboard-entry";
 import { DashboardProject } from "../type/dashboard-project.interface";
 
@@ -39,7 +39,7 @@ export const testWorkflowContent = (operatorTypes: string[]): WorkflowContent =>
   commentBoxes: [],
   links: [],
   operatorPositions: {},
-  settings: { dataTransferBatchSize: 400 },
+  settings: { dataTransferBatchSize: 400, executionMode: ExecutionMode.STREAMING },
 });
 
 export const testWorkflow1: Workflow = {

frontend/src/app/dashboard/component/user/user-workflow/user-workflow.component.ts

@@ -31,7 +31,7 @@ import { DashboardEntry, UserInfo } from "../../../type/dashboard-entry";
 import { UserService } from "../../../../common/service/user/user.service";
 import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy";
 import { NotificationService } from "../../../../common/service/notification/notification.service";
-import { WorkflowContent } from "../../../../common/type/workflow";
+import { ExecutionMode, WorkflowContent } from "../../../../common/type/workflow";
 import { NzUploadFile } from "ng-zorro-antd/upload";
 import * as JSZip from "jszip";
 import { FiltersComponent } from "../filters/filters.component";
@@ -230,7 +230,10 @@ export class UserWorkflowComponent implements AfterViewInit {
       commentBoxes: [],
       links: [],
       operatorPositions: {},
-      settings: { dataTransferBatchSize: this.config.env.defaultDataTransferBatchSize },
+      settings: {
+        dataTransferBatchSize: this.config.env.defaultDataTransferBatchSize,
+        executionMode: this.config.env.defaultExecutionMode,
+      },
     };
     let localPid = this.pid;
     this.workflowPersistService

frontend/src/app/workspace/component/left-panel/settings/settings.component.html

@@ -16,12 +16,25 @@
  specific language governing permissions and limitations
  under the License.
 -->
-
 <div class="settings-container">
-  <h4>Workflow Settings</h4>
   <form
     [formGroup]="settingsForm"
     class="form-inline">
+    <b>Execution Mode:</b>
+    <nz-radio-group formControlName="executionMode">
+      <label
+        nz-radio
+        [nzValue]="ExecutionMode.STREAMING"
+        >Streaming</label
+      >
+      <br />
+      <label
+        nz-radio
+        [nzValue]="ExecutionMode.BATCH"
+        >Batch</label
+      >
+    </nz-radio-group>
+    <br />
     <div class="form-group">
       <label for="dataTransferBatchSize">Data Transfer Batch Size:</label>
       <input

frontend/src/app/workspace/component/left-panel/settings/settings.component.scss

@@ -21,10 +21,6 @@
   padding: 10px;
 }
 
-h4 {
-  margin-bottom: 15px;
-}
-
 .form-inline {
   display: flex;
   flex-direction: column;