[Bug]: Prism raises error when running a Python SDK pipeline with Create transform

[shunping]

What happened?

When running a Python pipeline with Create transform on prism, prism raises an error.

with beam.Pipeline(options=options) as p:
  _ = (p | beam.Create([1]) | beam.Map(print))

Prism error:

"found 1 uses of features unimplemented in prism in job job:
unsupported feature "PTransform.Spec.Urn" set with value beam:transform:pickled_python:v1 Create/MaybeReshuffle"

The problem is related to MaybeReshuffle at https://github.com/apache/beam/blob/v2.64.0/sdks/python/apache_beam/transforms/core.py#L3971, where the exact same input pcollection is returned as the output pcollection, leading to an empty transform.

Here is another simple example to trigger the problem.

class EmptyTransform(beam.PTransform):
  def expand(self, input):
    return input

with beam.Pipeline(options=options) as p:
  _ = (p | beam.Impulse() | EmptyTransform() | beam.Map(print))

Prism error:

"found 1 uses of features unimplemented in prism in job job:
unsupported feature "PTransform.Spec.Urn" set with value beam:transform:pickled_python:v1 EmptyTransform"

Here is the proto of that transform in the pipeline proto:

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: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[shunping]

In the examples, there is an "empty" transform with zero subtransforms (not meeting the requirement of

beam/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go

Line 242 in 56e11aa

if len(t.GetSubtransforms()) > 0 {

), but non-empty payload (not meeting the requirement of

beam/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go

Line 248 in 56e11aa

if len(t.GetSpec().GetPayload()) == 0 {

). Therefore, it falls through to the check on

beam/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go

Line 253 in 56e11aa

check("PTransform.Spec.Urn", urn+" "+t.GetUniqueName(), "<doesn't exist>")

.

Only fixing the checking in jobservices/management is not enough, as then the empty transform and the transform prior to it will have the same output pcollection, which will lead to another error raised about "two producers on the same pcollection" (

beam/sdks/go/pkg/beam/runners/prism/internal/preprocess.go

Line 354 in 56e11aa

return nil, fmt.Errorf("computeFacts: two producers for one PCollection: %v and %v", p, link{Transform: tID, Local: local, Global: global})

)

To fix the second issue, we will have to skip the "empty" transform during graph preparation (prior to sorting the topology), since it is an no-op that causes prism confused.