feat(workflow-operator): read a column named in a different case the way the engine does

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.
     *

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,15 +102,20 @@ 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("")
     val entries = rawDict.split(",").toList.map(_.toLowerCase)
     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)
 

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())"""

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.

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)"
       }
     }
   }

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.
     //

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.
     //

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.

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

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"
+    }
+  }
 }

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 {

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"
+  }
 }

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.

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"
+  }
 }

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"
+  }
 }