diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala new file mode 100644 index 00000000000..1414f6b043c --- /dev/null +++ b/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 +} 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 cfb2b5e76fa..8660fcdc69d 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 @@ -29,9 +29,11 @@ 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 /** @@ -39,13 +41,19 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper * 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" + ) } 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 b0e202a1bb6..b968535bedd 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 @@ -22,14 +22,17 @@ package org.apache.texera.amber.operator.keywordSearch import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} 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, StandaloneHelpers} import org.apache.texera.amber.operator.filter.FilterOpDesc import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName 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 -class KeywordSearchOpDesc extends FilterOpDesc { +class KeywordSearchOpDesc extends FilterOpDesc with StandaloneCodeGenerator { @JsonProperty(required = true) @JsonSchemaTitle("attribute") @@ -37,10 +40,19 @@ class KeywordSearchOpDesc extends FilterOpDesc { @AutofillAttributeName var attribute: String = _ + // The value is a Lucene query, and its lexer needs double quotes in pairs: one on its + // own opens a phrase that never closes, and `parse` throws before a row is read. The + // pattern walks the value the way the lexer does, so a quote a backslash escapes stays + // a character in a term. The other syntactic characters are left alone, since which + // uses of them parse depends on what follows and a stricter pattern would reject the + // phrase and range queries that work today. Anchored, because the form validates with + // `test`, which searches. @JsonProperty(required = true) @JsonSchemaTitle("keywords") @JsonPropertyDescription("keywords") - @JsonSchemaInject(json = """{"minLength": 1}""") + @JsonSchemaInject( + json = """{"minLength": 1, "pattern": "^(?:[^\"\\\\]|\\\\.|\"(?:[^\"\\\\]|\\\\.)*\")*$"}""" + ) var keyword: String = _ @JsonProperty(required = true, defaultValue = "false") @@ -75,4 +87,38 @@ class KeywordSearchOpDesc extends FilterOpDesc { outputPorts = List(OutputPort()), supportReconfiguration = true ) + + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { + // The engine runs a Lucene query per row; a script matches the terms + // themselves, so a query that uses the syntax — a phrase, a boolean, a + // wildcard, a fuzzy match — reads here as the words it is written with. + val raw = Option(keyword).getOrElse("") + val terms = raw.trim.split("\\s+").filter(_.nonEmpty).toList + if (terms.isEmpty) return "out1df = in1df" + + val regexSpecials = Set('.', '^', '$', '*', '+', '?', '(', ')', '[', ']', '{', '}', '|', '\\') + 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) + // 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. + val caseArg = if (isCaseSensitive) "True" else "False" + + // 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. + // + // The terms are matched against the text the engine indexed, not against + // whatever pandas made of the column. See [[renderedAsText]]. + val declared = inputSchemas.values.headOption + .flatMap(schema => scala.util.Try(schema.getAttribute(attribute)).toOption) + .map(_.getType) + val column = renderedAsText(s"in1df[$attrLit]", declared) + s"""out1df = in1df[in1df[$attrLit].notna() & $column.str.contains($pyLiteral, regex=True, case=$caseArg, na=False)].reset_index(drop=True)""" + } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java new file mode 100644 index 00000000000..cb35c9aab91 --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java @@ -0,0 +1,45 @@ +/* + * 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.metadata.annotations; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Test-only metadata for transform verification: names the column in the shared + * verification fixture the operator runs on that should fill this + * {@code @AutofillAttributeName} field when the operator is auto-configured. + * + *

It lets a field declare a semantic sample that the column's + * {@code AttributeType} alone cannot express — e.g. a valid three-letter ISO + * country code, or a genuine OHLC price column — so the parity test exercises + * the operator on realistic input instead of a degenerate first-column pick + * (which can hide translation bugs and produce vacuous passes). + * + *

This has no effect on production: it is not a Jackson / JSON-schema + * annotation and is read only by the test-side ConfigGenerator. + */ +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.FIELD}) +public @interface SampleColumn { + String value(); +} 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 0f886ae53be..f3051311f9c 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 @@ -22,14 +22,17 @@ package org.apache.texera.amber.operator.regex import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle 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, StandaloneHelpers} import org.apache.texera.amber.operator.filter.FilterOpDesc import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName 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 -class RegexOpDesc extends FilterOpDesc { +class RegexOpDesc extends FilterOpDesc with StandaloneCodeGenerator { @JsonProperty(value = "attribute", required = true) @JsonPropertyDescription("column to search regex on") @@ -72,4 +75,27 @@ class RegexOpDesc extends FilterOpDesc { outputPorts = List(OutputPort()), supportReconfiguration = true ) + + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { + // JVM uses Java Pattern.matcher(v).find — partial match. pandas str.contains + // is also partial by default. Java-only regex syntax (\Q\E, possessive + // 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 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. + // + // The pattern is matched against the text the engine would have matched, not + // against whatever pandas made of the column. See [[renderedAsText]]. + val declared = inputSchemas.values.headOption + .flatMap(schema => scala.util.Try(schema.getAttribute(attribute)).toOption) + .map(_.getType) + val column = renderedAsText(s"in1df[$attrLit]", declared) + s"""out1df = in1df[in1df[$attrLit].notna() & $column.str.contains($pyLiteral, regex=True, case=$caseArg, na=False)].reset_index(drop=True)""" + } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpExec.scala index c8e840de5be..ebeb59fc57e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/regex/RegexOpExec.scala @@ -31,8 +31,11 @@ class RegexOpExec(descString: String) extends FilterOpExec { Pattern.compile(desc.regex, if (desc.caseInsensitive) Pattern.CASE_INSENSITIVE else 0) this.setFilterFunc(this.matchRegex) + // A row with nothing in the column matches nothing. The `Option` here used to + // wrap the `toString` rather than the field, so a null raised before it was + // ever consulted; an empty cell is ordinary input, since a blank in a CSV + // arrives as one. private def matchRegex(tuple: Tuple): Boolean = - Option[Any](tuple.getField(desc.attribute).toString) - .map(_.toString) - .exists(value => pattern.matcher(value).find) + Option(tuple.getField[Any](desc.attribute)) + .exists(value => pattern.matcher(value.toString).find) } 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 da4a7437b93..9170cbe9888 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 @@ -20,26 +20,39 @@ package org.apache.texera.amber.operator.substringSearch import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} -import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} 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, StandaloneHelpers} import org.apache.texera.amber.operator.filter.FilterOpDesc -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 -class SubstringSearchOpDesc extends FilterOpDesc { +class SubstringSearchOpDesc extends FilterOpDesc with StandaloneCodeGenerator { + // Verification reads a column holding lower-case, upper-case and letterless rows, + // so that flipping Case Sensitive changes WHICH rows match. On a single-case column + // it changes nothing and the sweep decides nothing. @JsonProperty(required = true) @JsonSchemaTitle("attribute") @JsonPropertyDescription("column to search substring on") @AutofillAttributeName + @SampleColumn("mixed_case") var attribute: String = _ + // A letter-bearing sample, for the same reason -- case cannot matter to a digit. @JsonProperty(required = true) @JsonSchemaTitle("Substring") @JsonPropertyDescription("substring") + @JsonSchemaInject(json = """ +{ + "examples": ["ab"] +} +""") var substring: String = _ @JsonProperty(required = true, defaultValue = "false") @@ -74,4 +87,27 @@ class SubstringSearchOpDesc extends FilterOpDesc { outputPorts = List(OutputPort()), supportReconfiguration = true ) + + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { + // JVM uses String.contains (case-sensitive) or toLowerCase.contains + // (case-insensitive). pandas str.contains with regex=False is the direct + // 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 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. + // + // The substring is searched in the text the engine would have searched, not + // in whatever pandas made of the column. See [[renderedAsText]]. + val declared = inputSchemas.values.headOption + .flatMap(schema => scala.util.Try(schema.getAttribute(attribute)).toOption) + .map(_.getType) + val column = renderedAsText(s"in1df[$attrLit]", declared) + s"""out1df = in1df[in1df[$attrLit].notna() & $column.str.contains($pyLiteral, regex=False, case=$caseArg, na=False)].reset_index(drop=True)""" + } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExec.scala index 5331a4601d0..3044bca1b14 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExec.scala @@ -30,11 +30,19 @@ class SubstringSearchOpExec(descString: String) extends FilterOpExec { this.setFilterFunc(findSubstring) private def findSubstring(tuple: Tuple): Boolean = { - val content = tuple.getField(desc.attribute).toString - if (desc.isCaseSensitive) { - content.contains(desc.substring) + val field = tuple.getField[Any](desc.attribute) + // A row with nothing in the column matches nothing. FilterPredicate answers the + // same way: once a field is null, every condition but IS_NULL / IS_NOT_NULL is + // false. An empty cell is ordinary input, since a blank in a CSV arrives as null. + if (field == null) { + false } else { - content.toLowerCase.contains(desc.substring.toLowerCase) + val content = field.toString + if (desc.isCaseSensitive) { + content.contains(desc.substring) + } else { + content.toLowerCase.contains(desc.substring.toLowerCase) + } } } } 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 82b0af60b36..c23c3dd66f0 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 @@ -22,14 +22,16 @@ package org.apache.texera.amber.operator.typecasting import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.{AttributeTypeUtils, Schema} +import org.apache.texera.amber.core.tuple.{AttributeType, AttributeTypeUtils, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow._ +import org.apache.texera.amber.operator.{StandaloneCodeGenerator, StandaloneHelpers} import org.apache.texera.amber.operator.map.MapOpDesc 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 -class TypeCastingOpDesc extends MapOpDesc { +class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { @JsonProperty(required = true) @JsonSchemaTitle("TypeCasting Units") @@ -72,4 +74,72 @@ class TypeCastingOpDesc extends MapOpDesc { List(OutputPort()) ) } + + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { + val units = Option(typeCastingUnits).getOrElse(List.empty) + if (units.isEmpty) return "out1df = in1df.copy()" + + // `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 + val effective = units.zipWithIndex.collect { + case (unit, i) if lastPerColumn(unit.attribute) == i => unit + } + + // What each column arrived as: no cast reads another's result. + val declared: Map[String, AttributeType] = inputSchemas.values.headOption + .map(_.getAttributes.map(a => a.getName -> a.getType).toMap) + .getOrElse(Map.empty) + + val lines = scala.collection.mutable.ArrayBuffer[String]("out1df = in1df.copy()") + effective.foreach { unit => + val colLit = pyStringLiteral(unit.attribute) + // Every cast goes through the transcription of AttributeTypeUtils rather + // than through Python's own conversions, which answer differently: a + // non-empty string is always a true boolean, and `int("6.7")` raises + // where the engine's NumberFormat reads 6. + // + // A timestamp is the one that stays approximate. The engine reads it with + // DateParserUtils, which accepts a set of formats no single pandas call + // states, so this coerces what it cannot read rather than claiming a + // match it does not have. + val expr = unit.resultType match { + case AttributeType.STRING => + // Not `astype(str)`, which renders an empty cell as "nan" and + // capitalises a boolean. See [[renderedAsText]]. + renderedAsText(s"out1df[$colLit]", declared.get(unit.attribute)) + case AttributeType.INTEGER => + // A hole survives the cast, because parseField returns a null field + // untouched; nullable "Int64" holds one where numpy's int cannot. Only + // a DOUBLE source saturates on the way to 32 bits, and only the + // declared type still says so: the values reach the helper as floats. + val wrap = !declared.get(unit.attribute).contains(AttributeType.DOUBLE) + val wrapArg = if (wrap) "True" else "False" + s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else _texera_cast_int32(x, $wrapArg)).astype("Int64")""" + case AttributeType.LONG => + s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else _texera_cast_integral(x)).astype("Int64")""" + case AttributeType.DOUBLE => + // NaN rather than pd.NA: float64 is how a double column is held here, + // and it carries its hole as NaN. pd.NA would not survive the astype. + s"""out1df[$colLit].apply(lambda x: float("nan") if pd.isna(x) else _texera_cast_double(x)).astype("float64")""" + case AttributeType.BOOLEAN => + // Nullable "boolean" for the same reason, and because `.astype(bool)` + // reads NaN as True: NaN is a non-zero float. + s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else _texera_cast_boolean(x)).astype("boolean")""" + case AttributeType.TIMESTAMP => + // A number is an instant in milliseconds and needs its own reading; + // see the helper. Text keeps the parser it already had. + if (declared.get(unit.attribute).contains(AttributeType.LONG)) + s"""_texera_epoch_millis_to_timestamp(out1df[$colLit])""" + else s"""pd.to_datetime(out1df[$colLit], errors="coerce")""" + case _ => s"""out1df[$colLit]""" + } + lines += s"""out1df[$colLit] = $expr""" + } + lines.mkString("\n") + } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts) } 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 ef742de3ae4..fa2e5c21511 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 @@ -21,20 +21,23 @@ package org.apache.texera.amber.operator.unneststring import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.AttributeType +import org.apache.texera.amber.core.tuple.{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, StandaloneHelpers} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.operator.flatmap.FlatMapOpDesc -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.util.JSONUtils.objectMapper -class UnnestStringOpDesc extends FlatMapOpDesc { +class UnnestStringOpDesc extends FlatMapOpDesc with StandaloneCodeGenerator { @JsonProperty(value = "Delimiter", required = true, defaultValue = ",") @JsonPropertyDescription("string that separates the data") var delimiter: String = _ @@ -42,6 +45,7 @@ class UnnestStringOpDesc extends FlatMapOpDesc { @JsonProperty(value = "Attribute", required = true) @JsonPropertyDescription("column of the string to unnest") @AutofillAttributeName + @SampleColumn("csv_list") var attribute: String = _ @JsonProperty(value = "Result attribute", required = true, defaultValue = "unnestResult") @@ -84,4 +88,32 @@ class UnnestStringOpDesc extends FlatMapOpDesc { }) ) } + + override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty) + + override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = { + if (resultAttribute == null || resultAttribute.trim.isEmpty) { + throw new RuntimeException("Result attribute cannot be empty") + } + // The JVM op uses Scala's `delimiter.r.split(...)`, so delimiter is a regex; it + // 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) + // What is split is the text the engine split, not whatever pandas made of the + // column. See [[renderedAsText]]. + val declared = inputSchemas.values.headOption + .flatMap(schema => scala.util.Try(schema.getAttribute(attribute)).toOption) + .map(_.getType) + val column = renderedAsText(s"out1df[$attributeLit]", declared) + s"""# Nothing in the column unnests to nothing, the way the operator answers a null + |# field with no rows at all. Dropped before the split rather than after: the + |# rendering would turn the empty cell into text and unnest that. + |out1df = in1df[in1df[$attributeLit].notna()].copy() + |out1df[$resultLit] = $column.str.split($delim, regex=True) + |out1df = out1df.explode($resultLit, ignore_index=True) + |out1df = out1df[(out1df[$resultLit].notna()) & (out1df[$resultLit] != "")].reset_index(drop=True)""".stripMargin + } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExec.scala index 1e59968217b..0a0e08d9a66 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExec.scala @@ -29,10 +29,18 @@ class UnnestStringOpExec(descString: String) extends FlatMapOpExec { setFlatMapFunc(splitByDelimiter) private def splitByDelimiter(tuple: Tuple): Iterator[TupleLike] = { - desc.delimiter.r - .split(tuple.getField(desc.attribute).toString) - .filter(_.nonEmpty) - .iterator - .map(split => TupleLike(tuple.getFields ++ Seq(split))) + val field = tuple.getField[Any](desc.attribute) + // Nothing in the column unnests to nothing, the same way the filter below drops + // the empty pieces a run of delimiters produces. An empty cell is ordinary input, + // since a blank in a CSV arrives as null. + if (field == null) { + Iterator.empty + } else { + desc.delimiter.r + .split(field.toString) + .filter(_.nonEmpty) + .iterator + .map(split => TupleLike(tuple.getFields ++ Seq(split))) + } } } 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 b56e04c8fd1..59b7bb456bf 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 @@ -22,11 +22,16 @@ package org.apache.texera.amber.operator.keywordSearch import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.operator.LogicalOp -import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorMetadataGenerator} import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.lucene.analysis.standard.StandardAnalyzer +import org.apache.lucene.queryparser.classic.QueryParser import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.util.regex.Pattern +import scala.util.Try + class KeywordSearchOpDescSpec extends AnyFlatSpec with Matchers { private val workflowId = WorkflowIdentity(1L) @@ -79,6 +84,58 @@ class KeywordSearchOpDescSpec extends AnyFlatSpec with Matchers { physical.outputPorts.keySet shouldBe op.operatorInfo.outputPorts.map(_.id).toSet } + /** The `pattern` the keyword field injects into its schema, read from the schema + * itself rather than restated here, so the test cannot pass against a stale copy. + */ + private val keywordPattern: String = { + val property = OperatorMetadataGenerator + .generateOperatorJsonSchema(classOf[KeywordSearchOpDesc]) + .path("properties") + .path("keyword") + property.path("pattern").asText() + } + + /** Whether the parser the executor builds accepts the value at all. */ + private def parses(keyword: String): Boolean = + Try(new QueryParser("col", new StandardAnalyzer()).parse(keyword)).isSuccess + + // A value the field should take, and whether the pattern is meant to admit it. Quoting + // is the only query syntax in play in each, so the verdict belongs to QueryParser, and + // every case is put to it below rather than decided here. + private val keywords: Seq[(String, Boolean)] = Seq( + "hello" -> true, + "\"a b\"" -> true, // a phrase query: the quotes are paired + "a \"b\" c" -> true, + "\"a\" \"b\"" -> true, + "he\\\"llo" -> true, // an escaped quote is a character in a term, not a delimiter + "\\\"unclosed" -> true, + "\"a \\\" b\"" -> true, // an escaped quote inside a phrase leaves the pair intact + "a\\\\\"b\"" -> true, // an escaped backslash, then a phrase + "he\"llo" -> false, + "\"unclosed" -> false, + "x\"y\"z\"" -> false, + "\\\\\"" -> false, // the backslash is escaped, so the quote is the bare one + "end\\" -> false // a backslash with nothing left to escape + ) + + behavior of "The keyword pattern" + + it should "be present in the generated schema" in { + keywordPattern should not be empty + } + + keywords.foreach { + case (value, isValid) => + val verb = if (isValid) "accept" else "reject" + it should s"$verb '$value', as QueryParser does" in { + // find() rather than matches(), because the form validates with + // `new RegExp().test`, which searches instead of anchoring. An unanchored + // pattern would pass every value here. + Pattern.compile(keywordPattern).matcher(value).find() shouldBe isValid + parses(value) shouldBe isValid + } + } + "KeywordSearchOpDesc JSON round-trip" should "preserve attribute, keyword, and isCaseSensitive via the polymorphic base" in { val json = objectMapper.writeValueAsString(newDesc("title", "apache", caseSensitive = true)) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpExecSpec.scala index 3d93539876c..b887b5f9e8f 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/regex/RegexOpExecSpec.scala @@ -127,6 +127,20 @@ class RegexOpExecSpec extends AnyFlatSpec { // Descriptor parse failure surfaces during construction // --------------------------------------------------------------------------- + it should "yield nothing when the column is empty" in { + val exec = new RegexOpExec(descJson(regex = "hello")) + // A blank CSV cell arrives as null. This used to throw a NullPointerException on + // the toString instead of answering the filter. + assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty) + } + + it should "yield nothing when the column is empty and the regex matches anything" in { + val exec = new RegexOpExec(descJson(regex = ".*")) + // `.*` finds a match in every value, including the empty string, but a row with + // no value has none to search, so it is filtered out rather than kept. + assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty) + } + "RegexOpExec construction" should "throw on malformed descriptor JSON" in { // The constructor calls objectMapper.readValue; mis-formed JSON must diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExecSpec.scala index 83fd90fee0f..bd7b947496c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExecSpec.scala @@ -94,6 +94,24 @@ class SubstringSearchOpExecSpec extends AnyFlatSpec { assert(exec.processTuple(t, port = 0).toList == List(t)) } + // --------------------------------------------------------------------------- + // Edge: empty cell + // --------------------------------------------------------------------------- + + it should "yield nothing when the column is empty" in { + val exec = new SubstringSearchOpExec(descJson(substring = "hello")) + // A blank CSV cell arrives as null. This used to throw a NullPointerException on + // the toString instead of answering the filter. + assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty) + } + + it should "yield nothing when the column is empty and the substring is empty too" in { + val exec = new SubstringSearchOpExec(descJson(substring = "")) + // The empty substring matches every value, but a row with no value has none to + // match, so it is filtered out rather than kept. + assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty) + } + // --------------------------------------------------------------------------- // Edge: empty substring // --------------------------------------------------------------------------- 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 ddf59a7500e..2aa9341e7c9 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 @@ -19,8 +19,9 @@ package org.apache.texera.amber.operator.typecasting +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, AttributeTypeUtils, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants @@ -28,6 +29,12 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.nio.charset.StandardCharsets +import java.nio.file.Files +import java.util.concurrent.TimeUnit +import scala.io.Source +import scala.util.Try + class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { private val workflowId = WorkflowIdentity(1L) @@ -90,4 +97,174 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { tc.typeCastingUnits.head.attribute shouldBe "n" tc.typeCastingUnits.head.resultType shouldBe AttributeType.STRING } + + // The values a cast reads differently on the two sides. Python's own `bool` + // answers true for every non-empty string, so "false" and "0" are where the + // script used to disagree with the run it came from; text that is neither a + // boolean nor a number, and an empty cell, are the two ends of the range. + private val boolCases = Seq("true", "false", "0", "1", "not a boolean", null) + + /** What the engine answers, as the string the Python side prints back: the + * literal, or `error` for a value `parseField` refuses. + */ + private def engineAnswer(value: Any, to: AttributeType): String = + Try(AttributeTypeUtils.parseField(value, to)) + .map(v => if (v == null) "null" else v.toString) + .getOrElse("error") + + it should "cast to boolean the way AttributeTypeUtils does" in { + val python = resolvePython().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val op = new TypeCastingOpDesc + op.typeCastingUnits = List(castUnit("v", AttributeType.BOOLEAN)) + + // The generated block reads `in1df` and writes `out1df`, so it becomes the + // body of a function the driver calls once per value. One row at a time, + // because a value the cast refuses would otherwise end the comparison at + // the first one. + val body = op.generateStandaloneCode().linesIterator.map(" " + _).mkString("\n") + val values = boolCases + .map(v => if (v == null) "None" else "\"" + v + "\"") + .mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |def cast(in1df): + |$body + | return out1df + | + | + |for value in $values: + | try: + | answer = cast(pd.DataFrame({"v": [value]}))["v"].iloc[0] + | print("null" if pd.isna(answer) else str(answer).lower()) + | except Exception: + | print("error") + |""".stripMargin + + val script = Files.createTempFile("typecast-bool-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val process = + new ProcessBuilder(python, script.toString).redirectErrorStream(true).start() + val out = Source.fromInputStream(process.getInputStream).mkString + process.waitFor(120, TimeUnit.SECONDS) + withClue(s"python said:\n$out\nscript:\n${Source.fromFile(script.toFile).mkString}") { + process.exitValue() shouldBe 0 + } + + val fromScript = out.trim.linesIterator.toSeq + val fromEngine = boolCases.map { v => + if (v == null) "null" else engineAnswer(v, AttributeType.BOOLEAN).toLowerCase + } + withClue(s"cases=${boolCases.mkString(", ")}\nscript said $fromScript\n") { + fromScript shouldBe fromEngine + } + // The pairs that made the review: "false" is not true, and "0" is not true. + fromEngine shouldBe Seq("true", "false", "false", "true", "error", "null") + } + + // One column per source type, since what the text looks like follows the + // column and not the value: a whole double keeps its point, an integer never + // grows one, and a boolean is lower case. + private val stringColumns: Seq[(String, String, Seq[AnyRef])] = Seq( + ("dbl", "float64", Seq(Double.box(6.0), Double.box(7.25))), + ("int", "int64", Seq(Int.box(6), Int.box(7))), + ("flag", "bool", Seq(Boolean.box(true), Boolean.box(false))) + ) + + it should "cast to string the way AttributeTypeUtils does" in { + val python = resolvePython().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val op = new TypeCastingOpDesc + op.typeCastingUnits = stringColumns.map { + case (name, _, _) => castUnit(name, AttributeType.STRING) + }.toList + + val frame = stringColumns + .map { + case (name, dtype, values) => + val cells = values + .map { + case b: java.lang.Boolean => if (b) "True" else "False" + case other => other.toString + } + .mkString("[", ", ", "]") + s""" "$name": pd.Series($cells, dtype="$dtype"),""" + } + .mkString("\n") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({ + |$frame + |}) + |${op.generateStandaloneCode()} + |for column in ${stringColumns.map(c => "\"" + c._1 + "\"").mkString("[", ", ", "]")}: + | for cell in out1df[column]: + | print("null" if pd.isna(cell) else cell) + |""".stripMargin + + val script = Files.createTempFile("typecast-string-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val process = + new ProcessBuilder(python, script.toString).redirectErrorStream(true).start() + val out = Source.fromInputStream(process.getInputStream).mkString + process.waitFor(120, TimeUnit.SECONDS) + withClue(s"python said:\n$out\nscript:\n$driver") { + process.exitValue() shouldBe 0 + } + + val fromEngine = stringColumns.flatMap(_._3).map(v => engineAnswer(v, AttributeType.STRING)) + withClue(s"script said ${out.trim.linesIterator.toSeq}\n") { + out.trim.linesIterator.toSeq shouldBe fromEngine + } + // The pair that made the review: a double keeps its point at 6.0, where the + // integer 6 has none. + fromEngine shouldBe Seq("6.0", "7.25", "6", "7", "true", "false") + } + + // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path + // (UDF_PYTHON_PATH), then python3 / python / py. + private def resolvePython(): Option[String] = { + def fromConfig: Option[String] = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + + def runnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable) + } + + private def canImportPandas(python: String): Boolean = + Try( + new ProcessBuilder(python, "-c", "import pandas").redirectErrorStream(true).start() + ).toOption + .exists { p => + if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } } 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 1db11e8302d..dc290e99479 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 @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.unneststring 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 @@ -82,6 +83,30 @@ class UnnestStringOpDescSpec extends AnyFlatSpec with Matchers { } } + // A hole widens an integer column to float, so the split would see "6.0" where + // the engine splits "6". A real DOUBLE holding 6.0 looks the same and keeps its + // point, so only the declared type can decide. + "UnnestStringOpDesc.generateStandaloneCode" should + "narrow a column the schema declares INTEGER before rendering it" in { + val op = newDesc(",", "n", "piece") + val schemas = Map(PortIdentity(0) -> Schema().add(new Attribute("n", AttributeType.INTEGER))) + op.generateStandaloneCode(schemas) should include( + """_texera_cast_string(out1df["n"].astype("Int64"))""" + ) + } + + it should "leave a DOUBLE column its decimal point" in { + val op = newDesc(",", "n", "piece") + val schemas = Map(PortIdentity(0) -> Schema().add(new Attribute("n", AttributeType.DOUBLE))) + op.generateStandaloneCode(schemas) should include("""_texera_cast_string(out1df["n"])""") + } + + it should "read the column as it arrives when no schema is given" in { + newDesc(",", "n", "piece").generateStandaloneCode() should include( + """_texera_cast_string(out1df["n"])""" + ) + } + "UnnestStringOpDesc" should "round-trip its fields through the polymorphic base" in { val restored = objectMapper.readValue( diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExecSpec.scala index a746180c55d..146d9c780eb 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExecSpec.scala @@ -108,6 +108,24 @@ class UnnestStringOpExecSpec extends AnyFlatSpec with BeforeAndAfter { opExec.close() } + it should "produce no rows when the attribute is empty" in { + opDesc.attribute = "field1" + opDesc.delimiter = "-" + opExec = new UnnestStringOpExec(objectMapper.writeValueAsString(opDesc)) + // A blank CSV cell arrives as null. This used to throw a NullPointerException on + // the toString instead of unnesting to nothing. + val tuple: Tuple = Tuple + .builder(tupleSchema) + .add(new Attribute("field1", AttributeType.STRING), null) + .add(new Attribute("field2", AttributeType.INTEGER), 1) + .add(new Attribute("field3", AttributeType.STRING), "a") + .build() + + opExec.open() + assert(opExec.processTuple(tuple, 0).isEmpty) + opExec.close() + } + it should "split by regex delimiter" in { opDesc.attribute = "field1" opDesc.delimiter = "<\\d*>"