Repository: beam Updated Branches: refs/heads/master e8fdd373c -> bf6dda320
Fix PValue input in _PubSubReadEvaluator Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/a75202f3 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/a75202f3 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/a75202f3 Branch: refs/heads/master Commit: a75202f344f22be5c5fdf62b3eb54a151ad29af6 Parents: e8fdd37 Author: Charles Chen <[email protected]> Authored: Wed Jul 5 16:18:51 2017 -0700 Committer: Ahmet Altay <[email protected]> Committed: Wed Jul 5 18:08:59 2017 -0700 ---------------------------------------------------------------------- sdks/python/apache_beam/runners/direct/transform_evaluator.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/a75202f3/sdks/python/apache_beam/runners/direct/transform_evaluator.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/runners/direct/transform_evaluator.py b/sdks/python/apache_beam/runners/direct/transform_evaluator.py index 641291d..cb2ace2 100644 --- a/sdks/python/apache_beam/runners/direct/transform_evaluator.py +++ b/sdks/python/apache_beam/runners/direct/transform_evaluator.py @@ -436,8 +436,9 @@ class _PubSubReadEvaluator(_TransformEvaluator): bundles = [bundle] else: bundles = [] - input_pvalue = self._applied_ptransform.inputs - if not input_pvalue: + if self._applied_ptransform.inputs: + input_pvalue = self._applied_ptransform.inputs[0] + else: input_pvalue = pvalue.PBegin(self._applied_ptransform.transform.pipeline) unprocessed_bundle = self._evaluation_context.create_bundle( input_pvalue)
