feat(workflow-operator): export the JSON Lines and Arrow sources as Python

amber/src/main/python/core/models/schema/attribute_type.py

@@ -16,6 +16,10 @@
 # under the License.
 
 import datetime
+import functools
+import os
+import time
+import zoneinfo
 import pyarrow as pa
 from bidict import bidict
 from enum import Enum
@@ -98,21 +102,67 @@ def _parse_bool(v):
 
 
 def _parse_timestamp(v):
+    # A TIMESTAMP reaches a worker as a wall clock with no zone, so a value
+    # parsed here holds none either and compares with a row's. An offset is
+    # moved to the machine's zone first, as the engine's DateParserUtils does.
     if _is_empty_value(v):
-        return datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+        return datetime.datetime(1970, 1, 1)
 
     normalized_value = str(v)
     if normalized_value.endswith("Z"):
         normalized_value = normalized_value[:-1] + "+00:00"
     parsed_value = datetime.datetime.fromisoformat(normalized_value)
-    if (
-        parsed_value.tzinfo is None
-        or parsed_value.tzinfo.utcoffset(parsed_value) is None
-    ):
-        return parsed_value.replace(tzinfo=datetime.timezone.utc)
+    if parsed_value.utcoffset() is not None:
+        return _local_wall_clock(parsed_value)
     return parsed_value
 
 
