Repository navigation
[Bug]: Latest.Globally() issues in global window #35934
Description
What happened?
I was trying to repro an issue with slowly updating side inputs, and I found that the following code only logs one element:
pipeline = beam.Pipeline(options=options)
from apache_beam.transforms import core
start_timestamp = time.time() # start timestamp of the periodic impulse
main_input_fire_interval = 60 # interval in seconds at which the main input PCollection is emitted.
side_input_fire_interval = 60 # interval in seconds at which the side input PCollection is emitted.
def add_timestamp(element, timestamp=core.DoFn.TimestampParam):
return [(element, timestamp)]
side_input = (
pipeline
| "SideInputImpulse" >> PeriodicImpulse(
start_timestamp=start_timestamp,
fire_interval=main_input_fire_interval)
| "Window" >> WindowInto(
GlobalWindows(),
# Define the trigger. Since the Global Window never # closes, we must tell the runner WHEN to emit a result. This trigger
# fires repeatedly for every new element that arrives.
trigger=Repeatedly(AfterCount(1)),
# Define the accumulation mode. DISCARDING tells the # runner to forget old values after a trigger fires. This prevents # state from growing indefinitely and ensures Latest.Globally() only
# considers the most recent element.
accumulation_mode=AccumulationMode.DISCARDING,
)
| 'Add timestamp' >> core.ParDo(add_timestamp)
)
side_input | "LocalLatest" >> Latest.Globally() | "show latest" >> beam.Map(logging.info)
However, if I change Latest.Globally() to Latest.Globally().without_defaults() it repeatedly yields a stream of elements as expected based on the trigger. I observed this on Dataflow (this pattern isn't supported on Prism, and I haven't validated it on other runners). As best I can tell, it is getting hung up on side inputs.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
- Component: Python SDK
- Component: Java SDK
- Component: Go SDK
- Component: Typescript SDK
- Component: IO connector
- Component: Beam YAML
- Component: Beam examples
- Component: Beam playground
- Component: Beam katas
- Component: Website
- Component: Infrastructure
- Component: Spark Runner
- Component: Flink Runner
- Component: Samza Runner
- Component: Twister2 Runner
- Component: Hazelcast Jet Runner
- Component: Google Cloud Dataflow Runner
Activity
ankitkumarrain commentedon Oct 1, 2025
Hello everyone here , i have written code is it mergable . import apache_beam as beam
from apache_beam.transforms import core
from apache_beam.transforms.window import GlobalWindows, WindowInto, Repeatedly, AfterCount, AccumulationMode
from apache_beam.transforms.periodicsequence import PeriodicImpulse
from apache_beam.transforms.combiners import Latest
import logging
import time
pipeline = beam.Pipeline(options=options)
start_timestamp = time.time()
main_input_fire_interval = 60
side_input_fire_interval = 60
def add_timestamp(element, timestamp=core.DoFn.TimestampParam):
return (element, timestamp)
side_input = (
pipeline
| "SideInputImpulse" >> PeriodicImpulse(
start_timestamp=start_timestamp,
fire_interval=main_input_fire_interval,
)
| "SideInputWindow" >> WindowInto(
GlobalWindows(),
trigger=Repeatedly(AfterCount(1)),
accumulation_mode=AccumulationMode.DISCARDING,
)
| "Add timestamp" >> core.ParDo(add_timestamp)
)
(
side_input
| "LocalLatest" >> Latest.Globally().without_defaults()
| "Show latest" >> beam.Map(logging.info)
)
paragkharad2007-star commentedon Nov 9, 2025
actually This difference is caused by how Latest.Globally() interacts with triggers and default values.
shunping commentedon Nov 18, 2025
I ran into a similar problem and tested a bit on FnAPI runner and Prism.
FnAPI Runner
Latest.Globally()orLatest.Globally().without_defaults()Prism
Latest.Globally()is not supported yet as it has a sideinput with trigger. ([prism] Triggers and Panes and Windowing Strategies #31438 (comment))Latest.Globally().without_defaults()can yield a stream of elements as expected.scwhittle commentedon Dec 16, 2025
I think that this could be due to the internal caching of global side-inputs within the python sdk. I previously added an experiment
disable_global_windowed_args_cachingto disable this, it would be interesting to see if using that matches your expectations. There was a performance impact so that wasn't enabled in all cases. The existing issue is #28776I had previously put together #31356 (reopened as #37123) to fix this by clearing per-bundle but it relied on the sdk cache which was disabled at the time and I lost track of it. Resurrecting that may be the fix to this bug.
[edit]: AFAICT the python sdk state cache is still disabled by default.
github-actions commentedon May 15, 2026
This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.
github-actions commentedon Jun 15, 2026
This issue has been closed due to lack of activity. If you think that is incorrect, you can reopen the issue at any time.
scwhittle commentedon Sep 1, 2026
@damccorm what is the status of the python state cache? If it is enabled, I think that the approach of #37123 is valid. For batch performance shouldn't be affected as there are long-lived bundles. For streaming, the value used per-bundle will be fixed and the cache would be used across bundles. If the cache is not enabled, the side input would have to be fetched from the runner on each bundle.
damccorm commentedon Sep 11, 2026
Missed this, sorry -
beam/sdks/python/apache_beam/options/pipeline_options.py
Line 1566 in 8b3370c
So I don't think this problem is caused by caching
scwhittle commentedon Sep 12, 2026
Sorry I was unclear. The problem is not due to the state cache itself but is due to caching of side inputs within dofninvokers (originally motivated by performance I believe). See files in #37123. I had a proposed fix to clear the DoFnInvoker cached window args between bundles, however without the overall state cache being enabled this would result in re-fetching side inputs from the runner for each bundle. This might have a negative impact on streaming performance in particular where bundles are generally short-lived.
One option could be to just perform such clearing within dofninvoker if the state cache is enabled, and recommend customers enable that flag if they want to observe updated side inputs. Or we could rework the caching within dofninvoker to be shared across dofninvokers and refreshed periodically.
damccorm commentedon Sep 14, 2026
Isn't this necessary for correctness, though? It seems like what we are doing today is basically just caching with bad refresh guarantees.
We could consider reenabling the cache by default, though we'd need to solve #30360
scwhittle commentedon Sep 15, 2026
I believe there is not real guarantee about side-input refreshes in the beam model it is just best-effort in the dataflow runner by sprinkling cache timeouts around to handle side-inputs refreshing. A side-input observed for a window is always "valid" for that window from the model. @kennknowles to keep me honest.
I originally just removed this dofninvoker cache but that led to 25% regression #31356 (comment)
what about we:
Then those who want refreshing side inputs can either enable the option or enable the cache. And if we roll the cache out completely everyone will pick up the improvement.
damccorm commentedon Sep 15, 2026
This SGTM, with a couple of notes:
We should definitely make this something that users can disable as an escape hatch for us if we regress this use case somehow.
We should also make sure that this ends up in CHANGES.md
I'm not sure how much this matters - if the expected behavior is unclear to some of us, and Latest.Globally doesn't actually give the latest element, I think its a bug.
Can we document Latest.globally as part of this work?
scwhittle commentedon Sep 15, 2026
Ack, I meant to say "change the default behavior if the state cache is enabled". This would still let users manually override this option separately.
I agree it should be fixed too, which is why I put together those changes originally.
But I think
Latest.globally()is perhaps a confusing name here, it is how to calculate the side input (latest value, it could just as easily be aCount.globally()). The refreshing part is due to the triggering re-evaluating the side-input. And there is not a guarantee on main inputs observing this updated triggered value. Even once we fix this there will be some latency waiting for cache timeouts, it won't be the absolutelatestvalue. For example, if the side-input triggering specifies to refresh it every second, it would likely take at least a minute. But this is still better than stuck on the original value forever :)I think a beam-native broadcast variable would be a better fit than refreshing global side inputs we currently have.