You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/05/31 23:59:28 UTC
(texera) branch gh-readonly-queue/main/pr-5022-e25c28225f25990a4c149b85fa48f8090900cf85 created (now a8afdc487a)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5022-e25c28225f25990a4c149b85fa48f8090900cf85
in repository https://gitbox.apache.org/repos/asf/texera.git
at a8afdc487a test(compiling-service): add direct unit tests for WorkflowCompiler (#5022)
This branch includes the following new commits:
new a8afdc487a test(compiling-service): add direct unit tests for WorkflowCompiler (#5022)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(compiling-service): add direct unit tests for WorkflowCompiler (#5022)
Posted by gi...@apache.org.
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-5022-e25c28225f25990a4c149b85fa48f8090900cf85
in repository https://gitbox.apache.org/repos/asf/texera.git
commit a8afdc487aaa8a1befca3cc4db61584b21d5a848
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Sun May 31 16:59:08 2026 -0700
test(compiling-service): add direct unit tests for WorkflowCompiler (#5022)
### What changes were proposed in this PR?
Add direct unit coverage for `workflow-compiling-service`'s
`WorkflowCompiler` lenient-mode contract: `compile` never throws,
per-operator failures accumulate into `operatorIdToError`, and
`physicalPlan` is `None` whenever any error occurred. Only the REST
happy path is covered today.
The spec invokes the compiler directly rather than the REST resource,
sidestepping a separate response-serialization NPE surfaced while
writing these tests (apache/texera#5021). The contract tests stay green
once that bug is fixed.
### Any related issues, documentation, discussions?
Closes #5020. References #5021.
### How was this PR tested?
`sbt "WorkflowCompilingService/testOnly
org.apache.texera.amber.compiler.WorkflowCompilerSpec"` — all green.
`scalafmtCheck` and `compile` clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.7 (1M context)
---------
Signed-off-by: Yicong Huang <17...@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../amber/compiler/WorkflowCompilerSpec.scala | 317 +++++++++++++++++++++
.../resource/WorkflowCompilationResourceSpec.scala | 260 ++++++-----------
2 files changed, 410 insertions(+), 167 deletions(-)
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala
new file mode 100644
index 0000000000..ee221728d3
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala
@@ -0,0 +1,317 @@
+/*
+ * 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.compiler
+
+import org.apache.texera.amber.compiler.model.{LogicalLink, LogicalPlanPojo}
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType}
+import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity
+import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
+import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.COMPILATION_ERROR
+import org.apache.texera.amber.operator.filter.{
+ ComparisonType,
+ FilterPredicate,
+ SpecializedFilterOpDesc
+}
+import org.apache.texera.amber.operator.limit.LimitOpDesc
+import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc}
+import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc
+import org.scalatest.flatspec.AnyFlatSpec
+
+/**
+ * Direct unit coverage for the editing-time [[WorkflowCompiler]].
+ *
+ * Owns *compiler-behavior* tests — schema propagation through multi-op
+ * chains, lenient-mode error accumulation, terminal-storage selection.
+ * `WorkflowCompilationResourceSpec` owns *resource-layer* tests — HTTP
+ * status, response type discriminator, JSON envelope. Drawing the line
+ * here keeps each spec focused on a single failure axis.
+ *
+ * Bypassing the resource layer also sidesteps a separate NPE in response
+ * serialization (apache/texera#5021); these compiler-level tests stay
+ * green once that bug is fixed.
+ */
+class WorkflowCompilerSpec extends AnyFlatSpec {
+
+ private def newContext(): WorkflowContext =
+ new WorkflowContext(workflowId = WorkflowIdentity(0))
+
+ private def csvOp(fileName: String): CSVScanSourceOpDesc = {
+ val op = new CSVScanSourceOpDesc()
+ op.fileName = Some(fileName)
+ op.customDelimiter = Some(",")
+ op.hasHeader = true
+ op
+ }
+
+ private def csvOpNoFile(): CSVScanSourceOpDesc = {
+ val op = new CSVScanSourceOpDesc()
+ op.customDelimiter = Some(",")
+ op.hasHeader = true
+ op
+ }
+
+ private def projectOp(columns: List[String]): ProjectionOpDesc = {
+ val op = new ProjectionOpDesc()
+ op.attributes = columns.map(name => new AttributeUnit(name, ""))
+ op.isDrop = false
+ op
+ }
+
+ private def filterOp(predicates: FilterPredicate*): SpecializedFilterOpDesc = {
+ val op = new SpecializedFilterOpDesc
+ op.predicates = predicates.toList
+ op
+ }
+
+ private def limitOp(limit: Int): LimitOpDesc = {
+ val op = new LimitOpDesc
+ op.limit = limit
+ op
+ }
+
+ private val realCsvPath =
+ "workflow-compiling-service/src/test/resources/country_sales_small.csv"
+
+ // -------------------- happy path --------------------
+
+ "WorkflowCompiler" should "produce a populated physicalPlan and no errors for a well-formed plan" in {
+ val csv = csvOp(realCsvPath)
+ val proj = projectOp(List("Region", "Total Profit"))
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(csv, proj),
+ links = List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(0),
+ proj.operatorIdentifier,
+ PortIdentity(0)
+ )
+ ),
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isDefined, "happy path should yield a physical plan")
+ assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}")
+ // Schema for both operators' output ports should be populated and non-null —
+ // this is the property whose violation triggers the resource-level NPE.
+ val projSchemas = result.operatorIdToOutputSchemas(proj.operatorIdentifier)
+ assert(projSchemas.values.forall(s => s.isDefined && s.get != null))
+ }
+
+ it should "propagate schemas through a csv -> projection -> limit -> filter -> filter -> limit chain" in {
+ // Real-world editing-shape: source then filter/limit/project ops. Asserts
+ // the compiler threads schema through every link so the frontend sees the
+ // projected columns at every downstream port. Previously this lived in
+ // WorkflowCompilationResourceSpec as an HTTP test, but the property being
+ // pinned is compiler-level (schema propagation) — the REST envelope adds
+ // no signal.
+ val csv = csvOp(realCsvPath)
+ val proj = projectOp(List("Region", "Total Profit"))
+ val limit1 = limitOp(10)
+ val filter1 =
+ filterOp(new FilterPredicate("Total Profit", ComparisonType.GREATER_THAN, "10000"))
+ val filter2 = filterOp(new FilterPredicate("Region", ComparisonType.NOT_EQUAL_TO, "JPN"))
+ val limit2 = limitOp(5)
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(csv, proj, limit1, filter1, filter2, limit2),
+ links = List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(0),
+ proj.operatorIdentifier,
+ PortIdentity(0)
+ ),
+ LogicalLink(
+ proj.operatorIdentifier,
+ PortIdentity(0),
+ limit1.operatorIdentifier,
+ PortIdentity(0)
+ ),
+ LogicalLink(
+ limit1.operatorIdentifier,
+ PortIdentity(0),
+ filter1.operatorIdentifier,
+ PortIdentity(0)
+ ),
+ LogicalLink(
+ filter1.operatorIdentifier,
+ PortIdentity(0),
+ filter2.operatorIdentifier,
+ PortIdentity(0)
+ ),
+ LogicalLink(
+ filter2.operatorIdentifier,
+ PortIdentity(0),
+ limit2.operatorIdentifier,
+ PortIdentity(0)
+ )
+ ),
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isDefined)
+ assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}")
+ // Projection narrowed [Region, Country, ..., Total Profit] down to two
+ // columns; every downstream op should see exactly those two attributes.
+ val filter2Schemas = result.operatorIdToOutputSchemas(filter2.operatorIdentifier)
+ val outputAttrs = filter2Schemas(PortIdentity(0)).get.attributes
+ assert(
+ outputAttrs == List(
+ new Attribute("Region", AttributeType.STRING),
+ new Attribute("Total Profit", AttributeType.DOUBLE)
+ ),
+ s"projected schema should reach filter2 unchanged, got $outputAttrs"
+ )
+ }
+
+ // -------------------- lenient-mode error accumulation --------------------
+
+ // The frontend relies on `compile` *never throwing*: a user mid-edit
+ // routinely produces an inconsistent plan and the editing UI must render
+ // structured per-operator errors. These tests pin the contract.
+
+ "WorkflowCompiler" should "accumulate, not throw, when a scan source has no fileName" in {
+ val orphan = csvOpNoFile()
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(orphan),
+ links = List.empty,
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isEmpty, "any error must clear the physical plan")
+ val err = result.operatorIdToError(orphan.operatorIdentifier)
+ assert(err.`type` == COMPILATION_ERROR)
+ assert(err.operatorId == orphan.operatorIdentifier.id)
+ assert(err.message.contains("no input file name"), s"unexpected message: ${err.message}")
+ assert(err.details.nonEmpty, "stack-trace details should be populated for UI display")
+ }
+
+ it should "accumulate when a scan source's fileName points to a non-existent path" in {
+ val broken = csvOp("/does/not/exist/missing.csv")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(broken),
+ links = List.empty,
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isEmpty)
+ assert(result.operatorIdToError.contains(broken.operatorIdentifier))
+ // FileResolver.resolve falls through both resolvers and rethrows
+ // org.apache.commons.vfs2.FileNotFoundException(fileName); its message bundle
+ // renders as `Could not read from "<path>" because it is not a file.`, so the
+ // only stable substring across that wording and any java.io.FileNotFoundException
+ // fallback is the bad path itself.
+ assert(
+ result.operatorIdToError(broken.operatorIdentifier).message.contains("missing.csv"),
+ s"unexpected message: ${result.operatorIdToError(broken.operatorIdentifier).message}"
+ )
+ }
+
+ it should "accumulate a per-operator error when projection references a non-existent attribute" in {
+ val csv = csvOp(realCsvPath)
+ val badProjection = projectOp(List("DoesNotExist"))
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(csv, badProjection),
+ links = List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(0),
+ badProjection.operatorIdentifier,
+ PortIdentity(0)
+ )
+ ),
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isEmpty)
+ assert(
+ result.operatorIdToError.contains(badProjection.operatorIdentifier),
+ s"projection should be in errors, got ${result.operatorIdToError.keySet}"
+ )
+ // The upstream csv ran fine, so its output schema should still be present
+ // — partial progress is the whole point of lenient mode.
+ assert(
+ result.operatorIdToOutputSchemas.contains(csv.operatorIdentifier),
+ "upstream csv's schemas should be retained even when downstream fails"
+ )
+ }
+
+ it should "not throw when given an empty plan" in {
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List.empty,
+ links = List.empty,
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+ assert(result.operatorIdToError.isEmpty)
+ assert(result.operatorIdToOutputSchemas.isEmpty)
+ assert(result.physicalPlan.isDefined, "an empty plan compiles to an empty physical plan")
+ assert(result.physicalPlan.get.operators.isEmpty)
+ assert(result.physicalPlan.get.links.isEmpty)
+ }
+
+ // -------------------- multi-error accumulation --------------------
+
+ // Re-anchor the subject after the sub-section.
+ "WorkflowCompiler" should "accumulate errors for multiple unrelated failing ops in one compile" in {
+ val orphan1 = csvOpNoFile()
+ val orphan2 = csvOpNoFile()
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ LogicalPlanPojo(
+ operators = List(orphan1, orphan2),
+ links = List.empty,
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+
+ assert(result.physicalPlan.isEmpty)
+ // Both ops must appear in the error map — the frontend renders per-op
+ // diagnostics in parallel, so swallowing all-but-one would silently break
+ // multi-error workflows.
+ assert(
+ result.operatorIdToError.contains(orphan1.operatorIdentifier) &&
+ result.operatorIdToError.contains(orphan2.operatorIdentifier),
+ s"expected both csvs in errors, got ${result.operatorIdToError.keySet}"
+ )
+ }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala
index 87246fd7f3..fd99c3d27d 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala
@@ -24,23 +24,25 @@ import io.dropwizard.testing.junit5.ResourceExtension
import jakarta.ws.rs.client.Entity
import jakarta.ws.rs.core.{MediaType, Response}
import org.apache.texera.amber.compiler.model.{LogicalLink, LogicalPlanPojo}
-import org.apache.texera.amber.core.tuple.{Attribute, AttributeType}
import org.apache.texera.amber.core.workflow.PortIdentity
-import org.apache.texera.amber.operator.filter.{
- ComparisonType,
- FilterOpDesc,
- FilterPredicate,
- SpecializedFilterOpDesc
-}
-import org.apache.texera.amber.operator.limit.LimitOpDesc
import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc}
import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc
import org.apache.texera.amber.util.JSONUtils.objectMapper
-import org.apache.texera.amber.util.serde.PortIdentityKeySerializer
import org.assertj.core.api.Assertions.assertThat
import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpec
+/**
+ * Resource-layer tests for `/compile`. Owns only what the REST envelope
+ * adds on top of compilation itself: HTTP status, the
+ * `@JsonTypeInfo` discriminator that lets the frontend route success vs
+ * failure, and the JSON shape the resource expects on the wire.
+ *
+ * All compiler-behavior assertions (schema propagation, lenient-mode
+ * error accumulation, multi-op chains) live in
+ * `org.apache.texera.amber.compiler.WorkflowCompilerSpec` so a regression
+ * lands on the right spec.
+ */
class WorkflowCompilationResourceSpec extends AnyFlatSpec with BeforeAndAfterAll {
private val resources: ResourceExtension = ResourceExtension
@@ -48,193 +50,117 @@ class WorkflowCompilationResourceSpec extends AnyFlatSpec with BeforeAndAfterAll
.addResource(new WorkflowCompilationResource())
.setMapper(objectMapper)
.build()
- override protected def beforeAll(): Unit = {
- resources.before()
- }
- override protected def afterAll(): Unit = {
- resources.after()
- }
+ override protected def beforeAll(): Unit = resources.before()
+ override protected def afterAll(): Unit = resources.after()
- // utility function to create a csv scan op
- private def getCsvScanOpDesc(
- fileName: String,
- header: Boolean
- ): CSVScanSourceOpDesc = {
- val csvOp = new CSVScanSourceOpDesc()
- csvOp.fileName = Some(fileName)
- csvOp.customDelimiter = Some(",")
- csvOp.hasHeader = header
- csvOp
- }
+ private val realCsvPath =
+ "workflow-compiling-service/src/test/resources/country_sales_small.csv"
- // utility function to create a projection op
- private def getProjectionOpDesc(
- attributeNames: List[String],
- isDrop: Boolean = false
- ): ProjectionOpDesc = {
- val projectionOpDesc = new ProjectionOpDesc()
- projectionOpDesc.attributes = attributeNames.map(name => new AttributeUnit(name, ""))
- projectionOpDesc.isDrop = isDrop
- projectionOpDesc
+ private def csvOp(fileName: String): CSVScanSourceOpDesc = {
+ val op = new CSVScanSourceOpDesc()
+ op.fileName = Some(fileName)
+ op.customDelimiter = Some(",")
+ op.hasHeader = true
+ op
}
- // utility function to create a limit op
- private def getLimitOpDesc(
- limit: Int
- ): LimitOpDesc = {
- val limitOpDesc = new LimitOpDesc
- limitOpDesc.limit = limit
- limitOpDesc
+ private def projectOp(columns: List[String]): ProjectionOpDesc = {
+ val op = new ProjectionOpDesc()
+ op.attributes = columns.map(name => new AttributeUnit(name, ""))
+ op.isDrop = false
+ op
}
- // utility function to create a filter op
- private def getFilterOpDesc(
- filterPredicates: List[FilterPredicate]
- ): FilterOpDesc = {
- val filterOpDesc = new SpecializedFilterOpDesc
- filterOpDesc.predicates = filterPredicates
- filterOpDesc
- }
-
- // utility function to transform a logical plan pojo to json that can be deserialized correctly by the compile endpoint
- private def transformLogicalPlanPojoToJsonString(logicalPlanPojo: LogicalPlanPojo): String = {
- val jsonNode = objectMapper.valueToTree[ObjectNode](logicalPlanPojo)
-
- // iterate over the "links" array and replace nested "id" fields
+ // The frontend serializes LogicalLink with `fromOpId` / `toOpId` as flat
+ // strings, but the Scala case class stores them as nested `OperatorIdentity`
+ // records. This helper mirrors the wire shape so the test exercises the
+ // resource's actual JSON contract instead of a Scala-only round trip.
+ private def encodePojoAsFrontendJson(pojo: LogicalPlanPojo): String = {
+ val jsonNode = objectMapper.valueToTree[ObjectNode](pojo)
val linksArray = jsonNode.withArray("links")
linksArray.forEach { linkNode =>
- // replace "fromOpId" with its "id" field value
val fromOpIdNode = linkNode.get("fromOpId")
linkNode.asInstanceOf[ObjectNode].put("fromOpId", fromOpIdNode.get("id").asText())
-
- // replace "toOpId" with its "id" field value if it exists
val toOpIdNode = linkNode.get("toOpId")
linkNode.asInstanceOf[ObjectNode].put("toOpId", toOpIdNode.get("id").asText())
}
-
- // convert the modified JSON node back to a string
objectMapper.writeValueAsString(jsonNode)
}
- // utility function for asserting the successful compilation
- private def assertSuccessfulCompilation(response: Response): WorkflowCompilationSuccess = {
- val responseBody = response.readEntity(classOf[String])
- val compilationResponse =
- objectMapper.readValue(responseBody, classOf[WorkflowCompilationResponse])
-
- assertThat(compilationResponse.asInstanceOf[WorkflowCompilationSuccess])
- compilationResponse.asInstanceOf[WorkflowCompilationSuccess]
+ private def postCompile(pojo: LogicalPlanPojo): Response =
+ resources
+ .target("/compile")
+ .request(MediaType.APPLICATION_JSON)
+ .post(Entity.json(encodePojoAsFrontendJson(pojo)))
+
+ "POST /compile" should "return HTTP 200 for a well-formed plan" in {
+ val csv = csvOp(realCsvPath)
+ val proj = projectOp(List("Region", "Total Profit"))
+ val response = postCompile(
+ LogicalPlanPojo(
+ operators = List(csv, proj),
+ links = List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(0),
+ proj.operatorIdentifier,
+ PortIdentity(0)
+ )
+ ),
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
+ )
+ assertThat(response.getStatus).isEqualTo(200)
}
- it should "compile workflow successfully with multiple filter and limit operations" in {
- // construct the LogicalPlan: CSVScan --> Projection --> Limit --> Filter (TotalProfit > 10000) --> Filter (Region != "JPN") --> Limit
- val localCsvFilePath =
- "workflow-compiling-service/src/test/resources/country_sales_small.csv"
- val csvSourceOp = getCsvScanOpDesc(localCsvFilePath, header = true)
- val projectionOpDesc = getProjectionOpDesc(List("Region", "Total Profit"))
- val limitOpDesc1 = getLimitOpDesc(10)
-
- // Create the filter predicate for TotalProfit > 10000
- val filterPredicate1 = new FilterPredicate("Total Profit", ComparisonType.GREATER_THAN, "10000")
- val filterOpDesc1 = getFilterOpDesc(List(filterPredicate1))
-
- // Create the filter predicate for Region != "JPN"
- val filterPredicate2 = new FilterPredicate("Region", ComparisonType.NOT_EQUAL_TO, "JPN")
- val filterOpDesc2 = getFilterOpDesc(List(filterPredicate2))
-
- // Add a second limit operation
- val limitOpDesc2 = getLimitOpDesc(5)
-
- val logicalPlanPojo = LogicalPlanPojo(
- operators = List(
- csvSourceOp,
- projectionOpDesc,
- limitOpDesc1,
- filterOpDesc1,
- filterOpDesc2,
- limitOpDesc2
- ),
- links = List(
- LogicalLink(
- csvSourceOp.operatorIdentifier,
- PortIdentity(0),
- projectionOpDesc.operatorIdentifier,
- PortIdentity(0)
+ it should "tag the response body with type=success so the frontend can route on the discriminator" in {
+ // The @JsonTypeInfo on WorkflowCompilationResponse writes a `type` field.
+ // Both polymorphic deserialization (round-tripping through the trait) and
+ // a raw-JSON `type == "success"` check need to hold so the Angular client
+ // can branch without depending on Scala class names.
+ val csv = csvOp(realCsvPath)
+ val proj = projectOp(List("Region", "Total Profit"))
+ val response = postCompile(
+ LogicalPlanPojo(
+ operators = List(csv, proj),
+ links = List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(0),
+ proj.operatorIdentifier,
+ PortIdentity(0)
+ )
),
- LogicalLink(
- projectionOpDesc.operatorIdentifier,
- PortIdentity(0),
- limitOpDesc1.operatorIdentifier,
- PortIdentity(0)
- ),
- LogicalLink(
- limitOpDesc1.operatorIdentifier,
- PortIdentity(0),
- filterOpDesc1.operatorIdentifier,
- PortIdentity(0)
- ),
- LogicalLink(
- filterOpDesc1.operatorIdentifier,
- PortIdentity(0),
- filterOpDesc2.operatorIdentifier,
- PortIdentity(0)
- ),
- LogicalLink(
- filterOpDesc2.operatorIdentifier,
- PortIdentity(0),
- limitOpDesc2.operatorIdentifier,
- PortIdentity(0)
- )
- ),
- opsToViewResult = List(),
- opsToReuseResult = List()
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
)
- // transform the LogicalPlanPojo to a modified JSON string
- val modifiedLogicalPlanJsonString = transformLogicalPlanPojoToJsonString(logicalPlanPojo)
-
- // send the request to compile endpoint
- val response = resources
- .target("/compile")
- .request(MediaType.APPLICATION_JSON)
- .post(Entity.json(modifiedLogicalPlanJsonString))
-
- assertThat(response.getStatus).isEqualTo(200)
-
- // verify the schema is correctly propagated for the final limit operator
- val compilationResult = assertSuccessfulCompilation(response)
- val finalLimitInputSchema =
- compilationResult.operatorOutputSchemas(filterOpDesc2.operatorIdentifier.id)
+ val body = response.readEntity(classOf[String])
+ val node = objectMapper.readTree(body)
assert(
- finalLimitInputSchema(
- PortIdentityKeySerializer.portIdToString(PortIdentity(id = 0, internal = false))
- ).get.equals(
- List(
- new Attribute("Region", AttributeType.STRING),
- new Attribute("Total Profit", AttributeType.DOUBLE)
- )
- )
+ node.has("type") && node.get("type").asText() == "success",
+ s"expected type:success discriminator, got $body"
)
+ val parsed = objectMapper.readValue(body, classOf[WorkflowCompilationResponse])
+ assert(parsed.isInstanceOf[WorkflowCompilationSuccess])
+ val success = parsed.asInstanceOf[WorkflowCompilationSuccess]
+ assert(success.physicalPlan != null)
+ assert(success.operatorOutputSchemas.nonEmpty)
}
it should "return WorkflowCompilationFailure (not HTTP 500) when a scan source file cannot be resolved" in {
- val brokenCsv = getCsvScanOpDesc("/does/not/exist/missing.csv", header = true)
-
- val logicalPlanPojo = LogicalPlanPojo(
- operators = List(brokenCsv),
- links = List(),
- opsToViewResult = List(),
- opsToReuseResult = List()
+ val response = postCompile(
+ LogicalPlanPojo(
+ operators = List(csvOp("/does/not/exist/missing.csv")),
+ links = List.empty,
+ opsToViewResult = List.empty,
+ opsToReuseResult = List.empty
+ )
)
- val modifiedLogicalPlanJsonString = transformLogicalPlanPojoToJsonString(logicalPlanPojo)
-
- val response = resources
- .target("/compile")
- .request(MediaType.APPLICATION_JSON)
- .post(Entity.json(modifiedLogicalPlanJsonString))
-
// Must not surface as HTTP 500 — the error must come back as a structured failure.
assertThat(response.getStatus).isEqualTo(200)