jonathansudhakar1 opened a new issue, #41245: URL: https://github.com/apache/beam/issues/41245
### What happened? In batch, `iobase.Write` (which backs `WriteToText`, `WriteToAvro`, `WriteToParquet` and `WriteToTFRecord`) returns a `PCollection` of the final file paths. Each path is emitted in the `GlobalWindow` with timestamp `MAX_TIMESTAMP` ([iobase.py#L1476](https://github.com/apache/beam/blob/1cabbd79927cf29dffe637e640ca7e438433029d/sdks/python/apache_beam/io/iobase.py#L1476)). That's past the end of the global window, since `GlobalWindow().max_timestamp()` is `MAX_TIMESTAMP` minus one day, so downstream aggregations treat the paths as data beyond their window's end: - **Prism** (the default local runner): a `GroupByKey` on the paths fails the job with `panic in ElementManager.Bundles watermark evaluation goroutine: generating bundle for stage stage-009 at watermark +inf panicked`. `assert_that(paths, equal_to([...]))` sees an empty `PCollection` and the job fails with `nothing in progress and no refreshes with non zero pending elements: 3`. - **BundleBasedDirectRunner**: `beam.combiners.Count.Globally()` on the paths returns `0` instead of `3`. - **FnApiRunner**: works. Repro on master (1cabbd79927), Python 3.13: ```python import tempfile import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions tmp = tempfile.mkdtemp() with beam.Pipeline(options=PipelineOptions(['--runner=PrismRunner'])) as p: paths = p | beam.Create(['a', 'b', 'c']) | beam.io.WriteToText( tmp + '/out', num_shards=3) _ = paths | beam.Map(lambda x: (0, x)) | beam.GroupByKey() | beam.Map(print) ``` The default `num_shards=0` fails the same way. With `--runner=BundleBasedDirectRunner`, `paths | beam.combiners.Count.Globally() | beam.Map(print)` prints `0`. **Cause.** BEAM-5250 (7581caa3a5d, 2018) changed `_WriteBundleDoFn` to emit at `GlobalWindow().max_timestamp()` ([L1346](https://github.com/apache/beam/blob/1cabbd79927cf29dffe637e640ca7e438433029d/sdks/python/apache_beam/io/iobase.py#L1346)). Two places still use `timestamp.MAX_TIMESTAMP`: - `_finalize_write` ([L1476](https://github.com/apache/beam/blob/1cabbd79927cf29dffe637e640ca7e438433029d/sdks/python/apache_beam/io/iobase.py#L1476)), which produces the transform's output. - `_WriteKeyedBundleDoFn` ([L1449](https://github.com/apache/beam/blob/1cabbd79927cf29dffe637e640ca7e438433029d/sdks/python/apache_beam/io/iobase.py#L1449)), whose results are only read as a side input. **Possible fix.** Emit at `window.GlobalWindow().max_timestamp()` in `_finalize_write`, and in `_WriteKeyedBundleDoFn` for consistency. I tried the `_finalize_write` change locally: the repro then works on all three runners (`GroupByKey` succeeds on Prism, and `Count.Globally()` returns `3` on Prism and on BundleBasedDirectRunner). Streaming writes go through `_finalize_windowed_write` instead and aren't covered by this report. Separately, Prism panicking on these elements, rather than handling them as late data, may be worth a look by its maintainers. I found this while working on #25769, which documents this output. A test that asserts on the output can't pass on Prism without this fix, so I plan to include the fix in that PR. ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [x] Component: Python SDK - [ ] Component: Java SDK - [ ] Component: Go SDK - [ ] Component: Typescript SDK - [x] Component: IO connector - [ ] Component: Beam YAML - [ ] Component: Beam examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Infrastructure - [ ] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Prism Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- 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]
