This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7554-cb6e5c649978215188ea92937ec583be9990c49e in repository https://gitbox.apache.org/repos/asf/texera.git
commit a353b71c231f89d4d33a5fffa69be84d1ef675c5 Author: Xinyuan Lin <[email protected]> AuthorDate: Tue Aug 11 20:57:48 2026 -0700 test(amber): cover the synchronous execution endpoint (#7554) ### What changes were proposed in this PR? `SyncExecutionResource` was the largest uncovered file in the repository — 406 instrumented lines, no spec, **0%**. It is the endpoint an external caller uses to run a workflow and get results back in one request, so its result assembly, error classification and cell-truncation rules are all user-visible. Adds 23 tests, taking the file to **56.7% of lines** (230/406). Nothing here needs infrastructure: the class is a zero-arg Jersey resource that constructs with no fixture, and MockTexeraDB plus a single `workflow_computing_unit` row is enough to drive the public endpoint end to end — `initExecutionService` absorbs the missing engine into a FAILED state, after which the whole result-assembly tail runs. Covered: the state mapping and terminal-state predicate, console error detection, symmetric cell truncation and tuple-size estimation, error classification, sub-DAG computation, the 100-line operator-info aggregator, console-log retrieval, and `executeWorkflowSync` itself. ### Verification 52 mutations applied one at a time and reverted, with the production file's md5 compared against its pre-mutation value after each revert. Production diff empty. Highlights: swapping input for output metrics in the aggregator, taking the console title instead of the longer message as the error, walking the sub-DAG by the wrong link end, dropping the visited-set early return, promoting a mid-line "WARNING:" mention to a real warning, and reporting KILLED as FAILED. **Two assertions were found vacuous in review; one was fixed and one is reported as unpinnable.** *Fixed* — the in-memory console fallback's `.filter(_.nonEmpty)` could be dropped and nothing noticed, because no fixture had an operator that was *present* in `operatorConsole` while carrying zero messages. That state is real: `ExecutionConsoleService` creates exactly that shape via `getOrElse(opId, OperatorConsole())` before it has anything to add. Added a `"silent"` operator and asserted `consoleLogs` is `None` rather than `Some(Nil)` — the frontend renders a console pane for `Some`, so `Some(Nil)` is an empty pane. That mutation is now red. *Reported, not papered over* — inverting the console-error arm at line 328 leaves the suite green. The one test that exercises `stateString` runs a path where `terminatedByConsoleError` is false **and** `stateToString(finalState.state)` is also `"Failed"`, so the assertion cannot tell the two arms apart. Distinguishing them needs a run whose final state is not FAILED, which is impossible without a live engine — with no coordinator, `initExecutionService` always stamps FAILED. This is stated in the spec at that test rather than left for the next reader to discover. One further mutation survived and is an **equivalent mutant** rather than a gap: `truncateSingleTuple`'s `text.length > maxCellChars` flipped to `>=`. The truncation it guards re-checks the same bound and returns the cell unchanged, so no input can distinguish the two spellings. The test was kept (the behaviour is real) and its comment corrected to say why the boundary is unobservable. ### Deliberately not included - **~139 lines** behind real Iceberg result storage (`collectOperatorResult`, the symmetric-truncation engine). `DocumentFactory.openDocument` is an un-displaceable static for `vfs://` URIs. `common/workflow-core`'s `LocalHadoopIcebergCatalog` would unlock it, but amber declares only `DAO % "test->test", Auth % "test->test"` (build.sbt:270), so it is not on amber's test classpath. Adding `WorkflowCore % "test->test"` would take this file to roughly 87% — worth doing, but as its own change. - **The `Observable.amb` wait and its timeout/termination arms**, which need an execution that is still non-terminal, i.e. a live engine. - **`validateWorkflow` (905–924)** — it has **zero call sites** repo-wide. It can be driven reflectively, which is not the same as being live; testing it would cement dead code. Delete instead. - **The `if (executionService == null)` early return**, effectively unreachable: `initExecutionService` publishes before calling `executeWorkflow()`, and the constructor is documented and verified side-effect-free. ### Known follow-up Most helpers are reached through `PrivateMethodTester` because 14 of the 16 methods are `private def` (there is repo precedent — eight existing specs use it). One case was avoidable: `handleExecutionError` is reachable from the public endpoint by passing a wid with no `workflow` row, which would additionally cover lines 343–346. Left as noted rather than restructured here. Widening the nine pure helpers to `private[resource]` would remove nearly all the reflection for a one-word change each. No production file is touched. ### Any related issues, documentation, discussions? Closes #7553 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.web.resource.SyncExecutionResourceSpec" ``` ``` [info] Total number of tests run: 23 [info] Tests: succeeded 23, failed 0, canceled 0, ignored 0, pending 0 ``` Coverage measured with sbt-jacoco filtered to this spec (a bare `jacoco` runs amber's `@IntegrationTest` specs, which hang on Windows): 0/406 before, 230/406 after. `Test/scalafmtCheck` and `Test/scalafix --check` both pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../web/resource/SyncExecutionResourceSpec.scala | 890 +++++++++++++++++++++ 1 file changed, 890 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala new file mode 100644 index 0000000000..ef0825a470 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala @@ -0,0 +1,890 @@ +/* + * 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.web.resource + +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, OperatorIdentity} +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext, WorkflowSettings} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + ConsoleMessage, + ConsoleMessageType +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState._ +import org.apache.texera.amber.engine.architecture.worker.statistics.{ + PortTupleMetricsMapping, + TupleMetrics +} +import org.apache.texera.amber.engine.common.executionruntimestate.{ + ExecutionConsoleStore, + OperatorConsole, + OperatorMetrics, + OperatorStatistics +} +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.auth.SessionUser +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.Tables.OPERATOR_EXECUTIONS +import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowComputingUnitDao, + WorkflowDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowComputingUnit, + WorkflowVersion +} +import org.apache.texera.web.model.websocket.request.WorkflowExecuteRequest +import org.apache.texera.web.service.{WarehouseUnavailableException, WorkflowExecutionService} +import org.apache.texera.web.storage.ExecutionStateStore +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, PrivateMethodTester} + +import java.net.URI +import java.sql.Timestamp + +/** + * Unit tests for the synchronous "run this workflow and hand me the results" endpoint used by + * agent-service. + * + * Two of the resource's methods are public (`executeWorkflowSync`, `healthCheck`); the other + * sixteen are `private def`. `executeWorkflowSync` is driven for real here — one end-to-end case + * that gets as far as a running engine can be taken without one — but the result-shaping helpers + * it calls cannot be reached that way: every one of them only produces an observable difference + * when the engine has actually populated the stats/console stores and written Iceberg results, + * which no unit test can arrange. They are therefore driven through `PrivateMethodTester`, and + * each such test says which engine state it is standing in for. + * + * Deliberately NOT covered, and why: + * - `validateWorkflow` (lines 905-924). It has zero call sites: nothing in the repository, in + * any language, invokes it (agent-service has a same-named TypeScript function of its own, + * which is unrelated). It is reachable by reflection, but reflective reachability is not + * liveness — a test here would only cement dead code and make its deletion look like a + * regression. Reported as dead instead. + * - `shutdownPreviousExecution` (350-361) and `killExecution`'s `client.shutdown()` line (368). + * Both need an `AmberClient` test double; amber's test scope has no mocking library, and + * `AmberClient`'s constructor needs a started `AmberRuntime`. `shutdownPreviousExecution` is + * additionally unpinnable in principle: its whole body sits inside `catch (Exception) => warn`, + * so removing either null guard turns a no-op into a *swallowed* NPE — no observable change. + * - the body of `collectOperatorResult` past its `case None` (lines 541-695) and the + * database-authoritative half of `collectConsoleLogs`. Both need a real Iceberg document + * behind a registered result/console URI. The tuple-shaping helpers that body delegates to + * (`truncateSingleTuple`, `estimateTupleSize`, `symmetricTruncateCellValue`, + * `isVisualizationTuple`) are pinned directly instead. + * - the `Observable.amb` wait and its timeout/error handlers (lines 231-277), plus the + * `ConsoleErrorDetected` / `TargetResultsReady` termination arms (284-298). Reaching them + * needs an execution that is still non-terminal when `executeWorkflowSync` looks at it, i.e. + * a live engine. + */ +class SyncExecutionResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB + with PrivateMethodTester { + + /** + * Repo-unique ids. `executeWorkflowSync` registers a `WorkflowService` in a JVM-wide map that + * nothing in a test run ever removes, and result/console storage URIs are derived from the + * workflow id, so a value another suite also uses could collide inside the single JVM sbt runs + * amber's suites in. 9317 appears nowhere else in the repository. + */ + private val testWid = 9317 + private val testUid = 9317 + private val testCuid = 9317 + + /** + * The console-URI fixtures below belong to a SECOND computing unit. `initExecutionService` + * treats the newest execution of (workflow, computing unit) as "the previous run" and cleans + * its registered storage up; parking the fixtures elsewhere keeps the `executeWorkflowSync` + * case from tripping over URIs that point at documents no test ever created. + */ + private val fixtureCuid = 9318 + + private val resource = new SyncExecutionResource() + + private val stateToString = PrivateMethod[String](Symbol("stateToString")) + private val isTerminalState = PrivateMethod[Boolean](Symbol("isTerminalState")) + private val hasConsoleError = PrivateMethod[Boolean](Symbol("hasConsoleError")) + private val symmetricTruncateCellValue = + PrivateMethod[String](Symbol("symmetricTruncateCellValue")) + private val truncateSingleTuple = PrivateMethod[ObjectNode](Symbol("truncateSingleTuple")) + private val estimateTupleSize = PrivateMethod[Int](Symbol("estimateTupleSize")) + private val isVisualizationTuple = PrivateMethod[Boolean](Symbol("isVisualizationTuple")) + private val handleExecutionError = + PrivateMethod[SyncExecutionResult](Symbol("handleExecutionError")) + private val computeSubDAGIfNeeded = + PrivateMethod[LogicalPlanPojo](Symbol("computeSubDAGIfNeeded")) + private val collectOperatorInfos = + PrivateMethod[Map[String, OperatorInfo]](Symbol("collectOperatorInfos")) + private val getConsoleMessageUri = PrivateMethod[Option[URI]](Symbol("getConsoleMessageUri")) + private val killExecution = PrivateMethod[Unit](Symbol("killExecution")) + + /** eid of the row `getConsoleMessageUri`'s cases hang their operator_executions rows off. */ + private var consoleEid: Int = 0 + + /** A second eid, so the query's execution-id predicate is not vacuous. */ + private var otherEid: Int = 0 + + override protected def beforeAll(): Unit = { + initializeDBAndReplaceDSLContext() + val cfg = getDSLContext.configuration() + + val user = new User + user.setUid(testUid) + user.setName("sync_exec_user") + user.setEmail("[email protected]") + new UserDao(cfg).insert(user) + + val workflow = new Workflow + workflow.setWid(testWid) + workflow.setName("sync_exec_workflow") + workflow.setContent("{}") + workflow.setCreationTime(new Timestamp(System.currentTimeMillis())) + workflow.setLastModifiedTime(new Timestamp(System.currentTimeMillis())) + new WorkflowDao(cfg).insert(workflow) + + val version = new WorkflowVersion + version.setWid(testWid) + version.setContent("{}") + version.setCreationTime(new Timestamp(System.currentTimeMillis())) + new WorkflowVersionDao(cfg).insert(version) + + // workflow_executions.cuid is a foreign key; without these rows initExecutionService cannot + // insert its execution and never reaches the engine at all. + val unitDao = new WorkflowComputingUnitDao(cfg) + List(testCuid -> "sync_exec_unit", fixtureCuid -> "sync_exec_fixture_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) + } + + consoleEid = insertExecution("console-fixture") + otherEid = insertExecution("console-fixture-other") + } + + override protected def afterAll(): Unit = closeConnectionPool() + + private def insertExecution(name: String): Int = { + import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW_EXECUTIONS, WORKFLOW_VERSION} + val vid = getDSLContext + .select(WORKFLOW_VERSION.VID) + .from(WORKFLOW_VERSION) + .where(WORKFLOW_VERSION.WID.eq(testWid)) + .fetchOneInto(classOf[Integer]) + getDSLContext + .insertInto(WORKFLOW_EXECUTIONS) + .set(WORKFLOW_EXECUTIONS.VID, vid) + .set(WORKFLOW_EXECUTIONS.UID, Integer.valueOf(testUid)) + .set(WORKFLOW_EXECUTIONS.CUID, Integer.valueOf(fixtureCuid)) + .set(WORKFLOW_EXECUTIONS.NAME, name) + .set(WORKFLOW_EXECUTIONS.ENVIRONMENT_VERSION, "test") + .returning(WORKFLOW_EXECUTIONS.EID) + .fetchOne() + .getEid + } + + private def insertConsoleUri(eid: Int, opId: String, uri: String): Unit = { + getDSLContext + .insertInto(OPERATOR_EXECUTIONS) + .set(OPERATOR_EXECUTIONS.WORKFLOW_EXECUTION_ID, Integer.valueOf(eid)) + .set(OPERATOR_EXECUTIONS.OPERATOR_ID, opId) + .set(OPERATOR_EXECUTIONS.CONSOLE_MESSAGES_URI, uri) + .execute() + } + + /** + * An execution with no coordinator, no result service and no client — the pattern + * `WorkflowExecutionServiceSpec` and `WorkflowServiceSpec` establish: construction does no + * external work, and `client` stays null until `executeWorkflow()` builds one. + */ + private def newExecution(stateStore: ExecutionStateStore): WorkflowExecutionService = + new WorkflowExecutionService( + null, + new WorkflowContext(), + null, + WorkflowExecuteRequest( + executionName = "test", + engineVersion = "test", + logicalPlan = LogicalPlanPojo(List.empty, List.empty, List.empty, List.empty), + replayFromExecution = None, + workflowSettings = WorkflowSettings(), + emailNotificationEnabled = false, + computingUnitId = testCuid, + warehouseId = None + ), + stateStore, + (_: Throwable) => (), + None, + new URI("vfs:///sync-execution-spec") + ) + + private def metrics( + state: WorkflowAggregatedState, + inputs: Seq[(Int, Long)], + outputs: Seq[(Int, Long)] + ): OperatorMetrics = + OperatorMetrics( + operatorState = state, + operatorStatistics = OperatorStatistics( + inputMetrics = inputs.map { + case (port, count) => PortTupleMetricsMapping(PortIdentity(port), TupleMetrics(count, 0)) + }, + outputMetrics = outputs.map { + case (port, count) => PortTupleMetricsMapping(PortIdentity(port), TupleMetrics(count, 0)) + } + ) + ) + + private def console( + messages: (ConsoleMessageType, String, String)* + ): OperatorConsole = + OperatorConsole(consoleMessages = messages.map { + case (msgType, title, message) => + ConsoleMessage.defaultInstance + .withMsgType(msgType) + .withTitle(title) + .withMessage(message) + }) + + private def op(id: String): LogicalOp = { + val logicalOp = new KeywordSearchOpDesc() + logicalOp.setOperatorId(id) + logicalOp + } + + private def link(from: String, to: String): LogicalLink = + LogicalLink(OperatorIdentity(from), PortIdentity(), OperatorIdentity(to), PortIdentity()) + + /** 30 characters, so cell budgets below it actually truncate. */ + private val thirtyChars = "0123456789ABCDEFGHIJKLMNOPQRST" + + "healthCheck" should "report ok under the key the caller polls" in { + resource.healthCheck shouldBe Map("status" -> "ok") + } + + "stateToString" should "name every aggregated state and fall back to Unknown" in { + // The whole mapping in one assertion: a swapped or dropped case shows up as a diff. + val named = List( + UNINITIALIZED, + READY, + RUNNING, + PAUSING, + PAUSED, + RESUMING, + COMPLETED, + FAILED, + KILLED, + TERMINATED + ).map(state => state -> (resource invokePrivate stateToString(state))).toMap + + named shouldBe Map( + UNINITIALIZED -> "Uninitialized", + READY -> "Ready", + RUNNING -> "Running", + PAUSING -> "Pausing", + PAUSED -> "Paused", + RESUMING -> "Resuming", + COMPLETED -> "Completed", + FAILED -> "Failed", + KILLED -> "Killed", + TERMINATED -> "Terminated" + ) + + // WorkflowAggregatedState also declares UNKNOWN, and protobuf can deliver a value this build + // does not know; both must land on the catch-all rather than crash the response. + resource invokePrivate stateToString(WorkflowAggregatedState.UNKNOWN) shouldBe "Unknown" + resource invokePrivate stateToString( + WorkflowAggregatedState.Unrecognized(99) + ) shouldBe "Unknown" + } + + "isTerminalState" should "treat exactly the four finished states as terminal" in { + // Partitioning every declared state, so neither adding nor dropping a case survives: a state + // wrongly called terminal makes executeWorkflowSync skip waiting for the run to finish. + val terminal = List( + UNINITIALIZED, + READY, + RUNNING, + PAUSING, + PAUSED, + RESUMING, + COMPLETED, + FAILED, + WorkflowAggregatedState.UNKNOWN, + KILLED, + TERMINATED + ).filter(state => resource invokePrivate isTerminalState(state)) + + terminal should contain theSameElementsAs List(COMPLETED, FAILED, KILLED, TERMINATED) + } + + "hasConsoleError" should "find an ERROR anywhere in any operator's console" in { + resource invokePrivate hasConsoleError(ExecutionConsoleStore()) shouldBe false + + val printsOnly = ExecutionConsoleStore(operatorConsole = + Map("op-1" -> console((ConsoleMessageType.PRINT, "hello", ""))) + ) + resource invokePrivate hasConsoleError(printsOnly) shouldBe false + + // The ERROR is the SECOND message of the operator, so reading only the first message survives + // neither this case nor production: a Python operator prints before it raises. + val errorAfterPrint = ExecutionConsoleStore(operatorConsole = + Map( + "op-1" -> console( + (ConsoleMessageType.PRINT, "hello", ""), + (ConsoleMessageType.ERROR, "boom", "") + ) + ) + ) + resource invokePrivate hasConsoleError(errorAfterPrint) shouldBe true + + // ... and it belongs to the SECOND operator, so scanning only one entry of the map fails too. + val errorInSecondOperator = ExecutionConsoleStore(operatorConsole = + Map( + "op-1" -> console((ConsoleMessageType.PRINT, "hello", "")), + "op-2" -> console((ConsoleMessageType.ERROR, "boom", "")) + ) + ) + resource invokePrivate hasConsoleError(errorInSecondOperator) shouldBe true + + // COMMAND and DEBUGGER must not be mistaken for failures. + val nonErrorTypes = ExecutionConsoleStore(operatorConsole = + Map( + "op-1" -> console( + (ConsoleMessageType.COMMAND, "cmd", ""), + (ConsoleMessageType.DEBUGGER, "dbg", "") + ) + ) + ) + resource invokePrivate hasConsoleError(nonErrorTypes) shouldBe false + } + + "symmetricTruncateCellValue" should "keep both ends of an oversized cell" in { + // 30 chars, budget 27: the notice costs 17, leaving 10 to split 5/5. + resource invokePrivate symmetricTruncateCellValue( + thirtyChars, + 27 + ) shouldBe "01234...[truncated]...PQRST" + + // An odd leftover budget is floored, not rounded: 28 - 17 = 11 -> 5 per side, same output. + resource invokePrivate symmetricTruncateCellValue( + thirtyChars, + 28 + ) shouldBe "01234...[truncated]...PQRST" + } + + it should "leave a cell that fits alone, including at exactly the budget" in { + resource invokePrivate symmetricTruncateCellValue("short", 27) shouldBe "short" + // Length == budget must NOT be truncated; `<` instead of `<=` would mangle it. + resource invokePrivate symmetricTruncateCellValue(thirtyChars, 30) shouldBe thirtyChars + } + + it should "hard-cut when the budget cannot even hold the notice" in { + // "...[truncated]..." is 17 chars, so a 17-char budget leaves nothing to keep on either side. + resource invokePrivate symmetricTruncateCellValue( + thirtyChars, + 17 + ) shouldBe "0123456789ABCDEFG" + resource invokePrivate symmetricTruncateCellValue(thirtyChars, 5) shouldBe "01234" + } + + "truncateSingleTuple" should "shorten only textual cells over the budget" in { + val mapper = new ObjectMapper() + val tuple = mapper.createObjectNode() + tuple.put("fits", "0123456789") + // Exactly at the budget: must survive untouched. Note that this pins the *behaviour*, not the + // comparison operator — flipping this method's `>` to `>=` changes nothing, because + // symmetricTruncateCellValue re-checks the same bound and returns the cell as-is. The two + // spellings are equivalent, so no input can tell them apart. + tuple.put("exact", "01234567890123456789") + tuple.put("oversized", "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ") + tuple.put("count", 42) + tuple.put("flag", true) + tuple.putNull("missing") + + val truncated = resource invokePrivate truncateSingleTuple(tuple, 20) + + // Asserted as serialized JSON so field order — which decides what the caller renders as + // columns — is pinned along with the values. + truncated.toString shouldBe + """{"fits":"0123456789","exact":"01234567890123456789",""" + + """"oversized":"0...[truncated]...Z","count":42,"flag":true,"missing":null}""" + + // A copy, not an edit in place: the caller still holds the untruncated tuple. + tuple.get("oversized").asText() shouldBe "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" + } + + "estimateTupleSize" should "count the serialized cell plus its array separator" in { + val mapper = new ObjectMapper() + val tuple = mapper.createObjectNode() + tuple.put("a", 1) + // {"a":1} is 7 characters; the extra 1 is the comma this tuple will need in the JSON array, + // and dropping it makes every truncation budget quietly too generous. + resource invokePrivate estimateTupleSize(tuple, mapper) shouldBe 8 + } + + "isVisualizationTuple" should "recognize exactly the two visualization payload columns" in { + def tupleWith(names: String*): Tuple = { + val schema = new Schema(names.map(new Attribute(_, AttributeType.STRING)).toList) + names + .foldLeft(Tuple.builder(schema))((builder, name) => + builder.add(name, AttributeType.STRING, "x") + ) + .build() + } + + resource invokePrivate isVisualizationTuple(tupleWith("html-content")) shouldBe true + resource invokePrivate isVisualizationTuple(tupleWith("json-content")) shouldBe true + // Present alongside ordinary columns, and not first, so a head-only check fails. + resource invokePrivate isVisualizationTuple(tupleWith("region", "json-content")) shouldBe true + + resource invokePrivate isVisualizationTuple(tupleWith("region", "sales")) shouldBe false + // Near-misses: the match is on the whole column name, not a substring of it — a table with a + // "content" or "html" column must still render as a table. + resource invokePrivate isVisualizationTuple(tupleWith("content", "html")) shouldBe false + resource invokePrivate isVisualizationTuple(tupleWith("my-html-content")) shouldBe false + + // The catch: a malformed tuple must degrade to "not a visualization" rather than fail the + // whole result assembly. Nothing else reaches it, since getSchema cannot throw on its own. + resource invokePrivate isVisualizationTuple(null.asInstanceOf[Tuple]) shouldBe false + } + + "handleExecutionError" should "report a compilation-flavoured failure separately" in { + // The four keywords the classifier looks for, each on its own message. + val compilationMessages = List( + "compilation of the plan failed", + "Compilation error at op-3", + "unknown operator scan-1", + "schema mismatch on port 0" + ) + + compilationMessages.foreach { message => + val result = resource invokePrivate handleExecutionError(new Exception(message)) + withClue(s"for message '$message': ") { + result shouldBe SyncExecutionResult( + success = false, + state = "CompilationFailed", + operators = Map.empty, + compilationErrors = Some(Map("error" -> message)), + errors = Some(List(message)) + ) + } + } + } + + it should "report any other failure as a plain error, with no compilation detail" in { + resource invokePrivate handleExecutionError( + new Exception("connection reset by peer") + ) shouldBe SyncExecutionResult( + success = false, + state = "Error", + operators = Map.empty, + compilationErrors = None, + errors = Some(List("connection reset by peer")) + ) + } + + it should "survive an exception with no message" in { + // A null message must take the plain-error arm (the keyword scan would NPE on it) and still + // produce something the caller can display. + resource invokePrivate handleExecutionError(new Exception()) shouldBe SyncExecutionResult( + success = false, + state = "Error", + operators = Map.empty, + compilationErrors = None, + errors = Some(List("Unknown error")) + ) + } + + "computeSubDAGIfNeeded" should "run only the target's upstream sub-DAG" in { + // a -> b -> c, d -> c, c -> z, and an unrelated e. + val plan = LogicalPlanPojo( + operators = List(op("a"), op("b"), op("c"), op("d"), op("e"), op("z")), + links = List(link("a", "b"), link("b", "c"), link("d", "c"), link("c", "z")), + opsToViewResult = List("z"), + opsToReuseResult = List("a", "e") + ) + + val subDag = resource invokePrivate computeSubDAGIfNeeded(plan, List("c")) + + // "z" is downstream of the target and "e" is unrelated: both must be dropped, which is the + // whole point of "Execute To". Ids, not instances, because dfs order is map-iteration order. + subDag.operators.map(_.operatorIdentifier.id) should contain theSameElementsAs + List("a", "b", "c", "d") + subDag.links should contain theSameElementsAs List( + link("a", "b"), + link("b", "c"), + link("d", "c") + ) + subDag.opsToViewResult shouldBe List("c") + // "e" was never visited, so reusing its cached result would reference an operator that is not + // in the plan handed to the compiler. + subDag.opsToReuseResult shouldBe List("a") + } + + it should "visit a diamond's shared ancestor exactly once" in { + // a -> b -> d and a -> c -> d: without the visited set, "a" is collected twice and the + // compiler sees a duplicate operator. + val plan = LogicalPlanPojo( + operators = List(op("a"), op("b"), op("c"), op("d")), + links = List(link("a", "b"), link("a", "c"), link("b", "d"), link("c", "d")), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + + val subDag = resource invokePrivate computeSubDAGIfNeeded(plan, List("d")) + + subDag.operators.map(_.operatorIdentifier.id) should contain theSameElementsAs + List("a", "b", "c", "d") + subDag.operators should have size 4 + subDag.links should have size 4 + } + + it should "keep the full plan unless exactly one target is named" in { + val plan = LogicalPlanPojo( + operators = List(op("a"), op("b")), + links = List(link("a", "b")), + opsToViewResult = List("b"), + opsToReuseResult = List("a") + ) + + // No target: a plain "run everything" request. + (resource invokePrivate computeSubDAGIfNeeded( + plan, + List.empty + )) should be theSameInstanceAs plan + // Two targets: sub-DAG semantics are undefined, so the full DAG runs. + (resource invokePrivate computeSubDAGIfNeeded( + plan, + List("a", "b") + )) should be theSameInstanceAs plan + // One target the plan does not contain: fall back rather than run an empty plan. + (resource invokePrivate computeSubDAGIfNeeded( + plan, + List("ghost") + )) should be theSameInstanceAs plan + } + + "collectOperatorInfos" should "report per-operator stats, summed across ports" in { + val stateStore = new ExecutionStateStore() + stateStore.statsStore.updateState( + _.withOperatorInfo( + Map( + "op-with-stats" -> metrics( + COMPLETED, + inputs = Seq(0 -> 3L, 1 -> 4L), + outputs = Seq(0 -> 5L, 1 -> 6L) + ), + "op-no-ports" -> metrics(RUNNING, inputs = Seq.empty, outputs = Seq.empty) + ) + ) + ) + val execution = newExecution(stateStore) + + val infos = resource invokePrivate collectOperatorInfos( + ExecutionIdentity(consoleEid.toLong), + execution, + List("op-with-stats", "op-no-ports", "op-never-ran"), + 100000, + 20000, + None + ) + + infos.keySet shouldBe Set("op-with-stats", "op-no-ports", "op-never-ran") + + val withStats = infos("op-with-stats") + withStats.state shouldBe "Completed" + // Distinct totals on purpose: 3+4 and 5+6 cannot be produced by summing the wrong side, by + // taking a single port, or by counting ports instead of tuples. + withStats.inputTuples shouldBe 7L + withStats.outputTuples shouldBe 11L + withStats.inputPortShapes shouldBe Some(List(PortShape(0, 3L), PortShape(1, 4L))) + + // An operator the engine knows but that reported no ports: an empty shape list is reported as + // absent, so the caller does not render a zero-column table header. + val noPorts = infos("op-no-ports") + noPorts.state shouldBe "Running" + noPorts.inputTuples shouldBe 0L + noPorts.outputTuples shouldBe 0L + noPorts.inputPortShapes shouldBe None + + // A requested operator with no stats at all — an upstream target that never started. + val neverRan = infos("op-never-ran") + neverRan.state shouldBe "Unknown" + neverRan.inputTuples shouldBe 0L + neverRan.outputTuples shouldBe 0L + neverRan.inputPortShapes shouldBe None + + // No result URI is registered for this execution, so every entry reports an absent table + // rather than an empty one — the caller distinguishes the two. + infos.values.foreach { info => + info.resultMode shouldBe "table" + info.result shouldBe None + info.totalRowCount shouldBe None + info.consoleLogs shouldBe None + info.error shouldBe None + info.warnings shouldBe None + } + } + + it should "report every operator the engine knows when no target is named" in { + val stateStore = new ExecutionStateStore() + stateStore.statsStore.updateState( + _.withOperatorInfo( + Map( + "op-1" -> metrics(COMPLETED, Seq(0 -> 1L), Seq(0 -> 2L)), + "op-2" -> metrics(FAILED, Seq(0 -> 3L), Seq(0 -> 4L)) + ) + ) + ) + + val infos = resource invokePrivate collectOperatorInfos( + ExecutionIdentity(consoleEid.toLong), + newExecution(stateStore), + List.empty, + 100000, + 20000, + None + ) + + // Two operators, not one: "all of them" has to mean the whole stats map. + infos.keySet shouldBe Set("op-1", "op-2") + infos("op-1").state shouldBe "Completed" + infos("op-2").state shouldBe "Failed" + } + + it should "surface console errors, warnings and operators that are not targets" in { + val stateStore = new ExecutionStateStore() + stateStore.statsStore.updateState( + _.withOperatorInfo(Map("target" -> metrics(FAILED, Seq(0 -> 1L), Seq.empty))) + ) + val consoleState = ExecutionConsoleStore(operatorConsole = + Map( + "target" -> console( + (ConsoleMessageType.PRINT, "WARNING: input looked empty", ""), + // A print that merely mentions the word: `contains` instead of `startsWith` would + // wrongly promote it to a warning. + (ConsoleMessageType.PRINT, "note: WARNING: appears mid-line", ""), + (ConsoleMessageType.ERROR, "short title", "a much longer python traceback body") + ), + // Not a target: an upstream operator whose failure is the reason the target failed. Its + // console must still reach the caller. + "upstream" -> console((ConsoleMessageType.ERROR, "scala style error", "")), + "chatty" -> console((ConsoleMessageType.PRINT, "just progress", "")), + // Present in the map but with no messages at all. ExecutionConsoleService creates exactly + // this shape -- `getOrElse(opId, OperatorConsole())` -- before it has anything to add, so + // without it nothing distinguishes "no console" (None) from "an empty console" (Some(Nil)). + "silent" -> OperatorConsole() + ) + ) + + val infos = resource invokePrivate collectOperatorInfos( + ExecutionIdentity(consoleEid.toLong), + newExecution(stateStore), + List("target"), + 100000, + 20000, + Some(consoleState) + ) + + infos.keySet shouldBe Set("target", "upstream", "chatty", "silent") + + val target = infos("target") + target.consoleLogs shouldBe Some( + List( + ConsoleMessageInfo("PRINT", "WARNING: input looked empty", ""), + ConsoleMessageInfo("PRINT", "note: WARNING: appears mid-line", ""), + ConsoleMessageInfo("ERROR", "short title", "a much longer python traceback body") + ) + ) + // Python puts the full error in `message`; the longer of the two fields wins. + target.error shouldBe Some("a much longer python traceback body") + target.warnings shouldBe Some(List("WARNING: input looked empty")) + + // Scala puts the error in `title` and a stack trace in `message`; here `message` is empty, so + // the title has to be used instead of an empty string. + infos("upstream").error shouldBe Some("scala style error") + infos("upstream").state shouldBe "Unknown" + + // Prints only: logs travel, but no error and — critically — no empty warnings list. + infos("chatty").consoleLogs shouldBe Some( + List(ConsoleMessageInfo("PRINT", "just progress", "")) + ) + infos("chatty").error shouldBe None + infos("chatty").warnings shouldBe None + + // An operator whose console exists but is empty must report None, not Some(Nil): the frontend + // renders a console pane for Some and nothing for None, so Some(Nil) is an empty pane. + infos("silent").consoleLogs shouldBe None + infos("silent").error shouldBe None + infos("silent").warnings shouldBe None + } + + it should "let a disabled-warehouse refusal reach the caller" in { + // #6930: every other storage failure degrades into an absent result, but a kill-switch + // refusal must not — "no data" is indistinguishable from data loss. The URI carries an + // unresolvable warehouse name, which WarehouseReadGuard refuses in either flag state, so this + // does not depend on how the deployment has the switch set. + insertConsoleUri( + otherEid, + "guarded", + s"vfs:///wh/a%2Fb/wid/$testWid/eid/$otherEid/opid/guarded/consolemessages" + ) + val stateStore = new ExecutionStateStore() + stateStore.statsStore.updateState( + _.withOperatorInfo(Map("guarded" -> metrics(COMPLETED, Seq.empty, Seq.empty))) + ) + + a[WarehouseUnavailableException] should be thrownBy { + resource invokePrivate collectOperatorInfos( + ExecutionIdentity(otherEid.toLong), + newExecution(stateStore), + List("guarded"), + 100000, + 20000, + None + ) + } + } + + "getConsoleMessageUri" should "find the console URI for one operator of one execution" in { + def consoleUri(eid: Int, opId: String) = + s"vfs:///wid/$testWid/eid/$eid/opid/$opId/consolemessages" + + insertConsoleUri(consoleEid, "op-a", consoleUri(consoleEid, "op-a")) + insertConsoleUri(consoleEid, "op-b", consoleUri(consoleEid, "op-b")) + // Same operator id under a different execution: the execution predicate is what keeps this + // out of the answer, and a run's console must never leak into an earlier run's report. + insertConsoleUri(otherEid, "op-a", consoleUri(otherEid, "op-a")) + // A row that exists but was never given a URI (the operator produced no console output). + insertConsoleUri(consoleEid, "op-empty", "") + + resource invokePrivate getConsoleMessageUri( + ExecutionIdentity(consoleEid.toLong), + OperatorIdentity("op-a") + ) shouldBe Some(URI.create(consoleUri(consoleEid, "op-a"))) + // A second operator of the SAME execution: without the operator predicate the query matches + // two rows and cannot answer at all. + resource invokePrivate getConsoleMessageUri( + ExecutionIdentity(consoleEid.toLong), + OperatorIdentity("op-b") + ) shouldBe Some(URI.create(consoleUri(consoleEid, "op-b"))) + resource invokePrivate getConsoleMessageUri( + ExecutionIdentity(otherEid.toLong), + OperatorIdentity("op-a") + ) shouldBe Some(URI.create(consoleUri(otherEid, "op-a"))) + + // An empty stored URI is "nothing to read", not a URI that resolves to the storage root. + resource invokePrivate getConsoleMessageUri( + ExecutionIdentity(consoleEid.toLong), + OperatorIdentity("op-empty") + ) shouldBe None + resource invokePrivate getConsoleMessageUri( + ExecutionIdentity(consoleEid.toLong), + OperatorIdentity("op-absent") + ) shouldBe None + } + + "killExecution" should "stamp an end time and mark the execution killed" in { + val stateStore = new ExecutionStateStore() + val execution = newExecution(stateStore) + stateStore.statsStore.getState.endTimeStamp shouldBe 0L + + val before = System.currentTimeMillis() + resource invokePrivate killExecution(execution) + val after = System.currentTimeMillis() + + // Bounded on both sides: a constant, or a stamp taken from somewhere other than the wall + // clock, would fall outside the window. The end time is what the UI shows as the run's + // duration, so it has to be the moment of the kill. + val endTimeStamp = stateStore.statsStore.getState.endTimeStamp + endTimeStamp should be >= before + endTimeStamp should be <= after + stateStore.metadataStore.getState.state shouldBe KILLED + } + + "executeWorkflowSync" should "report a run whose plan cannot compile as a failure" in { + // A scan source with no file selected: the engine rejects it during compilation, so this + // reaches initExecutionService, the execution row, the engine, and the whole result-assembly + // tail without needing a coordinator to come up. + // + // What this does NOT pin, deliberately: which arm produced `state`. On this path + // `terminatedByConsoleError` is false and `stateToString(finalState.state)` is also "Failed", + // so inverting the console-error branch at SyncExecutionResource.scala:328 leaves the suite + // green. Distinguishing the two arms needs a run whose final state is not FAILED, which is + // impossible without a live engine here -- with no coordinator, initExecutionService always + // absorbs the failure and stamps FAILED. Recorded rather than papered over. + val scan = new CSVScanSourceOpDesc() + scan.setOperatorId("scan-op") + val user = new User + user.setUid(testUid) + user.setEmail("[email protected]") + + val result = resource.executeWorkflowSync( + testWid.toLong, + testCuid, + SyncExecutionRequest( + executionName = "sync-spec-run", + logicalPlan = LogicalPlanPojo(List(scan), List.empty, List.empty, List.empty), + workflowSettings = None, + targetOperatorIds = List("scan-op"), + timeoutSeconds = 5, + maxOperatorResultCharLimit = 100000, + maxOperatorResultCellCharLimit = 20000 + ), + new SessionUser(user) + ) + + result.success shouldBe false + result.state shouldBe "Failed" + // compilationErrors is always None on this path — the endpoint reports compile failures that + // happen inside the engine as fatal errors, not as the compilationErrors map. + result.compilationErrors shouldBe None + val errors = + result.errors.getOrElse(fail("expected the fatal compilation error to be reported")) + errors should have size 1 + // The fatal error is rendered as "<type>: <message>"; dropping the type loses the only + // machine-readable part of it. + errors.head should startWith("EXECUTION_FAILURE: ") + errors.head should include("No file selected") + + // The requested target is still reported, with no stats, so the caller learns which operator + // it asked about rather than getting an empty map. + result.operators.keySet shouldBe Set("scan-op") + result.operators("scan-op").state shouldBe "Unknown" + result.operators("scan-op").error shouldBe None + } +}
