Details
-
Bug
-
Status: Resolved
-
P2
-
Resolution: Fixed
-
None
Description
When we have a TimestampValue on combine:
main_stream = (p | 'main TestStream' >> TestStream() .add_elements([window.TimestampedValue(('k', 100), 0)]) .add_elements([window.TimestampedValue(('k', 400), 9)]) .advance_watermark_to_infinity() | 'main windowInto' >> beam.WindowInto( window.FixedWindows(10), timestamp_combiner=TimestampCombiner.OUTPUT_AT_LATEST) | 'Combine' >> beam.CombinePerKey(sum)) The expect timestamp should be: LATEST: (('k', 500), Timestamp(9)), EARLIEST: (('k', 500), Timestamp(0)), END_OF_WINDOW: (('k', 500), Timestamp(10)), But current py streaming gives following results: LATEST: (('k', 500), Timestamp(10)), EARLIEST: (('k', 500), Timestamp(10)), END_OF_WINDOW: (('k', 500), Timestamp(9.99999999)), More details and discussions: https://lists.apache.org/thread.html/d3af1f2f84a2e59a747196039eae77812b78a991f0f293c717e5f4e1@%3Cdev.beam.apache.org%3E