Thanks for confirming. Since it is unexpected behavior, I shall look into jira if it is already on radar, if not, will create one.
On Mon, Nov 11, 2019 at 6:11 PM Robert Bradshaw <rober...@google.com> wrote: > The END_OF_WINDOW is indeed 9.999999 (or, in Java, 9.999000), but the > results for LATEST and EARLIEST should be 9 and 0 respectively. > > On Mon, Nov 11, 2019 at 5:34 PM Ruoyun Huang <ruo...@google.com> wrote: > > > > Hi, Folks, > > > > I am trying to understand the behavior of TimestampCombiner. I have > a test like this: > > > > class TimestampCombinerTest(unittest.TestCase): > > > > def test_combiner_latest(self): > > """Test TimestampCombiner with LATEST.""" > > options = PipelineOptions() > > options.view_as(StandardOptions).streaming = True > > p = TestPipeline(options=options) > > > > 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)) > > > > class RecordFn(beam.DoFn): > > def process(self, > > elm=beam.DoFn.ElementParam, > > ts=beam.DoFn.TimestampParam): > > yield (elm, ts) > > > > records = (main_stream | beam.ParDo(RecordFn())) > > > > expected_window_to_elements = { > > window.IntervalWindow(0, 10): [ > > (('k', 500), Timestamp(9)), > > ], > > } > > > > assert_that( > > records, > > equal_to_per_window(expected_window_to_elements), > > use_global_window=False, > > label='assert per window') > > > > p.run() > > > > > > I expect the result to be following (based on various TimestampCombiner > strategy): > > LATEST: (('k', 500), Timestamp(9)), > > EARLIEST: (('k', 500), Timestamp(0)), > > END_OF_WINDOW: (('k', 500), Timestamp(10)), > > > > The above outcome is partially confirmed by Java side test : [1] > > > > > > However, from beam python, the outcome is like this: > > LATEST: (('k', 500), Timestamp(10)), > > EARLIEST: (('k', 500), Timestamp(10)), > > END_OF_WINDOW: (('k', 500), Timestamp(9.99999999)), > > > > What did I miss? what should be the right expected behavior? or this > looks like a bug? > > > > [1]: > https://github.com/apache/beam/blob/master/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java#L390 > > > > Cheers, > > > -- ================ Ruoyun Huang