sjvanrossum commented on code in PR #34777:
URL: https://github.com/apache/beam/pull/34777#discussion_r2066550910
##########
sdks/python/apache_beam/io/iobase.py:
##########
@@ -1235,10 +1430,20 @@ def _finalize_write(
writer = sink.open_writer(init_result, str(uuid.uuid4()))
extra_shards.append(writer.close())
outputs = sink.finalize_write(
- init_result, write_results + extra_shards, pre_finalize_results)
+ init_result, write_results + extra_shards, pre_finalize_results, w)
+ outputs = list(outputs)
+ _LOGGER.info("*** _finalize_write outputs %s",outputs)
if outputs:
- return (
- window.TimestampedValue(v, timestamp.MAX_TIMESTAMP) for v in outputs)
+ if not isinstance(w, window.GlobalWindow):
+ #handle windowed finalize
+ yield (
+ window.TimestampedValue(v,
+ timestamp=self.window.start,
+ windows=[w]
+ ) for v in outputs)
+ else:
+ return (
+ window.TimestampedValue(v, timestamp.MAX_TIMESTAMP) for v in outputs)
Review Comment:
The constructor of `TimestampedValue` has no argument `windows` afaik and
you likely want to shift these resulting elements to the window's end time,
although looking at the Java SDK perhaps this should be `w.max_timestamp()`.
```suggestion
return (
window.TimestampedValue(v, w.end) for v in outputs)
```
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
