(texera) branch gh-readonly-queue/main/pr-7767-fb9f4e26d8680ebeb3da28881b27707645d59acd created (now 0f54e934f1)

[gi...@apache.org]

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-7767-fb9f4e26d8680ebeb3da28881b27707645d59acd
in repository https://gitbox.apache.org/repos/asf/texera.git


      at 0f54e934f1 test(workflow-operator): extend tests to cover the Projection drop mode (#7767)

This branch includes the following new commits:

     new 0f54e934f1 test(workflow-operator): extend tests to cover the Projection drop mode (#7767)

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(workflow-operator): extend tests to cover the Projection drop mode (#7767)

[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-7767-fb9f4e26d8680ebeb3da28881b27707645d59acd
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 0f54e934f147721ee53b04056808f194607333db
Author: Eugene Gu <eu...@outlook.com>
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))
+  }
 }