feat(workflow-operator): export the operators that work on text

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala

@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator
+
+/** Python definitions shared by several operators' standalone code, emitted
+  * once per script via [[StandaloneCodeGenerator.standaloneHelpers]].
+  */
+object StandaloneHelpers {
+
+  /**
+    * A Python transcription of `AttributeTypeUtils`, for operators that cast a
+    * column to a declared type.
+    *
+    * Python's own conversions answer differently on the values a spreadsheet
+    * column actually holds. `bool("false")` is true, because every non-empty
+    * string is, and `int("6.7")` raises where the engine reads 6: a cast goes
+    * through `parseField(force = true)`, whose numeric branch is
+    * `java.text.NumberFormat`, which truncates a decimal, drops a grouping
+    * comma and stops at trailing letters. The engine reads "false" as false and
+    * "0" as false too, so the script has to do the same rather than hand back a
+    * column the workflow never produced.
+    *
+    * Refusing is still part of the contract where the engine refuses: text with
+    * no leading digits raises, and a script that quietly wrote NaN instead
+    * would report an answer the run it was exported from never reached.
+    */
+  val AttributeCasts: String =
+    """# AttributeTypeUtils, transcribed so a cast answers as the engine does.
+      |def _texera_cast_boolean(x):
+      |    # toBoolean first, then `toInt == 1`: "0" and "2" are both false.
+      |    if isinstance(x, str):
+      |        text = x.strip()
+      |        lowered = text.lower()
+      |        if lowered == "true":
+      |            return True
+      |        if lowered == "false":
+      |            return False
+      |        return int(text) == 1
+      |    return x != 0
+      |
+      |
+      |def _texera_parse_number(text):
+      |    # java.text.NumberFormat for Locale.US, which a cast reaches through
+      |    # `parseField(force = true)`. Lenient where Python's int is not: it stops
+      |    # at the first character that cannot continue a number ("12abc" is 12),
+      |    # drops "," without checking group sizes ("1,23" is 123), and reads "."
+      |    # as a decimal point. It refuses a leading "+" and text with no digits.
+      |    #
+      |    # Also returns whether it came back as a Double, which decides between
+      |    # the two narrowings below.
+      |    import re
+      |
+      |    match = re.match(r"\s*(-?)([0-9,]*)(?:\.([0-9]*))?", text)
+      |    sign = -1 if match.group(1) == "-" else 1
+      |    digits = (match.group(2) or "").replace(",", "")
+      |    fraction = match.group(3) or ""
+      |    if not digits and not fraction:
+      |        raise ValueError("Unparseable number: " + repr(text))
+      |    if fraction:
+      |        return sign * float((digits or "0") + "." + fraction), True
+      |    value = sign * int(digits)
+      |    # A whole number past long's range comes back as a Double.
+      |    if -(2 ** 63) <= value <= 2 ** 63 - 1:
+      |        return value, False
+      |    return float(value), True
+      |
+      |
+      |def _texera_long_value(value, is_double):
+      |    # Number.longValue(): a Double truncates toward zero and saturates at
+      |    # long's bounds, where a Long is already itself.
+      |    if is_double:
+      |        return max(-(2 ** 63), min(2 ** 63 - 1, int(value)))
+      |    return value
+      |
+      |
+      |def _texera_cast_integral(x):
+      |    # The LONG target. Scala's toLong takes no decimal point of its own, so
+      |    # a fraction only ever arrives already parsed.
+      |    if isinstance(x, str):
+      |        return _texera_long_value(*_texera_parse_number(x))
+      |    if isinstance(x, bool):
+      |        return 1 if x else 0
+      |    if isinstance(x, float):
+      |        return _texera_long_value(x, True)
+      |    return int(x)
+      |
+      |
+      |def _texera_cast_int32(x, wrap):
+      |    # The two 32-bit narrowings differ: Long.toInt keeps the low bits, so
+      |    # 2147483648 comes back as -2147483648, while Double.toInt saturates.
+      |    #
+      |    # Text decides by what the parse returned; everything else is told by the
+      |    # SOURCE column's declared type, because `Series.apply` hands the cells of
+      |    # a nullable integer column over as floats.
+      |    if isinstance(x, str):
+      |        value, is_double = _texera_parse_number(x)
+      |        wrap = not is_double
+      |    elif isinstance(x, bool):
+      |        value = 1 if x else 0
+      |    else:
+      |        value = int(x)
+      |    if wrap:
+      |        return ((value + 2147483648) % 4294967296) - 2147483648
+      |    # int() first: Double.intValue truncates toward zero before it clamps.
+      |    return max(-2147483648, min(2147483647, int(value)))
+      |
+      |
+      |def _texera_epoch_millis_to_timestamp(s):
+      |    # `new Timestamp(long)` reads MILLISECONDS where pd.to_datetime defaults
+      |    # to nanoseconds, and renders in the JVM's default zone, so leaving the
+      |    # result in UTC would put it a whole offset away. Text needs neither: a
+      |    # parsed wall clock is already the wall clock.
+      |    # tzlocal() and not the current offset: the zone carries its daylight
+      |    # rules, and each instant needs the offset in force when it happened.
+      |    from dateutil.tz import tzlocal
+      |
+      |    return (
+      |        pd.to_datetime(s, unit="ms", errors="coerce", utc=True)
+      |        .dt.tz_convert(tzlocal())
+      |        .dt.tz_localize(None)
+      |    )
+      |
+      |
+      |def _texera_cast_double(x):
+      |    if isinstance(x, str):
+      |        return float(x.strip())
+      |    return float(x)
+      |
+      |
+      |def _texera_cast_string(s):
+      |    # `toString` on the field, so the COLUMN's type decides the text and
+      |    # not the shape of the value: a double keeps its point, whether or not
+      |    # it lands on a whole number, and an integer never grows one. A column
+      |    # of no single type is read value by value.
+      |    if pd.api.types.is_bool_dtype(s):
+      |        return s.map(lambda x: None if pd.isna(x) else ("true" if x else "false"))
+      |    if pd.api.types.is_integer_dtype(s):
+      |        return s.map(lambda x: None if pd.isna(x) else str(int(x)))
+      |    if pd.api.types.is_datetime64_any_dtype(s):
+      |        # java.sql.Timestamp.toString: trailing zeros dropped from the
+      |        # fraction, but never all of them. Python's own str() writes no
+      |        # fraction at all on a whole second and six digits otherwise.
+      |        def _ts(x):
+      |            if pd.isna(x):
+      |                return None
+      |            text = x.strftime("%Y-%m-%d %H:%M:%S.%f").rstrip("0")
+      |            return text + "0" if text.endswith(".") else text
+      |
+      |        return s.map(_ts)
+      |    return s.map(
+      |        lambda x: None
+      |        if pd.isna(x)
+      |        else ("true" if x else "false") if pd.api.types.is_bool(x) else str(x)
+      |    )""".stripMargin
+}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala

