Yicong-Huang commented on code in PR #5707:
URL: https://github.com/apache/texera/pull/5707#discussion_r3410865325
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala:
##########
@@ -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)
+ )
+ }
Review Comment:
ok, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
