[Bug]: Side effect during the for loop execution in beam Pipeline

[giovanni-prestipino]

What happened?

When using for loop, to execute PTransform, all loops seem to use the last value of an iterator.

import apache_beam as beam

def main():
    with beam.Pipeline() as p:
        p_f = p | beam.Create([1, 2, 3, 4, 5])

        for i in range(0, 3):
            p_f = p_f | f"{i}" >> beam.Map(lambda x: x + i)

        p_f | beam.Map(print)

if __name__ == "__main__":
    main()

As a result of the code, I expected to print

4
5
6
7
8

But I got:

7
8
9
10
11

Moreover, I found a workaround defining a new sum function:

import apache_beam as beam


def _sum(pcoll, i):
    return pcoll | f"{i}" >> beam.Map(lambda x: x + i)

def main():
    with beam.Pipeline() as p:
        p_f = p | beam.Create([1, 2, 3, 4, 5])

        for i in range(0, 3):
            p_f = _sum(p_f, i)

        p_f | beam.Map(print)

if __name__ == "__main__":
    main()

I am using the following versions:

• apache_beam==2.55.0
• python==3.10.12

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

[giovanni-prestipino]

Maybe the problem is that the loop parameter should be passed to the Map as a side input:

def main():
    with beam.Pipeline() as p:
        p_f = p | beam.Create([1, 2, 3, 4, 5])

        for i in range(0, 3):
            p_f = p_f | f"{i}" >> beam.Map(lambda x, k: x + k, k=i)

        p_f | beam.Map(print)

if __name__ == "__main__":
    main()
``

[liferoad]

For your first example, I think this is expected. the pipeline won't be executed after the entire pipeline is construct. After that, the variable i has the final value, which will be passed in for the final execution.