From 7dc3d0fece7834a7491623426402a6cba09b137f Mon Sep 17 00:00:00 2001 From: kary zheng Date: Mon, 19 Oct 2026 15:44:31 -0700 Subject: [PATCH 1/2] feat(workflow-operator): read a column named in a different case the way the engine does The engine looks a column up through Schema, which ignores case, so a setting that names `Name` reads the column `name`. The exported script indexed the frame with the name as written and raised a KeyError. StandaloneCodeGenerator.columnAsStored resolves a name against the input schema to the table's spelling. Projection, Filter, Keyword Search, Regular Expression, Substring Search, Dictionary Matcher and Unnest String read their column through it. Projection still names its output by the alias, as the engine does. Type Casting matches a column by its exact name in the engine, so a unit whose column differs in case casts nothing there. The script now skips that unit too. Closes #9004 Co-Authored-By: Claude Opus 5.5 --- .../operator/StandaloneCodeGenerator.scala | 14 +++++++++++ .../dictionary/DictionaryMatcherOpDesc.scala | 12 +++++++--- .../filter/SpecializedFilterOpDesc.scala | 11 ++++++--- .../keywordSearch/KeywordSearchOpDesc.scala | 3 ++- .../projection/ProjectionOpDesc.scala | 22 +++++++++++------- .../amber/operator/regex/RegexOpDesc.scala | 3 ++- .../SubstringSearchOpDesc.scala | 3 ++- .../typecasting/TypeCastingOpDesc.scala | 11 ++++++++- .../unneststring/UnnestStringOpDesc.scala | 3 ++- .../DictionaryMatcherOpDescSpec.scala | 14 +++++++++++ .../filter/SpecializedFilterOpDescSpec.scala | 17 ++++++++++++++ .../KeywordSearchOpDescSpec.scala | 15 ++++++++++++ .../projection/ProjectionOpDescSpec.scala | 21 +++++++++++++++++ .../operator/regex/RegexOpDescSpec.scala | 15 ++++++++++++ .../SubstringSearchOpDescSpec.scala | 14 +++++++++++ .../typecasting/TypeCastingOpDescSpec.scala | 23 +++++++++++++++++++ .../unneststring/UnnestStringOpDescSpec.scala | 11 +++++++++ 17 files changed, 193 insertions(+), 19 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala index 3853d1cdd67..83dc95e89eb 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala @@ -45,6 +45,20 @@ trait StandaloneCodeGenerator { def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = generateStandaloneCode() + /** + * A column's name as the input table spells it. + * + * The engine looks a column up ignoring case, `Schema` lowercasing the name, so a + * setting that names `Name` reads the column `name`. pandas indexes a frame by the + * exact name. A name the schema does not hold, or no schema at all, comes back as + * written, leaving the script to fail where the run would have. + */ + protected def columnAsStored(column: String, schema: Option[Schema]): String = + schema + .flatMap(s => scala.util.Try(s.getAttribute(column)).toOption) + .map(_.getName) + .getOrElse(column) + /** * A column as the text the engine's `toString` would have produced. * diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala index e617502657a..0bd9f5c79d5 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala @@ -22,12 +22,13 @@ package org.apache.texera.amber.operator.dictionary import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.state.NoLoopVariable -import org.apache.texera.amber.core.tuple.{Attribute, AttributeType} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{ InputPort, OutputPort, PhysicalOp, + PortIdentity, SchemaPropagationFunc } import org.apache.texera.amber.operator.StandaloneCodeGenerator @@ -101,7 +102,9 @@ class DictionaryMatcherOpDesc extends MapOpDesc with StandaloneCodeGenerator { supportReconfiguration = true ) - override def generateStandaloneCode(): String = { + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { // JVM splits the dictionary on "," and lowercases each entry — no trim, // so leading/trailing whitespace around a comma becomes part of the entry. val rawDict = Option(dictionary).getOrElse("") @@ -109,7 +112,10 @@ class DictionaryMatcherOpDesc extends MapOpDesc with StandaloneCodeGenerator { val resultCol = Option(resultAttribute) .filter(_.trim.nonEmpty) .getOrElse("matched") - val attrPy = pyStringLiteral(Option(attribute).getOrElse("")) + // The engine reads the field through Schema, which ignores case. See [[columnAsStored]]. + val attrPy = pyStringLiteral( + columnAsStored(Option(attribute).getOrElse(""), inputSchemas.values.headOption) + ) val resultPy = pyStringLiteral(resultCol) val mt = Option(matchingType).getOrElse(MatchingType.SCANBASED) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index e5d47ee5232..763e83b9ab0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -21,8 +21,9 @@ package org.apache.texera.amber.operator.filter import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} -import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp} +import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral @@ -63,13 +64,17 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator ) } - override def generateStandaloneCode(): String = { + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { // No predicate keeps no row: the executor's filter is `predicates.exists`, // which answers false on an empty list. Passing the frame through would be // the opposite answer. if (predicates.isEmpty) return "out1df = in1df.iloc[0:0].copy()" val conditions = predicates.map { p => - val colLit = pyStringLiteral(p.attribute) + // FilterPredicate reads the field through Schema, which ignores case. See + // [[columnAsStored]]. + val colLit = pyStringLiteral(columnAsStored(p.attribute, inputSchemas.values.headOption)) p.condition match { case ComparisonType.IS_NULL => s"""(in1df[$colLit].isna())""" case ComparisonType.IS_NOT_NULL => s"""(in1df[$colLit].notna())""" diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDesc.scala index b968535bedd..f54989b2c6f 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDesc.scala @@ -102,7 +102,8 @@ class KeywordSearchOpDesc extends FilterOpDesc with StandaloneCodeGenerator { val escaped = terms.map(_.flatMap(c => if (regexSpecials.contains(c)) s"\\$c" else c.toString)) val pattern = escaped.mkString("\\b(?:", "|", ")\\b") val pyLiteral = pyStringLiteral(pattern) - val attrLit = pyStringLiteral(attribute) + // The engine reads the field through Schema, which ignores case. See [[columnAsStored]]. + val attrLit = pyStringLiteral(columnAsStored(attribute, inputSchemas.values.headOption)) // Case follows the flag, which picks the analyzer the engine indexes with: // StandardAnalyzer lower-cases both the field and the query, CaseSensitiveAnalyzer // is the same tokenizer with that filter left out. diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala index 375f7e61e01..d5ab7377f8f 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala @@ -108,25 +108,31 @@ class ProjectionOpDesc extends MapOpDesc with StandaloneCodeGenerator { ) } - override def generateStandaloneCode(): String = { + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { val units = Option(attributes).getOrElse(List.empty) // The engine refuses an empty selection, so the script says so too. Passing // the frame through would hand back data where a run would have stopped. if (units.isEmpty) return """raise ValueError("Please select at least one attribute to project.")""" + // The column each unit reads, as the table spells it. See [[columnAsStored]]. + val stored = (u: AttributeUnit) => + columnAsStored(u.getOriginalAttribute, inputSchemas.values.headOption) + if (isDrop) { // Drop mode ignores aliases (matches ProjectionOpExec). - val cols = units.map(u => pyStringLiteral(u.getOriginalAttribute)).mkString("[", ", ", "]") + val cols = units.map(u => pyStringLiteral(stored(u))).mkString("[", ", ", "]") s"out1df = in1df.drop(columns=$cols)" } else { - val originals = - units.map(u => pyStringLiteral(u.getOriginalAttribute)).mkString("[", ", ", "]") - // AttributeUnit.getAlias returns originalAttribute when alias is blank, - // so an explicit rename is only needed when they differ. + val originals = units.map(u => pyStringLiteral(stored(u))).mkString("[", ", ", "]") + // The output column is named by AttributeUnit.getAlias, which returns + // originalAttribute as written when alias is blank: the engine keeps the + // setting's case, not the table's. A rename is only needed where they differ. val renames = units - .filter(u => u.getAlias != u.getOriginalAttribute) - .map(u => s"""${pyStringLiteral(u.getOriginalAttribute)}: ${pyStringLiteral(u.getAlias)}""") + .filter(u => u.getAlias != stored(u)) + .map(u => s"""${pyStringLiteral(stored(u))}: ${pyStringLiteral(u.getAlias)}""") if (renames.isEmpty) { s"out1df = in1df[$originals].copy()" } else { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpDesc.scala index f3051311f9c..f2a0b312d9b 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpDesc.scala @@ -84,7 +84,8 @@ class RegexOpDesc extends FilterOpDesc with StandaloneCodeGenerator { // quantifiers, etc.) may behave differently in Python's re engine. val pyLiteral = pyStringLiteral(Option(regex).getOrElse("")) val caseArg = if (caseInsensitive) "False" else "True" - val attrLit = pyStringLiteral(attribute) + // The engine reads the field through Schema, which ignores case. See [[columnAsStored]]. + val attrLit = pyStringLiteral(columnAsStored(attribute, inputSchemas.values.headOption)) // The rows with nothing in the column are dropped before the match rather than // left to `na=False`, which by then has no null to see. // diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDesc.scala index 9170cbe9888..8903738cd0e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDesc.scala @@ -96,7 +96,8 @@ class SubstringSearchOpDesc extends FilterOpDesc with StandaloneCodeGenerator { // equivalent — the substring is matched literally, not as a regex. val pyLiteral = pyStringLiteral(Option(substring).getOrElse("")) val caseArg = if (isCaseSensitive) "True" else "False" - val attrLit = pyStringLiteral(attribute) + // The engine reads the field through Schema, which ignores case. See [[columnAsStored]]. + val attrLit = pyStringLiteral(columnAsStored(attribute, inputSchemas.values.headOption)) // The rows with nothing in the column are dropped before the match rather than // left to `na=False`, which by then has no null to see. // diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala index 145b1750778..b4290564f73 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala @@ -86,8 +86,17 @@ class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { // `tupleCasting` takes a Map of column to target type and reads each column's // ORIGINAL value once, so two units naming one column collapse to the last. val lastPerColumn = units.zipWithIndex.groupBy(_._1.attribute).view.mapValues(_.last._2).toMap + // Unlike most operators, both `tupleCasting` and `SchemaCasting` match a column + // by its exact name, not through Schema, which ignores case. So a unit that + // names no column, or names one in a different case, casts nothing in the + // engine, where pandas would raise a KeyError. Without a schema there is + // nothing to check against, and every unit is kept. + val inputSchema = inputSchemas.values.headOption val effective = units.zipWithIndex.collect { - case (unit, i) if lastPerColumn(unit.attribute) == i => unit + case (unit, i) + if lastPerColumn(unit.attribute) == i && + inputSchema.forall(_.getAttributeNames.contains(unit.attribute)) => + unit } // What each column arrived as: no cast reads another's result. diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala index 3a90db3009c..bf022f21618 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala @@ -145,7 +145,8 @@ class UnnestStringOpDesc extends FlatMapOpDesc with StandaloneCodeGenerator { // and the two column names are rendered as escaped Python literals. val delim = pyStringLiteral(Option(delimiter).getOrElse("")) val resultLit = pyStringLiteral(resultAttribute) - val attributeLit = pyStringLiteral(attribute) + // The engine reads the field through Schema, which ignores case. See [[columnAsStored]]. + val attributeLit = pyStringLiteral(columnAsStored(attribute, inputSchemas.values.headOption)) // What is split is the text the engine split, not whatever pandas made of the // column. See [[renderedAsText]]. val declared = inputSchemas.values.headOption diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDescSpec.scala index fbafd6869c3..edfd348c566 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDescSpec.scala @@ -103,4 +103,18 @@ class DictionaryMatcherOpDescSpec extends AnyFlatSpec with Matchers { val tree = objectMapper.readTree(json) tree.get("Matching type").asText shouldBe "Scan" } + + // The engine reads `Text` from a column `text`, since Schema ignores case. pandas + // would raise a KeyError on `Text`, so the script reads the column as the table + // spells it, whichever way it matches. + "DictionaryMatcherOpDesc.generateStandaloneCode" should + "read a column named in a different case from the table" in { + val inputSchemas = + Map(PortIdentity() -> Schema().add(new Attribute("text", AttributeType.STRING))) + MatchingType.values().foreach { mt => + val code = newDesc("a", "Text", "matched", mt).generateStandaloneCode(inputSchemas) + code should include("""out1df["text"]""") + code should not include "Text" + } + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 39e1ec59988..a475c615acf 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -20,7 +20,9 @@ package org.apache.texera.amber.operator.filter import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -81,6 +83,21 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { code should include("out1df") } + // FilterPredicate reads `Name` from a column `name`, since Schema ignores case. + // pandas would raise a KeyError on `Name`, so the script reads the column as the + // table spells it. + it should "read a column named in a different case from the table" in { + val d = new SpecializedFilterOpDesc + d.predicates = List( + new FilterPredicate("Name", ComparisonType.EQUAL_TO, "a"), + new FilterPredicate("NAME", ComparisonType.IS_NULL, "") + ) + val inputSchemas = + Map(PortIdentity() -> Schema().add(new Attribute("name", AttributeType.STRING))) + d.generateStandaloneCode(inputSchemas) shouldBe + """out1df = in1df[(in1df["name"].notna() & (in1df["name"] == "a")) | (in1df["name"].isna())].reset_index(drop=True)""" + } + // The executor filters on `predicates.exists`, which answers false on an empty // list, so no predicate keeps no row. The columns survive; the rows do not. it should "keep no row when there is no predicate" in { diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDescSpec.scala index 59b7bb456bf..79c1d4ccae3 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/keywordSearch/KeywordSearchOpDescSpec.scala @@ -20,7 +20,9 @@ package org.apache.texera.amber.operator.keywordSearch import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorMetadataGenerator} import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -146,4 +148,17 @@ class KeywordSearchOpDescSpec extends AnyFlatSpec with Matchers { kw.keyword shouldBe "apache" kw.isCaseSensitive shouldBe true } + + // The engine reads `Title` from a column `title`, since Schema ignores case. pandas + // would raise a KeyError on `Title`, so the script reads the column as the table + // spells it. + "KeywordSearchOpDesc.generateStandaloneCode" should + "read a column named in a different case from the table" in { + val inputSchemas = + Map(PortIdentity() -> Schema().add(new Attribute("title", AttributeType.STRING))) + val code = + newDesc("Title", "apache", caseSensitive = false).generateStandaloneCode(inputSchemas) + code should include("""in1df["title"]""") + code should not include "Title" + } } 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 848a2b4a647..cd1ba862a2f 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 @@ -229,6 +229,27 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { assert(drop.generateStandaloneCode() == """out1df = in1df.drop(columns=["a"])""") } + // The engine reads `FIELD1` from a column `field1`, since Schema ignores case, and + // names the kept column as the setting writes it. pandas would raise a KeyError on + // `FIELD1`, so the script reads the column as the table spells it. + it should "read a column named in a different case from the table" in { + val inputSchemas = Map(PortIdentity() -> schema) + + val keep = new ProjectionOpDesc + keep.attributes = List(new AttributeUnit("FIELD1", ""), new AttributeUnit("Field2", "f2")) + assert( + keep.generateStandaloneCode(inputSchemas) == + """out1df = in1df[["field1", "field2"]].rename(columns={"field1": "FIELD1", "field2": "f2"})""" + ) + + val drop = new ProjectionOpDesc + drop.attributes = List(new AttributeUnit("FIELD1", "")) + drop.isDrop = true + assert( + drop.generateStandaloneCode(inputSchemas) == """out1df = in1df.drop(columns=["field1"])""" + ) + } + // Schema propagation and the executor both refuse an empty selection, so the // script stops where a run would have. Passing the frame through would answer // with data a run never produces. diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpDescSpec.scala index c1f2db6a1fb..35c489e622c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpDescSpec.scala @@ -20,7 +20,9 @@ package org.apache.texera.amber.operator.regex import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -75,4 +77,17 @@ class RegexOpDescSpec extends AnyFlatSpec with Matchers { re.regex shouldBe "^https?://" re.caseInsensitive shouldBe true } + + // The engine reads `Url` from a column `url`, since Schema ignores case. pandas + // would raise a KeyError on `Url`, so the script reads the column as the table + // spells it. + "RegexOpDesc.generateStandaloneCode" should + "read a column named in a different case from the table" in { + val inputSchemas = + Map(PortIdentity() -> Schema().add(new Attribute("url", AttributeType.STRING))) + val code = + newDesc("Url", "^https?://", caseInsensitive = false).generateStandaloneCode(inputSchemas) + code should include("""in1df["url"]""") + code should not include "Url" + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDescSpec.scala index 2d0655d1d40..d278b977aa3 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpDescSpec.scala @@ -20,7 +20,9 @@ package org.apache.texera.amber.operator.substringSearch import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -80,4 +82,16 @@ class SubstringSearchOpDescSpec extends AnyFlatSpec with Matchers { ss.substring shouldBe "lo" ss.isCaseSensitive shouldBe true } + + // The engine reads `Body` from a column `body`, since Schema ignores case. pandas + // would raise a KeyError on `Body`, so the script reads the column as the table + // spells it. + "SubstringSearchOpDesc.generateStandaloneCode" should + "read a column named in a different case from the table" in { + val inputSchemas = + Map(PortIdentity() -> Schema().add(new Attribute("body", AttributeType.STRING))) + val code = newDesc("Body", "lo", caseSensitive = false).generateStandaloneCode(inputSchemas) + code should include("""in1df["body"]""") + code should not include "Body" + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 2aa9341e7c9..f59261a0130 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala @@ -86,6 +86,29 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { ) } + // Type Casting matches a column by its exact name, unlike the operators that go + // through Schema, so `N` against a column `n` casts nothing in the engine. The + // script skips that unit too rather than raise a KeyError on `N`. + "TypeCastingOpDesc.generateStandaloneCode" should + "skip a unit whose column differs from the table's in case" in { + val op = new TypeCastingOpDesc + op.typeCastingUnits = + List(castUnit("N", AttributeType.STRING), castUnit("m", AttributeType.DOUBLE)) + val input = Schema() + .add(new Attribute("n", AttributeType.INTEGER)) + .add(new Attribute("m", AttributeType.INTEGER)) + val inputPort = op.operatorInfo.inputPorts.head.id + + op.getExternalOutputSchemas(Map(inputPort -> input)).values.head shouldBe Schema() + .add(new Attribute("n", AttributeType.INTEGER)) + .add(new Attribute("m", AttributeType.DOUBLE)) + + val code = op.generateStandaloneCode(Map(inputPort -> input)) + code should include("""out1df["m"] = """) + code should not include "\"N\"" + code should not include "out1df[\"n\"] = " + } + "TypeCastingOpDesc" should "round-trip its casting units through the polymorphic base" in { val op = new TypeCastingOpDesc op.typeCastingUnits = List(castUnit("n", AttributeType.STRING)) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDescSpec.scala index 74fde249a45..c77806956c7 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDescSpec.scala @@ -214,6 +214,17 @@ class UnnestStringOpDescSpec extends AnyFlatSpec with Matchers { ) } + // The engine reads `N` from a column `n`, since Schema ignores case. pandas would + // raise a KeyError on `N`, so the script reads the column as the table spells it. + it should "read a column named in a different case from the table" in { + val op = newDesc(",", "N", "piece") + val schemas = Map(PortIdentity(0) -> Schema().add(new Attribute("n", AttributeType.STRING))) + val code = op.generateStandaloneCode(schemas) + code should include("""in1df[in1df["n"].notna()]""") + code should include("""_texera_cast_string(out1df["n"])""") + code should not include "\"N\"" + } + "UnnestStringOpDesc" should "round-trip its fields through the polymorphic base" in { val restored = objectMapper.readValue( From 99224085bc814b8453764e3405ee50a8810e0bf4 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 22 Oct 2026 00:41:03 -0700 Subject: [PATCH 2/2] fix(workflow-operator): name Projection's output columns by position Two units can read one column: `Name` as `first` and `NAME` as `second` both resolve to the table's `Name`. A rename map keyed by that column kept only the last alias, so the script returned `second, second` where the engine returns `first, second`. The script now sets the output names in order with `set_axis`. Co-Authored-By: Claude Opus 5.5 --- .../projection/ProjectionOpDesc.scala | 12 ++++++------ .../projection/ProjectionOpDescSpec.scala | 19 ++++++++++++++++++- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala index d5ab7377f8f..7c50e7c07a1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala @@ -130,14 +130,14 @@ class ProjectionOpDesc extends MapOpDesc with StandaloneCodeGenerator { // The output column is named by AttributeUnit.getAlias, which returns // originalAttribute as written when alias is blank: the engine keeps the // setting's case, not the table's. A rename is only needed where they differ. - val renames = units - .filter(u => u.getAlias != stored(u)) - .map(u => s"""${pyStringLiteral(stored(u))}: ${pyStringLiteral(u.getAlias)}""") - if (renames.isEmpty) { + // The names are set by position: two units can read one column, `Name` as + // `first` and `NAME` as `second`, and a rename map keyed by that column keeps + // only the last alias. + if (units.forall(u => u.getAlias == stored(u))) { s"out1df = in1df[$originals].copy()" } else { - val renameMap = renames.mkString("{", ", ", "}") - s"out1df = in1df[$originals].rename(columns=$renameMap)" + val aliases = units.map(u => pyStringLiteral(u.getAlias)).mkString("[", ", ", "]") + s"out1df = in1df[$originals].set_axis($aliases, axis=1)" } } } 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 cd1ba862a2f..74ce2c759c4 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 @@ -239,7 +239,7 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { keep.attributes = List(new AttributeUnit("FIELD1", ""), new AttributeUnit("Field2", "f2")) assert( keep.generateStandaloneCode(inputSchemas) == - """out1df = in1df[["field1", "field2"]].rename(columns={"field1": "FIELD1", "field2": "f2"})""" + """out1df = in1df[["field1", "field2"]].set_axis(["FIELD1", "f2"], axis=1)""" ) val drop = new ProjectionOpDesc @@ -250,6 +250,23 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { ) } + // `field1` as `first` and `FIELD1` as `second` read the same column. The engine + // keeps both, in order; a rename map keyed by `field1` would keep only `second`, + // twice. So the script names the output columns by position. + it should "keep both aliases of one column read twice" in { + val keep = new ProjectionOpDesc + keep.attributes = + List(new AttributeUnit("field1", "first"), new AttributeUnit("FIELD1", "second")) + assert( + keep.getExternalOutputSchemas(Map(PortIdentity() -> schema)).values.head.getAttributeNames == + List("first", "second") + ) + assert( + keep.generateStandaloneCode(Map(PortIdentity() -> schema)) == + """out1df = in1df[["field1", "field1"]].set_axis(["first", "second"], axis=1)""" + ) + } + // Schema propagation and the executor both refuse an empty selection, so the // script stops where a run would have. Passing the frame through would answer // with data a run never produces.