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

[ad-momo]

Hello,

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)

[ad-momo]

Hello,

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

[ad-momo]

Re-create with report issue template #29196