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..d6ee4e9a913 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,52 @@ 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. 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 + | + | 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 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..155c73dd65f 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 @@ -239,6 +239,194 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { fromEngine shouldBe Seq("6.0", "7.25", "6", "7", "true", "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. + 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", + "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" + ) + + /** `java.sql.Timestamp.toString` always writes a fraction where Python writes + * one only when there is something to write, and every case here lands on a + * whole second. + */ + private def withoutFraction(text: String): String = text.stripSuffix(".0") + + /** 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" 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: 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({"v": $values}) + |${op.generateStandaloneCode()} + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-timestamp-", ".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 } + + // Only the printed cells: pandas writes a parsing warning to stderr, which + // this process merges into the same stream. + val fromScript = cellsOf(out) + val fromEngine = + timestampCases.map(v => withoutFraction(engineAnswer(v, AttributeType.TIMESTAMP))) + withClue(s"cases=${timestampCases.mkString(", ")}\nscript said $fromScript\n") { + fromScript shouldBe fromEngine + } + // The rows that made the issue: a moment either side of the nanosecond edge. + fromEngine.takeRight(3) shouldBe + Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59") + // 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" 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 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" 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 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 // (UDF_PYTHON_PATH), then python3 / python / py. private def resolvePython(): Option[String] = {