Abacn commented on code in PR #39090:
URL: https://github.com/apache/beam/pull/39090#discussion_r3667972282
##########
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:
Prefer enum over raw numbers
##########
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:
We should not encode the items when use cursors. The whole point of cursor
(with_timestamp) mode is that the saved state doesn't grow with the number of
items in poll therefore scalable
##########
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:
Please keep doc and comments concise thoughout this PR.
Agent has a tendancy of stacking chat history into doc. Sometimes need a
fresh session to rewrite things
--
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]
