Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Python] use get_buffer to fetch buffer when the buffer is None #27373

Merged
merged 9 commits into from
Jul 11, 2023
Merged

[Python] use get_buffer to fetch buffer when the buffer is None #27373

Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
aeb7661
Create empty ListBuffer when buffer is none
AnandInguva Jul 6, 2023
22c4cc2
Replace empty buffer with a List/GroupBuffer
AnandInguva Jul 10, 2023
a576094
Merge remote-tracking branch 'upstream/master' into direct_runner_bug
AnandInguva Jul 10, 2023
cac4316
Apply suggestions from code review
AnandInguva Jul 10, 2023
ff3019b
Fix mypy error
AnandInguva Jul 10, 2023
87634de
Update sdks/python/apache_beam/runners/portability/fn_api_runner/fn_r…
AnandInguva Jul 10, 2023
60e7ed1
Add test
AnandInguva Jul 11, 2023
aba5673
Add note to CHANGES.md
AnandInguva Jul 11, 2023
c77587f
Fix link in CHANGES.md
AnandInguva Jul 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -825,10 +825,11 @@ def _execute_bundle(self,

buffers_to_clean = set()
known_consumers = set()
for _, buffer_id in bundle_context_manager.stage_data_outputs.items():
for (consuming_stage_name, consuming_transform) in \
runner_execution_context.buffer_id_to_consumer_pairs.get(buffer_id,
[]):
for transform_id, buffer_id in (
bundle_context_manager.stage_data_outputs.items()):
for (consuming_stage_name, consuming_transform
) in runner_execution_context.buffer_id_to_consumer_pairs.get(
buffer_id, []):
buffer = runner_execution_context.pcoll_buffers.get(buffer_id, None)

if (buffer_id in runner_execution_context.pcoll_buffers and
Expand All @@ -840,6 +841,12 @@ def _execute_bundle(self,
# so we create a copy of the buffer for every new stage.
runner_execution_context.pcoll_buffers[buffer_id] = buffer.copy()
buffer = runner_execution_context.pcoll_buffers[buffer_id]
# When the buffer is not in the pcoll_buffers, it means that the
# it could be an empty PCollection. In this case, get the buffer using
# the buffer id and transform id
if buffer is None:
buffer = bundle_context_manager.get_buffer(buffer_id, transform_id)
assert buffer is not None
AnandInguva marked this conversation as resolved.
Show resolved Hide resolved
AnandInguva marked this conversation as resolved.
Show resolved Hide resolved

# If the buffer has already been added to be consumed by
# (stage, transform), then we don't need to add it again. This case
Expand All @@ -854,7 +861,7 @@ def _execute_bundle(self,
# MAX_TIMESTAMP for the downstream stage.
runner_execution_context.queues.watermark_pending_inputs.enque(
((consuming_stage_name, timestamp.MAX_TIMESTAMP),
DataInput({consuming_transform: buffer}, {}))) # type: ignore
DataInput({consuming_transform: buffer}, {})))

for bid in buffers_to_clean:
if bid in runner_execution_context.pcoll_buffers:
Expand Down