apache / apache/beam

Python typehints: with_output_types breaks multi-output dofns

Open
#19,912 0 comments 0 reactions 0 assignees View on GitHub
bug core P3 python types
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

```

def test_typed_multi_pardo(self):
p = TestPipeline()
res = (p
| beam.Create([1,
2, 3])
| beam.Map(lambda e: e).with_outputs().with_output_types(int))
self.assertIsNotNone(res[None].element_type)

res_main = (res[None]
| 'id_none' >> beam.ParDo(lambda e: [e]).with_input_types(int))

assert_that(res_main, equal_to([1, 2, 3]), label='none_check')
p.run()

```

Fails with:
```

typed_pipeline_test.py:212:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _ _
../pvalue.py:113: in __or__
return self.pipeline.apply(ptransform, self)
../pipeline.py:528:
in apply
transform.type_check_outputs(pvalueish_result)
../transforms/ptransform.py:386: in type_check_outputs

self.type_check_inputs_or_outputs(pvalueish, 'output')
../transforms/ptransform.py:401: in type_check_inputs_or_outputs

if pvalue_.element_type is None:
../pvalue.py:241: in __getattr__
return self[tag]
_ _ _ _
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = )]> at 0x7fa9513f3048>
tag
= 'element_type'

def __getitem__(self, tag):
# Accept int tags so that we can look at
Partition tags with the
# same ints that we used in the partition function.
# TODO(gildea):
Consider requiring string-based tags everywhere.
# This will require a partition function that
does not return ints.
if isinstance(tag, int):
tag = str(tag)
if tag == self._main_tag:

tag = None
elif self._tags and tag not in self._tags:
raise ValueError(

"Tag '%s' is neither the main tag '%s' "
"nor any of the tags %s" % (

tag, self._main_tag, self._tags))
# Check if we accessed this tag before.
if tag
in self._pcolls:
return self._pcolls[tag]

if tag is not None:
self._transform.output_tags.add(tag)

pcoll = PCollection(self._pipeline, tag=tag, element_type=typehints.Any)
# Transfer
the producer from the DoOutputsTuple to the resulting
# PCollection.
> pcoll.producer
= self.producer.parts[0]
E AttributeError: 'NoneType' object has no attribute 'parts'

../pvalue.py:266:
AttributeError

```

Imported from Jira [BEAM-8782](https://issues.apache.org/jira/browse/BEAM-8782). Original Jira may contain additional context.
Reported by: udim.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.