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