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,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,38 @@ 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. 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 +233,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
 }

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