Fix flakiness in fuzzy tests and support TimestampTypes in PeriodicImpulse.

[shunping]

A follow-up on #35412.

I've identified a flaky test in the test_fuzzy_interval for periodic impulse, which you can see in this failed test run. The test fails intermittently because the periodic impulse output is either missing an element or has an extra one.

I can reproduce this failure about once every 50-100 runs. The failure rate increases significantly if the interval is set to 1 microsecond.

Here's a comparison of a normal and a failed run after I added some debugging logs:

• Normal run:

  WARNING:root:random seed=1751056034036
  start=1751056297.165744 data_duration=7.9e-05 interval=1e-06, end=1751056297.165824
  input: 1751056297165744, 1751056297165824, 1
  WARNING:root:outputs=80.000000, start=1751056297165744.000000, start_micros=1751056297165744.000000, 
    end=1751056297165824.000000, end_micros=1751056297165824.000000, 
  end-start=80, interval=0.000001, delta_micros=80, interval_micros=1, delta_micros/interval=80.000000
  

• Failed run:

  WARNING:root:random seed=1751056034036
  start=1751056301.35949 data_duration=7.9e-05 interval=1e-06, end=1751056301.3595698
  input: 1751056301359490, 1751056301359569, 1
  WARNING:root:outputs=79.000000, start=1751056301359490.000000, start_micros=1751056301359490.000000, 
    end=1751056301359569.000000, end_micros=1751056301359569.000000, 
  end-start=79, interval=0.000001, delta_micros=79, interval_micros=1, delta_micros/interval=79.000000
  ERROR:apache_beam.runners.common:Failed assert: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79] == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78], missing elements [79] [while running 'assert_that/Match']
  

The root cause is a loss of precision with 64-bit floating-point numbers. As noted in the Wikipedia article on floating-point arithmetic, the precision is limited to about 15-17 decimal digits.

In the failing example, the start time is 1751056301.35949 (16 digits). However, the calculated end time is 1751056301.3595698 (17 digits). When converting the end time to microseconds (round(end * 1000000)), we get 1751056301359569 instead of the expected 1751056301359570. This rounding error causes delta_micros to be off by one, leading to the test failure.

This shows we can't rely on floating-point representations of start, end, and interval to accurately determine the number of elements to emit.

To fix this, I propose using integer-based timestamp arithmetic for all calculations. By performing these operations on microseconds directly, we can avoid floating-point inaccuracies and ensure the test's stability.

[shunping]

Looks like SlowlyChangingSideInputsTest.test_side_input_slow_update uses integer numbers in as start_timestamp and stop_timestamp when calling PeriodicImpulse.

beam/sdks/python/apache_beam/examples/snippets/snippets_test.py

Line 1428 in 4d58e6d

first_ts = math.floor(time.time()) - 30

beam/sdks/python/apache_beam/examples/snippets/snippets_test.py

Lines 1452 to 1454 in 4d58e6d

	pipeline, pipeline_result = snippets.side_input_slow_update(
	src_file_pattern, first_ts, last_ts, interval,
	sample_main_input_elements, main_input_windowing_interval)

beam/sdks/python/apache_beam/examples/snippets/snippets.py

Lines 1432 to 1433 in 4d58e6d

	| 'PeriodicImpulse' >> PeriodicImpulse(
	first_timestamp, last_timestamp, interval, True)

.

I suggest we support all TimestampTypes in PeriodicImpulse.

beam/sdks/python/apache_beam/utils/timestamp.py

Line 39 in 4d58e6d

TimestampTypes = Union[int, float, 'Timestamp']

@damccorm, WDYT?

[shunping]

remind me after tests pass

[github-actions]

Ok - I'll remind @shunping after tests pass

[github-actions]

All checks have passed: @shunping

[github-actions]

Assigning reviewers:

R: @liferoad for label python.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[damccorm]

Thanks, this LGTM