KevinGG commented on a change in pull request #12249:
URL: https://github.com/apache/beam/pull/12249#discussion_r454707919
##########
File path: sdks/python/apache_beam/runners/interactive/pipeline_instrument.py
##########
@@ -556,19 +556,26 @@ def _process(self, pcoll):
if not self._pin._user_pipeline:
# Retrieve a reference to the user defined pipeline instance.
self._pin._user_pipeline = user_pcoll.pipeline
- # Once user_pipeline is retrieved, check if the user pipeline
- # contains any source to cache. If so, current cache manager held
- # by current interactive environment might get wrapped into a
- # streaming cache, thus re-assign the reference to that cache
- # manager.
+ # Retrieve a reference to the cache manager for the user defined
+ # pipeline instance.
+ self._pin._cache_manager = ie.current_env().get_cache_manager(
Review comment:
This line is the "official" way within the module to initialize the
cache manager.
(Outside the module, each time a user appends a PTransform, a cache manager
will be created_if_absent for the user defined pipeline)
TL;DR: L577 is only a fallback when the user defined pipeline cannot be
identified from the given pipeline. It might not get executed either.
There are 2 scenarios that the user defined pipeline can not be identified:
- The given pipeline is empty without any PCollection in it.
- The given pipeline is the user defined pipeline (would not happen
internally except tests).
In either scenarios, use the given pipeline as the user pipeline as a
fallback last resort.
----------------------------------------------------------------
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:
us...@infra.apache.org
