feat(scheduling): reuse output storage across region re-executions

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

@@ -60,10 +60,43 @@ import org.apache.texera.web.SessionState
 import org.apache.texera.web.model.websocket.event.RegionStateEvent
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource
 
+import java.net.URI
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
 import scala.concurrent.duration.{Duration => ScalaDuration}
 
+object RegionExecutionCoordinator {
+
+  /**
+    * Decide whether to (re)create the output document at `uri`, then act.
+    *
+    * When `reuseExistingStorage` is set and the document already exists, the
+    * existing document is kept untouched -- this is how an operator whose
+    * region re-executes (e.g. LoopEnd, which accumulates output across loop
+    * iterations) avoids clobbering output an earlier run produced, since
+    * `createDocument` overrides any existing document. Otherwise the document
+    * is created.
+    *
+    * `documentExists` / `createDocument` are injected so the create-or-reuse
+    * decision can be unit-tested without an iceberg backend or a live region.
+    *
+    * @return true iff `createDocument` was invoked.
+    */
+  def provisionOutputDocument(
+      uri: URI,
+      reuseExistingStorage: Boolean,
+      documentExists: URI => Boolean,
+      createDocument: URI => Unit
+  ): Boolean = {
+    if (reuseExistingStorage && documentExists(uri)) {
+      false
+    } else {
+      createDocument(uri)
+      true
+    }
+  }
+}
+
 /**
   * The executor of a region.
   *
@@ -576,8 +609,29 @@ class RegionExecutionCoordinator(
           region.getOperator(outputPortId.opId).outputPorts(outputPortId.portId)._3
         val schema =
           schemaOptional.getOrElse(throw new IllegalStateException("Schema is missing"))
-        DocumentFactory.createDocument(resultURI, schema)
-        DocumentFactory.createDocument(stateURI, State.schema)
+        // Operators that reuse their output storage across region re-runs
+        // (e.g. LoopEnd, whose output accumulates across the iterations of its
+        // own loop) already have their result/state documents from a prior
+        // run; on re-execution `createDocument` (overrideIfExists=true) would
+        // clobber them, so reuse the existing document when it is already
+        // there. (The inner LoopEnd of a nested loop additionally drops its
+        // output once per outer iteration -- on the Python worker side in
+        // MainLoop._process_state_frame -- which is orthogonal to this
+        // region-provisioning reuse.)
+        // Decided per the operator that OWNS this port, not region-wide: a
+        // region mixing a reuse op (LoopEnd) with others must still recreate
+        // the others' documents on re-execution.
+        val reusesOutputStorage =
+          region.getOperator(outputPortId.opId).reusesOutputStorageOnReExecution
+        Seq((resultURI, schema), (stateURI, State.schema)).foreach {
+          case (uri, sch) =>
+            RegionExecutionCoordinator.provisionOutputDocument(
+              uri,
+              reusesOutputStorage,
+              DocumentFactory.documentExists,
+              u => DocumentFactory.createDocument(u, sch)
+            )
+        }
         if (!isRestart) {
           val (_, eid, _, _) = decodeURI(resultURI)
           WorkflowExecutionsResource.insertOperatorPortResultUri(

amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionOutputProvisioningSpec.scala

@@ -0,0 +1,105 @@
+/*
+ * 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.engine.architecture.scheduling
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.net.URI
+import scala.collection.mutable
+
+/**
+  * Unit tests for `RegionExecutionCoordinator.provisionOutputDocument`, the
+  * create-or-reuse decision behind output-port storage provisioning.
+  *
+  * This is the branch that lets a re-executing region (a loop body) keep the
+  * output an earlier run accumulated instead of clobbering it: a LoopEnd's
+  * region runs once per iteration, and `DocumentFactory.createDocument`
+  * overrides any existing document, so on a re-run we must reuse the existing
+  * document rather than recreate it.
+  *
+  * The decision was pulled out of the private `createOutputPortStorageObjects`
+  * (which needs a live controller + iceberg backend) into a pure function with
+  * injected `documentExists` / `createDocument`, so the four cases can be
+  * pinned directly with a spy -- no iceberg, no actor system.
+  */
+class RegionOutputProvisioningSpec extends AnyFlatSpec with Matchers {
+
+  private val uri = new URI("vfs:///wf/result/loop-end")
+
+  /** Run provisionOutputDocument and return (created?, number of create calls). */
+  private def provision(
+      reuseExistingStorage: Boolean,
+      exists: Boolean
+  ): (Boolean, Int) = {
+    val createCalls = mutable.ArrayBuffer.empty[URI]
+    val created = RegionExecutionCoordinator.provisionOutputDocument(
+      uri,
+      reuseExistingStorage,
+      _ => exists,
+      u => { createCalls += u; () }
+    )
+    (created, createCalls.size)
+  }
+
+  "provisionOutputDocument" should
+    "reuse (not recreate) an existing document when the operator reuses storage" in {
+    // The loop-iteration case: the document is already there from a prior
+    // region run, so createDocument must NOT be called -- otherwise the
+    // accumulated output would be clobbered.
+    val (created, createCalls) = provision(reuseExistingStorage = true, exists = true)
+    created shouldBe false
+    createCalls shouldBe 0
+  }
+
+  it should "create the document when the operator reuses storage but none exists yet" in {
+    // First iteration: nothing to reuse, so it must be created.
+    val (created, createCalls) = provision(reuseExistingStorage = true, exists = false)
+    created shouldBe true
+    createCalls shouldBe 1
+  }
+
+  it should "always (re)create when the operator does not reuse storage, even if a document exists" in {
+    // Non-loop operators get a fresh document every region execution; an
+    // existing one is intentionally overwritten.
+    val (created, createCalls) = provision(reuseExistingStorage = false, exists = true)
+    created shouldBe true
+    createCalls shouldBe 1
+  }
+
+  it should "create when the operator does not reuse storage and none exists" in {
+    val (created, createCalls) = provision(reuseExistingStorage = false, exists = false)
+    created shouldBe true
+    createCalls shouldBe 1
+  }
+
+  it should "not call documentExists when the operator does not reuse storage (create unconditionally)" in {
+    // Short-circuit: a non-reuse operator always recreates, so it must not
+    // even probe for existence.
+    var existsProbed = false
+    RegionExecutionCoordinator.provisionOutputDocument(
+      uri,
+      reuseExistingStorage = false,
+      _ => { existsProbed = true; true },
+      _ => ()
+    )
+    existsProbed shouldBe false
+  }
+}

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

