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..b8f7036bd1a 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 @@ -122,6 +122,54 @@ object StandaloneHelpers { | return max(-2147483648, min(2147483647, int(value))) | | + |def _texera_text_to_timestamp(s): + | # Read cell by cell, the way the engine reads a column: DateParserUtils + | # is handed one field at a time, so a row states its own format and the + | # rest of the column has no say in it. + | # + | # What is read here is held at microsecond resolution and not the + | # nanoseconds pandas parses into by default, which reach 1677 to 2262: + | # the engine holds a java.sql.Timestamp, where the year 2500 is an + | # ordinary moment and emptying it would answer for a row the run itself + | # had no trouble with. Text is cut to the millisecond, because + | # DateParserUtils reads it into a java.util.Date, which counts no finer. + | # A column that is already a moment is handed back + | # at the resolution it arrived in instead, because parseField returns a + | # java.sql.Timestamp untouched and that class counts nanoseconds. + | # + | # Still coerced, which the strict cast is not: the engine accepts a set + | # of formats no single pandas call states, so text neither can read is + | # answered with an empty cell rather than by ending the run. + | # + | # A reading that states an offset is moved to the zone the machine is + | # set to and then holds that wall clock, which is what the engine does + | # with one: DateParserUtils reads the offset and java.sql.Timestamp + | # keeps no zone of its own. Left alone, the offset travels as far as + | # the conversion below and stops the cast on a value the run reads. + | # tzlocal() and not a fixed offset, so each instant gets the one in + | # force when it happened. + | from dateutil.parser import parse as _parse_date + | from dateutil.tz import tzlocal + | + | if pd.api.types.is_datetime64_any_dtype(s): + | if getattr(s.dtype, "tz", None) is not None: + | return s.dt.tz_convert(tzlocal()).dt.tz_localize(None) + | return s + | + | def _one(x): + | if pd.isna(x): + | return None + | try: + | parsed = _parse_date(str(x).strip()) + | except (ValueError, OverflowError): + | return None + | if parsed.tzinfo is not None: + | parsed = parsed.astimezone(tzlocal()).replace(tzinfo=None) + | 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 @@ -144,6 +192,40 @@ object StandaloneHelpers { | return float(x) | | + |def _texera_java_text(x): + | # The text Java's toString gives the value Texera holds. A double past + | # 1e7 or under 1e-3 is written in E notation, where Python's str() + | # waits until 1e16 and 1e-4. Its digits are the shortest that read back + | # as the same value, as from JDK 19; an earlier JDK sometimes writes + | # more, 9.999999999999999E22 for 1e23. A java.sql.Timestamp writes every + | # digit of its nanoseconds, trailing zeros dropped but never all of them, + | # where Python writes no fraction on a whole second and six digits + | # otherwise. A boolean is written in lower case. + | 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 @@ -153,20 +235,5 @@ object StandaloneHelpers { | 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 + | return s.map(lambda x: None if pd.isna(x) else _texera_java_text(x))""".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..7a1422c7154 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 @@ -129,11 +129,13 @@ class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { // 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. + // 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..60e27ac2856 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,76 @@ 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))) + // 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" ) - it should "cast to string the way AttributeTypeUtils does" in { + /** 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 = 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") + op.typeCastingUnits = List(castUnit("v", AttributeType.TIMESTAMP)) + + // One frame rather than a row at a time, unlike the casts that refuse a + // value: this one coerces, 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({ - |$frame - |}) + |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 +247,122 @@ 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 script = Files.createTempFile("typecast-timestamp-nanos-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) - 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 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 + } + + // The one place the script is meant to differ, and the reason it cannot simply + // parse strictly: the engine accepts a set of formats no single pandas call + // states, so text neither can read leaves an empty cell instead of ending an + // exported run halfway. + it should "leave a cell it cannot read empty rather than refusing it" 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() shouldBe 0 + cellsOf(out) shouldBe Seq("null", "null", "2024-03-05 14:09:07") } - // 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") + // The engine refuses the same text, which is the difference this coercion is. + engineAnswer("not a date", AttributeType.TIMESTAMP) shouldBe "error" } // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path