diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala index 3853d1cdd67..09dc5551b6c 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala @@ -109,4 +109,40 @@ trait StandaloneCodeGenerator { * to start. */ def standaloneImports(): Seq[String] = Seq.empty + + /** + * A helper each input frame goes through before this operator's code runs, + * for an operator the engine hands its table in a different form than the + * script's upstream does. The translator defines it once and feeds the + * operator what it returns, leaving the upstream frame as it was. + * + * A Python operator runs in pyamber, which hands it a DOUBLE column as + * float64, where a NaN and a null are one value. A source can hand the script + * a Float64 column that keeps them apart, and an operator written against + * float64 then answers differently. + */ + def standaloneInputAdapter(): Option[StandaloneCodeGenerator.InputAdapter] = + this match { + case _: PythonOperatorDescriptor => Some(StandaloneCodeGenerator.PythonOperatorInput) + case _ => None + } +} + +object StandaloneCodeGenerator { + + /** A helper's name and the definition that binds it. */ + final case class InputAdapter(name: String, definition: String) + + /** `_texera_python_operator_input`: a copy with Float64 columns as float64, the way pyamber + * hands them over, so the upstream frame another operator reads stays as it was. + */ + val PythonOperatorInput: InputAdapter = InputAdapter( + "_texera_python_operator_input", + """def _texera_python_operator_input(df): + | df = df.copy() + | for _c in df.columns: + | if isinstance(df[_c].dtype, (pd.Float64Dtype, pd.Float32Dtype)): + | df[_c] = df[_c].astype("float64") + | return df""".stripMargin + ) } 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 index 1414f6b043c..83505f05d93 100644 --- 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 @@ -25,26 +25,26 @@ package org.apache.texera.amber.operator object StandaloneHelpers { /** - * A Python transcription of `AttributeTypeUtils`, for operators that cast a - * column to a declared type. + * `AttributeTypeUtils` in Python, so a cast gives the engine's answer. * - * 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. + * - `_texera_cast_boolean`: "true"/"false" in any case, else a number where only 1 is true. + * - `_texera_parse_number`: Java's NumberFormat, stopping at a non-digit and dropping commas. + * - `_texera_saturate`: a double narrowed as Java does, NaN to 0 and the rest to the bounds. + * - `_texera_long_value`: a double clamped to long's range. + * - `_texera_cast_integral`: the LONG cast. + * - `_texera_cast_int32`: the INTEGER cast, where a long wraps and a double clamps. + * - `_texera_text_to_timestamp`: each cell parsed alone, any year; unreadable text raises. + * - `_texera_epoch_millis_to_timestamp`: epoch milliseconds shown in the machine's zone. + * - `_texera_local_zone`: the machine's zone by its IANA name. + * - `_texera_standard_offset`: the zone's standard offset, which Java uses before 1900. + * - `_texera_local_wall_clock(s)`: the machine's wall clock at an instant, per value or column. + * - `_texera_cast_double(s)`: the DOUBLE cast, keeping NaN apart from null. + * - `_texera_java_text`: a value as Java's toString writes it. + * - `_texera_cast_string`: a column as text, decided by the column's type. + * - `_texera_cast_cells`: a cast applied cell by cell, reading nulls from the column's mask. */ 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. + """def _texera_cast_boolean(x): | if isinstance(x, str): | text = x.strip() | lowered = text.lower() @@ -57,14 +57,6 @@ object StandaloneHelpers { | | |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) @@ -76,23 +68,28 @@ object StandaloneHelpers { | 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_saturate(value, low, high): + | if value != value: + | return 0 + | if value <= low: + | return low + | if value >= high: + | return high + | return int(value) + | + | |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 _texera_saturate(value, -(2 ** 63), 2 ** 63 - 1) | 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): @@ -103,39 +100,119 @@ object StandaloneHelpers { | | |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) + | value = 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))) + | return ((int(value) + 2147483648) % 4294967296) - 2147483648 + | return _texera_saturate(value, -2147483648, 2147483647) + | + | + |def _texera_text_to_timestamp(s): + | from dateutil.parser import parse as _parse_date + | + | if pd.api.types.is_datetime64_any_dtype(s): + | if getattr(s.dtype, "tz", None) is not None: + | return _texera_local_wall_clocks(s) + | return s + | + | zone = _texera_local_zone() + | + | def _one(x): + | if pd.isna(x): + | return None + | try: + | parsed = _parse_date(str(x).strip()) + | except (ValueError, OverflowError) as e: + | raise ValueError(f"Failed to parse type str to Timestamp: {x}") from e + | if parsed.tzinfo is not None: + | parsed = _texera_local_wall_clock(parsed, zone) + | return parsed.replace(microsecond=parsed.microsecond // 1000 * 1000) + | + | return s.map(_one).astype("datetime64[us]") | | |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. + | return _texera_local_wall_clocks(pd.to_datetime(s, unit="ms", errors="coerce", utc=True)) + | + | + |def _texera_local_zone(): + | import os + | import zoneinfo + | + | names = [os.environ.get("TZ", "").lstrip(":")] + | try: + | import tzlocal + | + | names.insert(0, str(tzlocal.get_localzone())) + | except Exception: + | pass + | link = os.path.realpath("/etc/localtime") + | if "zoneinfo/" in link: + | names.append(link.split("zoneinfo/", 1)[1]) + | for name in names: + | try: + | return zoneinfo.ZoneInfo(name) + | except (ValueError, KeyError): + | continue + | return None + | + | + |def _texera_standard_offset(zone): + | import datetime + | import time + | + | if zone is None: + | return datetime.timedelta(seconds=-time.timezone) + | now = datetime.datetime.now(zone) + | return now.utcoffset() - now.dst() + | + | + |def _texera_local_wall_clock(moment, zone): + | import datetime + | | from dateutil.tz import tzlocal | - | return ( - | pd.to_datetime(s, unit="ms", errors="coerce", utc=True) - | .dt.tz_convert(tzlocal()) - | .dt.tz_localize(None) - | ) + | utc = moment.astimezone(datetime.timezone.utc).replace(tzinfo=None) + | if utc < datetime.datetime(1900, 1, 1): + | return utc + _texera_standard_offset(zone) + | if zone is not None: + | return moment.astimezone(zone).replace(tzinfo=None) + | try: + | return moment.astimezone(tzlocal()).replace(tzinfo=None) + | except (OSError, OverflowError): + | return utc + _texera_standard_offset(zone) + | + | + |def _texera_local_wall_clocks(s): + | import datetime + | + | from dateutil.tz import tzlocal + | + | zone = _texera_local_zone() + | utc = s.dt.tz_convert("UTC").dt.tz_localize(None) + | if zone is not None: + | wall = s.dt.tz_convert(zone).dt.tz_localize(None) + | before = utc < pd.Timestamp(1900, 1, 1) + | return wall.where(~before, utc + _texera_standard_offset(zone)) + | try: + | return s.dt.tz_convert(tzlocal()).dt.tz_localize(None) + | except (OSError, OverflowError): + | pass + | + | def _offset(t): + | if pd.isna(t): + | return 0 + | moment = t.to_pydatetime(warn=False) + | utc = moment.astimezone(datetime.timezone.utc).replace(tzinfo=None) + | return int((_texera_local_wall_clock(moment, None) - utc).total_seconds()) + | + | seconds = pd.Series([_offset(t) for t in s], index=s.index) + | return utc + pd.to_timedelta(seconds, unit="s").astype("timedelta64[s]") | | |def _texera_cast_double(x): @@ -144,29 +221,52 @@ object StandaloneHelpers { | return float(x) | | + |def _texera_cast_doubles(s): + | holes = s.isna().to_numpy() + | values = _texera_cast_cells(s, _texera_cast_double, 0.0).to_numpy(dtype="float64") + | return pd.Series(pd.arrays.FloatingArray(values, holes), index=s.index, name=s.name) + | + | + |def _texera_java_text(x): + | import datetime + | import decimal + | + | if pd.api.types.is_bool(x): + | return "true" if x else "false" + | if isinstance(x, datetime.datetime): + | nanos = x.microsecond * 1000 + getattr(x, "nanosecond", 0) + | return x.strftime("%Y-%m-%d %H:%M:%S") + "." + (f"{nanos:09d}".rstrip("0") or "0") + | if pd.api.types.is_integer(x): + | return str(int(x)) + | if pd.api.types.is_float(x): + | value = float(x) + | if value != value: + | return "NaN" + | if value in (float("inf"), float("-inf")): + | return "Infinity" if value > 0 else "-Infinity" + | if value == 0 or 1e-3 <= abs(value) < 1e7: + | return repr(value) + | sign, digits, exponent = decimal.Decimal(repr(value)).as_tuple() + | power = len(digits) + exponent - 1 + | kept = "".join(map(str, digits)).rstrip("0") or "0" + | return ("-" if sign else "") + kept[0] + "." + (kept[1:] or "0") + "E" + str(power) + | return str(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) + | return _texera_cast_cells(s, _texera_java_text, None) + | + | + |def _texera_cast_cells(s, cast, hole): + | holes = s.isna().to_numpy() + | return pd.Series( + | [hole if empty else cast(x) for x, empty in zip(s, holes)], + | index=s.index, + | name=s.name, + | dtype=object, | )""".stripMargin } 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 c23c3dd66f0..461d0dd0df8 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 @@ -117,23 +117,25 @@ class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { // 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")""" + // The column's mask decides the holes, so a stored NaN is cast like + // any other double; see `_texera_cast_cells`. + s"""_texera_cast_cells(out1df[$colLit], lambda x: _texera_cast_int32(x, $wrapArg), pd.NA).astype("Int64")""" case AttributeType.LONG => - s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else _texera_cast_integral(x)).astype("Int64")""" + s"""_texera_cast_cells(out1df[$colLit], _texera_cast_integral, pd.NA).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")""" + s"""_texera_cast_doubles(out1df[$colLit])""" 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")""" + s"""_texera_cast_cells(out1df[$colLit], _texera_cast_boolean, pd.NA).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. + // A number is an instant in milliseconds and needs its own reading. + // Text is read a cell at a time, and over the years the engine + // reaches rather than the ones pandas parses into by default; see + // the helpers. 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")""" + else s"""_texera_text_to_timestamp(out1df[$colLit])""" case _ => s"""out1df[$colLit]""" } lines += s"""out1df[$colLit] = $expr""" diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 2aa9341e7c9..cfc60326cba 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 @@ -25,7 +25,9 @@ import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, AttributeTy 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.tags.IntegrationTest import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.Tag import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -37,6 +39,8 @@ import scala.util.Try class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { + private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName) + private val workflowId = WorkflowIdentity(1L) private val executionId = ExecutionIdentity(1L) @@ -98,11 +102,10 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { 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) + // Text that is neither a boolean nor a number, which the engine refuses. The + // verification only compares runs that both produced something, so it cannot + // ask this; the values both sides read are its `str_to_bool` column. + private val boolCases = Seq("not a boolean") /** What the engine answers, as the string the Python side prints back: the * literal, or `error` for a value `parseField` refuses. @@ -112,7 +115,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { .map(v => if (v == null) "null" else v.toString) .getOrElse("error") - it should "cast to boolean the way AttributeTypeUtils does" in { + it should "refuse a boolean the way AttributeTypeUtils does" taggedAs NeedsPythonPackages in { val python = resolvePython().getOrElse( cancel("No runnable python executable (udf.conf python.path, python3, python, py)") ) @@ -167,58 +170,354 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { 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") + fromEngine shouldBe Seq("error") } - // 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))) - ) + // A double column read from Arrow into a nullable float keeps a stored NaN and + // an empty cell apart. The engine casts a NaN like any other double: "NaN" as + // text, 0 as an integer, true as a boolean. The infinities and 3e9 sit past + // the integer bounds, which a Double's toInt and toLong saturate at. + private val doubleCases: Seq[java.lang.Double] = + Seq( + Double.NaN, + null, + 1.0, + -2.7, + 3.0e9, + Double.PositiveInfinity, + Double.NegativeInfinity, + -0.0 + ) - it should "cast to string the way AttributeTypeUtils does" in { + private val doubleTargets = + Seq(AttributeType.STRING, AttributeType.INTEGER, AttributeType.LONG, AttributeType.BOOLEAN) + + it should "cast a stored NaN the way AttributeTypeUtils does" taggedAs NeedsPythonPackages 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") + // One column per target, each holding the same doubles. + val columns = doubleTargets.map(_.name().toLowerCase) val op = new TypeCastingOpDesc - op.typeCastingUnits = stringColumns.map { - case (name, _, _) => castUnit(name, AttributeType.STRING) - }.toList + op.typeCastingUnits = doubleTargets.zip(columns).map { case (to, c) => castUnit(c, to) }.toList + val input = columns.foldLeft(Schema())((s, c) => s.add(new Attribute(c, AttributeType.DOUBLE))) + val generated = op.generateStandaloneCode(Map(op.operatorInfo.inputPorts.head.id -> input)) - val frame = stringColumns + val values = doubleCases .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"),""" + case null => "None" + case d if d.isNaN => "float(\"nan\")" + case d if d.isInfinite() => if (d > 0) "float(\"inf\")" else "float(\"-inf\")" + case d => d.toString } - .mkString("\n") + .mkString("[", ", ", "]") + // Read the way the Arrow source reads a file, so the column arrives with its + // validity mask and not with NaN standing in for every empty cell. val driver = s"""import pandas as pd + |import pyarrow as pa + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |_columns = ${columns.map("\"" + _ + "\"").mkString("[", ", ", "]")} + |_table = pa.table({c: pa.array($values, pa.float64()) for c in _columns}) + |in1df = _table.to_pandas(types_mapper={pa.float64(): pd.Float64Dtype()}.get) + |$generated + | + |for c in _columns: + | for answer in out1df[c]: + | if answer is None or answer is pd.NA: + | answer = "null" + | elif pd.api.types.is_bool(answer): + | answer = str(answer).lower() + | print("cell", c, answer) + |""".stripMargin + + val script = Files.createTempFile("typecast-double-nan-", ".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 fromScript = cellsOf(out) + val fromEngine = doubleTargets.zip(columns).flatMap { + case (to, c) => doubleCases.map(v => s"$c ${engineAnswer(v, to)}") + } + withClue(s"script said $fromScript\n") { fromScript shouldBe fromEngine } + // The stored NaN in each column, which the script used to empty. + doubleTargets.map(to => engineAnswer(Double.NaN, to)) shouldBe Seq("NaN", "0", "0", "true") + } + + // Text read as a double and then written back as text, by two operators. A + // NaN read here has to stay apart from an empty cell for the second cast to + // write it as "NaN". + private val doubleTextCases = Seq("NaN", null, "1.25", "Infinity") + + it should "keep a NaN read from text apart from an empty cell" taggedAs NeedsPythonPackages 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 toDouble = new TypeCastingOpDesc + toDouble.typeCastingUnits = List(castUnit("v", AttributeType.DOUBLE)) + val toText = new TypeCastingOpDesc + toText.typeCastingUnits = List(castUnit("v", AttributeType.STRING)) + val port = toText.operatorInfo.inputPorts.head.id + val first = toDouble.generateStandaloneCode( + Map(port -> Schema().add(new Attribute("v", AttributeType.STRING))) + ) + val second = toText.generateStandaloneCode( + Map(port -> Schema().add(new Attribute("v", AttributeType.DOUBLE))) + ) + + val values = doubleTextCases + .map(v => if (v == null) "None" else "\"" + v + "\"") + .mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${toDouble.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": $values}) + |$first + |in1df = out1df + |$second + | + |for answer in out1df["v"]: + | print("cell", "null" if answer is None else answer) + |""".stripMargin + + val script = Files.createTempFile("typecast-double-text-", ".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 = doubleTextCases.map { v => + engineAnswer(AttributeTypeUtils.parseField(v, AttributeType.DOUBLE), AttributeType.STRING) + } + withClue(s"script said ${cellsOf(out)}\n") { cellsOf(out) shouldBe fromEngine } + fromEngine shouldBe Seq("NaN", "null", "1.25", "Infinity") + } + + // Moments either side of 1970, read three ways: as epoch milliseconds, as + // text that states an offset, and from a column that states its zone. Each + // asks the machine's zone for the wall clock. Windows' localtime refuses an + // instant before 1970, which no other platform does, so the driver refuses + // it the same way here. The engine writes such an instant at the zone's + // standard offset, as java.util.Date does before 1900. + private val epochYears = Seq(1600, 2024, 2500) + + it should "read a moment before 1970 where localtime refuses it" taggedAs NeedsPythonPackages in { + readMomentsAsTheEngineDoes( + epochYears.map(y => + java.time.ZonedDateTime.of(y, 6, 1, 0, 0, 0, 0, java.time.ZoneOffset.UTC) + ), + zone = None + ) + } + + // A zone whose offset has moved, read with the offset in force at each instant. + // Mexico City was seven hours behind in January 1931, kept summer time until + // 2022, and before 1922 kept its local mean time; before 1900 java.util.Date + // takes today's standard offset instead. Only today's two offsets would put + // the first three an hour or more away. + private val mexicoCityInstants = + Seq( + "1931-01-01T00:00:00Z", + "2021-07-01T00:00:00Z", + "1921-01-01T00:00:00Z", + "1899-12-31T00:00:00Z" + ) + .map(java.time.ZonedDateTime.parse(_)) + + it should "read a moment with the offset its zone had then" taggedAs NeedsPythonPackages in { + val fromEngine = + readMomentsAsTheEngineDoes(mexicoCityInstants, zone = Some("America/Mexico_City")) + // What the pair has to agree on: each offset the zone had, not today's. + fromEngine.map(moment) shouldBe Seq( + "1930-12-31 17:00:00", + "2021-06-30 19:00:00", + "1920-12-31 17:23:24", + "1899-12-30 18:00:00" + ).map(moment) + } + + /** Casts each instant three ways, as epoch milliseconds, as text that states an + * offset and from a column that states its zone, with localtime refusing + * every instant before 1970 as Windows' does, and compares the wall clocks + * with the engine's. A given zone is set for both sides; otherwise each uses + * the machine's. + */ + private def readMomentsAsTheEngineDoes( + instants: Seq[java.time.ZonedDateTime], + zone: Option[String] + ): Seq[String] = { + 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("ms", AttributeType.TIMESTAMP), + castUnit("text", AttributeType.TIMESTAMP), + castUnit("zoned", AttributeType.TIMESTAMP) + ) + val input = Schema() + .add(new Attribute("ms", AttributeType.LONG)) + .add(new Attribute("text", AttributeType.STRING)) + .add(new Attribute("zoned", AttributeType.TIMESTAMP)) + val generated = op.generateStandaloneCode(Map(op.operatorInfo.inputPorts.head.id -> input)) + + val millis = instants.map(_.toInstant.toEpochMilli) + val texts = instants.map(_.toLocalDateTime.plusMinutes(330).toString + ":00+05:30") + val driver = + s"""import time + | + |import numpy as np + |import pandas as pd | |${op.standaloneHelpers().mkString("\n\n")} | | + |_localtime = time.localtime + | + | + |def _windows_localtime(seconds=None): + | if seconds is not None and seconds < 0: + | raise OSError(22, "Invalid argument") + | return _localtime() if seconds is None else _localtime(seconds) + | + | + |time.localtime = _windows_localtime + |_utc = ${instants.map(i => "\"" + i.toLocalDateTime + "\"").mkString("[", ", ", "]")} |in1df = pd.DataFrame({ - |$frame + | "ms": pd.array(${millis.mkString("[", ", ", "]")} + [None], dtype="Int64"), + | "text": ${texts.map("\"" + _ + "\"").mkString("[", ", ", "]")} + [None], + | "zoned": pd.Series(np.array(_utc + ["NaT"], dtype="datetime64[us]")).dt.tz_localize("UTC"), |}) + |$generated + | + |for c in ["ms", "text", "zoned"]: + | for answer in out1df[c]: + | print("cell", c, "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-before-1970-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val builder = new ProcessBuilder(python, script.toString).redirectErrorStream(true) + zone.foreach(z => builder.environment().put("TZ", z)) + val process = builder.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 } + + // Every column names the same instants, so each has the engine's answer + // for the milliseconds. The text is also asked directly. + val machineZone = java.util.TimeZone.getDefault + zone.foreach(z => java.util.TimeZone.setDefault(java.util.TimeZone.getTimeZone(z))) + val (fromMillis, fromText) = + try { + ( + millis.map(m => + engineAnswer(java.lang.Long.valueOf(m), AttributeType.TIMESTAMP) + ) :+ "null", + texts.map(t => engineAnswer(t, AttributeType.TIMESTAMP)) + ) + } finally java.util.TimeZone.setDefault(machineZone) + fromText.map(moment) shouldBe fromMillis.init.map(moment) + val fromScript = cellsOf(out).map(_.split(" ", 2)).map(p => (p(0), moment(p(1)))) + withClue(s"script said ${cellsOf(out)}\n") { + fromScript shouldBe Seq("ms", "text", "zoned").flatMap(c => + fromMillis.map(v => (c, moment(v))) + ) + } + fromMillis.init + } + + // The moments where the two sides used to part. pandas parses into nanoseconds + // by default, which reach only 1677 to 2262, so everything past that edge was + // emptied where the engine holds a java.sql.Timestamp and reads it like any + // other moment. The first three rows are ordinary ones, and they are also three + // different formats in one column: the engine hands DateParserUtils a field at + // a time, so a row states its own format rather than the column's first one. + // Two of them state an offset, which DateParserUtils reads and java.sql.Timestamp + // then keeps no zone for: the moment is held as the wall clock of the machine's + // own zone. The expectation is taken from the engine rather than written down, + // so the pair says the same thing wherever the suite runs. Two carry digits + // past the millisecond, which DateParserUtils reads into a java.util.Date and + // so drops. + private val timestampCases = Seq( + "2024-03-05 14:09:07", + "2024-03-05T14:09:07", + "March 5, 2024", + "2024-03-05T14:09:07Z", + "2024-03-05T14:09:07+05:30", + "2024-03-05 14:09:07.123456", + "2024-03-05 14:09:07.9999", + "1677-09-22 00:12:44", + "2262-04-11 23:47:16", + "2500-01-01 00:00:00", + "1500-06-15 08:30:00", + "9999-12-31 23:59:59" + ) + + /** A printed moment as the wall clock it names. `java.sql.Timestamp.toString` + * trims its fraction and writes ".0" for none, where Python pads it to six + * digits or leaves it out. + */ + private def moment(text: String): Any = + Try(java.time.LocalDateTime.parse(text.replace(' ', 'T'))).getOrElse(text) + + /** The cells the driver printed, told apart from anything pandas wrote to + * stderr, which this process merges into the same stream. + */ + private def cellsOf(out: String): Seq[String] = + out.linesIterator.filter(_.startsWith("cell ")).map(_.drop("cell ".length)).toSeq + + it should "cast text to a timestamp the way AttributeTypeUtils does" taggedAs NeedsPythonPackages 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.TIMESTAMP)) + + // One frame rather than a row at a time, unlike the casts that refuse a + // value: every case here reads, and the column is where a format that is not + // the first row's would be lost. + val values = timestampCases.map(v => "\"" + v + "\"").mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": $values}) |${op.generateStandaloneCode()} - |for column in ${stringColumns.map(c => "\"" + c._1 + "\"").mkString("[", ", ", "]")}: - | for cell in out1df[column]: - | print("null" if pd.isna(cell) else cell) + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) |""".stripMargin - val script = Files.createTempFile("typecast-string-", ".py") + val script = Files.createTempFile("typecast-timestamp-", ".py") script.toFile.deleteOnExit() Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) @@ -226,17 +525,120 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { 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 + withClue(s"python said:\n$out\nscript:\n$driver") { process.exitValue() shouldBe 0 } + + // Only the printed cells, so nothing else written to the merged stderr is + // read as one. + val fromScript = cellsOf(out) + val fromEngine = timestampCases.map(v => engineAnswer(v, AttributeType.TIMESTAMP)) + withClue(s"cases=${timestampCases.mkString(", ")}\nscript said $fromScript\n") { + fromScript.map(moment) shouldBe fromEngine.map(moment) } + // The rows that made the issue: a moment either side of the nanosecond edge. + fromEngine.takeRight(3).map(moment) shouldBe + Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59").map(moment) + // And the pair that states an offset, which reaches the same moment by two + // spellings: five and a half hours apart in the text, and so in the reading. + val zoned = fromScript.slice(3, 5) + java.time.Duration + .between( + java.time.LocalDateTime.parse(zoned(1).replace(' ', 'T')), + java.time.LocalDateTime.parse(zoned(0).replace(' ', 'T')) + ) + .toMinutes shouldBe 330 + } + + // A column that is already a moment is not text and is not re-read: parseField + // hands a java.sql.Timestamp back untouched, and that class counts nanoseconds, + // so narrowing the column here would fold two moments the run tells apart into + // one. The pair below differs only past the microsecond. + private val nanosecondCases = Seq( + "2024-03-05 14:09:07.123456789", + "2024-03-05 14:09:07.123456001" + ) + + it should "keep the resolution a timestamp column arrived in" taggedAs NeedsPythonPackages 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.TIMESTAMP)) + // The declared type is what sends this down the branch under test: a column + // the engine already holds as a moment, cast to the type it already has. + val input = Schema().add(new Attribute("v", AttributeType.TIMESTAMP)) + val generated = op.generateStandaloneCode(Map(op.operatorInfo.inputPorts.head.id -> input)) + + val values = nanosecondCases.map(v => "\"" + v + "\"").mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": pd.to_datetime($values)}) + |$generated + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin - 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 + val script = Files.createTempFile("typecast-timestamp-nanos-", ".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 fromScript = cellsOf(out) + val fromEngine = + nanosecondCases.map(v => engineAnswer(java.sql.Timestamp.valueOf(v), AttributeType.TIMESTAMP)) + withClue(s"script said $fromScript\n") { fromScript shouldBe fromEngine } + // What the two answers have to carry: the rows stay apart. + fromEngine shouldBe nanosecondCases + } + + // Text no format reads ends the run on both sides. The engine's parseField + // throws, and the script raises the engine's message rather than leaving the + // cell empty. + it should "refuse a cell it cannot read, as the engine does" taggedAs NeedsPythonPackages 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.TIMESTAMP)) + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": ["not a date", None, "2024-03-05 14:09:07"]}) + |${op.generateStandaloneCode()} + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-timestamp-unreadable-", ".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() should not be 0 + out should include("Failed to parse type str to Timestamp: not a date") } - // 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") + engineAnswer("not a date", AttributeType.TIMESTAMP) shouldBe "error" } // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala index 8e873403787..6defb00f2da 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala +++ b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala @@ -83,15 +83,17 @@ class WorkflowToPythonTranslator extends LazyLogging { // Helper definitions the operator bodies below refer to. Collected across the // whole plan and deduplicated by text, so a workflow holding two operators // that share one helper still emits it once. Order follows the topological - // order, which keeps the script stable for a given plan. + // order, which keeps the script stable for a given plan. They go in once the + // bodies are written, since which of them the script calls is only known then. val helpers = topoOrder .map(logicalPlan.getOperator) - .collect { case gen: StandaloneCodeGenerator => gen.standaloneHelpers() } + .collect { + case gen: StandaloneCodeGenerator => + gen.standaloneHelpers() ++ gen.standaloneInputAdapter().map(_.definition) + } .flatten .distinct - if (helpers.nonEmpty) { - helpers.foreach { helper => script += helper; script += "" } - } + val helpersAt = script.length for (opIdentity <- topoOrder) { val opId = opIdentity.id @@ -140,11 +142,24 @@ class WorkflowToPythonTranslator extends LazyLogging { } .toMap + // An operator the engine hands its table in another form reads it in + // that form here too, under a name of its own, so the upstream frame + // another operator reads is not the one converted. + val fedVars = gen.standaloneInputAdapter() match { + case Some(adapter) => + inVars.map { v => + val fed = s"${v}_${adapter.name.stripPrefix("_texera_")}" + script += s"$fed = ${adapter.name}($v)" + fed + } + case None => inVars + } + // generateStandaloneCode() returns a code block using in{N}df / out{N}df // placeholders; substituteVars() replaces them with the assigned vars. script += substituteVars( gen.generateStandaloneCode(inputSchemas), - inVars, + fedVars, outVars, fileBase(displayName, fileBaseCounts), displayName @@ -199,9 +214,61 @@ class WorkflowToPythonTranslator extends LazyLogging { } } + script.insertAll( + helpersAt, + calledHelpers(helpers, script.drop(helpersAt).mkString("\n")).flatMap(Seq(_, "")) + ) script.mkString("\n") } + // The helpers with only the definitions the script calls, directly or through + // another helper. A helper block serves several operators, so it can hold far + // more functions than one plan uses, and those would only be text for the + // reader to wade through. Only a top-level `def _texera_...` is ever dropped: + // anything else a block holds is kept, and what it calls counts as called. A + // name is read from code only, so a comment or a string naming a helper keeps + // nothing alive. A block left with nothing is dropped whole. + private def calledHelpers(helpers: Seq[String], body: String): Seq[String] = { + val blocks = helpers.map(topLevelUnits) + val defined = blocks.flatten.collect { case (Some(name), text) => name -> text }.toMap + val called = mutable.Set[String]() + def visit(text: String): Unit = { + val code = splitOffLiterals(text).collect { case (true, segment) => segment } + """\b_texera_\w+""".r.findAllIn(code.mkString("\n")).foreach { name => + if (defined.contains(name) && called.add(name)) visit(defined(name)) + } + } + visit(body) + blocks.flatten.collect { case (None, text) => text }.foreach(visit) + blocks + .map(_.collect { case (name, text) if name.forall(called) => text }.mkString) + .map(_.replaceAll("\\s+$", "")) + .filter(_.nonEmpty) + } + + // A helper block cut at its top-level statements, each with the comment and + // blank lines just above it and the indented lines below it, and named when it + // is a `def _texera_...`. + private def topLevelUnits(block: String): List[(Option[String], String)] = { + val units = ArrayBuffer[(Option[String], StringBuilder)]() + val lead = new StringBuilder + block.split("\n", -1).foreach { line => + val topLevel = line.nonEmpty && !line.head.isWhitespace + if (topLevel && !line.startsWith("#") && !line.startsWith("@")) { + val name = """def (_texera_\w+)\s*\(""".r.findPrefixMatchOf(line).map(_.group(1)) + units += ((name, new StringBuilder(lead.toString))) + lead.setLength(0) + units.last._2 ++= line + "\n" + } else if (topLevel || units.isEmpty || lead.nonEmpty) { + lead ++= line + "\n" + } else { + units.last._2 ++= line + "\n" + } + } + val cut = units.map { case (name, text) => (name, text.toString) }.toList + if (lead.isEmpty) cut else cut :+ ((None, lead.toString)) + } + // The stem of the files one operator writes, taken from its name so a reader // can tell whose picture is whose. Numbered from one even when the plan holds // a single chart, so the name a script writes does not depend on what else diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala index ba0c3f3ae70..3ff7cc9fd73 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala @@ -22,7 +22,12 @@ package org.apache.texera.amber.translator import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} -import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.apache.texera.amber.core.tuple.Schema +import org.apache.texera.amber.operator.{ + LogicalOp, + PythonOperatorDescriptor, + StandaloneCodeGenerator +} import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -63,6 +68,21 @@ class WorkflowToPythonTranslatorSpec extends AnyFlatSpec with Matchers { op } + /** A stub the engine would run in pyamber, which is all the input adapter asks. */ + private class PythonStubOp(block: String) extends StubOp(block) with PythonOperatorDescriptor { + override def generatePythonCode(): String = "" + override def getOutputSchemas( + inputSchemas: Map[PortIdentity, Schema] + ): Map[PortIdentity, Schema] = + Map.empty + } + + private def pythonStub(id: String, block: String): LogicalOp = { + val op = new PythonStubOp(block) + op.setOperatorId(id) + op + } + private def upstream(id: String): LogicalOp = stub(id, "out1df = in1df.copy()") private def link(from: LogicalOp, to: LogicalOp): LogicalLink = @@ -134,6 +154,40 @@ class WorkflowToPythonTranslatorSpec extends AnyFlatSpec with Matchers { script should include("df2 = df1.copy()") } + /** pyamber hands a Python operator float64, where a NaN and a null are one value, + * and a source can hand the script Float64, which keeps them apart. The + * operator reads a converted copy, so the frame another operator reads from the + * same upstream is the one the source wrote. + */ + it should "feed a Python operator a converted copy of each input" in { + val source = upstream("source") + val python = pythonStub("python", "out1df = in1df.copy()") + val script = new WorkflowToPythonTranslator().translate( + LogicalPlan(List(source, python), List(link(source, python))) + ) + script should include("df1_python_operator_input = _texera_python_operator_input(df1)") + script should include("df2 = df1_python_operator_input.copy()") + } + + it should "leave an operator the engine runs on the JVM reading its upstream as it is" in { + val first = upstream("first") + val second = upstream("second") + val script = new WorkflowToPythonTranslator().translate( + LogicalPlan(List(first, second), List(link(first, second))) + ) + script should not include "_texera_python_operator_input" + } + + it should "define the Python operator input helper once and leave the upstream frame alone" in { + val source = upstream("source") + val pythons = List("a", "b").map(id => pythonStub(id, "out1df = in1df.copy()")) + val script = new WorkflowToPythonTranslator().translate( + LogicalPlan(source :: pythons, pythons.map(link(source, _))) + ) + script.linesIterator.count(_ == "def _texera_python_operator_input(df):") shouldBe 1 + script should not include "df1 = _texera_python_operator_input(" + } + /** Nothing stops a column from being named after a placeholder. The * substitution rewrites the variable a block reads with, never the column * name it asks that variable for. @@ -163,6 +217,64 @@ class WorkflowToPythonTranslatorSpec extends AnyFlatSpec with Matchers { script should include("""fig.write_html("stub_2.html")""") } + /** A stub that declares `helper` and runs `block`. */ + private def withHelper(id: String, helper: String, block: String): LogicalOp = { + val op = new StubOp(block) { + override def standaloneHelpers(): Seq[String] = Seq(helper) + } + op.setOperatorId(id) + op + } + + private val sharedHelper = + """import math + | + | + |def _texera_called(x): + | return _texera_reached(x) + | + | + |# Only ever named, never called. + |def _texera_named(x): + | return x + | + | + |def _texera_reached(x): + | return math.floor(x) + | + | + |def _texera_unused(x): + | return x""".stripMargin + + /** A helper block serves several operators, so one plan rarely calls all of + * it. What it does not call has no place in the script. + */ + it should "emit only the helper definitions the script calls" in { + val op = withHelper("op", sharedHelper, "out1df = in1df.apply(_texera_called)") + val script = new WorkflowToPythonTranslator().translate(LogicalPlan(List(op), List.empty)) + script should include("def _texera_called(x):") + script should include("def _texera_reached(x):") + script should not include "def _texera_unused(x):" + script should include("import math") + } + + it should "not keep a helper alive for a comment or a string naming it" in { + val op = withHelper( + "op", + sharedHelper, + "# _texera_named\nout1df = in1df.assign(note=\"_texera_named\")" + ) + val script = new WorkflowToPythonTranslator().translate(LogicalPlan(List(op), List.empty)) + script should not include "def _texera_named(x):" + script should not include "# Only ever named, never called." + } + + it should "emit nothing of a helper block whose definitions the script never calls" in { + val op = withHelper("op", "def _texera_unused(x):\n return x", "out1df = in1df.copy()") + val script = new WorkflowToPythonTranslator().translate(LogicalPlan(List(op), List.empty)) + script should not include "_texera_unused" + } + /** The translator's own contract when it meets an operator it cannot render: * a comment rather than a silently wrong line. */