@@ -29,23 +29,31 @@ import org.apache.texera.amber.core.workflow.{
   PhysicalOp,
   SchemaPropagationFunc
 }
+import org.apache.texera.amber.operator.StandaloneCodeGenerator
 import org.apache.texera.amber.operator.map.MapOpDesc
-import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName
+import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn}
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 /**
   * Dictionary matcher operator matches a tuple if the specified column is in the given dictionary.
   * It outputs an extra column to label the tuple if it is matched or not
   * This is the description of the operator
   */
-class DictionaryMatcherOpDesc extends MapOpDesc {
+class DictionaryMatcherOpDesc extends MapOpDesc with StandaloneCodeGenerator {
   @JsonProperty(value = "Dictionary", required = true)
   @JsonPropertyDescription("dictionary values separated by a comma") var dictionary: String = _
 
+  // Verification matches against a text column holding the value it fills a dictionary
+  // with, in rows that are single words a stemmer leaves alone -- the conjunction
+  // branch runs Lucene's stemmer on one side and nothing on the other, so a word that
+  // stems agrees either way.
   @JsonProperty(value = "Attribute", required = true)
   @JsonPropertyDescription("column name to match")
-  @AutofillAttributeName var attribute: String = _
+  @AutofillAttributeName
+  @SampleColumn("name")
+  var attribute: String = _
 
   @JsonProperty(value = "result attribute", required = true, defaultValue = "matched")
   @JsonPropertyDescription("column name of the matching result") var resultAttribute: String = _
@@ -89,4 +97,127 @@ class DictionaryMatcherOpDesc extends MapOpDesc {
       outputPorts = List(OutputPort()),
       supportReconfiguration = true
     )
+
+  override def generateStandaloneCode(): 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(""))
+    val resultPy = pyStringLiteral(resultCol)
+    val mt = Option(matchingType).getOrElse(MatchingType.SCANBASED)
+
+    val entriesLiteral = entries.map(pyStringLiteral).mkString("[", ", ", "]")
+
+    mt match {
+      case MatchingType.SCANBASED =>
+        // Exact case-insensitive equality between the (lowercased) cell value
+        // and any dictionary entry. Null or empty text never matches, matching
+        // JVM behavior.
+        s"""out1df = in1df.copy()
+           |out1df[$resultPy] = out1df[$attrPy].apply(
+           |    lambda v, _entries=$entriesLiteral: (
+           |        False if pd.isna(v)
+           |        else (str(v).lower() != "" and str(v).lower() in _entries)
+           |    )
+           |)""".stripMargin
+
+      case MatchingType.SUBSTRING =>
+        // JVM checks dictionaryEntries.exists(entry => entry.contains(text)) —
+        // the cell value (lowercased) is a substring of some dictionary entry,
+        // NOT the other way around.
+        s"""out1df = in1df.copy()
+           |out1df[$resultPy] = out1df[$attrPy].apply(
+           |    lambda v, _entries=$entriesLiteral: (
+           |        False if pd.isna(v)
+           |        else (str(v).lower() != "" and any(str(v).lower() in _e for _e in _entries))
+           |    )
+           |)""".stripMargin
+
+      case MatchingType.CONJUNCTION_INDEXBASED =>
+        // JVM tokenizes via Lucene EnglishAnalyzer (StandardTokenizer + lowercase
+        // + English stop words + possessive filter + Porter2 stemmer) and matches
+        // when an entry's token set is a subset of the text's token set. Best-
+        // effort reproduction: regex \w+ tokens, lowercase, drop the same stop
+        // word lists, but NO stemming — morphological variants ("book" vs "books")
+        // will diverge from JVM behavior.
+        val tokenSetsLiteral = entries
+          .map(tokenizeForConjunction)
+          .map(toks => toks.toList.sorted.map(pyStringLiteral).mkString("frozenset({", ", ", "})"))
+          .mkString("[", ", ", "]")
+        val stopWordsLiteral = DictionaryMatcherOpDesc.STOP_WORDS.toList.sorted
+          .map(pyStringLiteral)
+          .mkString("frozenset({", ", ", "})")
+        s"""import re as _texera_dm_re
+           |_TEXERA_DM_STOPWORDS = $stopWordsLiteral
+           |def _texera_dm_tokenize(text):
+           |    return frozenset(t for t in _texera_dm_re.findall(r"\\w+", text.lower()) if t not in _TEXERA_DM_STOPWORDS)
+           |out1df = in1df.copy()
+           |out1df[$resultPy] = out1df[$attrPy].apply(
+           |    lambda v, _entries=$tokenSetsLiteral: (
+           |        False if pd.isna(v)
+           |        else (lambda _t: bool(_t) and any(_e.issubset(_t) for _e in _entries))(_texera_dm_tokenize(str(v)))
+           |    )
+           |)""".stripMargin
+    }
+  }
+
+  private def tokenizeForConjunction(text: String): Set[String] = {
+    val wordRe = "\\w+".r
+    wordRe
+      .findAllIn(text.toLowerCase)
+      .toSet
+      .filterNot(DictionaryMatcherOpDesc.STOP_WORDS.contains)
+  }
+}
+
+private object DictionaryMatcherOpDesc {
+  // Mirrors Lucene's EnglishAnalyzer.ENGLISH_STOP_WORDS_SET plus the URL stop
+  // words filtered by DictionaryMatcherOpExec.
+  val STOP_WORDS: Set[String] = Set(
+    "a",
+    "an",
+    "and",
+    "are",
+    "as",
+    "at",
+    "be",
+    "but",
+    "by",
+    "for",
+    "if",
+    "in",
+    "into",
+    "is",
+    "it",
+    "no",
+    "not",
+    "of",
+    "on",
+    "or",
+    "such",
+    "that",
+    "the",
+    "their",
+    "then",
+    "there",
+    "these",
+    "they",
+    "this",
+    "to",
+    "was",
+    "will",
+    "with",
+    "http",
+    "https",
+    "org",
+    "net",
+    "com",
+    "store",
+    "www",
+    "html"
+  )
 }