Eliaaazzz commented on code in PR #39090:
URL: https://github.com/apache/beam/pull/39090#discussion_r3670805988
##########
sdks/python/apache_beam/io/watch.py:
##########
@@ -50,13 +51,53 @@ def poll(prefix) -> PollResult[str]:
poll_interval=Duration(seconds=5),
termination=after_total_of(60))
+Watermark and event-time contract
Review Comment:
You are right, that is exactly what happened here. I use AI tooling for
drafting and I let the docs absorb the review discussion instead of keeping
them clean. I rewrote the docstrings and comments this PR adds from a clean
slate, each fact is stated once now, and I have gone through every line myself.
##########
sdks/python/apache_beam/io/watch.py:
##########
@@ -348,6 +421,14 @@ def decode(self, encoded: bytes) -> _GrowthState:
if tag == 1:
watermark, outputs = self._non_polling_coder.decode(payload)
return _NonPollingGrowthState(PollResult(tuple(outputs), watermark))
+ if tag == 2:
Review Comment:
Done, replaced the raw tag numbers with an IntEnum.
##########
sdks/python/apache_beam/io/watch.py:
##########
@@ -322,18 +382,31 @@ def __init__(self, output_coder: Coder, termination:
TerminationCondition):
nullable_ts,
coders.ListCoder(TupleCoder([coders.BytesCoder(), TimestampCoder()])),
])
+ self._cursor_polling_coder = TupleCoder([
+ termination.state_coder(),
+ nullable_ts,
+ coders.ListCoder(TupleCoder([coders.BytesCoder(), TimestampCoder()])),
+ TimestampCoder(),
+ ])
self._non_polling_coder = TupleCoder([
nullable_ts,
coders.ListCoder(_TimestampedValueCoder(output_coder)),
])
def encode(self, state: _GrowthState) -> bytes:
if isinstance(state, _PollingGrowthState):
- payload = self._polling_coder.encode((
+ if state.cursor is None:
+ payload = self._polling_coder.encode((
+ state.termination_state,
+ state.poll_watermark,
+ list(state.completed.items())))
+ return self._envelope_coder.encode((0, payload))
+ payload = self._cursor_polling_coder.encode((
state.termination_state,
state.poll_watermark,
- list(state.completed.items())))
- return self._envelope_coder.encode((0, payload))
+ list(state.completed.items()),
Review Comment:
Agreed, the cursor state should not carry the items at all. The completed
map was already empty at runtime in cursor mode, but the byte format still
reserved a slot for it. Cursor states now encode only the termination state,
the poll watermark and the cursor, and a new test asserts the encoded size
stays constant no matter how many outputs a round claims.
--
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]
