diff --git a/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json b/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json index c5309eebb070..e3d6056a5de9 100644 --- a/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 17 + "modification": 1 } diff --git a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json b/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json index d6a91b7e2e86..e3d6056a5de9 100644 --- a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 7 + "modification": 1 } diff --git a/CHANGES.md b/CHANGES.md index c5da08bd1a5f..75a362d1580b 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -73,6 +73,9 @@ * (Java) Supported acknowledge mode for JmsIO ([#39253](https://github.com/apache/beam/issues/39253)). * (Python) Staged files directory is now automatically added to `sys.path` on the Python SDK worker at startup. This makes Python files provided via the '--files_to_stage' pipeline option importable in the pipeline code and makes it easier to initialize Python SDK harness at startup via the `--beam_plugins` pipeline option. For more information, see the [Staging Individual Files](https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/#staging-files) section of the dependency management docs. This behavior can be disabled by passing the '--experiments=no_staged_dir_in_sys_path' pipeline option ([#39431](https://github.com/apache/beam/issues/39431)). * (Python) Added `equal_to_approx`, an `assert_that` matcher that compares numeric pipeline outputs with a configurable tolerance ([#18028](https://github.com/apache/beam/issues/18028)). +* (Python) `Timestamp` now supports variable subsecond precision, up to nanoseconds. The portable + `beam:logical_type:timestamp:v1` logical type now maps to Python's `Timestamp` ([#39344](https://github.com/apache/beam/issues/39344)). +* X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Breaking Changes diff --git a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml index 8ce9d2751545..eb7999680074 100644 --- a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml +++ b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml @@ -626,3 +626,34 @@ examples: "\u0001\u0003\u0061\u0062\u0063" : "abc" "\u0001\u000a\u006d\u006f\u0072\u0065\u0020\u0062\u0079\u0074\u0065\u0073" : "more bytes" "\u0000" : null + +--- + +coder: + urn: "beam:coder:row:v1" + # f_timestamp: logical(timestamp:v1, precision=3, representation row) + payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x02\x12$5f2fbf6c-8b8e-4a30-a9be-e0f65f0f79c3\"\x02\x10\x03*\x04\x0a\x02\x18\x03\x12$9c7c1b6c-2a5e-4c53-9b32-91b71c1f8ac3" +examples: + "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x03\xe7": {f_timestamp: {seconds: 1597328054, subseconds: 999}} + "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\x01\xf4": {f_timestamp: {seconds: -2, subseconds: 500}} + "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x00\x00": {f_timestamp: {seconds: 1597328054, subseconds: 0}} + +--- + +coder: + urn: "beam:coder:row:v1" + # f_timestamp: logical(timestamp:v1, precision=6, representation row) + payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x03\x12$b8b3faf8-63f8-4b40-9f4d-3c1e83d4e07b\"\x02\x10\x03*\x04\x0a\x02\x18\x06\x12$e0a72e10-b214-4fd2-a4dc-fca0e7f7d924" +examples: + "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\xc0\xc4\x07": {f_timestamp: {seconds: 1597328054, subseconds: 123456}} + "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\xa0\xc2\x1e": {f_timestamp: {seconds: -2, subseconds: 500000}} + +--- + +coder: + urn: "beam:coder:row:v1" + # f_timestamp: logical(timestamp:v1, precision=9, representation row) + payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x03\x12$3d6f7b81-2f13-4a41-8be3-4d8cd8f30f42\"\x02\x10\x03*\x04\x0a\x02\x18\x09\x12$cf03a3c1-4a3a-4f77-8e2b-6e5b19c19f27" +examples: + "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x95\x9a\xef:": {f_timestamp: {seconds: 1597328054, subseconds: 123456789}} + "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\xff\x93\xeb\xdc\x03": {f_timestamp: {seconds: -2, subseconds: 999999999}} diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto index d514be101be5..c114c35f2d90 100644 --- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto +++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto @@ -234,6 +234,19 @@ message LogicalTypes { // since the epoch. DATE = 8 [(org.apache.beam.model.pipeline.v1.beam_urn) = "beam:logical_type:date:v1"]; + + // A URN for Timestamp type + // - Representation type: ROW for + // precision < 5, otherwise ROW + // - Argument type: INT32. + // A timestamp without a timezone, represented as seconds since the + // epoch plus a non-negative fraction of a second. The argument is + // the precision: the number of decimal digits of the subseconds + // field (e.g. 3 for milliseconds, 6 for micros, 9 for nanos). + // subseconds stays non-negative even for timestamps before the epoch, + // e.g. -1.5s at precision 6 is {seconds: -2, subseconds: 500000}. + TIMESTAMP = 9 [(org.apache.beam.model.pipeline.v1.beam_urn) = + "beam:logical_type:timestamp:v1"]; } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java index d527b2949235..8122704e6436 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java @@ -50,6 +50,7 @@ import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant; import org.apache.beam.sdk.schemas.logicaltypes.PythonCallable; import org.apache.beam.sdk.schemas.logicaltypes.SchemaLogicalType; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.schemas.logicaltypes.UnknownLogicalType; import org.apache.beam.sdk.schemas.logicaltypes.VariableBytes; import org.apache.beam.sdk.schemas.logicaltypes.VariableString; @@ -115,6 +116,7 @@ private static String getLogicalTypeUrn(String identifier) { .put(FixedString.IDENTIFIER, FixedString.class) .put(VariableString.IDENTIFIER, VariableString.class) .put(Date.IDENTIFIER, Date.class) + .put(Timestamp.IDENTIFIER, Timestamp.class) .build(); public static SchemaApi.Schema schemaToProto(Schema schema, boolean serializeLogicalType) { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/Timestamp.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/Timestamp.java index 87e47f5961e3..51fa9008b749 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/Timestamp.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/Timestamp.java @@ -22,6 +22,8 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import java.time.Instant; +import org.apache.beam.model.pipeline.v1.RunnerApi; +import org.apache.beam.model.pipeline.v1.SchemaApi; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.values.Row; import org.checkerframework.checker.nullness.qual.NonNull; @@ -56,7 +58,12 @@ * } */ public class Timestamp implements Schema.LogicalType { - public static final String IDENTIFIER = "beam:logical_type:timestamp:v1"; + public static final String IDENTIFIER = + SchemaApi.LogicalTypes.Enum.TIMESTAMP + .getValueDescriptor() + .getOptions() + .getExtension(RunnerApi.beamUrn); + static final int MIN_PRECISION = 0; static final int MAX_PRECISION = 9; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaTranslationTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaTranslationTest.java index 8a382c4ad1b8..925b1e02b786 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaTranslationTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaTranslationTest.java @@ -51,6 +51,7 @@ import org.apache.beam.sdk.schemas.logicaltypes.PythonCallable; import org.apache.beam.sdk.schemas.logicaltypes.SchemaLogicalType; import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.schemas.logicaltypes.UnknownLogicalType; import org.apache.beam.sdk.schemas.logicaltypes.VariableBytes; import org.apache.beam.sdk.schemas.logicaltypes.VariableString; @@ -460,6 +461,12 @@ public static Iterable data() { .add(FieldType.logicalType(FixedString.of(10))) .add(FieldType.logicalType(VariableString.of(10))) .add(FieldType.logicalType(FixedPrecisionNumeric.of(10))) + .add(FieldType.logicalType(Timestamp.MILLIS)) + .add(FieldType.logicalType(Timestamp.MICROS)) + .add(FieldType.logicalType(Timestamp.NANOS)) + // INT16/INT32 subseconds representation boundary. + .add(FieldType.logicalType(Timestamp.of(4))) + .add(FieldType.logicalType(Timestamp.of(5))) .add(FieldType.logicalType(new PortableNullArgLogicalType())) .add(FieldType.logicalType(new NullArgumentLogicalType())) .build(); diff --git a/sdks/python/apache_beam/coders/row_coder_test.py b/sdks/python/apache_beam/coders/row_coder_test.py index 396cdc9b3f66..28170721c28b 100644 --- a/sdks/python/apache_beam/coders/row_coder_test.py +++ b/sdks/python/apache_beam/coders/row_coder_test.py @@ -203,6 +203,60 @@ def test_create_row_coder_from_schema(self): for test_case in self.PEOPLE: self.assertEqual(test_case, coder.decode(coder.encode(test_case))) + def test_create_row_coder_from_schema_with_parameterized_timestamp(self): + # Simulates schemas produced by the Java SDK's parameterized Timestamp + # logical type. Subseconds field uses INT16 for + # precision < 5 and INT32 otherwise. + for precision, subseconds_type in [(3, schema_pb2.INT16), + (4, schema_pb2.INT16), + (5, schema_pb2.INT32), + (6, schema_pb2.INT32), + (9, schema_pb2.INT32)]: + schema = schema_pb2.Schema( + id="timestamp_precision_%d" % precision, + fields=[ + schema_pb2.Field( + name="event_time", + type=schema_pb2.FieldType( + logical_type=schema_pb2.LogicalType( + urn="beam:logical_type:timestamp:v1", + representation=schema_pb2.FieldType( + row_type=schema_pb2.RowType( + schema=schema_pb2.Schema( + id="timestamp_repr_%d" % precision, + fields=[ + schema_pb2.Field( + name="seconds", + type=schema_pb2.FieldType( + atomic_type=schema_pb2.INT64) + ), + schema_pb2.Field( + name="subseconds", + type=schema_pb2.FieldType( + atomic_type=subseconds_type)), + ]))), + argument_type=schema_pb2.FieldType( + atomic_type=schema_pb2.INT32), + argument=schema_pb2.FieldValue( + atomic_value=schema_pb2.AtomicTypeValue( + int32=precision))))), + ]) + coder = RowCoder(schema) + row = named_tuple_from_schema(schema) + + subseconds = 10**precision - 1 + for value in [ + Timestamp(seconds=1500000000, + subseconds=subseconds, + precision=precision), + Timestamp(seconds=-2, subseconds=subseconds, precision=precision), + Timestamp(seconds=1500000000, precision=precision), + ]: + test_case = row(event_time=value) + decoded = coder.decode(coder.encode(test_case)) + self.assertEqual(test_case, decoded) + self.assertEqual(decoded.event_time.precision(), precision) + def test_row_coder_negative_varint(self): schema = schema_pb2.Schema( id="negative", diff --git a/sdks/python/apache_beam/portability/common_urns.py b/sdks/python/apache_beam/portability/common_urns.py index 7777f63ffbe9..913fb282f86b 100644 --- a/sdks/python/apache_beam/portability/common_urns.py +++ b/sdks/python/apache_beam/portability/common_urns.py @@ -93,3 +93,4 @@ fixed_char = LogicalTypes.Enum.FIXED_CHAR var_char = LogicalTypes.Enum.VAR_CHAR date = LogicalTypes.Enum.DATE +timestamp = LogicalTypes.Enum.TIMESTAMP diff --git a/sdks/python/apache_beam/typehints/schemas.py b/sdks/python/apache_beam/typehints/schemas.py index ca33e1eefcbe..084ecc93581c 100644 --- a/sdks/python/apache_beam/typehints/schemas.py +++ b/sdks/python/apache_beam/typehints/schemas.py @@ -977,6 +977,92 @@ def _from_typing(cls, typ): 'MicrosInstantRepresentation', [('seconds', np.int64), ('micros', np.int64)]) +ParameterizedTimestampRepresentation = NamedTuple( + 'ParameterizedTimestampRepresentation', [('seconds', np.int64), + ('subseconds', np.int32)]) + +# The subseconds field is INT16 for precision < 5 +_TIMESTAMP_SHORT_PRECISION_LIMIT = 5 +ParameterizedTimestampShortRepresentation = NamedTuple( + 'ParameterizedTimestampShortRepresentation', [('seconds', np.int64), + ('subseconds', np.int16)]) + + +@LogicalType._register_internal +class ParameterizedTimestamp(LogicalType[Timestamp, + ParameterizedTimestampRepresentation, + np.int32]): + """Timestamp logical type parameterized by subsecond precision. + + The argument is the precision: the number of decimal digits used to + represent the fraction of a second, e.g. 3 for milliseconds, 6 for + microseconds, 9 for nanoseconds. + + Values are represented as a row of ``seconds`` (INT64, floored seconds + since the epoch) and ``subseconds`` (units of 10**-precision seconds, + always in ``[0, 10**precision)``. + ``subseconds`` is an INT16 field for precision < 5 and an INT32 + field otherwise. + + Note: Timestamp originating from Python to xlang still defaults to + MicrosInstant for backwards compatibility. To override the mapping of + Timestamp to this logical type, re-register using + :func:`~LogicalType.register_logical_type(ParameterizedTimestamp)`. + """ + def __init__(self, precision: int = Timestamp.MICROS_PRECISION) -> None: + # The argument arrives as np.int32 when decoded from a schema proto. + precision = int(precision) + if not 0 <= precision <= Timestamp.NANOS_PRECISION: + raise ValueError( + 'Timestamp precision must be between 0 and %d (inclusive), ' + 'but was %d.' % (Timestamp.NANOS_PRECISION, precision)) + self._precision = precision + + @classmethod + def urn(cls): + return common_urns.timestamp.urn + + def representation_type(self) -> type: # type: ignore[override] + # Unlike other logical types, the representation depends on the + # argument, so this is an instance method rather than a classmethod. + if self._precision < _TIMESTAMP_SHORT_PRECISION_LIMIT: + return ParameterizedTimestampShortRepresentation + return ParameterizedTimestampRepresentation + + @classmethod + def language_type(cls): + return Timestamp + + def to_representation_type(self, value: Timestamp): + # Verify that the value can be represented exactly at this type's precision + if value.precision() != self._precision: + value = value.to_precision(self._precision) + return self.representation_type()(value.seconds(), value.subseconds()) + + def to_language_type(self, value) -> Timestamp: + subseconds = int(value.subseconds) + # Match Java's toInputType: out-of-range subseconds indicate data + # corruption or a precision mismatch. + if not 0 <= subseconds < 10**self._precision: + raise ValueError( + 'Invalid subseconds %d for Timestamp with precision %d.' % + (subseconds, self._precision)) + return Timestamp( + seconds=int(value.seconds), + subseconds=subseconds, + precision=self._precision) + + @classmethod + def argument_type(cls): + return np.int32 + + def argument(self): + return self._precision + + @classmethod + def _from_typing(cls, typ): + return cls() + @LogicalType._register_internal class MillisInstant(NoArgumentLogicalType[Timestamp, np.int64]): @@ -1019,9 +1105,11 @@ def to_language_type(self, value): return Timestamp(micros=millis * 1000) -# Make sure MicrosInstant is registered after MillisInstant so that it -# overwrites the mapping of Timestamp language type representation choice and -# thus does not lose microsecond precision inside python sdk. +# Make sure MicrosInstant is registered after MillisInstant and +# ParameterizedTimestamp so that it overwrites the mapping of Timestamp +# language type representation choice: plain Timestamp typehints keep their +# historical micros_instant encoding and do not lose microsecond precision +# inside the python sdk. @LogicalType._register_internal class MicrosInstant(NoArgumentLogicalType[Timestamp, MicrosInstantRepresentation]): diff --git a/sdks/python/apache_beam/typehints/schemas_test.py b/sdks/python/apache_beam/typehints/schemas_test.py index a8db60c9260a..c2c21a7ce391 100644 --- a/sdks/python/apache_beam/typehints/schemas_test.py +++ b/sdks/python/apache_beam/typehints/schemas_test.py @@ -770,6 +770,100 @@ def test_union(self): ('unique', Any)]) +class ParameterizedTimestampTest(unittest.TestCase): + def test_urn(self): + self.assertEqual( + schemas.ParameterizedTimestamp.urn(), 'beam:logical_type:timestamp:v1') + + def test_timestamp_typehint_still_maps_to_micros_instant(self): + # Existing pipelines rely on plain Timestamp fields being encoded as + # micros_instant; registering ParameterizedTimestamp must not change + # the default. + field_type = typing_to_runner_api(Timestamp) + self.assertEqual( + field_type.logical_type.urn, common_urns.micros_instant.urn) + + def test_from_runner_api_reconstructs_precision(self): + logical_type_proto = schema_pb2.LogicalType( + urn=common_urns.timestamp.urn, + representation=typing_to_runner_api( + schemas.ParameterizedTimestampRepresentation), + argument_type=schema_pb2.FieldType(atomic_type=schema_pb2.INT32), + argument=schema_pb2.FieldValue( + atomic_value=schema_pb2.AtomicTypeValue(int32=9))) + logical_type = schemas.LogicalType.from_runner_api(logical_type_proto) + self.assertIsInstance(logical_type, schemas.ParameterizedTimestamp) + self.assertEqual(logical_type.argument(), 9) + value = logical_type.to_language_type( + schemas.ParameterizedTimestampRepresentation( + seconds=np.int64(1234), subseconds=np.int32(123456789))) + self.assertEqual( + value, Timestamp(seconds=1234, subseconds=123456789, precision=9)) + self.assertEqual(value.precision(), 9) + + def test_representation_type_matches_java(self): + # The Java SDK uses an INT16 subseconds field for precision < 5 and + # INT32 otherwise; the wire formats differ, so we must match. + for precision in range(0, 5): + self.assertEqual( + schemas.ParameterizedTimestamp(precision).representation_type(), + schemas.ParameterizedTimestampShortRepresentation) + for precision in range(5, 10): + self.assertEqual( + schemas.ParameterizedTimestamp(precision).representation_type(), + schemas.ParameterizedTimestampRepresentation) + + def test_precision_validation(self): + with self.assertRaises(ValueError): + schemas.ParameterizedTimestamp(10) + with self.assertRaises(ValueError): + schemas.ParameterizedTimestamp(-1) + + def test_value_round_trip(self): + for precision, subseconds in [(3, 500), (6, 500000), (9, 123456789)]: + logical_type = schemas.ParameterizedTimestamp(precision) + value = Timestamp( + seconds=1234, subseconds=subseconds, precision=precision) + representation = logical_type.to_representation_type(value) + self.assertEqual(representation.seconds, 1234) + self.assertEqual(representation.subseconds, subseconds) + self.assertEqual(logical_type.to_language_type(representation), value) + + def test_negative_timestamps_use_floored_seconds(self): + # -1.5s is represented as {seconds: -2, subseconds: 500000} at + # microsecond precision, matching the Java SDK and java.time.Instant. + logical_type = schemas.ParameterizedTimestamp(6) + representation = logical_type.to_representation_type(Timestamp(-1.5)) + self.assertEqual(representation.seconds, -2) + self.assertEqual(representation.subseconds, 500000) + self.assertEqual( + logical_type.to_language_type(representation), Timestamp(-1.5)) + + def test_to_language_type_rejects_out_of_range_subseconds(self): + # Java's toInputType rejects these as likely data corruption. + logical_type = schemas.ParameterizedTimestamp(3) + with self.assertRaises(ValueError): + logical_type.to_language_type( + schemas.ParameterizedTimestampShortRepresentation( + np.int64(10), np.int16(5000))) + with self.assertRaises(ValueError): + logical_type.to_language_type( + schemas.ParameterizedTimestampShortRepresentation( + np.int64(10), np.int16(-1))) + + def test_to_representation_type_guards_against_precision_loss(self): + # Mirrors the Java SDK's toBaseType check: a value that cannot be + # represented exactly at the logical type's precision is an error. + logical_type = schemas.ParameterizedTimestamp(6) + nanos_value = Timestamp(seconds=1, subseconds=123456789, precision=9) + with self.assertRaises(ValueError): + logical_type.to_representation_type(nanos_value) + # Lower-precision values are converted losslessly. + millis_value = Timestamp(seconds=1, subseconds=500, precision=3) + representation = logical_type.to_representation_type(millis_value) + self.assertEqual(representation.subseconds, 500000) + + class HypothesisTest(unittest.TestCase): # There is considerable variablility in runtime for this test, disable # deadline. diff --git a/sdks/python/apache_beam/utils/timestamp.py b/sdks/python/apache_beam/utils/timestamp.py index 3f585eecae08..2953541b42f4 100644 --- a/sdks/python/apache_beam/utils/timestamp.py +++ b/sdks/python/apache_beam/utils/timestamp.py @@ -24,7 +24,9 @@ # mypy: disallow-untyped-defs import datetime +import re import time +from typing import Optional from typing import Union from typing import overload @@ -41,28 +43,82 @@ DurationTypes = Union[int, float, 'Duration'] TimestampDurationTypes = Union[int, float, 'Duration', 'Timestamp'] +# Powers of ten indexed by exponent +_POW_10 = {i: 10**i for i in range(10)} + class Timestamp(object): - """Represents a Unix second timestamp with microsecond granularity. + """Represents a Unix second timestamp with configurable subsecond precision. Can be treated in common timestamp arithmetic operations as a numeric type. - Internally stores a time interval as an int of microseconds. This strategy - is necessary since floating point values lose precision when storing values, - especially after arithmetic operations (for example, 10000000 % 0.1 evaluates - to 0.0999999994448885). + Internally stores the timestamp as an int of floored seconds since the + epoch plus a non-negative int subsecond value interpreted with the + specified precision where (``0 <= subseconds < 10**precision``). + + Integer storage is necessary since floating point + values lose precision when storing values, especially after arithmetic + operations (for example, 10000000 % 0.1 evaluates to 0.0999999994448885). + + ``precision`` is the number of decimal digits used to represent the + fraction of a second (e.g. 3 for millis, 6 for micros, 9 for + nanos). Defaults to microseconds. + If ``seconds`` is a float, the fractional part will be captured up + to ``precision`` digits. + + Lossy conversion operations will throw an error unless + ``allow_lossy_conversion=True`` is specified (e.g. see ``to_utc_datetime``). """ + MICROS_PRECISION = 6 + NANOS_PRECISION = 9 + def __init__( self, seconds: Union[int, float] = 0, - micros: Union[int, float] = 0) -> None: + subseconds: Union[int, float] = 0, + precision: int = MICROS_PRECISION, + *, + micros: Optional[Union[int, float]] = None) -> None: if not isinstance(seconds, (int, float)): raise TypeError( 'Cannot interpret %s %s as seconds.' % (seconds, type(seconds))) - if not isinstance(micros, (int, float)): + if not isinstance(subseconds, (int, float)): raise TypeError( - 'Cannot interpret %s %s as micros.' % (micros, type(micros))) - self.micros = int(seconds * 1000000) + int(micros) + 'Cannot interpret %s %s as subseconds.' % + (subseconds, type(subseconds))) + if not isinstance(precision, int): + raise TypeError( + 'Cannot interpret %s %s as precision.' % (precision, type(precision))) + if not 0 <= precision <= Timestamp.NANOS_PRECISION: + raise ValueError( + 'Timestamp precision must be between 0 and %d (inclusive), ' + 'but was %d.' % (Timestamp.NANOS_PRECISION, precision)) + if micros is not None: + if not isinstance(micros, (int, float)): + raise TypeError( + 'Cannot interpret %s %s as micros.' % (micros, type(micros))) + if subseconds: + raise ValueError( + 'micros and subseconds are mutually exclusive, got micros=%s, ' + 'subseconds=%s.' % (micros, subseconds)) + if precision != Timestamp.MICROS_PRECISION: + raise ValueError( + 'micros implies microsecond precision (6) but precision was %d; ' + 'use subseconds instead.' % precision) + subseconds = micros + self._precision = precision + total = int(seconds * _POW_10[precision]) + int(subseconds) + self._seconds, self._subseconds = divmod(total, _POW_10[precision]) + + def _total(self, precision: int) -> int: + """Returns the total time since the epoch in units of 10**-precision + seconds. + + ``precision`` must be greater than or equal to this timestamp's + precision, so that scaling up is always lossless. + """ + return self._seconds * _POW_10[precision] + ( + self._subseconds * _POW_10[precision - self._precision]) @staticmethod def of(seconds: TimestampTypes) -> 'Timestamp': @@ -116,6 +172,10 @@ def from_utc_datetime(cls, dt: datetime.datetime) -> 'Timestamp': def from_rfc3339(cls, rfc3339: str) -> 'Timestamp': """Create a ``Timestamp`` instance from an RFC 3339 compliant string. + Fractional seconds up to microseconds produce a microsecond-precision + Timestamp; a longer fraction (up to nanoseconds) produces a Timestamp + whose precision matches the number of fractional digits. + .. note:: All timezones are implicitly converted to UTC. @@ -128,33 +188,108 @@ def from_rfc3339(cls, rfc3339: str) -> 'Timestamp': raise ValueError( "Could not parse RFC 3339 string '{}' due to error: '{}'.".format( rfc3339, e)) - return cls.from_utc_datetime(dt) + timestamp = cls.from_utc_datetime(dt) + # dateutil silently truncates fractional seconds to microseconds; parse + # any sub-microsecond digits ourselves to avoid losing precision. + fraction = re.search(r'[0-9]{2}[.,]([0-9]{7,})', rfc3339) + if fraction: + digits = fraction.group(1) + if len(digits) > cls.NANOS_PRECISION: + raise ValueError( + "Could not parse RFC 3339 string '%s': fractional seconds " + 'beyond nanosecond precision are not supported.' % rfc3339) + precision = len(digits) + sub_micro = int(digits[cls.MICROS_PRECISION:]) + return Timestamp( + timestamp.seconds(), + timestamp.subseconds() * _POW_10[precision - cls.MICROS_PRECISION] + + sub_micro, + precision) + return timestamp def seconds(self) -> int: """Returns the timestamp in seconds.""" - return self.micros // 1000000 + return self._seconds + + def subseconds(self) -> int: + """Returns the fraction of a second, in units of 10**-precision seconds. + + Always non-negative and less than 10**precision + """ + return self._subseconds + + def precision(self) -> int: + """Returns the precision of this Timestamp.""" + return self._precision + + @property + def micros(self) -> int: + """Returns the total number of microseconds since the epoch.""" + if self._precision > Timestamp.MICROS_PRECISION: + raise ValueError( + '%r has greater than microsecond precision, converting it to ' + 'micros may lose precision. Use to_precision(6, ' + 'allow_lossy_conversion=True) to explicitly truncate it first, ' + 'or use nanos instead.' % self) + return self._total(Timestamp.MICROS_PRECISION) + + @property + def nanos(self) -> int: + """Returns the total number of nanoseconds since the epoch.""" + return self._total(Timestamp.NANOS_PRECISION) + + def to_precision( + self, + precision: int, + allow_lossy_conversion: bool = False) -> 'Timestamp': + """Returns this Timestamp converted to the given precision. + + Increasing precision is always lossless. Decreasing precision raises + ValueError if this timestamp has a non-zero component below the target + precision, unless allow_lossy_conversion is True, in which case the + timestamp is truncated (floored) to the target precision. + """ + if precision == self._precision: + return self + if not 0 <= precision <= Timestamp.NANOS_PRECISION: + raise ValueError( + 'Timestamp precision must be between 0 and %d (inclusive), ' + 'but was %d.' % (Timestamp.NANOS_PRECISION, precision)) + if precision > self._precision: + scale = _POW_10[precision - self._precision] + return Timestamp(self._seconds, self._subseconds * scale, precision) + scale = _POW_10[self._precision - precision] + remainder = self._subseconds % scale + if remainder and not allow_lossy_conversion: + raise ValueError( + '%r cannot be represented exactly at precision %d. Set ' + 'allow_lossy_conversion=True to truncate it.' % (self, precision)) + return Timestamp(self._seconds, self._subseconds // scale, precision) def predecessor(self) -> 'Timestamp': - """Returns the largest timestamp smaller than self.""" - return Timestamp(micros=self.micros - 1) + """Returns the largest timestamp smaller than self, at this precision.""" + return Timestamp(self._seconds, self._subseconds - 1, self._precision) def successor(self) -> 'Timestamp': - """Returns the smallest timestamp larger than self.""" - return Timestamp(micros=self.micros + 1) + """Returns the smallest timestamp larger than self, at this precision.""" + return Timestamp(self._seconds, self._subseconds + 1, self._precision) def __repr__(self) -> str: - micros = self.micros + total = self._total(self._precision) sign = '' - if micros < 0: + if total < 0: sign = '-' - micros = -micros - int_part = micros // 1000000 - frac_part = micros % 1000000 + total = -total + int_part, frac_part = divmod(total, _POW_10[self._precision]) if frac_part: - return 'Timestamp(%s%d.%06d)' % (sign, int_part, frac_part) + return 'Timestamp(%s%d.%0*d)' % ( + sign, int_part, self._precision, frac_part) return 'Timestamp(%s%d)' % (sign, int_part) - def to_utc_datetime(self, has_tz: bool = False) -> datetime.datetime: + def to_utc_datetime( + self, + has_tz: bool = False, + allow_lossy_conversion: bool = False) -> datetime.datetime: """Returns a ``datetime.datetime`` object of UTC for this Timestamp. Note that this method returns a ``datetime.datetime`` object without a @@ -164,74 +299,112 @@ def to_utc_datetime(self, has_tz: bool = False) -> datetime.datetime: Args: has_tz: whether the timezone info is attached, default to False. + allow_lossy_conversion: must be set to True to convert a timestamp + with precision above microseconds, since ``datetime.datetime`` only + supports microsecond resolution; the result is truncated (floored) + to whole microseconds. Returns: a ``datetime.datetime`` object of UTC for this Timestamp. - """ + Raises: + ValueError: if this timestamp has precision above microseconds and + allow_lossy_conversion is not True. + """ + if self._precision > Timestamp.MICROS_PRECISION: + if not allow_lossy_conversion: + raise ValueError( + 'Converting %r to datetime truncates it to microseconds. Set ' + 'allow_lossy_conversion=True to allow this conversion.' % self) + micros_of_second = self._subseconds // _POW_10[self._precision - + Timestamp.MICROS_PRECISION] + else: + micros_of_second = self._subseconds * _POW_10[Timestamp.MICROS_PRECISION - + self._precision] # We can't easily construct a datetime object from microseconds, so we # create one at the epoch and add an appropriate timedelta interval. epoch = self._epoch_datetime_utc() if not has_tz: epoch = epoch.replace(tzinfo=None) - return epoch + datetime.timedelta(microseconds=self.micros) + return epoch + datetime.timedelta( + seconds=self._seconds, microseconds=micros_of_second) def to_rfc3339(self) -> str: - # Append 'Z' for UTC timezone. - return self.to_utc_datetime().isoformat() + 'Z' + """Returns an RFC 3339 string for this Timestamp.""" + if self._precision <= Timestamp.MICROS_PRECISION: + # Append 'Z' for UTC timezone. + return self.to_utc_datetime().isoformat() + 'Z' + # format the fractional second manually + whole_second_datetime = self._epoch_datetime_utc().replace( + tzinfo=None) + datetime.timedelta(seconds=self._seconds) + result = whole_second_datetime.isoformat() + if self._subseconds: + result = result + '.%0*d' % (self._precision, self._subseconds) + + return result + 'Z' def to_proto(self) -> timestamp_pb2.Timestamp: """Returns the `google.protobuf.timestamp_pb2` representation.""" - secs = self.micros // 1000000 - nanos = (self.micros % 1000000) * 1000 - return timestamp_pb2.Timestamp(seconds=secs, nanos=nanos) + return timestamp_pb2.Timestamp( + seconds=self._seconds, + nanos=self._subseconds * + _POW_10[Timestamp.NANOS_PRECISION - self._precision]) @staticmethod def from_proto(timestamp_proto: timestamp_pb2.Timestamp) -> 'Timestamp': """Creates a Timestamp from a `google.protobuf.timestamp_pb2`. - Note that the google has a sub-second resolution of nanoseconds whereas this - class has a resolution of microsends. This class will truncate the - nanosecond resolution down to the microsecond. + The returned Timestamp has microsecond precision if the proto's nanos + are microsecond-aligned, and nanosecond precision otherwise. """ - if timestamp_proto.nanos % 1000 != 0: - # TODO(https://github.com/apache/beam/issues/19922): Better define - # timestamps. - raise ValueError( - "Cannot convert from nanoseconds to microseconds " + - "because this loses precision. Please make sure that " + - "this is the correct behavior you want and manually " + - "truncate the precision to the nearest microseconds. " + - "See [https://github.com/apache/beam/issues/19922] for " + - "more information.") - + return Timestamp( + timestamp_proto.seconds, + timestamp_proto.nanos, + Timestamp.NANOS_PRECISION) return Timestamp( - seconds=timestamp_proto.seconds, micros=timestamp_proto.nanos // 1000) + timestamp_proto.seconds, + timestamp_proto.nanos // 1000, + Timestamp.MICROS_PRECISION) def __float__(self) -> float: # Note that the returned value may have lost precision. - return self.micros / 1000000 + return self._total(Timestamp.NANOS_PRECISION) / 1000000000 def __int__(self) -> int: # Note that the returned value may have lost precision. - return self.micros // 1000000 + return self._seconds def __eq__(self, other: object) -> bool: - # Allow comparisons between Duration and Timestamp values. - if isinstance(other, (Duration, Timestamp)): - return self.micros == other.micros + if isinstance(other, Timestamp): + if self._precision == other._precision: + return ( + self._seconds == other._seconds and + self._subseconds == other._subseconds) + precision = max(self._precision, other._precision) + return self._total(precision) == other._total(precision) + elif isinstance(other, Duration): + # Allow comparisons between Duration and Timestamp values. + return self._total(Timestamp.NANOS_PRECISION) == other.micros * 1000 elif isinstance(other, (int, float)): - return self.micros == Timestamp.of(other).micros + return self == Timestamp.of(other) else: # Support equality with other types return NotImplemented def __lt__(self, other: TimestampDurationTypes) -> bool: # Allow comparisons between Duration and Timestamp values. - if not isinstance(other, Duration): - other = Timestamp.of(other) - return self.micros < other.micros + if isinstance(other, Duration): + return self._total(Timestamp.NANOS_PRECISION) < other.micros * 1000 + other = Timestamp.of(other) + if self._seconds != other._seconds: + return self._seconds < other._seconds + if self._precision == other._precision: + return self._subseconds < other._subseconds + precision = max(self._precision, other._precision) + return ( + self._subseconds * _POW_10[precision - self._precision] + < other._subseconds * _POW_10[precision - other._precision]) def __gt__(self, other: TimestampDurationTypes) -> bool: return not (self < other or self == other) @@ -243,11 +416,16 @@ def __ge__(self, other: TimestampDurationTypes) -> bool: return not self < other def __hash__(self) -> int: - return hash(self.micros) + # Normalized to max precision + return hash(self._total(Timestamp.NANOS_PRECISION)) def __add__(self, other: DurationTypes) -> 'Timestamp': other = Duration.of(other) - return Timestamp(micros=self.micros + other.micros) + precision = max(self._precision, Timestamp.MICROS_PRECISION) + return Timestamp( + subseconds=self._total(precision) + + other.micros * _POW_10[precision - Timestamp.MICROS_PRECISION], + precision=precision) def __radd__(self, other: DurationTypes) -> 'Timestamp': return self + other @@ -264,13 +442,33 @@ def __sub__( self, other: Union[DurationTypes, 'Timestamp']) -> Union['Timestamp', 'Duration']: if isinstance(other, Timestamp): - return Duration(micros=self.micros - other.micros) + diff_nanos = ( + self._total(Timestamp.NANOS_PRECISION) - + other._total(Timestamp.NANOS_PRECISION)) + if diff_nanos % 1000 != 0: + raise ValueError( + 'The difference of %r and %r has sub-microsecond precision, ' + 'which Duration cannot represent. Truncate the operands with ' + 'to_precision(6, allow_lossy_conversion=True) first.' % + (self, other)) + return Duration(micros=diff_nanos // 1000) other = Duration.of(other) - return Timestamp(micros=self.micros - other.micros) + precision = max(self._precision, Timestamp.MICROS_PRECISION) + return Timestamp( + subseconds=self._total(precision) - + other.micros * _POW_10[precision - Timestamp.MICROS_PRECISION], + precision=precision) def __mod__(self, other: DurationTypes) -> 'Duration': other = Duration.of(other) - return Duration(micros=self.micros % other.micros) + remainder_nanos = self._total(Timestamp.NANOS_PRECISION) % ( + other.micros * 1000) + if remainder_nanos % 1000 != 0: + raise ValueError( + 'The remainder of %r modulo %r has sub-microsecond precision, ' + 'which Duration cannot represent. Truncate this timestamp with ' + 'to_precision(6, allow_lossy_conversion=True) first.' % (self, other)) + return Duration(micros=remainder_nanos // 1000) MIN_TIMESTAMP = Timestamp( @@ -361,8 +559,10 @@ def __float__(self) -> float: def __eq__(self, other: object) -> bool: # Allow comparisons between Duration and Timestamp values. - if isinstance(other, (Duration, Timestamp)): + if isinstance(other, Duration): return self.micros == other.micros + elif isinstance(other, Timestamp): + return self.micros * 1000 == other.nanos elif isinstance(other, (int, float)): return self.micros == Duration.of(other).micros else: @@ -371,8 +571,9 @@ def __eq__(self, other: object) -> bool: def __lt__(self, other: TimestampDurationTypes) -> bool: # Allow comparisons between Duration and Timestamp values. - if not isinstance(other, Timestamp): - other = Duration.of(other) + if isinstance(other, Timestamp): + return self.micros * 1000 < other.nanos + other = Duration.of(other) return self.micros < other.micros def __gt__(self, other: TimestampDurationTypes) -> bool: @@ -385,7 +586,9 @@ def __ge__(self, other: TimestampDurationTypes) -> bool: return not self < other def __hash__(self) -> int: - return hash(self.micros) + # Timestamps hash on their total nanoseconds. Hash equivalently so that + # a Duration and Timestamp that compare equal also hash equal. + return hash(self.micros * 1000) def __neg__(self) -> 'Duration': return Duration(micros=-self.micros) diff --git a/sdks/python/apache_beam/utils/timestamp_test.py b/sdks/python/apache_beam/utils/timestamp_test.py index f8d6cfdeafee..e1d120da471f 100644 --- a/sdks/python/apache_beam/utils/timestamp_test.py +++ b/sdks/python/apache_beam/utils/timestamp_test.py @@ -172,11 +172,13 @@ def test_from_proto(self): expected_ts = Timestamp(seconds=1234, micros=56) self.assertEqual(actual_ts, expected_ts) - def test_from_proto_fails_with_truncation(self): - # TODO(https://github.com/apache/beam/issues/19922): Better define - # timestamps. - with self.assertRaises(ValueError): - Timestamp.from_proto(timestamp_pb2.Timestamp(seconds=1234, nanos=56789)) + def test_from_proto_with_sub_micro_nanos(self): + # Sub-microsecond protos produce a nanosecond-precision Timestamp + # instead of losing precision (or raising, as this method used to). + actual_ts = Timestamp.from_proto( + timestamp_pb2.Timestamp(seconds=1234, nanos=56789)) + self.assertEqual(actual_ts.precision(), Timestamp.NANOS_PRECISION) + self.assertEqual(actual_ts.nanos, 1234 * 10**9 + 56789) def test_to_proto(self): ts = Timestamp(seconds=1234, micros=56) @@ -193,6 +195,181 @@ def test_equality(self): self.assertTrue(max_val >= min_val, "%s >= %s" % (max_val, min_val)) +class TimestampPrecisionTest(unittest.TestCase): + def test_constructor(self): + ts = Timestamp(seconds=1234, subseconds=123456789, precision=9) + self.assertEqual(ts.seconds(), 1234) + self.assertEqual(ts.subseconds(), 123456789) + self.assertEqual(ts.precision(), 9) + self.assertEqual(ts.nanos, 1234123456789) + + # Subseconds overflowing a second carry into seconds. + ts = Timestamp(seconds=1, subseconds=1500, precision=3) + self.assertEqual(ts.seconds(), 2) + self.assertEqual(ts.subseconds(), 500) + + # Negative timestamps floor seconds so subseconds stay non-negative. + ts = Timestamp(seconds=-2, subseconds=500000, precision=6) + self.assertEqual(ts.seconds(), -2) + self.assertEqual(ts.subseconds(), 500000) + self.assertEqual(ts, Timestamp(-1.5)) + + def test_constructor_validation(self): + with self.assertRaises(ValueError): + Timestamp(0, 0, precision=10) + with self.assertRaises(ValueError): + Timestamp(0, 0, precision=-1) + with self.assertRaises(TypeError): + Timestamp(0, 0, precision=6.0) # type: ignore[arg-type] + # micros is an alias for subseconds at the default precision only. + with self.assertRaises(ValueError): + Timestamp(0, subseconds=1, micros=1) + with self.assertRaises(ValueError): + Timestamp(0, precision=9, micros=1) + self.assertEqual(Timestamp(1, micros=500000), Timestamp(1.5)) + + def test_default_precision_is_micros(self): + self.assertEqual(Timestamp(1.5).precision(), Timestamp.MICROS_PRECISION) + self.assertEqual(Timestamp.now().precision(), Timestamp.MICROS_PRECISION) + + def test_equality_across_precisions(self): + self.assertEqual( + Timestamp(seconds=1, subseconds=500, precision=3), + Timestamp(seconds=1, subseconds=500000000, precision=9)) + self.assertEqual( + hash(Timestamp(seconds=1, subseconds=500, precision=3)), + hash(Timestamp(seconds=1, subseconds=500000000, precision=9))) + # An equal Duration must hash equal too. + self.assertEqual(hash(Timestamp(micros=5)), hash(Duration(micros=5))) + self.assertNotEqual( + Timestamp(seconds=1, subseconds=500000001, precision=9), + Timestamp(seconds=1, subseconds=500000, precision=6)) + + def test_comparison(self): + self.assertLess( + Timestamp(seconds=1, subseconds=500000000, precision=9), + Timestamp(seconds=1, subseconds=500001, precision=6)) + self.assertGreater( + Timestamp(seconds=1, subseconds=500000001, precision=9), + Timestamp(seconds=1, subseconds=500000, precision=6)) + self.assertLess(Timestamp(2.1), Timestamp(3)) + self.assertGreater(Timestamp(3), Timestamp(2.1)) + + def test_str(self): + self.assertEqual('Timestamp(1.500)', str(Timestamp(1, 500, precision=3))) + self.assertEqual( + 'Timestamp(1.123456789)', str(Timestamp(1, 123456789, precision=9))) + self.assertEqual( + 'Timestamp(-1.500000000)', str(Timestamp(-1.5, precision=9))) + self.assertEqual('Timestamp(1)', str(Timestamp(1, precision=9))) + self.assertEqual('Timestamp(1)', str(Timestamp(1, precision=0))) + + def test_precision_conversion(self): + ts = Timestamp(seconds=1, subseconds=123456789, precision=9) + self.assertIs(ts.to_precision(9), ts) + # Lossless upward conversion. + up = Timestamp(1.5).to_precision(9) + self.assertEqual(up.precision(), 9) + self.assertEqual(up, Timestamp(1.5)) + # Lossy downward conversion requires explicit permission and floors. + with self.assertRaises(ValueError): + ts.to_precision(6) + truncated = ts.to_precision(6, allow_lossy_conversion=True) + self.assertEqual(truncated.precision(), 6) + self.assertEqual(truncated.micros, 1123456) + # A lossless downward conversion doesn't require the flag. + self.assertEqual(Timestamp(1.5, precision=9).to_precision(3).precision(), 3) + # Truncation of negative timestamps floors towards negative infinity. + negative = Timestamp( + seconds=-1, precision=9).predecessor().to_precision( + 6, allow_lossy_conversion=True) + self.assertEqual(negative, Timestamp(-1) - Duration(micros=1)) + + def test_micros_guard_rail(self): + ts = Timestamp(seconds=1, subseconds=123456789, precision=9) + with self.assertRaises(ValueError): + _ = ts.micros + self.assertEqual(ts.nanos, 1123456789) + self.assertEqual(Timestamp(1, 500, precision=3).micros, 1500000) + + def test_predecessor_successor(self): + ts = Timestamp(seconds=10, precision=9) + self.assertEqual(ts.predecessor().nanos, 10 * 10**9 - 1) + self.assertEqual(ts.successor().nanos, 10 * 10**9 + 1) + self.assertEqual(ts.predecessor().precision(), 9) + self.assertEqual(ts.successor().seconds(), 10) + self.assertEqual(ts.predecessor().seconds(), 9) + # Micros-precision timestamps keep their historical 1-micro step. + self.assertEqual(Timestamp(10).predecessor().micros, 10 * 10**6 - 1) + + def test_to_utc_datetime_guard_rail(self): + ts = Timestamp(seconds=1234, subseconds=123456789, precision=9) + with self.assertRaises(ValueError): + ts.to_utc_datetime() + dt = ts.to_utc_datetime(allow_lossy_conversion=True) + self.assertEqual(dt.microsecond, 123456) + # Timestamps at or below microsecond precision convert freely. + Timestamp(1.5).to_utc_datetime() + + def test_to_rfc3339_is_lossless(self): + ts = Timestamp(seconds=1458343379, subseconds=123456789, precision=9) + self.assertEqual(ts.to_rfc3339(), '2016-03-18T23:22:59.123456789Z') + self.assertEqual(Timestamp.from_rfc3339(ts.to_rfc3339()), ts) + # Whole seconds have no fractional digits, as before. + self.assertEqual( + Timestamp(1458343379, precision=9).to_rfc3339(), '2016-03-18T23:22:59Z') + + def test_from_rfc3339_with_nanos(self): + ts = Timestamp.from_rfc3339('2016-03-18T23:22:59.123456789Z') + self.assertEqual(ts.precision(), 9) + self.assertEqual(ts.subseconds(), 123456789) + # Precision matches the number of fractional digits (above 6). + ts = Timestamp.from_rfc3339('2016-03-18T23:22:59.1234567Z') + self.assertEqual(ts.precision(), 7) + self.assertEqual(ts.subseconds(), 1234567) + self.assertEqual(Timestamp.from_rfc3339(ts.to_rfc3339()), ts) + self.assertEqual( + Timestamp.from_rfc3339('2016-03-18T23:22:59.123Z').precision(), 6) + with self.assertRaises(ValueError): + Timestamp.from_rfc3339('2016-03-18T23:22:59.1234567891Z') + + def test_from_rfc3339_with_comma_separator(self): + # ISO 8601 also allows ',' as the decimal separator. + ts = Timestamp.from_rfc3339('2016-03-18T23:22:59,123456789Z') + self.assertEqual( + ts, Timestamp.from_rfc3339('2016-03-18T23:22:59.123456789Z')) + self.assertEqual(ts.subseconds(), 123456789) + + def test_proto_round_trip_with_nanos(self): + ts = Timestamp(seconds=1234, subseconds=123456789, precision=9) + self.assertEqual(Timestamp.from_proto(ts.to_proto()), ts) + # Negative timestamps use non-negative proto nanos. + ts = Timestamp(-1.5, precision=9).predecessor() + proto = ts.to_proto() + self.assertEqual(proto.seconds, -2) + self.assertEqual(proto.nanos, 499999999) + self.assertEqual(Timestamp.from_proto(proto), ts) + + def test_arithmetic_preserves_precision(self): + ts = Timestamp(seconds=1, subseconds=123456789, precision=9) + self.assertEqual((ts + 1).nanos, ts.nanos + 10**9) + self.assertEqual((ts + 1).precision(), 9) + self.assertEqual((ts - Duration(micros=1)).nanos, ts.nanos - 1000) + # Sub-micros precision arithmetic results are widened to micros. + self.assertEqual((Timestamp(1, 500, precision=3) + 0.5).precision(), 6) + + def test_duration_arithmetic_guard_rail(self): + ts = Timestamp(seconds=1, subseconds=123456789, precision=9) + # Exact differences are representable as a (micros) Duration. + self.assertEqual(ts - (ts - Duration(micros=3)), Duration(micros=3)) + self.assertEqual(ts + Duration(micros=5) - ts, Duration(micros=5)) + # Sub-microsecond differences are not. + with self.assertRaises(ValueError): + _ = ts - ts.predecessor() + with self.assertRaises(ValueError): + _ = ts % Duration(seconds=1) + + class DurationTest(unittest.TestCase): def test_of(self): interval = Duration(123)