lukecwik commented on a change in pull request #12289:
URL: https://github.com/apache/beam/pull/12289#discussion_r456745203
##########
File path: sdks/python/apache_beam/transforms/environments.py
##########
@@ -595,6 +595,7 @@ def _python_sdk_capabilities_iter():
yield common_urns.protocols.LEGACY_PROGRESS_REPORTING.urn
yield common_urns.protocols.WORKER_STATUS.urn
yield 'beam:version:sdk_base:' + DockerEnvironment.default_docker_image()
+ yield common_urns.sdf_components.TRUNCATE_SIZED_RESTRICTION.urn
Review comment:
Lets not add this URN until we are sure with internal testing with
Dataflow just in case we find that we need to fix/change something.
##########
File path: sdks/python/apache_beam/runners/worker/operations.py
##########
@@ -743,6 +743,20 @@ def pcollection_count_monitoring_infos(self,
tag_to_pcollection_id):
return infos
+class SdfTruncateSizedRestrictions(DoOperation):
+ def __init__(self, *args, **kwargs):
+ super(SdfTruncateSizedRestrictions, self).__init__(*args, **kwargs)
+
+ def current_element_progress(self):
+ # type: () -> Optional[iobase.RestrictionProgress]
+ return self.consumers[0][0].current_element_progress()
+
+ def try_split(
+ self, fraction_of_remainder
+ ): # type: (...) -> Optional[Tuple[Iterable[SdfSplitResultsPrimary],
Iterable[SdfSplitResultsResidual]]]
+ return self.consumers[0][0].try_split(fraction_of_remainder)
Review comment:
```suggestion
return self.receivers[0].try_split(fraction_of_remainder)
```
We should be relying on the ConsumerSet to properly forward the
split/progress call. See:
https://github.com/apache/beam/blob/af2d6b0379d64b522ecb769d88e9e7e7b8900208/sdks/python/apache_beam/runners/worker/operations.py#L150
https://github.com/apache/beam/blob/af2d6b0379d64b522ecb769d88e9e7e7b8900208/sdks/python/apache_beam/runners/worker/operations.py#L198
##########
File path: sdks/python/apache_beam/runners/worker/operations.py
##########
@@ -743,6 +743,20 @@ def pcollection_count_monitoring_infos(self,
tag_to_pcollection_id):
return infos
+class SdfTruncateSizedRestrictions(DoOperation):
+ def __init__(self, *args, **kwargs):
+ super(SdfTruncateSizedRestrictions, self).__init__(*args, **kwargs)
+
+ def current_element_progress(self):
+ # type: () -> Optional[iobase.RestrictionProgress]
+ return self.consumers[0][0].current_element_progress()
Review comment:
```suggestion
return self.receivers[0].current_element_progress()
```
We should be relying on the ConsumerSet to properly forward the
split/progress call. See:
https://github.com/apache/beam/blob/af2d6b0379d64b522ecb769d88e9e7e7b8900208/sdks/python/apache_beam/runners/worker/operations.py#L150
https://github.com/apache/beam/blob/af2d6b0379d64b522ecb769d88e9e7e7b8900208/sdks/python/apache_beam/runners/worker/operations.py#L198
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
