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-7767-fb9f4e26d8680ebeb3da28881b27707645d59acd in repository https://gitbox.apache.org/repos/asf/texera.git
commit 0f54e934f147721ee53b04056808f194607333db Author: Eugene Gu <[email protected]> AuthorDate: Wed Aug 19 04:38:04 2026 +0000 test(workflow-operator): extend tests to cover the Projection drop mode (#7767) ### What changes were proposed in this PR? The Projection operator's "Drop Option" (`isDrop`) inverts what the operator does: with it on, the listed attributes are removed and every other attribute is kept under its original name, with aliases ignored. Both the runtime tuple rewrite (`ProjectionOpExec`) and the compile-time output-schema derivation (`ProjectionOpDesc`) branch on this flag, but all 19 existing Projection tests leave it at its `false` default, so the entire drop half ran uncovered. This PR adds 14 tests, all with `isDrop = true`, and changes no production code. `ProjectionOpExecSpec` gains 8 tests: dropping one and multiple attributes, aliases being ignored, dropping a non-existent attribute, dropping every attribute, exec output matching the descriptor-derived schema for the same config, case-sensitive name matching, and duplicate drop entries. `ProjectionOpDescSpec` gains 6 mirror tests on the schema-derivation side: remaining names/types/order, aliases ignored, non-existent attribute, dropping everything, case-insensitive matching, and duplicate entries. The degenerate-case tests pin three exec-vs-descriptor asymmetries as they exist today rather than fixing them, since unifying them is a behavior decision for a separate issue: unknown names (the exec's `diff`-based rewrite silently ignores them, `Schema.remove` throws `IllegalArgumentException`), case handling (the exec matches exactly and keeps a column the descriptor-derived schema says was dropped, with no error on either side), and duplicate entries (the exec tolerates them, the descriptor throws). Each side's test carries a comment pointing at the other side of the asymmetry. ### Any related issues, documentation, discussions? Resolves #7765 ### How was this PR tested? This PR is test-only. Both suites pass locally: `sbt "WorkflowOperator/testOnly *ProjectionOpExecSpec *ProjectionOpDescSpec"` runs 33 tests (19 pre-existing + 14 new), all green, and `WorkflowOperator/Test/scalafmtCheck` passes. The full `WorkflowOperator/test` suite was also run locally to confirm the new tests introduce no cross-test interference. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Fable 5) --- .../operator/projection/ProjectionOpDescSpec.scala | 74 +++++++++++++ .../operator/projection/ProjectionOpExecSpec.scala | 121 +++++++++++++++++++++ 2 files changed, 195 insertions(+) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala index 2927401ef5..4c1b48b796 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala @@ -116,6 +116,80 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { } + it should "derive the drop-mode schema with original names, types and order" in { + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("field2", "") + ) + val outputSchema = + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + assert(outputSchema.getAttributes.length == 2) + assert(outputSchema.getIndex("field1") == 0) + assert(outputSchema.getIndex("field3") == 1) + assert(outputSchema.getAttribute("field1").getType == AttributeType.STRING) + assert(outputSchema.getAttribute("field3").getType == AttributeType.BOOLEAN) + } + + it should "ignore aliases when deriving the drop-mode schema" in { + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("field1", "renamed") + ) + val outputSchema = + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + assert(outputSchema.getAttributeNames == List("field2", "field3")) + } + + it should "raise IllegalArgumentException when dropping a non-existent attribute" in { + // Unlike the exec, whose diff-based rewrite silently ignores unknown names, + // Schema.remove rejects them at schema-derivation time. + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("field---5", "f5") + ) + assertThrows[IllegalArgumentException] { + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + } + } + + it should "derive an empty schema when dropping every attribute" in { + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("field1", ""), + new AttributeUnit("field2", ""), + new AttributeUnit("field3", "") + ) + val outputSchema = + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + assert(outputSchema.getAttributes.isEmpty) + } + + it should "match drop names case-insensitively when deriving the schema" in { + // Unlike the exec, whose diff-based rewrite matches names exactly and + // would keep field2, Schema.remove lowercases both sides. + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("FIELD2", "") + ) + val outputSchema = + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + assert(outputSchema.getAttributeNames == List("field1", "field3")) + } + + it should "raise IllegalArgumentException on duplicate entries in the drop list" in { + // The exec's multiset diff tolerates duplicates; the schema derivation folds + // Schema.remove one entry at a time, so the second removal of the same name + // rejects a now non-existent attribute. + projectionOpDesc.isDrop = true + projectionOpDesc.attributes ++= List( + new AttributeUnit("field2", ""), + new AttributeUnit("field2", "") + ) + assertThrows[IllegalArgumentException] { + projectionOpDesc.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head + } + } + it should "preserve a HashPartition when its attributes are non-empty" in { val out = projectionOpDesc.derivePartition()(List(HashPartition(List("field1")))) assert(out == HashPartition(List("field1"))) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala index a514f0ab3e..b89df6ed14 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala @@ -20,6 +20,7 @@ package org.apache.texera.amber.operator.projection import org.apache.texera.amber.core.tuple._ +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.BeforeAndAfter import org.scalatest.flatspec.AnyFlatSpec @@ -153,4 +154,124 @@ class ProjectionOpExecSpec extends AnyFlatSpec with BeforeAndAfter { assert(outputTuple.getField[String](0) == "hello") assert(outputTuple.getField[Int](1) == 1) } + + it should "drop a single attribute and keep the rest under their original names" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field2", "") + ) + val outputSchema = Schema() + .add(new Attribute("field1", AttributeType.STRING)) + .add(new Attribute("field3", AttributeType.BOOLEAN)) + + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings.keySet == Set("field1", "field3")) + + val outputTuple = output.enforceSchema(outputSchema) + assert(outputTuple.length == 2) + assert(outputTuple.getField[String](0) == "hello") + assert(outputTuple.getField[Boolean](1)) + } + + it should "drop multiple attributes" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field1", ""), + new AttributeUnit("field3", "") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings == Map("field2" -> 1)) + } + + it should "ignore aliases in drop mode" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field2", "renamed") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings.keySet == Set("field1", "field3")) + assert(!output.fieldMappings.contains("renamed")) + } + + it should "silently ignore dropping a non-existent attribute" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field---5", "f5") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings == Map("field1" -> "hello", "field2" -> 1, "field3" -> true)) + } + + it should "emit an empty tuple when dropping every attribute" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field1", ""), + new AttributeUnit("field2", ""), + new AttributeUnit("field3", "") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings.isEmpty) + } + + it should "emit exactly the attributes the descriptor derives for the same drop config" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field2", "") + ) + val derivedSchema = + opDesc.getExternalOutputSchemas(Map(PortIdentity() -> tupleSchema)).values.head + + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings.keySet == derivedSchema.getAttributeNames.toSet) + + val outputTuple = output.enforceSchema(derivedSchema) + assert(outputTuple.length == 2) + assert(outputTuple.getField[String]("field1") == "hello") + assert(outputTuple.getField[Boolean]("field3")) + } + + it should "match drop names case-sensitively" in { + // Unlike the descriptor, whose Schema.remove lowercases both sides and + // would drop field2, the diff-based rewrite matches names exactly. + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("FIELD2", "") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings == Map("field1" -> "hello", "field2" -> 1, "field3" -> true)) + } + + it should "tolerate duplicate entries in the drop list" in { + opDesc.isDrop = true + opDesc.attributes = List( + new AttributeUnit("field2", ""), + new AttributeUnit("field2", "") + ) + val projectionOpExec = new ProjectionOpExec(objectMapper.writeValueAsString(opDesc)) + projectionOpExec.open() + + val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] + assert(output.fieldMappings == Map("field1" -> "hello", "field3" -> true)) + } }
