Eliaaazzz commented on code in PR #39461:
URL: https://github.com/apache/beam/pull/39461#discussion_r3767107365
##########
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:
Review Comment:
Renamed it to _ensure_mtime and dropped the option argument, which only
existed to name the flag in the message, so the two callers now read
metadata.path, _ensure_mtime(metadata) and
Timestamp.of(_ensure_mtime(metadata)). I kept one helper rather than repeating
the raise at both sites. Happy to inline it if you would rather not have the
indirection. Commit afa436c.
##########
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.
That first commit touches merged code in a fileio PR. Say the word and I
will move it to its own PR.
##########
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:
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.
##########
sdks/python/apache_beam/io/fileio.py:
##########
@@ -261,10 +344,11 @@ class MatchContinuously(beam.PTransform):
guarantees.
Matching continuously scales poorly, as it is stateful, and requires storing
- file ids in memory. In addition, because it is memory-only, if a pipeline is
- 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``
Review 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?
--
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]
