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