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..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 @@ -108,30 +108,36 @@ 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 renames = units - .filter(u => u.getAlias != u.getOriginalAttribute) - .map(u => s"""${pyStringLiteral(u.getOriginalAttribute)}: ${pyStringLiteral(u.getAlias)}""") - if (renames.isEmpty) { + 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. + // 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/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 4a13a20db2c..fd2009a5b65 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 @@ -91,7 +91,8 @@ class RegexOpDesc extends FilterOpDesc with StandaloneCodeGenerator { RegexOpDesc.unicodeBoundaries(RegexOpDesc.unicodeCase(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 3d4358ede2c..45ff58fe2ff 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 @@ -99,7 +99,8 @@ class SubstringSearchOpDesc extends FilterOpDesc with StandaloneCodeGenerator { // call the engine makes, and the column by Python's lower(), which follows // the same Unicode mappings. val text = Option(substring).getOrElse("") - 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..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 @@ -229,6 +229,44 @@ 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"]].set_axis(["FIELD1", "f2"], axis=1)""" + ) + + val drop = new ProjectionOpDesc + drop.attributes = List(new AttributeUnit("FIELD1", "")) + drop.isDrop = true + assert( + drop.generateStandaloneCode(inputSchemas) == """out1df = in1df.drop(columns=["field1"])""" + ) + } + + // `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. 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 b7af76f6234..4c7befcb7a0 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 @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.regex import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp @@ -207,4 +207,17 @@ class RegexOpDescSpec extends AnyFlatSpec with Matchers { if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false } else p.exitValue() == 0 } + + // 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 a3a5c90680e..5e3fa32c0c8 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 @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.substringSearch import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp @@ -180,4 +180,16 @@ class SubstringSearchOpDescSpec extends AnyFlatSpec with Matchers { if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false } else p.exitValue() == 0 } + + // 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(