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-7929-efb00420ac7ec136b256be2ccf8a6ee6a3eed0cc in repository https://gitbox.apache.org/repos/asf/texera.git
commit b5e2558fe7d0a6949b799634caff0d2e67ca144c Author: Eugene Gu <[email protected]> AuthorDate: Tue Aug 25 00:49:23 2026 +0000 refactor(workflow-operator): match drop-mode attribute names case-insensitively in Projection (#7929) ### What changes were proposed in this PR? The Projection operator's drop mode (`isDrop = true`) decides "which attributes to drop" in two places, and the two places used different rules for comparing attribute names. At compile time, the descriptor derives the output schema through `Schema.remove`, which ignores case, just like every other lookup in the `Schema` class. At runtime, the executor computed the kept attributes with `List.diff`, which matches names exactly. So the operator contradicted itself whenever a drop entry differed from the schema attribute only in case. Dropping `Field1` against a schema containing `field1`: the declared output schema removes the attribute, but the executor keeps it. A misspelled name is caught at plan time (`Schema.remove` rejects non-existent attributes), but a name that differs only in case passes that check and diverges silently. Users never see the divergence, because the engine rebuilds every output tuple against the declared schema by attribute name (`DataProcessor` calling `MapTupleLike.enforceSchema`), which discards the extra field — the final output is correct today, but by coincidence rather than by design. This PR makes the executor compute the same answer the descriptor declares, so correctness no longer depends on that coincidence. The change: the executor lowercases the drop list into a set and keeps an attribute unless its lowercased name is in that set — the same case-insensitive rule `Schema` uses everywhere. Kept attributes keep their original spelling and order; unknown names are still silently ignored; duplicate drop entries are still tolerated. The spec that pinned the old case-sensitive behavior is flipped to assert the unified semantics, and the stale cross-reference comments in `ProjectionOpDescSpec`/`ProjectionOpExecSpec` are updated. ### Any related issues, documentation, discussions? Fixes #7925. ### How was this PR tested? Existing Projection specs cover the change. The drop-mode case-matching spec in `ProjectionOpExecSpec` previously asserted that dropping `FIELD2` leaves `field2` in the output; it is renamed to "match drop names case-insensitively" and now asserts `field2` is removed, guarding against regression. Ran `sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.projection.*"` locally: 3 suites, 37 tests, all passed (0 failed, 0 canceled, 0 ignored), covering drop mode (case matching, unknown names, duplicates, ordering) and the unchanged keep mode. ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) --- .../apache/texera/amber/operator/projection/ProjectionOpExec.scala | 5 +++-- .../texera/amber/operator/projection/ProjectionOpDescSpec.scala | 7 +++---- .../texera/amber/operator/projection/ProjectionOpExecSpec.scala | 7 +++---- 3 files changed, 9 insertions(+), 10 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala index 158d15e274..b0512cc826 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala @@ -39,8 +39,9 @@ class ProjectionOpExec( val fields = mutable.LinkedHashMap[String, Any]() if (desc.isDrop) { val allAttribute = tuple.schema.getAttributeNames - val selectedAttributes = desc.attributes.map(_.getOriginalAttribute) - val keepAttributes = allAttribute.diff(selectedAttributes) + val selectedAttributes = desc.attributes.map(_.getOriginalAttribute.toLowerCase).toSet + val keepAttributes = + allAttribute.filterNot(attribute => selectedAttributes.contains(attribute.toLowerCase)) keepAttributes.foreach { attribute => val newList = List( 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 4c1b48b796..66e6f556af 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 @@ -141,7 +141,7 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { } it should "raise IllegalArgumentException when dropping a non-existent attribute" in { - // Unlike the exec, whose diff-based rewrite silently ignores unknown names, + // Unlike the exec, which silently ignores unknown names, // Schema.remove rejects them at schema-derivation time. projectionOpDesc.isDrop = true projectionOpDesc.attributes ++= List( @@ -165,8 +165,7 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { } 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. + // Schema.remove lowercases both sides, matching the exec's behavior. projectionOpDesc.isDrop = true projectionOpDesc.attributes ++= List( new AttributeUnit("FIELD2", "") @@ -177,7 +176,7 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { } it should "raise IllegalArgumentException on duplicate entries in the drop list" in { - // The exec's multiset diff tolerates duplicates; the schema derivation folds + // The exec 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 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 b89df6ed14..1d8ee9717b 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 @@ -248,9 +248,8 @@ class ProjectionOpExecSpec extends AnyFlatSpec with BeforeAndAfter { 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. + it should "match drop names case-insensitively" in { + // Matches the descriptor, whose Schema.remove lowercases both sides. opDesc.isDrop = true opDesc.attributes = List( new AttributeUnit("FIELD2", "") @@ -259,7 +258,7 @@ class ProjectionOpExecSpec extends AnyFlatSpec with BeforeAndAfter { projectionOpExec.open() val output = projectionOpExec.processTuple(tuple, 0).next().asInstanceOf[MapTupleLike] - assert(output.fieldMappings == Map("field1" -> "hello", "field2" -> 1, "field3" -> true)) + assert(output.fieldMappings == Map("field1" -> "hello", "field3" -> true)) } it should "tolerate duplicate entries in the drop list" in {