@@ -198,6 +198,12 @@ case class PhysicalOp(
     // schema propagation function
     propagateSchema: SchemaPropagationFunc = SchemaPropagationFunc(schemas => schemas),
     isOneToManyOp: Boolean = false,
+    // Whether to reuse this operator's existing output storage instead of
+    // recreating it when its region re-executes, so output accumulated by
+    // earlier runs (e.g. across loop iterations) survives. Named after the
+    // behavior the scheduler checks, not the operator that sets it, so any
+    // future operator needing the same treatment can reuse it.
+    reusesOutputStorageOnReExecution: Boolean = false,
     // hint for number of workers
     suggestedWorkerNum: Option[Int] = None,
     // name of the PVE to execute within
@@ -316,6 +322,17 @@ case class PhysicalOp(
   def withIsOneToManyOp(isOneToManyOp: Boolean): PhysicalOp =
     this.copy(isOneToManyOp = isOneToManyOp)
 
+  /**
+    * Creates a copy specifying whether this operator's output storage is
+    * reused rather than recreated when its region re-executes (see the field
+    * doc). The region scheduler uses it to preserve iceberg output across
+    * loop iterations instead of overwriting it on every region invocation.
+    */
+  def withReusesOutputStorageOnReExecution(
+      reusesOutputStorageOnReExecution: Boolean
+  ): PhysicalOp =
+    this.copy(reusesOutputStorageOnReExecution = reusesOutputStorageOnReExecution)
+
   /**
     * Creates a copy of the PhysicalOp with the schema of a specified input port updated.
     * The schema can either be a successful schema definition or an error represented as a Throwable.