[Bug]: The Python DoOutputsTuple force Tagged PCollection to is_bounded=True

[ad-momo]

What happened?

When you apply a beam.Partition on a unbounded PCollection, the tagged PCollections from DoOutputsTuple have is_bounded=True.

Normally, if the source is unbounded then the Pcollections from partition are unbounded too, no ?

beam/sdks/python/apache_beam/pvalue.py

Line 307 in bc06581

pcoll = PCollection(self._pipeline, tag=tag, element_type=typehints.Any)

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[ad-momo]

A workaround waiting for support response :

# The Taggedoutput have to have same is_bounded state like the source
initial_get_item = DoOutputsTuple.__getitem__


def new_get_item(self: DoOutputsTuple, tag: int | str | None):
    tag_not_in_pcolls = tag not in self._pcolls

    pcoll = initial_get_item(self, tag)

    if tag_not_in_pcolls:
        assert self.producer is not None
        pval = self.producer.parts[0].outputs[None]

        # pass bounded state
        pcoll.is_bounded = pval.is_bounded

    return pcoll


DoOutputsTuple.__getitem__ = new_get_item

[damccorm]

Looks like we output a DoOutputsTuple here -

beam/sdks/python/apache_beam/transforms/core.py

Line 1827 in b7acf57

return pvalue.DoOutputsTuple(

but in that object we have no logic for saying if a transform is bounded -

beam/sdks/python/apache_beam/pvalue.py

Line 307 in 39d3e36

pcoll = PCollection(self._pipeline, tag=tag, element_type=typehints.Any)

I think we should be able to iterate over self.producer.main_inputs to make a determination about boundedness

[ad-momo]

Thanks for response and fix 😃

In my workaround, I supposed the tagged output have the same bounded state as the main output self.producer.parts[0].outputs[None] because I don’t see any case contrary to that.

By the way, I don't know how to explain why the main output has the correct bounded state 😮

[tvalentyn]

@damccorm Thanks for investigating and fixing this!