Copilot commented on code in PR #7712:
URL: https://github.com/apache/texera/pull/7712#discussion_r3801461537
##########
amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala:
##########
@@ -121,14 +209,221 @@ class WorkflowServiceSpec extends AnyFlatSpec with
Matchers {
}
}
- private final class TestWorkflowService(id: Long)
- extends WorkflowService(WorkflowIdentity(id), computingUnitId = 1,
cleanUpTimeoutSecs) {
+ private final class TestWorkflowService(id: Long, cuid: Int = 1)
+ extends WorkflowService(WorkflowIdentity(id), computingUnitId = cuid,
cleanUpTimeoutSecs) {
val lifecycle = new RecordingLifecycleManager
override val lifeCycleManager: WorkflowLifecycleManager = lifecycle
val results = new RecordingResultService(workflowId, computingUnitId,
stateStore)
override val resultService: ExecutionResultService = results
}
+ //
---------------------------------------------------------------------------
+ // Fixtures for the previous-run clean-up case. `initExecutionService` needs
the whole
+ // user / workflow / workflow_version / workflow_computing_unit chain before
it can insert
+ // its own execution row: workflow_executions.uid, .vid and .cuid are all
foreign keys.
+ //
---------------------------------------------------------------------------
+
+ // Four id domains, four different literals. `getLatestExecutionID(wid,
cuid)` binds two bare
+ // `Integer`s into one predicate and `insertNewExecution` takes wid and uid
side by side, so a
+ // fixture that reused one number for all of them would make every
transposition of those
+ // arguments produce byte-identical SQL.
+ private val testWid = 9411
+
+ /** A second workflow that shares `testCuid`, so the WID leg of the
predicate is not vacuous. */
+ private val otherWid = 9412
+ private val testUid = 9413
+
+ /** The computing unit the run under test executes on. */
+ private val testCuid = 9414
+
+ /** A second computing unit of the same workflow, so the clean-up's scope is
not vacuous. */
+ private val otherCuid = 9415
+
+ /** The unit the read-before-delete case runs on, kept apart from every
other case's rows. */
+ private val probeCuid = 9416
+
+ // Executions carry explicit eids rather than the SERIAL's: see the header on
+ // LargeBinaryManager. The relative order is load-bearing and is stated here
rather than
+ // inherited from insertion order.
+
+ /** An older, superseded execution of (`testWid`, `testCuid`). Must survive
untouched. */
+ private val olderEid = 941100
+
+ /** The newest execution of (`testWid`, `testCuid`): the run whose registry
must be cleared. */
+ private val previousEid = 941101
+
+ /**
+ * A LATER execution of the same workflow on a different computing unit.
Its eid is larger than
+ * `previousEid` on purpose: `getLatestExecutionID` picks the maximum eid
*among the rows of
+ * the requested computing unit*, so without the CUID leg this row would be
chosen instead.
+ */
+ private val otherUnitEid = 941102
+
+ /**
+ * A LATER execution of a *different* workflow on the SAME computing unit.
`workflow_computing_unit`
+ * is keyed by uid and carries no wid, so one unit legitimately runs many
workflows; without the
+ * WID leg this row is what `getLatestExecutionID` returns, and a new run
of `testWid` would
+ * wipe an unrelated workflow's registry.
+ */
+ private val otherWorkflowEid = 941103
+
+ /** The read-before-delete case's execution; the only fixture with a
non-NULL URI column. */
+ private val probeEid = 941104
+
+ private val fixtureEids: Seq[Integer] =
+ Seq(olderEid, previousEid, otherUnitEid, otherWorkflowEid,
probeEid).map(Integer.valueOf)
+
+ private var executingUser: User = _
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+ val cfg = getDSLContext.configuration()
+
+ executingUser = new User
+ executingUser.setUid(testUid)
+ executingUser.setName("workflow_service_spec_user")
+
executingUser.setEmail(s"workflow-service-${UUID.randomUUID()}@example.com")
+ new UserDao(cfg).insert(executingUser)
+
+ val workflowDao = new WorkflowDao(cfg)
+ val versionDao = new WorkflowVersionDao(cfg)
+ List(testWid -> "workflow_service_spec_workflow", otherWid ->
"workflow_service_spec_other")
+ .foreach {
+ case (wid, name) =>
+ val workflow = new Workflow
+ workflow.setWid(wid)
+ workflow.setName(name)
+ workflow.setContent("{}")
+ workflow.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ workflow.setLastModifiedTime(new
Timestamp(System.currentTimeMillis()))
+ workflowDao.insert(workflow)
+
+ val version = new WorkflowVersion
+ version.setWid(wid)
+ version.setContent("{}")
+ version.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ versionDao.insert(version)
+ }
+
+ val unitDao = new WorkflowComputingUnitDao(cfg)
+ List(
+ testCuid -> "workflow_service_spec_unit",
+ otherCuid -> "workflow_service_spec_other_unit",
+ probeCuid -> "workflow_service_spec_probe_unit"
+ ).foreach {
+ case (cuid, name) =>
+ val unit = new WorkflowComputingUnit
+ unit.setCuid(cuid)
+ unit.setUid(testUid)
+ unit.setName(name)
+ unit.setType(WorkflowComputingUnitTypeEnum.local)
+ unit.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ unitDao.insert(unit)
+ }
+
+ insertExecution(olderEid, testWid, testCuid, "older-run")
+ insertExecution(previousEid, testWid, testCuid, "previous-run")
+ insertExecution(otherUnitEid, testWid, otherCuid, "other-unit-run")
+ insertExecution(otherWorkflowEid, otherWid, testCuid, "other-workflow-run")
+ insertExecution(probeEid, testWid, probeCuid, "probe-run")
Review Comment:
Because `workflow_executions.eid` is a `SERIAL` (see
`sql/texera_ddl.sql:286-307`), inserting explicit fixture eids (941100+) does
**not** advance the underlying sequence. The next production insert in
`initExecutionService` will therefore typically allocate `eid = 1`, which
reintroduces the `DEFAULT_EXECUTION_ID`/MinIO blast-radius risk and also makes
any later “latest execution = max(eid)” logic behave unlike production.
Consider advancing the sequence past the highest fixture eid after seeding
the fixture executions.
--
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]
