carloea2 commented on code in PR #8709:
URL: https://github.com/apache/texera/pull/8709#discussion_r4362899533
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDescSpec.scala:
##########
@@ -108,6 +119,465 @@ class AggregateOpDescSpec extends AnyFlatSpec with
Matchers {
}
}
+ // The integers behind a timestamp column mean microseconds or nanoseconds
+ // depending on the resolution it was read at, and a nanosecond total leaves
+ // the range of a 64-bit integer after a handful of modern dates.
+ it should "read a timestamp as epoch milliseconds at either resolution"
taggedAs NeedsPythonPackages in {
+ val python = resolvePython().getOrElse(cancel("No runnable python
executable"))
+ if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+ val input = Schema().add("t", AttributeType.TIMESTAMP)
+ val desc = descWith(
+ List.empty,
+ aggOp(AggregationFunction.SUM, "t", "ts_total"),
+ aggOp(AggregationFunction.AVERAGE, "t", "ts_avg")
+ )
+ val block = desc.generateStandaloneCode(Map(PortIdentity() -> input))
+
+ def report(unit: String): String =
+ s"""in1df = pd.DataFrame({
+ | "t": pd.to_datetime(["2024-01-01"] *
6).astype("datetime64[$unit]"),
+ |})
+ |$block
+ |print(pd.Timestamp(out1df.iloc[0]["ts_total"]).isoformat())
+ |print(repr(float(out1df.iloc[0]["ts_avg"])))""".stripMargin
+
+ val driver =
+ s"""import pandas as pd
+ |${desc.standaloneHelpers().mkString("\n")}
+ |${report("ns")}
+ |${report("us")}
+ |""".stripMargin
+
+ val out = runPython(python, driver, "aggregate-timestamp-units-", "UTC")
+
+ withClue(s"python said:\n$out\nscript:\n$driver") {
+ // Six times 1704067200000 milliseconds, read back as a timestamp: a date
+ // no nanosecond count can hold, and the same one at either resolution.
+ out.trim.linesIterator.toSeq shouldBe Seq(
+ "2293-12-31T00:00:00",
+ "1704067200000.0",
+ "2293-12-31T00:00:00",
+ "1704067200000.0"
+ )
+ }
+ }
+
+ /** `Timestamp.getTime` answers for the instant a wall clock names in the
+ * JVM's default zone, and `new Timestamp(long)` renders one back there, so
a
+ * script that read the column as UTC was a whole offset out on every value
+ * it added. Checked against Java 17 in the same zone: the sum below is
+ * 2078-01-01 06:00:00 and the average 1704132000000.
+ */
+ it should "add timestamps in the zone the script runs in" taggedAs
NeedsPythonPackages in {
+ val python = resolvePython().getOrElse(cancel("No runnable python
executable"))
+ if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+ val input = Schema().add("t", AttributeType.TIMESTAMP)
+ val desc = descWith(
+ List.empty,
+ aggOp(AggregationFunction.SUM, "t", "ts_total"),
+ aggOp(AggregationFunction.AVERAGE, "t", "ts_avg")
+ )
+ val block = desc.generateStandaloneCode(Map(PortIdentity() -> input))
+
+ val driver =
+ s"""import pandas as pd
+ |${desc.standaloneHelpers().mkString("\n")}
+ |in1df = pd.DataFrame({
+ | "t": pd.to_datetime(["2024-01-01 00:00:00", "2024-01-02
00:00:00"]),
+ |})
+ |$block
+ |print(pd.Timestamp(out1df.iloc[0]["ts_total"]).isoformat())
+ |print(repr(float(out1df.iloc[0]["ts_avg"])))
+ |""".stripMargin
+
+ // A zone six hours behind UTC and with no daylight saving of its own, so
+ // one offset covers both dates.
+ val out = runPython(python, driver, "aggregate-timestamp-zone-",
"America/Mexico_City")
+
+ withClue(s"python said:\n$out\nscript:\n$driver") {
+ out.trim.linesIterator.toSeq shouldBe Seq("2078-01-01T06:00:00",
"1704132000000.0")
+ }
+ }
+
+ /** With no zone named, as on Windows without tzlocal, the script asks
+ * tzlocal(), which asks localtime, and Windows refuses an instant before
+ * 1970: the column raised and the sum came back in UTC. The driver stands
in
+ * for both. Checked against Java 17 in the same zone: the sum below is
+ * 1653-12-31 18:00:00 and the average -4.9859712E12.
+ */
+ it should "add timestamps before 1970 where localtime refuses them" taggedAs
NeedsPythonPackages in {
+ val python = resolvePython().getOrElse(cancel("No runnable python
executable"))
+ if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+ val input = Schema().add("t", AttributeType.TIMESTAMP)
+ val desc = descWith(
+ List.empty,
+ aggOp(AggregationFunction.SUM, "t", "ts_total"),
+ aggOp(AggregationFunction.AVERAGE, "t", "ts_avg")
+ )
+ val block = desc.generateStandaloneCode(Map(PortIdentity() -> input))
+
+ val driver =
+ s"""import time
+ |import numpy as np
+ |import pandas as pd
+ |${desc.standaloneHelpers().mkString("\n")}
+ |_localtime = time.localtime
+ |def _refusing(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 = _refusing
+ |_texera_local_zone = lambda: None
Review Comment:
The Windows fallback probe also needs a deterministic time.timezone.
TZ=America/Mexico_City leaves Python's C-runtime timezone at 0 here, so forcing
_texera_local_zone to None gives 1653-12-31T12:00:00 instead of the asserted
18:00. Adding time.timezone = 6 * 3600 to this mocked setup makes the test
pass. This is a fixture portability issue, not a reproduced aggregation defect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
