[Python] Refactor MatchContinuously onto the Watch transform - #39461
[Python] Refactor MatchContinuously onto the Watch transform#39461Eliaaazzz wants to merge 15 commits into
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Assigning reviewers: R: @tvalentyn 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). |
|
R: @Abacn |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
Route MatchContinuously through Watch when deduplication is enabled. The polling loop and the set of already-matched file ids now live in the splittable DoFn restriction, replacing the per-key state DoFns. Because the matched ids are part of the restriction, a runner with checkpointing enabled restores them after a restart and does not reprocess files. The docstring is updated accordingly. Verified on Flink 1.20: after killing the TaskManager mid stream the job restored from a checkpoint and every file was still emitted exactly once. has_deduplication=False keeps the previous PeriodicImpulse behaviour.
1361ca1 to
3760698
Compare
registry.get_coder receives typing and native generic annotations such as tuple[str, float] unconverted and falls back to pickling. Watch now converts hints with convert_to_beam_type before the registry lookup, so MatchContinuously's annotated key functions infer the same StrUtf8Coder and TupleCoder the explicit settings supplied. Also trims the docstrings and comments this PR adds.
…ports Annotates _MatchContinuouslyPollFn with PollResult[FileMetadata], covers typing.Tuple key inference alongside the native form, reorders the third-party test imports, and drops the remaining Java references from test comments.
| restarted, already processed files will be reprocessed. Consider an alternate | ||
| technique, such as Pub/Sub Notifications | ||
| (https://cloud.google.com/storage/docs/pubsub-notifications) | ||
| file ids for every file the pattern has matched. With ``has_deduplication`` |
There was a problem hiding this comment.
Just a side note, with #39090 in we should be able to eliminate this comment
Matching continuously scales poorly, as it is stateful, and requires storing
There was a problem hiding this comment.
Follow up: now #39090 has been merged. We can introduce an option for FileIO.matchContinuously (timestamp_cursor), if set True, then it's backed by watch transform's timestamp_cursor mode
There was a problem hiding this comment.
One transition detail I would like to confirm with you while wiring this option up. When Watch switches from hash dedup to the cursor, it seeds the cursor from the greatest timestamp in the old completed map. In MatchContinuously those are poll times, while the cursor compares last-modified times, so the two are not the same domain. For now the option's doc says it needs a pipeline started fresh.
The seeding lives in the merged _cursor_of and has its own tests, so I did not change it here. Would you rather keep the seeding as it is, or reject the switch outright when a restriction carries hash state?
watch.py and watch_test.py return to master; the inference fix lands separately so it can make the release cut. The annotated key functions meanwhile fall back to the deterministic FastPrimitivesCoder form, which keeps dedup correct.
Opt-in timestamp_cursor=True backs deduplication with the Watch transform's cursor mode, so the restriction holds one timestamp rather than an id per matched file. The poll stamps each match with its last-modified time, which is what the cursor dedups on, and the watermark stays at the poll time. Those event times are floored to the millisecond, the resolution a runner keeps for element timestamps. A cursor taken from finer mtimes is persisted truncated and returns below the outputs it came from, matching every file again on the next poll. The class docstring no longer calls matching continuously stateful without qualification, since the cursor bounds the state, and the startup warning is skipped in that mode.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #39461 +/- ##
============================================
+ Coverage 55.96% 58.83% +2.87%
- Complexity 2272 7638 +5366
============================================
Files 1113 1663 +550
Lines 175101 216472 +41371
Branches 1458 6536 +5078
============================================
+ Hits 97988 127361 +29373
- Misses 74654 84400 +9746
- Partials 2459 4711 +2252
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
| # Floored to the millisecond a runner keeps for element timestamps, so the | ||
| # cursor compares against the same resolution it is persisted at. | ||
| micros = Timestamp.of(_mtime_of(metadata, 'timestamp_cursor')).micros | ||
| return Timestamp(micros=micros - micros % 1000) |
There was a problem hiding this comment.
Why need this special handling here? It should be as correct just pass micros. If there is a watermark issue, check comment below.
There was a problem hiding this comment.
You were right that fileio should just pass micros, and taking the flooring out surfaced the real problem a level down. Watch persists the cursor with TimestampCoder, which encodes milliseconds, so the cursor came back up to a millisecond below the outputs it was taken from and the next poll matched them all again. With the flooring gone and nothing else changed, test_timestamp_cursor_emits_files_modified_past_the_cursor fails with the first file emitted twice. The same rounding hit the checkpoint primary, so a replayed output could repeat its emission in a different window.
Commit 462d2f5 fixes both where they belong. The restriction now encodes microseconds at a fixed width, so the cursor state size test still holds, and the two envelope tags whose payload changed are retired rather than reused, since a millisecond payload is the same width and would otherwise decode to a wrong timestamp instead of failing. fileio then stamps Timestamp.of(mtime) as you asked, commit afa436c.
This PR is otherwise fileio only, and commit 462d2f5 changes watch.py, which is already merged. I put it here because the cursor dedup in MatchContinuously is not correct without it. If you would rather keep the separation you asked for earlier, I will move it into its own PR and rebase this one on top.
There was a problem hiding this comment.
Please do not introduce new timestamp coder, we already have many duplications in the code base.
Watch persists the cursor with TimestampCoder, which encodes milliseconds, so the cursor came back up to a millisecond below the outputs it was taken from and the next poll matched them all again
This is a good observation, and suggests an existing bug in timestamp_cursor implementation. There is a risk of a racing condition such that an upcoming element having same timestamp, it might either get dropped, or there are duplicates, depending on how do we round the timestamp.
I think "allowed_lateness" we have discussed earlier is the correct solution here. Instead of introducign a new coder (technically micro precision still has risk on sub-micro rounding), we should always put the elements already seen that having timestamp >= (cursor_in_millis_precision - allowed_lateness_duration) into the restriction. The default allowed_lateness is zero.
Then, on next poll, it automatically dedups the latest results last time already emitted, and can handle upcoming elements that have same timestamp
There was a problem hiding this comment.
Done, and the coder is gone. You were right that precision was the wrong lever: rounding only moves where the tie falls, it does not remove it.
Deduplication goes back to hashing the output key, and the cursor now bounds that state instead of standing in for it. A key is retired once the greatest emitted event time has moved allowed_lateness past it, so the restriction holds a trailing window rather than every key ever seen. allowed_lateness defaults to zero.
Two things this fixed beyond the tie. match_updated_files was silently ignored under timestamp_cursor, because the cursor branch skipped the key function entirely; both modes now share one path, so the option composes again. And the hash to cursor transition I asked about on the other thread no longer exists: a restriction resumed in cursor mode keeps the hashes its hash rounds recorded, so there is nothing to seed and no poll time read as a last-modified time.
The tie is covered at the Watch level and through MatchContinuously, and I checked both tests fail when the boundary goes back to a strict comparison, with the second file at the same last-modified time dropped silently. GCS reports to the millisecond, so two objects written in the same millisecond reach this.
One question. I put allowed_lateness on Watch only and did not surface it on MatchContinuously, since adding a parameter to a public IO transform seemed yours to decide. For a filesystem clock running behind the local one it is the knob that matters. Would you like it on MatchContinuously as well?
Commits 9df6713 and 9ce7c97.
There was a problem hiding this comment.
One cost of the retention window I should have called out when I proposed it. A key is retired by the event time it was recorded with, so a file modified after the cursor moved past it has nothing left to prove it was seen and is matched a second time, whatever match_updated_files says. The MatchContinuously test I added for the update case never advanced the cursor, so it stopped short of the retirement and read as a guarantee it is not.
Documented on the option and pinned with a Watch test rather than changed. Removing it means recording a key's current event time every round, not just the round it first appeared, which means carrying the already-seen outputs' timestamps through the claim. Happy to do that if you would rather the option not weaken match_updated_files. Commit 0c00e43.
There was a problem hiding this comment.
I put allowed_lateness on Watch only and did not surface it on MatchContinuously
This looks fine to me. It's fine filesystem clock and pipeline runtime clock has offset. If modified files timestamp is one-way, it's fine then
| ] | ||
| return PollResult.incomplete(outputs).with_watermark(now) | ||
| return PollResult.incomplete( | ||
| match_result.metadata_list, timestamp=now).with_watermark(now) |
There was a problem hiding this comment.
We need to consider watermark more carefully here:
There is chance that the remote file system clock and local machine has offset. We don't want to advance the watermark prematurely (if filesystem clock is slower than the machine running SDK harness)
if there is new element seen in this poll, watermark should be no later than min(max(poll timestamp), now); if there isn't new element seen in this poll, set to Now sounds fine.
There was a problem hiding this comment.
Adopted, with one deviation I would like your call on.
Under timestamp_cursor the poll now returns the newest last-modified time it matched, capped at the poll time, so a filesystem clock behind the local one holds the watermark back and one running ahead cannot carry it past now.
The deviation is the empty poll. Going to now there also advances past a lagging filesystem clock, and I could reproduce a file arriving behind the watermark after a single empty poll, so the poll now returns no watermark and leaves the estimator where it is. That makes the rule purely evidence based, at the cost of the watermark not moving while a directory is quiet. If you would rather have windows keep closing on idle I will put now back.
Commit afa436c, with three poll fn tests for the newest match, the cap, and the empty match.
There was a problem hiding this comment.
The adjusted logic could have stale watermark slowly updated sources. When a poll has no new result, watermark stuck. We have a similar issue for PeriodicImpulse: #39026
In FileIO.matchContinuously, are the timestamped elements on their mtime, or the poll time? This affects how watermark should be handled. My first comment assume they were timestamped on mtime, which introduces tricky scenarios. If it's actually on poll time, things would be simpler. Just advance watermark to now() everytime would suffice.
There was a problem hiding this comment.
They carry their last-modified time under timestamp_cursor, and the poll time otherwise.
That is not only a choice. The retention window compares an output's event time against the cursor, and a poll time would never grow stale, so every re-listed file would look recent and the state would never be bounded. The last-modified time is what lets the state stay bounded, so the simpler poll-time option is not open to us here.
The watermark is back to what you asked for: the newest last-modified time the poll matched, capped at the poll time, and the poll time when the poll matched nothing. I dropped the hold I had added, so a quiet directory no longer stalls the watermark, which was the shape of #39026 you pointed at.
That leaves the case you raised first, a filesystem clock behind the local one. With the watermark at the poll time, a file written while that clock lags arrives behind it. allowed_lateness now covers the deduplication half of that, and I have asked on the other thread whether it should be exposed on MatchContinuously too.
Commit 9ce7c97.
There was a problem hiding this comment.
Correction to what I said last round: a quiet directory did still stall. I only released the watermark on an empty match, and a directory that keeps the same files is not one. Every poll re-lists them, the newest last-modified time never moves, and the watermark sits there while the poll time runs away.
The hold now follows the evidence instead. A poll that turns up a last-modified time newer than any before it has just read the filesystem clock, so the watermark stops there, capped at the poll time. A poll that finds nothing newer takes the poll time. A directory being fed trails the filesystem clock throughout, a quiet one catches up within a poll.
That leaves a file written while the clock lags, in the interval after arrivals stop. allowed_lateness is the knob for it, which is the open question on the other thread.
Two poll fn tests; the release one fails on 9ce7c97 with the watermark still at the file's last-modified time. Commit 0c00e43.
TimestampCoder encodes milliseconds, so a persisted cursor returned up to a millisecond below the outputs it was taken from and the next poll handed them out again. A replayed checkpoint primary lost the same precision and repeated its emission at a different event time. Both now encode microseconds at a fixed width, and a payload of any other width is rejected rather than read as a smaller timestamp. The two envelope tags whose payload changed are retired rather than reused, since a millisecond payload is the width of a microsecond one and would otherwise decode to a wrong timestamp.
The poll stamps a match with Timestamp.of(mtime) now that the restriction keeps microseconds, so the millisecond flooring is gone. The watermark under timestamp_cursor tracks the filesystem clock: the newest last-modified time matched, capped at the poll time, and left where it is by a poll that matches nothing. A clock behind the local one cannot make a later file late, and one ahead cannot carry the watermark with it. _mtime_of is renamed _ensure_mtime and no longer takes the option name. The option's doc states the resolution it compares at and that it needs a fresh pipeline, since an in-place update would seed the cursor with poll times the old state recorded.
A cursor alone cannot tell two outputs at one event time apart: comparing strictly drops the second, comparing loosely repeats both. Rounding only moves where that tie falls, so the microsecond coder goes with it. Dedup goes back to hashing the output key, and the cursor now bounds that state rather than standing in for it. A key is retired once the greatest emitted event time has moved allowed_lateness past it, so the restriction holds a trailing window instead of every key ever seen. allowed_lateness defaults to zero and widens the window for a source whose outputs arrive out of order. The cursor is orthogonal to the key, so it no longer conflicts with output_key_fn, and a restriction resumed in cursor mode keeps the hashes its hash rounds recorded rather than seeding a cursor from them.
The cursor bounds the deduplication state rather than replacing the key, so the transform passes the same key function in both modes and match_updated_files decides whether a changed file counts as new again. The watermark under timestamp_cursor is the newest last-modified time matched, capped at the poll time, and the poll time when nothing matched, so a quiet directory no longer holds it back.
Holding the watermark at the newest last-modified time matched stalls a directory that is quiet rather than empty: every poll re-lists the same files, so the newest one never moves and the watermark sits at its last-modified time while the poll time runs away from it. Only an empty match released it, which is not the case that stalls. The hold now follows the evidence. A poll that turns up a last-modified time newer than any before it has just read the filesystem clock, so the watermark stops there and files still in flight behind a clock that lags the local one are not late. A poll that finds nothing newer has no fresh reading to go on, so the watermark takes the poll time and event-time windows keep closing. A continuously fed directory therefore trails the filesystem clock throughout, and a quiet one catches up. Also records what bounding the deduplication state costs: a file modified after its id was retired reads as new and is matched a second time, whatever match_updated_files says. The MatchContinuously test that covered the update case never advanced the cursor, so it did not reach the retirement; the Watch test added here pins it.
0c00e43 to
0526f7d
Compare
This reverts commit 0526f7d.
Holding the watermark at the newest last-modified time matched stalls a directory that is quiet rather than empty: every poll re-lists the same files, so the newest one never moves and the watermark sits at its last-modified time while the poll time runs away from it. Only an empty match released it, which is not the case that stalls. The hold now follows the evidence. A poll that turns up a last-modified time newer than any before it has just read the filesystem clock, so the watermark stops there and files still in flight behind a clock that lags the local one are not late. A poll that finds nothing newer has no fresh reading to go on, so the watermark takes the poll time and event-time windows keep closing. A continuously fed directory therefore trails the filesystem clock throughout, and a quiet one catches up. Also records what bounding the deduplication state costs: a file modified after its id was retired reads as new and is matched a second time, whatever match_updated_files says. The MatchContinuously test that covered the update case never advanced the cursor, so it did not reach the retirement; the Watch test added here pins it.
|
Apologies for the force push earlier. I realized this conflicts with the requested review For auditability: the force push only changed commit metadata/trailers. The reviewed code tree from 0c00e43 and the current head I then avoided further history rewriting and pushed two normal commits: The existing review threads are still preserved in the conversation/outdated threads. I won't force-push this PR branch again. |
|
I added a reviewer guide at the top of the PR description to make the current review path easier to follow after the history correction. It includes the audit note, current entry points, and the remaining reviewer decisions. |
Ran the Dataflow integration check you asked for. The job is still running, so you can inspect it live rather than only reading numbers from me. Job: Markers: open in Cloud Logging, or paste Watched prefix:
Project DesignThe sequence is built so the outcome isolates key retirement. Only
The two overwrites are the same operation on the same path. The only thing that differs is whether the key was still retained. Result
Totals:
Because the branches share a worker, the three matched each file within 79ms, 71ms and 682ms of one another, so the comparison is not sensitive to poll skew between branches. Raw markers, prefix stripped. Event timesDefaults and ScopeEach marker is written by a Tell me when you have finished looking and I will cancel the job. |
Keying on the path alone left an update out while its key was retained and let one through once the cursor had retired the key, so whether an updated file matched depended on cursor timing. The cursor now keys on the path and the last-modified time. An update is a new key and is always matched. Retiring a key can no longer duplicate a match either, since the only file that would recreate a retired key carries the same last-modified time and is skipped by the same mark.
|
Good catch, and it was worse than an edge case. Fixed in The bugThe cursor keyed on the path alone: output_key_fn=(_file_path_and_mtime_key if self.match_upd else _file_path_key),So an update was matched or not depending on whether the cursor had happened to retire the key yet. While the key was retained the update was dropped, which is what you saw. Once a newer file moved the cursor past it, the same update went through. The outcome depended on cursor timing rather than on anything the user asked for. The fixThe cursor now keys on the path and the last-modified time, so an update is a new key and is always matched. It also removes the weakness I had documented as the cost of bounding the state. A retired key can no longer produce a duplicate: the only file that would recreate key What remains is the intended limitation: a file whose last-modified time is already below the floor when it first appears is skipped, which is the backfill and copy-preserving-mtime case, and is what One consequence worth your call: Re-validationJob: Markers: open in Cloud Logging, or paste Watched prefix Three modes run as three roots in one pipeline, each with its own Same sequence as before. Only
Totals: The cursor column now equals the Defaults and Unit coverage: As before, each marker is a |
| file with timestamp changes. | ||
| apply_windowing: Whether each element should be assigned to | ||
| individual window. If false, all elements will reside in global window. | ||
| timestamp_cursor: (When has_deduplication is set to True) bound the |
There was a problem hiding this comment.
I'll cut it down this
| if timestamp_cursor and not has_deduplication: | ||
| raise ValueError( | ||
| 'MatchContinuously(timestamp_cursor=True) deduplicates, so it ' | ||
| 'requires has_deduplication=True.') |
There was a problem hiding this comment.
| if timestamp_cursor and not has_deduplication: | |
| raise ValueError( | |
| 'MatchContinuously(timestamp_cursor=True) deduplicates, so it ' | |
| 'requires has_deduplication=True.') | |
| if timestamp_cursor: | |
| if not has_deduplication: | |
| raise ValueError( | |
| 'MatchContinuously(timestamp_cursor=True) deduplicates, so it ' | |
| 'requires has_deduplication=True.') | |
| if not match_updated_files: | |
| _LOGGER.warning("timestamp_cursor implies match_upd, set match_upd=True") | |
| self.match_upd = True |
| termination=termination, | ||
| output_key_fn=( | ||
| _file_path_and_mtime_key if self.match_upd else _file_path_key), | ||
| _file_path_and_mtime_key |
There was a problem hiding this comment.
We don't need to change this line if we guarantee match_upd=True when timestamp_cursor is set
3 participants
Reviewer guide after history correction
Apologies for the earlier force push. I understand this makes review harder and will not force-push this PR branch again.
For auditability, the reviewed code tree from
0c00e43b937912577939ec9eab3e0cb6d7772095and the current head626ee74c97ee9c225edd6463cfc56c5e5760bfc8are identical:git diff 0c00e43b93 626ee74c97is empty.I then restored normal PR history with regular commits only:
f184d2f365: revert0526f7d9b5626ee74c97: re-apply the same code changeSuggested current review path:
sdks/python/apache_beam/io/fileio.py:_MatchContinuouslyPollFn, especially timestamping and watermark handling.sdks/python/apache_beam/io/fileio.py:timestamp_cursordocs and interaction withmatch_updated_files.sdks/python/apache_beam/io/fileio_test.py: cursor and watermark coverage.sdks/python/apache_beam/io/watch_test.py: retained-key / allowed-lateness behavior.Open reviewer decisions:
allowed_latenessshould also be exposed onMatchContinuously.match_updated_filesundertimestamp_cursoris acceptable, or whether already-seen output timestamps should be carried through claim processing.Relevant existing review threads:
match_updated_filestradeoff undertimestamp_cursor: [Python] Refactor MatchContinuously onto the Watch transform #39461 (comment)Existing review threads are preserved in the conversation/outdated threads; this guide is only to make the current review path easier to follow.
Routes
fileio.MatchContinuouslythrough theWatchtransform when deduplication is enabled.Built on the
Watchtransform, which merged in #39023.What changes
The polling loop and the set of already-matched file ids move into the splittable DoFn restriction. The per-key state DoFns
_RemoveDuplicatesand_RemoveOldDuplicatesare removed, sinceWatchperforms the deduplication.has_deduplication=Falsekeeps the previousPeriodicImpulsebehaviour, so that path is unchanged.Behaviour change worth calling out
Because the matched ids are part of the restriction, a runner with checkpointing enabled restores them after a restart and does not reprocess files. The class docstring previously stated the opposite, that already processed files are reprocessed on restart, which was accurate for the earlier memory-only implementation. The docstring is updated in this PR.
Validation
Fault tolerance on Flink 1.20 with checkpointing enabled: two files present at start, two added while running, then the TaskManager was killed mid stream. The JobManager restored the job from checkpoint 3 and every file was still emitted exactly once, with no reprocessing.
Also exercised on Dataflow Runner v2 reading a real GCS prefix: files present at startup and files added to the bucket mid run were each emitted exactly once.
Unit tests:
watch_test.py28 passed,fileio_test.pyMatchContinuously tests 8 passed. Formatted with yapf 0.43.0 and isort 7.0.0.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).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