[BEAM-8657] Not doing Combiner lifting for data-driven triggers

• Type: Bug
• Status: Triage Needed
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.18.0
• Component/s: sdk-py-core
• Assignee: Robert Bradshaw
• Reporter: wendy liu
• Created: 13/Nov/19 23:38
• Updated: 13/Apr/23 10:58
• Resolved: 15/Nov/19 19:47

Description

The test below will always produce empty output, due to combiner lifting, which already combines all input values of one shard into one before grouping. To fix this, we shall not do Combiner lifting for the data-driven triggers.

l = [window.TimestampedValue(('a', 1), 1),
window.TimestampedValue(('b', 3), 3),
window.TimestampedValue(('a', 2), 2),
window.TimestampedValue(('a', 5), 5),]

result = (p | Map(lambda x : x) | 'window' >> beam.WindowInto(FixedWindows(6), trigger=trigger.AfterCount(2), accumulation_mode=trigger.AccumulationMode.DISCARDING) | beam.CombinePerKey(combine.Largest(1)))

Issue Links

• links to: GitHub Pull Request #10098
• links to: GitHub Pull Request #10109