Abacn commented on code in PR #39461:
URL: https://github.com/apache/beam/pull/39461#discussion_r3767209402
##########
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:
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
--
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]
