From 79560eee61c05b44985e0cde39d4a96406511387 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Wed, 24 Jun 2026 21:32:57 +1000 Subject: [PATCH] [Python] Bound and harden the Watch transform Follow-up to the experimental Watch transform that documents the event-time contract, makes late data observable, and adds an opt-in bound on the per-input dedup state. Watermark/event-time: - Document the event-time/watermark contract on the module and PollResult: explicit watermark, else the earliest event time of the round's new outputs, held when a round is empty, released to MAX on complete(). - Warn (throttled) when an output is emitted behind the current watermark, so out-of-order late data is observable rather than silently dropped, and point to PollResult.with_watermark for out-of-order sources. Scalability: - Add opt-in Watch.with_dedup_horizon(Duration): drop and collect outputs more than the horizon behind the watermark, so the dedup set tracks a sliding event-time window instead of the full history. A re-listed aged output is suppressed rather than re-emitted: the suppression cutoff equals the prior round's eviction cutoff, so there is no resurrection gap, and a full-relisting poll stays exactly-once with an advancing watermark. Off by default; horizon validated non-negative; suppression reported via a throttled warning. - Reuse the parent dedup map on idle rounds to drop the O(N) per-round copy. Tests cover the watermark contract, the out-of-order late warning, dedup horizon eviction/retention/boundary/order-independence/suppression and re-emission, idle-round reuse, and an end-to-end relisting + suppression case. 30 tests and 5 subtests pass. --- sdks/python/apache_beam/io/watch.py | 228 +++++++++++++- sdks/python/apache_beam/io/watch_test.py | 370 +++++++++++++++++++++++ 2 files changed, 584 insertions(+), 14 deletions(-) diff --git a/sdks/python/apache_beam/io/watch.py b/sdks/python/apache_beam/io/watch.py index 8ccbd4f8e6e5..d9f3ff7a2bcf 100644 --- a/sdks/python/apache_beam/io/watch.py +++ b/sdks/python/apache_beam/io/watch.py @@ -48,6 +48,45 @@ def poll(prefix): .with_poll_interval(Duration(seconds=5)) .with_termination_per_input(after_total_of(60))) +Watermark and event-time contract +--------------------------------- +Each emitted output carries the event time the poll function reported for it. +Raw (non-``TimestampedValue``) outputs default to processing time +(``Timestamp.now()`` at poll), so wrap outputs in ``TimestampedValue`` or pass +``timestamp=`` to :meth:`PollResult.incomplete`/:meth:`PollResult.complete` when +the data has a real event time. The watermark for an input is derived per round +as: (a) the explicit ``with_watermark`` if the poll supplies one; else (b) the +earliest event time of this round's new outputs; else (c) held unchanged when a +round yields nothing; and it is released to ``MAX_TIMESTAMP`` on +:meth:`PollResult.complete`. The watermark only ever advances. + +Policy (b) is safe only for poll functions that enumerate outputs in +non-decreasing event-time order. If a later round returns a brand-new output +whose event time is *below* the already-advanced watermark, that output is +emitted at its true (earlier) time and is therefore late: downstream event-time +windowing may drop it. Watch logs a throttled warning when this happens. For +out-of-order sources, have the poll supply an explicit watermark via +:meth:`PollResult.with_watermark` that bounds the earliest event time any future +output may have. Dedup is by output value identity (a re-seen value keeps its +first-seen timestamp), and the output coder must be deterministic for dedup to +hold across workers and restarts. + +Scalability +----------- +Parallelism is per input element: each input is one restriction watched on one +worker, with no intra-key splitting, so throughput scales with the number of +input elements, not with a single input's growth. The +per-input dedup set retains one hash per distinct output and is not +garbage-collected by default, so it grows with the number of distinct outputs an +input ever produces. For high-cardinality, long-lived inputs, bound it with +:meth:`Watch.with_dedup_horizon`, which drops outputs more than the horizon +behind the watermark and collects their dedup state, so memory tracks a sliding +event-time window. A re-surfaced aged output is suppressed rather than +re-emitted, so this is safe for full-relisting polls; the trade-off is that a +genuinely new output arriving more than the horizon behind the watermark is +dropped as late (see its docstring). The poll function runs synchronously on the +watch path, so keep it bounded and timeout-safe. + This API is experimental and may change in backwards-incompatible ways. """ @@ -113,6 +152,11 @@ def is_complete(self) -> bool: @staticmethod def _normalize(outputs, timestamp) -> Tuple[TimestampedValue, ...]: + # The default timestamp is computed once per call (not per output) so all + # raw outputs in one poll share an event time and the inferred watermark is + # not jittered by wall-clock reads. ``timestamp=None`` means "use processing + # time"; pass ``timestamp=`` or wrap outputs in ``TimestampedValue`` to set + # a real event time. if timestamp is None: default_ts = Timestamp.now() else: @@ -130,7 +174,10 @@ def incomplete(outputs: Iterable, timestamp=None) -> 'PollResult': """Reports outputs and expects more; the transform infers the watermark. A raw (non-:class:`TimestampedValue`) output is stamped with ``timestamp`` - when given, else with the current processing time. + when given, else with the current processing time. With no explicit + watermark, the transform holds the watermark at the earliest event time of + this poll's new outputs, which is only safe for non-decreasing event-time + enumerations; out-of-order sources should call :meth:`with_watermark`. """ return PollResult(PollResult._normalize(outputs, timestamp), watermark=None) @@ -139,12 +186,16 @@ def complete(outputs: Iterable, timestamp=None) -> 'PollResult': """Reports the final outputs for an input, after which polling stops. A raw (non-:class:`TimestampedValue`) output is stamped with ``timestamp`` - when given, else with the current processing time. + when given, else with the current processing time. The watermark is released + to ``MAX_TIMESTAMP`` so downstream event-time windows for this input close. """ return PollResult( PollResult._normalize(outputs, timestamp), watermark=MAX_TIMESTAMP) def with_watermark(self, watermark) -> 'PollResult': + """Sets an explicit watermark: a promise that no future output for this + input will have an event time below ``watermark``. Use this for sources that + can surface outputs out of event-time order across poll rounds.""" return dataclasses.replace(self, watermark=Timestamp.of(watermark)) @@ -374,12 +425,14 @@ def __init__( poll_fn: Callable[[Any], PollResult], key_coder: Coder, termination: TerminationCondition, - now_fn: Callable[[], float]): + now_fn: Callable[[], float], + dedup_horizon: Optional[Duration] = None): self._restriction = restriction self._poll_fn = poll_fn self._key_coder = key_coder self._termination = termination self._now = now_fn + self._dedup_horizon = dedup_horizon self._should_stop = False self._primary = None # type: Optional[_GrowthState] self._residual = None # type: Optional[_GrowthState] @@ -409,10 +462,23 @@ def try_claim(self, holder: list) -> bool: now = Timestamp.of(self._now()) result = self._poll_fn(element) + # An output more than the dedup horizon behind the committed watermark is + # past its dedup lifetime: its hash has been (or is about to be) evicted, so + # it is dropped as late instead of re-emitted. Without this, a re-listing + # poll that re-surfaces an aged output every round would resurrect the + # evicted hash, re-emit a duplicate, and pin the inferred watermark to the + # output's old event time. The suppression cutoff equals the prior round's + # eviction cutoff, so an evicted output is suppressed on its next appearance + # with no resurrection gap. None when no horizon or no committed watermark. + suppress_cutoff = self._suppress_cutoff(restriction.poll_watermark) new_outputs = [] # type: List[TimestampedValue] claimed = [] # type: List[Tuple[bytes, Timestamp]] seen_this_round = set() # type: set + suppressed = 0 for output in result.outputs: + if suppress_cutoff is not None and output.timestamp < suppress_cutoff: + suppressed += 1 + continue key_hash = self._hash_output(output.value) if key_hash in restriction.completed or key_hash in seen_this_round: continue @@ -447,17 +513,26 @@ def try_claim(self, holder: list) -> bool: # initiated or via defer_remainder) resumes a state that emits nothing. self._residual = _NonPollingGrowthState(PollResult((), watermark)) else: - merged = collections.OrderedDict(restriction.completed) - for key_hash, first_seen in claimed: - merged[key_hash] = first_seen residual_watermark = self._max_watermark( restriction.poll_watermark, watermark) + completed = self._build_completed( + restriction.completed, claimed, residual_watermark) self._residual = _PollingGrowthState( - merged, residual_watermark, termination_state) - holder[1] = ('poll', new_outputs, watermark, stop) + completed, residual_watermark, termination_state) + holder[1] = ('poll', new_outputs, watermark, stop, suppressed) self._should_stop = True return True + def _suppress_cutoff(self, + watermark: Optional[Timestamp]) -> Optional[Timestamp]: + # Event-time floor below which an output is past its dedup horizon. Derived + # from the committed (prior-round) watermark, the one value known before + # this round's outputs are inspected, so it matches the prior round's + # eviction cutoff. + if self._dedup_horizon is None or watermark is None: + return None + return watermark - self._dedup_horizon + @staticmethod def _max_watermark(left: Optional[Timestamp], right: Optional[Timestamp]) -> Optional[Timestamp]: @@ -467,6 +542,43 @@ def _max_watermark(left: Optional[Timestamp], return left return max(left, right) + def _build_completed( + self, + base: 'collections.OrderedDict[bytes, Timestamp]', + claimed: List[Tuple[bytes, Timestamp]], + watermark: Optional[Timestamp], + ) -> 'collections.OrderedDict[bytes, Timestamp]': + """Builds the residual dedup map. + + Reuses the parent map untouched when a round adds nothing and no dedup + horizon is configured, so a steady-state idle round avoids the O(N) copy. + Otherwise copies, appends this round's new hashes, and (when a horizon is + set) evicts entries whose event time has fallen behind the watermark. + """ + horizon = self._dedup_horizon + if not claimed and horizon is None: + return base + merged = collections.OrderedDict(base) + for key_hash, first_seen in claimed: + merged[key_hash] = first_seen + if horizon is not None and watermark is not None: + self._evict_stale(merged, watermark - horizon) + return merged + + @staticmethod + def _evict_stale( + completed: 'collections.OrderedDict[bytes, Timestamp]', + cutoff: Timestamp) -> None: + # Drop every entry first seen at an event time below the cutoff, regardless + # of insertion order. Insertion order is not event-time order in general: + # with an explicit held-back watermark a later round can add an + # earlier-timestamped output after a future one, so a prefix scan would miss + # genuinely-stale entries and fail to bound the state. This full scan is + # O(N) in the (horizon-bounded) map size, paid only when a horizon is set. + stale = [h for h, first_seen in completed.items() if first_seen < cutoff] + for h in stale: + del completed[h] + def try_split(self, fraction_of_remainder): # Only self-checkpoint (fraction 0) is supported; decline dynamic splits. if fraction_of_remainder != 0: @@ -521,14 +633,20 @@ def __init__( termination: TerminationCondition, poll_interval: Duration, output_coder: Coder, - now_fn: Optional[Callable[[], float]] = None): + now_fn: Optional[Callable[[], float]] = None, + dedup_horizon: Optional[Duration] = None): self._poll_fn = poll_fn self._termination = termination self._poll_interval = poll_interval self._output_coder = output_coder self._key_coder = output_coder self._now = now_fn or time.time + self._dedup_horizon = dedup_horizon self._restriction_coder = _GrowthStateCoder(output_coder, termination) + # Counts of late emissions and horizon-suppressed outputs seen on this + # worker, for throttled warnings. + self._late_count = 0 + self._suppressed_count = 0 def initial_restriction(self, element) -> _PollingGrowthState: now = Timestamp.of(self._now()) @@ -543,7 +661,8 @@ def create_tracker(self, restriction) -> _GrowthRestrictionTracker: self._poll_fn, self._key_coder, self._termination, - self._now) + self._now, + self._dedup_horizon) def split(self, element, restriction): # Watch fans out by input element, so each restriction stays whole. @@ -581,8 +700,18 @@ def process( for output in work[1].outputs: yield TimestampedValue((element, output.value), output.timestamp) return - new_outputs, watermark, stop = work[1], work[2], work[3] + new_outputs, watermark, stop, suppressed = ( + work[1], work[2], work[3], work[4]) + if suppressed: + self._warn_suppressed(element, suppressed) + # Emit outputs first, then advance the watermark (emit-then-advance), so a + # round that reports both outputs and a higher watermark cannot push the + # watermark past those outputs' event times. An output behind the current + # watermark is already late; warn rather than silently dropping it. + current_watermark = watermark_estimator.current_watermark() for output in new_outputs: + if current_watermark is not None and output.timestamp < current_watermark: + self._warn_late(element, output.timestamp, current_watermark) yield TimestampedValue((element, output.value), output.timestamp) if stop: # The input is finished, so release the watermark hold to MAX. @@ -592,6 +721,39 @@ def process( _set_watermark_if_greater(watermark_estimator, watermark) tracker.defer_remainder(self._poll_interval) + def _warn_late(self, element, output_timestamp, watermark) -> None: + # Throttle to powers of two so an ongoing problem stays visible without + # flooding worker logs. + self._late_count += 1 + if self._late_count & (self._late_count - 1) == 0: + _LOGGER.warning( + 'Watch emitted an output for input %r at event time %s, behind the ' + 'current watermark %s, so downstream event-time windowing may drop ' + 'it as late. This happens when the poll returns outputs out of ' + 'event-time order across rounds without an explicit watermark; call ' + 'PollResult.with_watermark(...) for out-of-order sources. ' + '(%d late emissions so far on this worker)', + element, + output_timestamp, + watermark, + self._late_count) + + def _warn_suppressed(self, element, count) -> None: + # Throttle to powers of two so an ongoing drop stays visible without + # flooding worker logs. + self._suppressed_count += count + if self._suppressed_count & (self._suppressed_count - 1) == 0: + _LOGGER.warning( + 'Watch suppressed %d output(s) for input %r more than the dedup ' + 'horizon behind the watermark, dropping them as late. For a ' + 're-listing poll this is expected for already-emitted outputs; for a ' + 'source that surfaces genuinely new late data, widen ' + 'with_dedup_horizon(...) or supply PollResult.with_watermark(...). ' + '(%d suppressed so far on this worker)', + count, + element, + self._suppressed_count) + def _set_watermark_if_greater(watermark_estimator, new_watermark) -> None: # set_watermark raises on regression, so only ever advance the watermark. @@ -617,13 +779,15 @@ def __init__( termination: Optional[TerminationCondition] = None, poll_interval: Optional[Duration] = None, output_coder: Optional[Coder] = None, - now_fn: Optional[Callable[[], float]] = None): + now_fn: Optional[Callable[[], float]] = None, + dedup_horizon: Optional[Duration] = None): super().__init__() self._poll_fn = poll_fn self._termination = termination or never() self._poll_interval = poll_interval self._output_coder = output_coder self._now = now_fn + self._dedup_horizon = dedup_horizon @classmethod def growth_of(cls, poll_fn: Callable[[Any], PollResult]) -> 'Watch': @@ -635,7 +799,8 @@ def _replace(self, **changes) -> 'Watch': termination=self._termination, poll_interval=self._poll_interval, output_coder=self._output_coder, - now_fn=self._now) + now_fn=self._now, + dedup_horizon=self._dedup_horizon) spec.update(changes) return Watch(**spec) @@ -649,6 +814,40 @@ def with_termination_per_input( def with_output_coder(self, output_coder: Coder) -> 'Watch': return self._replace(output_coder=output_coder) + def with_dedup_horizon(self, horizon) -> 'Watch': + """Bounds the per-input dedup state by event time. + + By default the dedup set is exact and unbounded: it retains one hash per + distinct output forever, so its size grows with the number of distinct + outputs an input produces. With a horizon set, Watch drops every output + whose event time is more than ``horizon`` (a :class:`Duration` or seconds) + behind the input's watermark and collects its dedup entry, so the set tracks + a sliding event-time window instead of the full history. This bounds memory, + the per-round copy, and the per-checkpoint encode to the horizon window. + + An aged-out output that the poll re-surfaces at its original (now + below-horizon) event time is suppressed, not re-emitted: the suppression + cutoff equals the prior round's eviction cutoff, so a re-listing poll that + returns the whole collection every round stays exactly-once and its + watermark keeps advancing. The trade-off is that a genuinely new output + arriving more than ``horizon`` behind the watermark is also dropped as late; + Watch logs a throttled warning whenever it suppresses an output. For + out-of-order sources that must keep such outputs, widen the horizon or + supply an explicit :meth:`PollResult.with_watermark`. + + Residual hazard: an aged-out value that the poll re-returns at a fresh, + at-or-after-watermark timestamp is on time, so it is not suppressed and is + re-emitted as a duplicate. Cursor or time-windowed polls that never re-emit + an aged-out value at a fresh timestamp avoid this. + + ``horizon`` must be non-negative. + """ + horizon = _as_duration(horizon) + if horizon < Duration(0): + raise ValueError( + 'dedup horizon must be non-negative, got %s' % (horizon, )) + return self._replace(dedup_horizon=horizon) + def expand(self, pcoll): if self._poll_interval is None: raise ValueError('Watch requires with_poll_interval(...)') @@ -669,7 +868,8 @@ def expand(self, pcoll): self._termination, self._poll_interval, output_coder, - self._now)) + self._now, + self._dedup_horizon)) def _as_duration(value) -> Duration: diff --git a/sdks/python/apache_beam/io/watch_test.py b/sdks/python/apache_beam/io/watch_test.py index 07f2eadfbd91..586babbb39b6 100644 --- a/sdks/python/apache_beam/io/watch_test.py +++ b/sdks/python/apache_beam/io/watch_test.py @@ -18,6 +18,7 @@ """Tests for the Watch transform.""" import collections +import hashlib import unittest import apache_beam as beam @@ -60,6 +61,47 @@ def _initial_polling(termination=None, now=Timestamp(0)): collections.OrderedDict(), None, termination.for_new_input(now, 'input')) +def _horizon_tracker(restriction, poll_fn, horizon, now=0.0): + return _GrowthRestrictionTracker( + restriction, + poll_fn, + StrUtf8Coder(), + never(), lambda: now, + Duration(horizon)) + + +def _hash(value): + return hashlib.blake2b(StrUtf8Coder().encode(value), digest_size=16).digest() + + +class PollResultTest(unittest.TestCase): + def test_normalize_stamps_one_processing_time_when_timestamp_none(self): + before = Timestamp.now() + result = PollResult.incomplete(['a', 'b']) + after = Timestamp.now() + # Raw outputs share a single processing-time stamp (no per-output jitter). + stamps = {o.timestamp for o in result.outputs} + self.assertEqual(1, len(stamps)) + ts = stamps.pop() + self.assertTrue(before <= ts <= after) + + def test_normalize_preserves_timestamped_and_applies_explicit_default(self): + result = PollResult.incomplete([_ts('a', 1), 'b'], timestamp=7) + by_value = {o.value: o.timestamp for o in result.outputs} + self.assertEqual(Timestamp(1), by_value['a']) # TimestampedValue preserved + self.assertEqual(Timestamp(7), by_value['b']) # raw stamped with default + + def test_complete_releases_watermark_to_max(self): + self.assertEqual( + MAX_TIMESTAMP, PollResult.complete([_ts('a', 1)]).watermark) + self.assertTrue(PollResult.complete([]).is_complete) + + def test_with_watermark_overrides(self): + self.assertEqual( + Timestamp(0), + PollResult.incomplete([_ts('a', 9)]).with_watermark(0).watermark) + + class GrowthStateCoderTest(unittest.TestCase): def test_polling_round_trip_preserves_resume_state(self): termination = after_total_of(Duration(30)) @@ -217,6 +259,267 @@ def poll(unused_element): self.assertIsInstance(residual, _PollingGrowthState) self.assertEqual(2, len(residual.completed)) + def test_out_of_order_new_output_infers_its_earlier_event_time(self): + # Round 1 surfaces a@10; round 2 surfaces a brand-new b@5 (earlier). The + # round-2 inferred watermark is b's own time; the process() estimator's + # monotonic guard is what leaves b late. This locks in the + # reference-consistent behavior the design documents and warns about. + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 10)]) + return PollResult.incomplete([_ts('b', 5)]) + + tracker = _new_tracker(_initial_polling(), poll) + holder = ['input', None] + tracker.try_claim(holder) + self.assertEqual(Timestamp(10), holder[1][2]) + _, residual = tracker.try_split(0) + + resumed = _new_tracker(residual, poll) + holder = ['input', None] + self.assertTrue(resumed.try_claim(holder)) + self.assertEqual(['b'], [o.value for o in holder[1][1]]) + self.assertEqual(Timestamp(5), holder[1][2]) + + def test_explicit_watermark_holds_below_output_time(self): + # An explicit watermark below the output's own event time is honored, so a + # later earlier-timestamped output stays on time (the out-of-order-safe + # path). + def poll(unused_element): + return PollResult.incomplete([_ts('a', 10)]).with_watermark(0) + + tracker = _new_tracker(_initial_polling(), poll) + holder = ['input', None] + tracker.try_claim(holder) + self.assertEqual(Timestamp(0), holder[1][2]) + _, residual = tracker.try_split(0) + self.assertEqual(Timestamp(0), residual.poll_watermark) + + def test_idle_round_reuses_completed_map_object(self): + # A round that discovers nothing (and no dedup horizon) must reuse the + # parent dedup map rather than copying it O(N). + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 1)]) + return PollResult.incomplete([]) + + tracker = _new_tracker(_initial_polling(), poll) + holder = ['input', None] + tracker.try_claim(holder) + _, residual1 = tracker.try_split(0) + + resumed = _new_tracker(residual1, poll) + holder = ['input', None] + resumed.try_claim(holder) + _, residual2 = resumed.try_split(0) + self.assertIs(residual1.completed, residual2.completed) + + def test_dedup_horizon_evicts_outputs_behind_the_watermark(self): + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 10), _ts('b', 20)]) + return PollResult.incomplete([_ts('c', 30)]) + + tracker = _horizon_tracker(_initial_polling(), poll, horizon=5) + holder = ['input', None] + tracker.try_claim(holder) + _, residual1 = tracker.try_split(0) + self.assertEqual(2, len(residual1.completed)) # nothing stale yet + + resumed = _horizon_tracker(residual1, poll, horizon=5) + holder = ['input', None] + resumed.try_claim(holder) + _, residual2 = resumed.try_split(0) + # Watermark advanced to 30; a@10 and b@20 are older than 30-5=25, evicted. + self.assertEqual([_hash('c')], list(residual2.completed)) + self.assertEqual(Timestamp(30), residual2.completed[_hash('c')]) + + def test_dedup_horizon_unset_retains_all_hashes(self): + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 10), _ts('b', 20)]) + return PollResult.incomplete([_ts('c', 30)]) + + tracker = _new_tracker(_initial_polling(), poll) + holder = ['input', None] + tracker.try_claim(holder) + _, residual1 = tracker.try_split(0) + resumed = _new_tracker(residual1, poll) + holder = ['input', None] + resumed.try_claim(holder) + _, residual2 = resumed.try_split(0) + self.assertEqual(3, len(residual2.completed)) + + def test_dedup_horizon_still_dedups_recent_outputs(self): + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 10)]) + # 'a' re-appears but is well within the (large) horizon, so it stays in + # the dedup set and only the genuinely-new 'b' is emitted. + return PollResult.incomplete([_ts('a', 10), _ts('b', 12)]) + + tracker = _horizon_tracker(_initial_polling(), poll, horizon=100) + holder = ['input', None] + tracker.try_claim(holder) + _, residual1 = tracker.try_split(0) + resumed = _horizon_tracker(residual1, poll, horizon=100) + holder = ['input', None] + resumed.try_claim(holder) + self.assertEqual(['b'], [o.value for o in holder[1][1]]) + + def test_dedup_horizon_reemits_a_value_resurfaced_at_a_fresh_timestamp(self): + # Documents the residual with_dedup_horizon hazard: a value re-returned at a + # NEWER, at-or-after-watermark timestamp is on time, so suppression does not + # drop it; with its entry already evicted it is seen as new and re-emitted. + # (A re-list at the original below-horizon timestamp is suppressed instead; + # see test_dedup_horizon_suppresses_aged_relist_and_advances_watermark.) + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('x', 10)]) + if len(polls) == 2: + return PollResult.incomplete([_ts('y', 30)]) # advances watermark to 30 + return PollResult.incomplete([_ts('x', 40)]) # x re-surfaces, on time + + tracker = _horizon_tracker(_initial_polling(), poll, horizon=5) + holder = ['input', None] + tracker.try_claim(holder) + _, residual = tracker.try_split(0) + + resumed = _horizon_tracker(residual, poll, horizon=5) + holder = ['input', None] + resumed.try_claim(holder) # round 2: y@30, watermark->30, evict x@10 (<25) + _, residual = resumed.try_split(0) + self.assertNotIn(_hash('x'), residual.completed) + + resumed2 = _horizon_tracker(residual, poll, horizon=5) + holder = ['input', None] + resumed2.try_claim(holder) # round 3: x@40 re-emitted as a duplicate + self.assertEqual(['x'], [o.value for o in holder[1][1]]) + + def test_dedup_horizon_evicts_regardless_of_insertion_order(self): + # An explicit held-back watermark lets a later round insert an + # earlier-timestamped output after a future-timestamped one, so insertion + # order is not event-time order. Eviction must still drop the aged-out + # entry (a full scan, not a prefix scan). + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('future', 1000000)]).with_watermark(0) + if len(polls) == 2: + return PollResult.incomplete([_ts('old', 900)]).with_watermark(900) + return PollResult.incomplete([_ts('marker', 1005)]).with_watermark(1005) + + state = _initial_polling() + for _ in range(3): + tracker = _horizon_tracker(state, poll, horizon=100) + holder = ['input', None] + tracker.try_claim(holder) + _, state = tracker.try_split(0) + # cutoff = 1005 - 100 = 905; old@900 is evicted even though 'future' + # precedes it in insertion order, while future@1000000 is retained. + self.assertNotIn(_hash('old'), state.completed) + self.assertIn(_hash('future'), state.completed) + self.assertIn(_hash('marker'), state.completed) + + def test_dedup_horizon_keeps_entry_exactly_at_cutoff(self): + # Eviction is strict (< cutoff): an entry whose event time equals + # watermark - horizon is retained, not evicted. + polls = [] + + def poll(unused_element): + polls.append(len(polls)) + if len(polls) == 1: + return PollResult.incomplete([_ts('a', 25)]) + return PollResult.incomplete([_ts('b', 30)]) # watermark->30, cutoff=25 + + tracker = _horizon_tracker(_initial_polling(), poll, horizon=5) + holder = ['input', None] + tracker.try_claim(holder) + _, residual = tracker.try_split(0) + resumed = _horizon_tracker(residual, poll, horizon=5) + holder = ['input', None] + resumed.try_claim(holder) + _, residual = resumed.try_split(0) + self.assertIn(_hash('a'), residual.completed) # a@25 == cutoff, kept + self.assertIn(_hash('b'), residual.completed) + + def test_dedup_horizon_suppresses_aged_relist_and_advances_watermark(self): + # A full-relisting poll re-surfaces every output at its original event time + # every round. Once an output ages past the horizon its hash is evicted; + # suppression then drops the re-listed copy instead of re-emitting it, so + # the watermark keeps advancing (no freeze) and the dedup set stays bounded. + def poll_for(round_index): + def poll(unused_element): + return PollResult.incomplete( + [_ts('f%d' % i, i) for i in range(round_index + 1)]) + + return poll + + state = _initial_polling() + emitted = collections.Counter() + watermarks = [] + sizes = [] + for round_index in range(10): + tracker = _horizon_tracker(state, poll_for(round_index), horizon=3) + holder = ['input', None] + tracker.try_claim(holder) + for output in holder[1][1]: + emitted[output.value] += 1 + _, state = tracker.try_split(0) + watermarks.append(state.poll_watermark) + sizes.append(len(state.completed)) + # Every file emitted exactly once: the aged re-lists are suppressed, not + # re-emitted, so there is no duplicate storm. + self.assertEqual([1] * 10, [emitted['f%d' % i] for i in range(10)]) + # The watermark advances to the final round instead of freezing at the + # horizon: GC-only (no suppression) would pin it at horizon + 1. + self.assertEqual(Timestamp(9), watermarks[-1]) + # Dedup state is bounded by the horizon window, not the full history. + self.assertLessEqual(max(sizes), 5) + + def test_dedup_horizon_reports_suppressed_count(self): + # An aged output below the cutoff is suppressed and surfaced as a count so + # the DoFn can warn; an on-time output in the same round is still emitted. + def poll_for(round_index): + def poll(unused_element): + if round_index == 0: + return PollResult.incomplete([_ts('seed', 100)]) + return PollResult.incomplete([_ts('aged', 1), _ts('fresh', 101)]) + + return poll + + tracker = _horizon_tracker(_initial_polling(), poll_for(0), horizon=5) + holder = ['input', None] + tracker.try_claim(holder) + _, state = tracker.try_split(0) # watermark now 100, cutoff 95 + + resumed = _horizon_tracker(state, poll_for(1), horizon=5) + holder = ['input', None] + resumed.try_claim(holder) + work = holder[1] + self.assertEqual(['fresh'], [o.value for o in work[1]]) # aged@1 dropped + self.assertEqual(1, work[4]) # one suppression reported + # Module-level so the poll function pickles by reference; the call counter is # shared within the single in-memory DirectRunner process. @@ -236,10 +539,47 @@ def _complete_poll(prefix): return PollResult.complete([_ts(prefix + 'a', 1), _ts(prefix + 'b', 2)]) +def _out_of_order_poll(prefix): + # Round 1 emits a late_after@10 (advances the watermark to 10); round 2 emits + # early@5, which is behind the watermark and therefore late. + _POLL_CALLS[prefix] += 1 + count = _POLL_CALLS[prefix] + if count == 1: + return PollResult.incomplete([_ts(prefix + 'late_after', 10)]) + return PollResult.complete([_ts(prefix + 'early', 5)]) + + +def _aging_relist_poll(prefix): + # Re-lists the whole growing directory every round at the files' original + # event times, then completes. With a tight dedup horizon the oldest files + # age out and their re-lists must be suppressed, not re-emitted. + _POLL_CALLS[prefix] += 1 + count = _POLL_CALLS[prefix] + outputs = [_ts('%s%d' % (prefix, i), i) for i in range(count)] + if count >= 5: + return PollResult.complete(outputs) + return PollResult.incomplete(outputs) + + def _windowed_group(kv, window=beam.DoFn.WindowParam): return ((window.start, window.end), sorted(kv[1])) +class WatchBuilderTest(unittest.TestCase): + def test_with_dedup_horizon_rejects_negative(self): + watch = Watch.growth_of(_complete_poll) + with self.assertRaises(ValueError): + watch.with_dedup_horizon(-1) + with self.assertRaises(ValueError): + watch.with_dedup_horizon(Duration(-0.5)) + + def test_with_dedup_horizon_accepts_zero_and_duration(self): + watch = Watch.growth_of(_complete_poll) + self.assertEqual(Duration(0), watch.with_dedup_horizon(0)._dedup_horizon) + self.assertEqual( + Duration(5), watch.with_dedup_horizon(Duration(5))._dedup_horizon) + + class WatchEndToEndTest(unittest.TestCase): def _in_memory_pipeline(self): return TestPipeline( @@ -288,6 +628,36 @@ def test_multi_round_dedups_stops_and_is_per_input(self): self.assertEqual(3, _POLL_CALLS['x:']) self.assertEqual(3, _POLL_CALLS['y:']) + def test_out_of_order_across_rounds_warns_about_late_output(self): + _POLL_CALLS.clear() + with self.assertLogs(level='WARNING') as logs: + with self._in_memory_pipeline() as p: + output = ( + p | beam.Create(['k:']) + | Watch.growth_of(_out_of_order_poll).with_poll_interval( + Duration(0.05))) + # Watch emits every new output; the lateness is a downstream concern. + assert_that( + output, equal_to([('k:', 'k:late_after'), ('k:', 'k:early')])) + self.assertTrue( + any('behind the current watermark' in line for line in logs.output), + 'expected a late-emission warning, got: %s' % logs.output) + + def test_relisting_with_dedup_horizon_suppresses_and_dedups(self): + _POLL_CALLS.clear() + with self.assertLogs(level='WARNING') as logs: + with self._in_memory_pipeline() as p: + output = ( + p | beam.Create(['k:']) + | Watch.growth_of(_aging_relist_poll).with_poll_interval( + Duration(0.05)).with_dedup_horizon(Duration(1))) + # Each file is emitted exactly once even though every round re-lists the + # whole directory: aged re-lists are suppressed, recent ones deduped. + assert_that(output, equal_to([('k:', 'k:%d' % i) for i in range(5)])) + self.assertTrue( + any('more than the dedup horizon' in line for line in logs.output), + 'expected a suppression warning, got: %s' % logs.output) + if __name__ == '__main__': unittest.main()