[Python] Refactor MatchContinuously to use Watch#39352
Draft
Eliaaazzz wants to merge 4 commits into
Draft
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.
- Replace growth_of and the with_* builders with constructor arguments. - Infer the output coder from PollFn.default_output_coder() or a PollResult[V] return annotation via the coder registry; require determinism only of the dedup key coder, converting it GroupByKey-style with as_deterministic_coder, and add output_key_fn/output_key_coder mirroring Java's outputKeyFn/outputKeyCoder. - Rework the restriction tracker to Java's design: try_claim takes the (PollResult, termination_state) round and validates it against the restriction; try_split derives the replay primary and merged residual. - Match Java's watermark handling: seed the estimator from the input timestamp and advance it to the poll watermark or earliest new output when resuming.
Eliaaazzz
force-pushed
the
agent/match-continuously-watch
branch
from
July 16, 2026 14:23
bb35430 to
c7364c5
Compare
Eliaaazzz
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR is stacked on #39023 (
[Python] Add Watch transform with growth_of polling SDF). Please review #39023 first; this PR contains theFileIO.MatchContinuouslyintegration on top of that Watch transform.This refactors the deduplicated
FileIO.MatchContinuouslypath to useWatch.MatchContinuouslynow usesWatchwhenhas_deduplication=True. Normal deduplication is still by file path. Whenmatch_updated_files=True, the poll function timestamps eachFileMetadataoutput withlast_updated_in_seconds, andWatch(use_timestamp=True)emits the same path again when that timestamp changes. Thehas_deduplication=Falsepath remainsPeriodicImpulseplusMatchAllso every poll can re-emit all matches.This PR also adds the small Watch API hook needed by the FileIO integration:
use_timestamp, an opt-in dedup mode that includes output event time in the dedup key.TerminationCondition.on_poll_complete(now, state), so the FileIO stop window can count only polls at or afterstart_timestamp.Validation:
python -m pytest -q sdks/python/apache_beam/io/watch_test.py sdks/python/apache_beam/io/fileio_test.py::MatchContinuouslyTestgit diff --check origin/python-watch-transform..HEADThank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. Stacked on [Python] Add Watch transform with growth_of polling SDF #39023.CHANGES.mdwith noteworthy changes. Not updated; this is an experimental Python SDK transform/refactor.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
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.