You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ro...@apache.org on 2017/09/19 22:23:51 UTC
[2/3] beam git commit: Support multiple materializations of the same
pvalue.
Support multiple materializations of the same pvalue.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/07c08cc8
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/07c08cc8
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/07c08cc8
Branch: refs/heads/master
Commit: 07c08cc80135ff37c1224e0e659306f65481ad65
Parents: 72960b3
Author: Robert Bradshaw <ro...@gmail.com>
Authored: Fri Sep 15 15:40:10 2017 -0700
Committer: Robert Bradshaw <ro...@gmail.com>
Committed: Tue Sep 19 15:23:28 2017 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/runner.py | 19 ++++++++++---------
sdks/python/apache_beam/transforms/ptransform.py | 17 +++++++++++------
.../apache_beam/transforms/ptransform_test.py | 9 +++++++++
3 files changed, 30 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/07c08cc8/sdks/python/apache_beam/runners/runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/runner.py b/sdks/python/apache_beam/runners/runner.py
index 131d54f..bdabd81 100644
--- a/sdks/python/apache_beam/runners/runner.py
+++ b/sdks/python/apache_beam/runners/runner.py
@@ -247,17 +247,18 @@ class PValueCache(object):
self._cache[
self.to_cache_key(transform, tag)] = [value, transform.refcounts[tag]]
- def get_pvalue(self, pvalue):
+ def get_pvalue(self, pvalue, decref=True):
"""Gets the value associated with a PValue from the cache."""
self._ensure_pvalue_has_real_producer(pvalue)
try:
value_with_refcount = self._cache[self.key(pvalue)]
- value_with_refcount[1] -= 1
- logging.debug('PValue computed by %s (tag %s): refcount: %d => %d',
- pvalue.real_producer.full_label, self.key(pvalue)[1],
- value_with_refcount[1] + 1, value_with_refcount[1])
- if value_with_refcount[1] <= 0:
- self.clear_pvalue(pvalue)
+ if decref:
+ value_with_refcount[1] -= 1
+ logging.debug('PValue computed by %s (tag %s): refcount: %d => %d',
+ pvalue.real_producer.full_label, self.key(pvalue)[1],
+ value_with_refcount[1] + 1, value_with_refcount[1])
+ if value_with_refcount[1] <= 0:
+ self.clear_pvalue(pvalue)
return value_with_refcount[0]
except KeyError:
if (pvalue.tag is not None
@@ -268,8 +269,8 @@ class PValueCache(object):
else:
raise
- def get_unwindowed_pvalue(self, pvalue):
- return [v.value for v in self.get_pvalue(pvalue)]
+ def get_unwindowed_pvalue(self, pvalue, decref=True):
+ return [v.value for v in self.get_pvalue(pvalue, decref)]
def clear_pvalue(self, pvalue):
"""Removes a PValue from the cache."""
http://git-wip-us.apache.org/repos/asf/beam/blob/07c08cc8/sdks/python/apache_beam/transforms/ptransform.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/ptransform.py b/sdks/python/apache_beam/transforms/ptransform.py
index f630977..7cf1441 100644
--- a/sdks/python/apache_beam/transforms/ptransform.py
+++ b/sdks/python/apache_beam/transforms/ptransform.py
@@ -75,10 +75,12 @@ class _PValueishTransform(object):
"""
def visit_nested(self, node, *args):
if isinstance(node, (tuple, list)):
- # namedtuples require unpacked arguments in their constructor,
- # but do have a _make method that takes a sequence.
- return getattr(node.__class__, '_make', node.__class__)(
- [self.visit(x, *args) for x in node])
+ args = [self.visit(x, *args) for x in node]
+ if isinstance(node, tuple) and hasattr(node.__class__, '_make'):
+ # namedtuples require unpacked arguments in their constructor
+ return node.__class__(*args)
+ else:
+ return node.__class__(args)
elif isinstance(node, dict):
return node.__class__(
{key: self.visit(value, *args) for (key, value) in node.items()})
@@ -102,7 +104,9 @@ class _MaterializedDoOutputsTuple(pvalue.DoOutputsTuple):
self._pvalue_cache = pvalue_cache
def __getitem__(self, tag):
- return self._pvalue_cache.get_unwindowed_pvalue(self._deferred[tag])
+ # Simply accessing the value should not use it up.
+ return self._pvalue_cache.get_unwindowed_pvalue(
+ self._deferred[tag], decref=False)
class _MaterializePValues(_PValueishTransform):
@@ -111,7 +115,8 @@ class _MaterializePValues(_PValueishTransform):
def visit(self, node):
if isinstance(node, pvalue.PValue):
- return self._pvalue_cache.get_unwindowed_pvalue(node)
+ # Simply accessing the value should not use it up.
+ return self._pvalue_cache.get_unwindowed_pvalue(node, decref=False)
elif isinstance(node, pvalue.DoOutputsTuple):
return _MaterializedDoOutputsTuple(node, self._pvalue_cache)
else:
http://git-wip-us.apache.org/repos/asf/beam/blob/07c08cc8/sdks/python/apache_beam/transforms/ptransform_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/ptransform_test.py b/sdks/python/apache_beam/transforms/ptransform_test.py
index c328cb1..8237c52 100644
--- a/sdks/python/apache_beam/transforms/ptransform_test.py
+++ b/sdks/python/apache_beam/transforms/ptransform_test.py
@@ -686,6 +686,15 @@ class PTransformTest(unittest.TestCase):
flat = res | beam.Flatten()
self.assertEqual(sorted(flat), [1, 8])
+ def test_tuple_twice(self):
+ class Duplicate(PTransform):
+ def expand(self, pcoll):
+ return pcoll, pcoll
+
+ res1, res2 = [1, 2, 4, 8] | Duplicate()
+ self.assertEqual(sorted(res1), [1, 2, 4, 8])
+ self.assertEqual(sorted(res2), [1, 2, 4, 8])
+
@beam.ptransform_fn
def SamplePTransform(pcoll):