Eliaaazzz commented on code in PR #39461:
URL: https://github.com/apache/beam/pull/39461#discussion_r3771887808
##########
sdks/python/apache_beam/io/fileio.py:
##########
@@ -292,32 +293,51 @@ def state_coder(self):
return VarIntCoder()
+def _mtime_of(metadata: filesystem.FileMetadata, option: str) -> float:
+ # A missing (zero) timestamp is rejected because every file would then carry
+ # the same one, and updates could never be told apart.
+ if not metadata.last_updated_in_seconds:
+ raise BeamIOError(
+ 'MatchContinuously(%s=True) requires file last-modified times, but '
+ '%s reports none.' % (option, metadata.path))
+ return metadata.last_updated_in_seconds
+
+
+def _mtime_timestamp(metadata: filesystem.FileMetadata) -> Timestamp:
+ # 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)
Review 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.
##########
sdks/python/apache_beam/io/fileio.py:
##########
@@ -330,6 +350,12 @@ def __call__(self, file_pattern: str) ->
PollResult[filesystem.FileMetadata]:
self._empty_match_treatment)):
raise BeamIOError(
'Empty match for pattern %s. Disallowed.' % file_pattern)
+ if self._mtime_timestamps:
+ outputs = [
+ TimestampedValue(metadata, _mtime_timestamp(metadata))
+ for metadata in match_result.metadata_list
+ ]
+ return PollResult.incomplete(outputs).with_watermark(now)
return PollResult.incomplete(
match_result.metadata_list, timestamp=now).with_watermark(now)
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
