Merge branch 'main' into feat/handle-lakefs-exception

Author: xuang7

.github/workflows/build-and-push-images.yml

@@ -49,6 +49,11 @@ on:
           - both
           - amd64
           - arm64
+      with_r_support:
+        description: 'Enable R support for workflow-execution-coordinator'
+        required: false
+        default: false
+        type: boolean
   schedule:
     # Run nightly at 2:00 AM UTC
     - cron: '0 2 * * *'
@@ -70,6 +75,7 @@ jobs:
       docker_registry: ${{ steps.set-params.outputs.docker_registry }}
       services: ${{ steps.set-params.outputs.services }}
       platforms: ${{ steps.set-params.outputs.platforms }}
+      with_r_support: ${{ steps.set-params.outputs.with_r_support }}
     steps:
       - name: Set build parameters
         id: set-params
@@ -82,13 +88,15 @@ jobs:
             echo "docker_registry=apache" >> $GITHUB_OUTPUT
             echo "services=*" >> $GITHUB_OUTPUT
             echo "platforms=both" >> $GITHUB_OUTPUT
+            echo "with_r_support=false" >> $GITHUB_OUTPUT
           else
             echo "Manual workflow_dispatch - using user inputs"
             echo "branch=${{ github.event.inputs.branch || 'main' }}" >> $GITHUB_OUTPUT
             echo "image_tag=${{ github.event.inputs.image_tag }}" >> $GITHUB_OUTPUT
             echo "docker_registry=${{ github.event.inputs.docker_registry || 'apache' }}" >> $GITHUB_OUTPUT
             echo "services=${{ github.event.inputs.services || '*' }}" >> $GITHUB_OUTPUT
             echo "platforms=${{ github.event.inputs.platforms || 'both' }}" >> $GITHUB_OUTPUT
+            echo "with_r_support=${{ github.event.inputs.with_r_support || 'false' }}" >> $GITHUB_OUTPUT
           fi
 
   # Step 1: Generate JOOQ code once and share it
@@ -350,6 +358,8 @@ jobs:
           tags: ${{ env.DOCKER_REGISTRY }}/${{ matrix.image_name }}:${{ needs.set-parameters.outputs.image_tag }}-amd64
           cache-from: type=gha,scope=${{ matrix.image_name }}-amd64
           cache-to: type=gha,mode=max,scope=${{ matrix.image_name }}-amd64
+          build-args: |
+            ${{ (matrix.service == 'computing-unit-master' || matrix.service == 'computing-unit-worker') && needs.set-parameters.outputs.with_r_support == 'true' && 'WITH_R_SUPPORT=true' || '' }}
           labels: |
             org.opencontainers.image.title=${{ matrix.image_name }}
             org.opencontainers.image.description=Apache Texera ${{ matrix.image_name }} (AMD64)
@@ -427,6 +437,8 @@ jobs:
           tags: ${{ env.DOCKER_REGISTRY }}/${{ matrix.image_name }}:${{ needs.set-parameters.outputs.image_tag }}-arm64
           cache-from: type=gha,scope=${{ matrix.image_name }}-arm64
           cache-to: type=gha,mode=max,scope=${{ matrix.image_name }}-arm64
+          build-args: |
+            ${{ (matrix.service == 'computing-unit-master' || matrix.service == 'computing-unit-worker') && needs.set-parameters.outputs.with_r_support == 'true' && 'WITH_R_SUPPORT=true' || '' }}
           labels: |
             org.opencontainers.image.title=${{ matrix.image_name }}
             org.opencontainers.image.description=Apache Texera ${{ matrix.image_name }} (ARM64)
@@ -487,6 +499,7 @@ jobs:
           echo "- **Tag:** \`${{ needs.set-parameters.outputs.image_tag }}\`" >> $GITHUB_STEP_SUMMARY
           echo "- **Services:** ${{ needs.set-parameters.outputs.services }}" >> $GITHUB_STEP_SUMMARY
           echo "- **Platforms:** ${{ needs.set-parameters.outputs.platforms }}" >> $GITHUB_STEP_SUMMARY
+          echo "- **R Support:** ${{ needs.set-parameters.outputs.with_r_support }}" >> $GITHUB_STEP_SUMMARY
           echo "" >> $GITHUB_STEP_SUMMARY
           echo "## Build Method" >> $GITHUB_STEP_SUMMARY
           echo "**Parallel platform builds** (faster)" >> $GITHUB_STEP_SUMMARY

amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGenerator.scala

@@ -304,10 +304,15 @@ class CostBasedScheduleGenerator(
     */
   private def createRegionDAG(): DirectedAcyclicGraph[Region, RegionLink] = {
     val searchResultFuture: Future[SearchResult] = Future {
-      if (ApplicationConfig.useTopDownSearch)
-        topDownSearch(globalSearch = ApplicationConfig.useGlobalSearch)
-      else
-        bottomUpSearch(globalSearch = ApplicationConfig.useGlobalSearch)
+      workflowContext.workflowSettings.executionMode match {
+        case ExecutionMode.MATERIALIZED =>
+          getFullyMaterializedSearchState
+        case ExecutionMode.PIPELINED =>
+          if (ApplicationConfig.useTopDownSearch)
+            topDownSearch(globalSearch = ApplicationConfig.useGlobalSearch)
+          else
+            bottomUpSearch(globalSearch = ApplicationConfig.useGlobalSearch)
+      }
     }
     val searchResult = Try(
       Await.result(searchResultFuture, ApplicationConfig.searchTimeoutMilliseconds.milliseconds)
@@ -477,6 +482,29 @@ class CostBasedScheduleGenerator(
     )
   }
 
+  /** Constructs a baseline fully materialized region plan (one operator per region) and evaluates its cost. */
+  def getFullyMaterializedSearchState: SearchResult = {
+    val startTime = System.nanoTime()
+
+    val (regionDAG, cost) =
+      tryConnectRegionDAG(physicalPlan.links) match {
+        case Left(dag) => (dag, allocateResourcesAndEvaluateCost(dag))
+        case Right(_) =>
+          (
+            new DirectedAcyclicGraph[Region, RegionLink](classOf[RegionLink]),
+            Double.PositiveInfinity
+          )
+      }
+
+    SearchResult(
+      state = Set.empty,
+      regionDAG = regionDAG,
+      cost = cost,
+      searchTimeNanoSeconds = System.nanoTime() - startTime,
+      numStatesExplored = 1
+    )
+  }
+
   /**
     * Another direction to perform the search. Depending on the configuration, either a global search or a greedy search
     * will be performed to find an optimal plan. The search starts from a plan where all edges are materialized, and

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

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