Round float seconds to nearest subsecond unit in Timestamp constructor

sdks/python/apache_beam/utils/timestamp.py

@@ -64,7 +64,7 @@ class Timestamp(object):
   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.
+  to ``precision`` digits, rounded to the nearest subsecond unit.
 
   Lossy conversion operations will throw an error unless
   ``allow_lossy_conversion=True`` is specified (e.g. see ``to_utc_datetime``).
@@ -107,7 +107,10 @@ def __init__(
             'use subseconds instead.' % precision)
       subseconds = micros
     self._precision = precision
-    total = int(seconds * _POW_10[precision]) + int(subseconds)
+    # Round (rather than truncate) so that float inputs like 2.000002, which
+    # cannot be represented exactly in binary, land on the nearest subsecond
+    # unit instead of one unit low.
+    total = round(seconds * _POW_10[precision]) + int(subseconds)
     self._seconds, self._subseconds = divmod(total, _POW_10[precision])
 
   def _total(self, precision: int) -> int:

sdks/python/apache_beam/utils/timestamp_test.py

@@ -214,6 +214,22 @@ def test_constructor(self):
     self.assertEqual(ts.subseconds(), 500000)
     self.assertEqual(ts, Timestamp(-1.5))
 
+  def test_constructor_float_rounds_to_nearest(self):
+    # Float seconds that are not exactly representable must round to the
+    # nearest subsecond unit rather than truncating one unit low.
+    self.assertEqual(Timestamp(2.000002).micros, 2000002)
+    self.assertEqual(Timestamp(1.000001).micros, 1000001)
+    self.assertEqual(Timestamp.of(2.000002).micros, 2000002)
+    # Rounding is symmetric about zero (truncation was not).
+    self.assertEqual(Timestamp(-2.000002).micros, -2000002)
+    self.assertEqual(Timestamp(-2.000002).micros, -Timestamp(2.000002).micros)
+    # Exact values are unaffected.
+    self.assertEqual(Timestamp(1.5).micros, 1500000)
+    self.assertEqual(Timestamp(2).micros, 2000000)
+    # Rounding at finer precisions.
+    ts = Timestamp(1.234567895, precision=9)
+    self.assertEqual(ts.nanos, 1234567895)
+
   def test_constructor_validation(self):
     with self.assertRaises(ValueError):
       Timestamp(0, 0, precision=10)