[Python] Bound Watch state with a timestamp cursor#39090
Conversation
Add an experimental Watch transform that watches a growing set of outputs per input element. Watch.growth_of(poll_fn) runs a periodic poll loop as a splittable DoFn and emits an unbounded PCollection of (input, output) pairs. Each process() performs one poll round and self-checkpoints via defer_remainder. New outputs are deduplicated with a stable 128-bit blake2b hash of the encoded output, and a manual watermark estimator advances per poll. Per-input termination supports never() and after_total_of(); polling also stops when a poll returns PollResult.complete(). The DoFn is its own RestrictionProvider, and restriction state serializes through a tagged GrowthState coder. Tests cover termination conditions, coder round-trips, the restriction tracker claim/checkpoint/dedup logic, and DirectRunner end-to-end runs.
Follow-up to the experimental Watch transform that documents the event-time contract, makes late data observable, and adds an opt-in O(1) bound on the per-input state. Watermark/event-time: - Document the event-time/watermark contract on the module and PollResult. - Warn (throttled) when an output is emitted behind the current watermark, so out-of-order late data is observable instead of silently dropped, and point to PollResult.with_watermark for out-of-order sources. Scalability: - Add opt-in Watch.with_timestamp_cursor(): dedup by a high-water-mark timestamp instead of by value identity. Watch keeps only the greatest event time emitted per input and emits the polled outputs strictly past it, so the per-input state is a single timestamp that never grows and the poll result is not hashed, giving O(1) state and per-checkpoint encoding. For sources whose outputs carry strictly increasing event-time timestamps; an output at or below the cursor is treated as already seen, so out-of-order or equal-timestamp outputs are dropped (use the default exact dedup otherwise). - Reuse the parent dedup map on idle rounds to drop the O(N) per-round copy. Tests cover the watermark contract, the out-of-order late warning, cursor emit-after-high-water-mark, O(1) state, relist-once, complete, cursor coder round-trip, and end-to-end cursor dedup. 26 tests and 5 subtests pass.
Eliaaazzz
force-pushed
the
python-watch-timestamp-cursor
branch
from
85476e4 to
71d6282
Compare
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This PR introduces a more scalable deduping strategy for the Watch transform in the Python SDK. By allowing users to opt into a timestamp-based cursor, the transform can now handle high-cardinality sources with O(1) state growth, avoiding the performance degradation associated with the default hash-based deduping set. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an experimental Watch transform for the Python SDK, which continuously polls and watches a growing set of outputs for each input element. It supports exact deduplication using 128-bit hashes or O(1) state deduplication using a high-water-mark timestamp cursor. The feedback suggests two improvements: first, ensuring that timestamps are normalized to Timestamp objects before comparison in cursor mode to prevent TypeError exceptions in Python 3; second, converting the input outputs iterable to a list during normalization to safely handle single-use generators or iterators.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| new_outputs = [ | ||
| output for output in result.outputs | ||
| if cursor is None or output.timestamp > cursor | ||
| ] |
There was a problem hiding this comment.
In _GrowthRestrictionTracker.try_claim, when self._cursor_mode is enabled, the code compares output.timestamp > cursor. However, output.timestamp can be a raw float or integer if the user directly instantiated TimestampedValue without using Timestamp objects (as TimestampedValue accepts int or float for its timestamp). Since cursor is a Timestamp object (or None), comparing a Timestamp with a float or integer will raise a TypeError in Python 3. To prevent comparison errors, normalize output.timestamp to a Timestamp object using Timestamp.of(output.timestamp) before comparison.
| new_outputs = [ | |
| output for output in result.outputs | |
| if cursor is None or output.timestamp > cursor | |
| ] | |
| cursor = restriction.cursor | |
| new_outputs = [ | |
| output for output in result.outputs | |
| if cursor is None or Timestamp.of(output.timestamp) > cursor | |
| ] |
| for output in outputs: | ||
| if isinstance(output, TimestampedValue): | ||
| normalized.append(output) | ||
| else: | ||
| normalized.append(TimestampedValue(output, default_ts)) |
There was a problem hiding this comment.
When iterating over outputs, if the user passes a single-use generator or iterator, iterating over it to normalize elements will consume it. Since outputs is iterated again when constructing the PollResult or during normalization, this can lead to empty outputs or unexpected behavior. Consider converting outputs to a list or tuple first before normalization.
| for output in outputs: | |
| if isinstance(output, TimestampedValue): | |
| normalized.append(output) | |
| else: | |
| normalized.append(TimestampedValue(output, default_ts)) | |
| outputs = list(outputs) | |
| normalized = [] | |
| for output in outputs: | |
| if isinstance(output, TimestampedValue): | |
| normalized.append(output) | |
| else: | |
| normalized.append(TimestampedValue(output, default_ts)) |
|
Assigning reviewers: R: @jrmccluskey for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
1 participant
Stacked on #39023 (the Watch transform MVP); the first commit belongs to that PR, so the change to review is the second commit. Relates to #18459 (the unbounded
completeddedup set).By default Watch dedups by value identity and keeps one hash per distinct output, so the per-input state grows without bound. This adds an opt-in
Watch.with_timestamp_cursor()that dedups by a high-water-mark timestamp instead: Watch keeps only the greatest event time it has emitted for an input and emits the polled outputs strictly past it, then advances the cursor. No hash set is kept and the poll result is not hashed, so the per-input state and per-checkpoint encoding are O(1) regardless of how many outputs the input produces. It is for sources whose outputs carry strictly increasing event-time timestamps; an output at or below the cursor is treated as already seen, so the default exact dedup remains for arbitrary-relisting or out-of-order sources. This also documents the event-time/watermark contract and adds a throttled late-emission warning.Testing: 28 unit tests and 5 subtests on the in-memory DirectRunner; yapf 0.43.0, isort 7.0.0, and pylint clean. Load tested on Dataflow Runner v2 (us-east1) against the default hash dedup, 4 inputs emitting 15000 outputs per 1s round over a 240s budget: the cursor processed 6,825,000 outputs versus the default's 1,650,000 in the same budget (4.1x), because the default re-serializes its growing dedup set on every checkpoint (it reached about 300-495K entries per input) while the cursor writes one timestamp; the gap widens with runtime. Both stayed exactly-once and reached JOB_STATE_DONE. Jobs on apache-beam-testing: cursor
2026-06-24_08_45_05-17501676526076065162, default2026-06-24_08_45_23-5060488636914380336.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.