This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8340-5865f2af0b88129144b25eeddef53842a885fed5 in repository https://gitbox.apache.org/repos/asf/texera.git
commit ddb5305bdf4784c8ef043876191e71ba85b589cf Author: Kary Zheng <[email protected]> AuthorDate: Thu Sep 17 03:50:51 2026 +0000 feat(workflow-operator): export the operators that work on text (#8340) ### What changes were proposed in this PR? Six operators that read or rewrite a text column implement `StandaloneCodeGenerator`: Type Casting, Keyword Search, Substring Search, Unnest String, Regex and Dictionary Matcher. The sixteen that rearrange rows rather than read inside one are in #8506. Each is written against what its executor does rather than against what its name suggests, and the verification that came with the export runs both and compares them. Some of what that turned up is visible in the code: a cast reads the column's declared type rather than the value in front of it, because a hole widens an integer column to float and "6" would otherwise come back "6.0"; the 32-bit narrowing wraps for an integral source and saturates for a double, as `Long.toInt` and `Double.toInt` do; Dictionary Matcher's substring branch asks whether the cell sits inside a dictionary entry, which is the direction the executor tests and not the one the name suggests. `StandaloneHelpers` holds what these six share: a transcription of `AttributeTypeUtils`, emitted once near the top of a script rather than inlined per operator. A cast goes through `parseField(force = true)`, whose numeric branch is `java.text.NumberFormat`, so the engine reads "12abc" as 12 and "false" as false where Python's own conversions answer differently. Four of these operators change what they do, not only how they export. Regex, Substring Search and Unnest String answered a null cell by raising, which a generated script has no way to reproduce and no reason to: nothing in a column matches nothing, and unnests to no rows. They answer it that way now, and their specs say so. Keyword Search states which keywords its query parser will take, which a user had no way to know from the field alone. ### Any related issues, documentation, discussions? Part of #8325, 10 of 27; that issue lists the set in order. #8327 has since merged, so the trait is in place and this branch compiles on its own. The rows these operators add to the verification runner follow with the harness rather than as whole new files here. Closes #8074, closes #7548. Closes #8414, the task this change is the whole of. ### How was this PR tested? Each operator asserts the block it emits in its own spec. Type Casting's spec goes further: it runs the block it emits through a real Python and compares every answer against `AttributeTypeUtils`, so the two findings the review raised are held there, and that part does run on this diff's CI. Once the verification lands each operator is also run through the engine and through its generated script, on every configuration its schema offers, and the two answers compared; this branch is cut from main and does not carry that machinery, so those runs are not on this diff's CI. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> Co-authored-by: Xinyuan Lin <[email protected]> --- .../texera/amber/operator/StandaloneHelpers.scala | 172 ++++++++++++++++++++ .../dictionary/DictionaryMatcherOpDesc.scala | 137 +++++++++++++++- .../keywordSearch/KeywordSearchOpDesc.scala | 52 +++++- .../metadata/annotations/SampleColumn.java | 45 ++++++ .../texera/amber/operator/regex/RegexOpDesc.scala | 30 +++- .../texera/amber/operator/regex/RegexOpExec.scala | 9 +- .../substringSearch/SubstringSearchOpDesc.scala | 44 ++++- .../substringSearch/SubstringSearchOpExec.scala | 16 +- .../operator/typecasting/TypeCastingOpDesc.scala | 74 ++++++++- .../operator/unneststring/UnnestStringOpDesc.scala | 38 ++++- .../operator/unneststring/UnnestStringOpExec.scala | 18 ++- .../keywordSearch/KeywordSearchOpDescSpec.scala | 59 ++++++- .../amber/operator/regex/RegexOpExecSpec.scala | 14 ++ .../SubstringSearchOpExecSpec.scala | 18 +++ .../typecasting/TypeCastingOpDescSpec.scala | 179 ++++++++++++++++++++- .../unneststring/UnnestStringOpDescSpec.scala | 25 +++ .../unneststring/UnnestStringOpExecSpec.scala | 18 +++ 17 files changed, 917 insertions(+), 31 deletions(-) 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 0000000000..1414f6b043 --- /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 cfb2b5e76f..8660fcdc69 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 b0e202a1bb..b968535bed 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 0000000000..cb35c9aab9 --- /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. + * + * <p>It lets a field declare a <em>semantic</em> 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). + * + * <p>This has <em>no effect on production</em>: 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 0f886ae53b..f3051311f9 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 c8e840de5b..ebeb59fc57 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 da4a7437b9..9170cbe988 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 5331a4601d..3044bca1b1 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 82b0af60b3..c23c3dd66f 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 ef742de3ae..fa2e5c2151 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 1e59968217..0a0e08d9a6 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 b56e04c8fd..59b7bb456b 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 3d93539876..b887b5f9e8 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 83fd90fee0..bd7b947496 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 ddf59a7500..2aa9341e7c 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 1db11e8302..dc290e9947 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 a746180c55..146d9c780e 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*>"
