This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new e344ec03fb7 Python timestamp fixes. (#39722)
e344ec03fb7 is described below
commit e344ec03fb7b3bdac74471d7356a687905da6008
Author: claudevdm <[email protected]>
AuthorDate: Tue Aug 11 14:03:06 2026 -0400
Python timestamp fixes. (#39722)
---
sdks/python/apache_beam/io/gcp/pubsub_test.py | 42 ++++++++++++++++++++++
.../runners/direct/transform_evaluator.py | 5 +++
sdks/python/apache_beam/typehints/schemas.py | 10 ++++--
sdks/python/apache_beam/typehints/schemas_test.py | 10 ++++++
sdks/python/apache_beam/utils/timestamp.py | 5 ++-
sdks/python/apache_beam/utils/timestamp_test.py | 17 +++++++++
6 files changed, 86 insertions(+), 3 deletions(-)
diff --git a/sdks/python/apache_beam/io/gcp/pubsub_test.py
b/sdks/python/apache_beam/io/gcp/pubsub_test.py
index 050a69aff6c..5a8f2bf617b 100644
--- a/sdks/python/apache_beam/io/gcp/pubsub_test.py
+++ b/sdks/python/apache_beam/io/gcp/pubsub_test.py
@@ -25,6 +25,7 @@ import unittest
import hamcrest as hc
import mock
+import pytest
import apache_beam as beam
from apache_beam import Pipeline
@@ -712,6 +713,47 @@ class TestReadFromPubSub(unittest.TestCase):
mock_pubsub.return_value.close.assert_not_called()
+ @pytest.mark.timeout(60)
+ def test_read_messages_timestamp_attribute_sub_micro_rfc3339(
+ self, mock_pubsub):
+ # Publishers may emit 7-9 fractional digits. Sub-microsecond digits
+ # must be truncated when the attribute is parsed; element timestamps
+ # are limited to microsecond resolution and messages are acked before
+ # the bundle is output.
+ data = b'data'
+ attributes = {'time': '2018-03-12T13:37:01.2345678Z'}
+ publish_time_secs = 1337000000
+ publish_time_nanos = 133700000
+ ack_id = 'ack_id'
+ pull_response = test_utils.create_pull_response([
+ test_utils.PullResponseMessage(
+ data, attributes, publish_time_secs, publish_time_nanos, ack_id)
+ ])
+ expected_elements = [
+ TestWindowedValue(
+ PubsubMessage(data, attributes),
+ timestamp.Timestamp(1520861821, micros=234567),
+ [window.GlobalWindow()]),
+ ]
+ mock_pubsub.return_value.pull.return_value = pull_response
+
+ options = PipelineOptions([])
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+ pcoll = (
+ p
+ | ReadFromPubSub(
+ 'projects/fakeprj/topics/a_topic',
+ None,
+ None,
+ with_attributes=True,
+ timestamp_attribute='time'))
+ assert_that(pcoll, equal_to(expected_elements), reify_windows=True)
+ mock_pubsub.return_value.acknowledge.assert_has_calls(
+ [mock.call(subscription=mock.ANY, ack_ids=[ack_id])])
+
+ mock_pubsub.return_value.close.assert_not_called()
+
def test_read_messages_timestamp_attribute_missing(self, mock_pubsub):
data = b'data'
attributes = {}
diff --git a/sdks/python/apache_beam/runners/direct/transform_evaluator.py
b/sdks/python/apache_beam/runners/direct/transform_evaluator.py
index 6702ec3362b..d60c840e8c7 100644
--- a/sdks/python/apache_beam/runners/direct/transform_evaluator.py
+++ b/sdks/python/apache_beam/runners/direct/transform_evaluator.py
@@ -724,6 +724,11 @@ class _PubSubReadEvaluator(_TransformEvaluator):
timestamp = Timestamp.from_rfc3339(rfc3339_or_milli)
except ValueError as e:
raise ValueError('Bad timestamp value: %s' % e)
+ if timestamp.precision() > Timestamp.MICROS_PRECISION:
+ # Element timestamps are limited to microsecond resolution, so
+ # ignore sub-microsecond digits, as the Dataflow service does.
+ timestamp = timestamp.to_precision(
+ Timestamp.MICROS_PRECISION, allow_lossy_conversion=True)
else:
if message.publish_time is None:
raise ValueError('No publish time present in message: %s' % message)
diff --git a/sdks/python/apache_beam/typehints/schemas.py
b/sdks/python/apache_beam/typehints/schemas.py
index 2fd3c22e1e5..80bee60ec95 100644
--- a/sdks/python/apache_beam/typehints/schemas.py
+++ b/sdks/python/apache_beam/typehints/schemas.py
@@ -1025,7 +1025,12 @@ class ParameterizedTimestamp(LogicalType[Timestamp,
Timestamp to this logical type, re-register using
:func:`~LogicalType.register_logical_type(ParameterizedTimestamp)`.
"""
- def __init__(self, precision: int = Timestamp.MICROS_PRECISION) -> None:
+ def __init__(self, precision: Optional[int] = None) -> None:
+ if precision is None:
+ # A timestamp:v1 proto without its precision argument is malformed;
+ # decoding at a guessed precision would silently misscale subseconds.
+ raise ValueError(
+ 'beam:logical_type:timestamp:v1 requires a precision argument.')
# The argument arrives as np.int32 when decoded from a schema proto.
precision = int(precision)
if not 0 <= precision <= Timestamp.NANOS_PRECISION:
@@ -1077,7 +1082,8 @@ class ParameterizedTimestamp(LogicalType[Timestamp,
@classmethod
def _from_typing(cls, typ):
- return cls()
+ # A bare Timestamp typehint has no precision; default to micros.
+ return cls(Timestamp.MICROS_PRECISION)
@LogicalType._register_internal
diff --git a/sdks/python/apache_beam/typehints/schemas_test.py
b/sdks/python/apache_beam/typehints/schemas_test.py
index 327fe7947ca..5e66a491090 100644
--- a/sdks/python/apache_beam/typehints/schemas_test.py
+++ b/sdks/python/apache_beam/typehints/schemas_test.py
@@ -879,6 +879,16 @@ class ParameterizedTimestampTest(unittest.TestCase):
representation = logical_type.to_representation_type(millis_value)
self.assertEqual(representation.subseconds, 500000)
+ def test_from_runner_api_rejects_missing_argument(self):
+ # A proto without the precision argument must be rejected; guessing a
+ # default precision would silently misscale subseconds.
+ proto = schema_pb2.LogicalType(
+ urn=common_urns.timestamp.urn,
+ representation=typing_to_runner_api(
+ schemas.ParameterizedTimestampShortRepresentation))
+ with self.assertRaises(ValueError):
+ schemas.LogicalType.from_runner_api(proto)
+
class HypothesisTest(unittest.TestCase):
# There is considerable variablility in runtime for this test, disable
diff --git a/sdks/python/apache_beam/utils/timestamp.py
b/sdks/python/apache_beam/utils/timestamp.py
index 2953541b42f..925467044f3 100644
--- a/sdks/python/apache_beam/utils/timestamp.py
+++ b/sdks/python/apache_beam/utils/timestamp.py
@@ -166,7 +166,10 @@ class Timestamp(object):
if dt.tzinfo != pytz.utc and dt.tzinfo != datetime.timezone.utc:
raise ValueError('dt not in UTC: %s' % dt)
duration = dt - cls._epoch_datetime_utc()
- return Timestamp(duration.total_seconds())
+ # Avoid total_seconds(): its float result can be off by a microsecond.
+ return Timestamp(
+ seconds=duration.days * 86400 + duration.seconds,
+ micros=duration.microseconds)
@classmethod
def from_rfc3339(cls, rfc3339: str) -> 'Timestamp':
diff --git a/sdks/python/apache_beam/utils/timestamp_test.py
b/sdks/python/apache_beam/utils/timestamp_test.py
index e1d120da471..ec1c3604652 100644
--- a/sdks/python/apache_beam/utils/timestamp_test.py
+++ b/sdks/python/apache_beam/utils/timestamp_test.py
@@ -369,6 +369,23 @@ class TimestampPrecisionTest(unittest.TestCase):
with self.assertRaises(ValueError):
_ = ts % Duration(seconds=1)
+ def test_from_rfc3339_fraction_is_exact(self):
+ # Expected values are integers taken from the string, never
+ # Timestamp(float): both sides would share the same lossy float path.
+ # Seconds just above a power of two maximize float error.
+ for rfc, want_sec, want_sub, want_p in [
+ ('2038-01-19T03:14:08.510215590Z', 2147483648, 510215590, 9),
+ ('2004-01-10T13:37:04.611178002Z', 1073741824, 611178002, 9),
+ ('1970-01-01T00:00:00.1252641Z', 0, 1252641, 7),
+ ('1970-01-01T00:00:00.254229935Z', 0, 254229935, 9),
+ ('1969-12-31T23:59:59.746939251Z', -1, 746939251, 9),
+ ('9999-12-31T23:59:59.389694109Z', 253402300799, 389694109, 9),
+ ]:
+ ts = Timestamp.from_rfc3339(rfc)
+ self.assertEqual((ts.seconds(), ts.subseconds(), ts.precision()),
+ (want_sec, want_sub, want_p),
+ rfc)
+
class DurationTest(unittest.TestCase):
def test_of(self):