+def _local_wall_clock(moment):
+    # The wall clock the machine's zone shows at an instant that states its
+    # offset. Windows' localtime refuses an instant before 1970, so the zone is
+    # asked by name where it has one, and otherwise such an instant takes the
+    # machine's standard offset. Before 1900 the engine reads every zone at
+    # its standard offset, where a named zone would give its local mean time.
+    zone = _named_local_zone()
+    utc = moment.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+    if utc < datetime.datetime(1900, 1, 1):
+        return utc + _standard_offset(zone)
+    if zone is not None:
+        return moment.astimezone(zone).replace(tzinfo=None)
+    try:
+        return moment.astimezone().replace(tzinfo=None)
+    except (OSError, OverflowError):
+        return utc + _standard_offset(None)
+
+
+def _standard_offset(zone):
+    if zone is None:
+        return datetime.timedelta(seconds=-time.timezone)
+    now = datetime.datetime.now(zone)
+    return now.utcoffset() - now.dst()
+
+
+def _named_local_zone():
+    return _zone_by_name(os.environ.get("TZ"))
+
+
+@functools.lru_cache(maxsize=None)
+def _zone_by_name(tz):
+    # TZ names the zone when it is set, and /etc/localtime links to it when
+    # not. A zone neither of them names is left to localtime.
+    if tz is not None:
+        name = tz.lstrip(":")
+    else:
+        link = os.path.realpath("/etc/localtime")
+        if "zoneinfo/" not in link:
+            return None
+        name = link.split("zoneinfo/", 1)[1]
+    try:
+        return zoneinfo.ZoneInfo(name)
+    except (ValueError, KeyError):
+        return None
+
+
 FROM_STRING_PARSER_MAPPING = {
     AttributeType.STRING: str,
     AttributeType.INT: lambda v: 0 if _is_empty_value(v) else int(v),

amber/src/test/python/core/models/schema/test_attribute_type.py

@@ -16,9 +16,12 @@
 # under the License.
 
 import datetime
+import time
+import types
 
 import pytest
 
+from core.models.schema import attribute_type
 from core.models.schema.attribute_type import (
     AttributeType,
     FROM_STRING_PARSER_MAPPING,
@@ -31,7 +34,42 @@
 parse_bool = FROM_STRING_PARSER_MAPPING[AttributeType.BOOL]
 parse_timestamp = FROM_STRING_PARSER_MAPPING[AttributeType.TIMESTAMP]
 
-EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+EPOCH = datetime.datetime(1970, 1, 1)
+
+
+@pytest.fixture
+def machine_zone(monkeypatch):
+    # An offset is moved to the machine's zone, so the tests set one.
+    # Windows has no tzset, so there the zone cannot be changed for a test.
+    if not hasattr(time, "tzset"):
+        pytest.skip("time.tzset is not available on this platform")
+
+    def set_zone(name):
+        monkeypatch.setenv("TZ", name)
+        time.tzset()
+
+    yield set_zone
+    monkeypatch.undo()
+    time.tzset()
+
+
+@pytest.fixture
+def windows_localtime(monkeypatch):
+    # Windows has no zone file to read and its localtime refuses an instant
+    # before 1970. Both are imitated here so any platform runs the case.
+    class RefusingDatetime(datetime.datetime):
+        def astimezone(self, tz=None):
+            if tz is None and self.timestamp() < 0:
+                raise OSError(22, "Invalid argument")
+            return super().astimezone(tz)
+
+    patched = types.SimpleNamespace(**vars(datetime))
+    patched.datetime = RefusingDatetime
+    monkeypatch.setattr(attribute_type, "datetime", patched)
+    monkeypatch.setattr(
+        attribute_type, "_named_local_zone", lambda: None, raising=False
+    )
+    monkeypatch.setattr(time, "timezone", -7200)
 
 
 class TestParseBool:
@@ -60,35 +98,77 @@ def test_a_non_numeric_non_literal_value_is_rejected(self):
 
 class TestParseTimestamp:
     @pytest.mark.parametrize("empty", [None, "", "   ", "\t\n"])
-    def test_an_absent_value_parses_as_the_utc_epoch(self, empty):
+    def test_an_absent_value_parses_as_the_epoch_with_no_zone(self, empty):
         parsed = parse_timestamp(empty)
-        # Assert the exact instant *and* the tzinfo: a naive 1970-01-01 would
-        # compare unequal here, so dropping the timezone is caught too.
         assert parsed == EPOCH
-        assert parsed.tzinfo == datetime.timezone.utc
-        assert (parsed.year, parsed.month, parsed.day) == (1970, 1, 1)
-
-    def test_a_zulu_suffix_yields_a_utc_aware_instant(self):
-        # Named for the observable outcome, not for the code that produces it.
-        # `_parse_timestamp` rewrites a trailing "Z" into "+00:00" before
-        # calling `fromisoformat`, but `fromisoformat` has accepted "Z" itself
-        # since Python 3.11 and the pyamber CI matrix is 3.11/3.12/3.13 -- so
-        # deleting that rewrite leaves the entire suite green, and this test
-        # must not be credited with pinning it. What it does pin is the
-        # resulting instant and its offset (and, uniquely in this file, the
-        # "+00:00" constant, should the rewrite ever run on an older runtime).
-        assert parse_timestamp("2024-05-06T07:08:09Z") == datetime.datetime(
-            2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc
-        )
+        assert parsed.tzinfo is None
 
-    def test_a_naive_value_is_assumed_to_be_utc(self):
-        assert parse_timestamp("2024-05-06T07:08:09") == datetime.datetime(
-            2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc
+    def test_a_value_with_no_offset_keeps_its_wall_clock(self, machine_zone):
+        machine_zone("Asia/Kolkata")
+        parsed = parse_timestamp("2024-05-06T07:08:09")
+        assert parsed == datetime.datetime(2024, 5, 6, 7, 8, 9)
+        assert parsed.tzinfo is None
+
+    def test_a_zulu_suffix_is_moved_to_the_machine_zone(self, machine_zone):
+        machine_zone("Asia/Kolkata")
+        assert parse_timestamp("2024-05-06T07:08:09Z") == datetime.datetime(
+            2024, 5, 6, 12, 38, 9
         )
 
-    def test_an_explicit_offset_is_preserved(self):
+    def test_an_explicit_offset_is_moved_to_the_machine_zone(self, machine_zone):
+        machine_zone("UTC")
         parsed = parse_timestamp("2024-05-06T07:08:09+02:00")
-        assert parsed.utcoffset() == datetime.timedelta(hours=2)
-        assert parsed == datetime.datetime(
-            2024, 5, 6, 5, 8, 9, tzinfo=datetime.timezone.utc
-        )
+        assert parsed == datetime.datetime(2024, 5, 6, 5, 8, 9)
+        assert parsed.tzinfo is None
+
+    @pytest.mark.parametrize(
+        "text, expected",
+        [
+            ("1960-01-01T12:00:00Z", datetime.datetime(1960, 1, 1, 17, 30)),
+            ("1960-01-01T12:00:00+02:00", datetime.datetime(1960, 1, 1, 15, 30)),
+            ("2024-05-06T07:08:09Z", datetime.datetime(2024, 5, 6, 12, 38, 9)),
+            ("3000-01-01T12:00:00Z", datetime.datetime(3000, 1, 1, 17, 30)),
+        ],
+    )
+    def test_an_offset_follows_the_named_zone_from_1900_on(
+        self, machine_zone, text, expected
+    ):
+        machine_zone("Asia/Kolkata")
+        assert parse_timestamp(text) == expected
+
+    @pytest.mark.parametrize(
+        "zone, expected",
+        [
+            ("America/Mexico_City", datetime.datetime(1600, 1, 1, 6, 0)),
+            ("Asia/Kolkata", datetime.datetime(1600, 1, 1, 17, 30)),
+        ],
+    )
+    def test_an_offset_before_1900_takes_the_standard_offset(
+        self, machine_zone, zone, expected
+    ):
+        # The engine's DateParserUtils and Timestamp on JDK 17 give these,
+        # where the zone's own rules would give its local mean time.
+        machine_zone(zone)
+        assert parse_timestamp("1600-01-01T12:00:00Z") == expected
+
+    @pytest.mark.parametrize(
+        "text, expected",
+        [
+            ("1960-01-01T12:00:00Z", datetime.datetime(1960, 1, 1, 14, 0)),
+            ("1960-01-01T12:00:00+02:00", datetime.datetime(1960, 1, 1, 12, 0)),
+            ("1600-01-01T12:00:00Z", datetime.datetime(1600, 1, 1, 14, 0)),
+        ],
+    )
+    def test_an_instant_before_1970_parses_where_localtime_refuses_it(
+        self, windows_localtime, text, expected
+    ):
+        # With no zone to name, an instant localtime refuses takes the
+        # machine's standard offset.
+        parsed = parse_timestamp(text)
+        assert parsed == expected
+        assert parsed.tzinfo is None
+
+    def test_an_instant_localtime_accepts_still_asks_it(self, windows_localtime):
+        assert parse_timestamp("2024-05-06T07:08:09+02:00") == datetime.datetime(
+            2024, 5, 6, 5, 8, 9
+        ).replace(tzinfo=datetime.timezone.utc).astimezone().replace(tzinfo=None)

amber/src/test/python/pytexera/udf/test_udf_operator.py

@@ -206,7 +206,7 @@ def test_injected_values_are_applied_before_open(self):
         assert operator.count_parameter.value == 7
         assert operator.enabled_parameter.value is True
         assert operator.created_at_parameter.value == datetime.datetime(
-            2024, 1, 1, 0, 0, tzinfo=datetime.timezone.utc
+            2024, 1, 1, 0, 0
         )
 
     def test_duplicate_parameter_names_with_conflicting_types_raise(self):
@@ -238,12 +238,7 @@ def test_duplicate_parameter_names_with_same_type_succeed(self):
             (
                 "2024-01-01T00:00:00",
                 AttributeType.TIMESTAMP,
-                datetime.datetime(2024, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
-            ),
-            (
-                "2024-01-01T00:00:00Z",
-                AttributeType.TIMESTAMP,
-                datetime.datetime(2024, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
+                datetime.datetime(2024, 1, 1, 0, 0),
             ),
         ],
     )

common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtils.scala

@@ -144,8 +144,14 @@ object AttributeTypeUtils extends Serializable {
           } else {
             str.trim.toInt
           }
-        case int: Integer               => int
-        case long: java.lang.Long       => long.toInt
+        case int: Integer         => int
+        case long: java.lang.Long => long.toInt
+        // An Arrow file states the width of its own integers, and a narrow
+        // column hands its values over as Shorts or Bytes. Without this the parse
+        // threw, the Arrow source caught it, and every value in the column
+        // arrived null.
+        case short: java.lang.Short     => short.toInt
+        case byte: java.lang.Byte       => byte.toInt
         case double: java.lang.Double   => double.toInt
         case boolean: java.lang.Boolean => if (boolean) 1 else 0
         // Timestamp and Binary are considered to be illegal here.
@@ -232,10 +238,15 @@ object AttributeTypeUtils extends Serializable {
   def parseDouble(fieldValue: Any): java.lang.Double = {
     val attempt: Try[Double] = Try {
       fieldValue match {
-        case str: String                => str.trim.toDouble
-        case int: Integer               => int.toDouble
-        case long: java.lang.Long       => long.toDouble
-        case double: java.lang.Double   => double
+        case str: String              => str.trim.toDouble
+        case int: Integer             => int.toDouble
+        case long: java.lang.Long     => long.toDouble
+        case double: java.lang.Double => double
+        // A single-precision column hands its values over as Floats, which only
+        // an Arrow file produces: Texera writes every double it owns as eight
+        // bytes. Without this the parse threw, the Arrow source caught it, and
+        // the whole column arrived null.
+        case float: java.lang.Float     => float.toDouble
         case boolean: java.lang.Boolean => if (boolean) 1 else 0
         // Timestamp and Binary are considered to be illegal here.
         case _ =>