feat(operator): read a timestamp cast over the years the engine reaches

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

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"""

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] = {