[Bug]: Beam YAML pipeline fails when using ReadFromPubSub with downstream Join transform

[Polber]

What happened?

Beam YAML pipelines fail during expansion when using a Join transform with a ReadFromPubSub source. The issue lies with Join expecting a Row schema instead of BeamSchema (which are fundamentally the same but implemented differently), and ReadFromPubSub uses the latter.

The following pipeline can be used to reproduce the error:

pipeline:
  transforms:
    - type: ReadFromPubSub
      config:
        topic: projects/daui-interpod-p2-management/topics/ryanmadden-topic
        format: json
        schema:
          type: object
          properties:
            foo: {type: number}
    - type: Create
      config:
        elements:
          - bar: 1
            name: baz
    - type: Join
      name: Transform
      config:
        type: inner
        equalities:
          - input0: foo
            input1: bar
      input:
        input0: ReadFromPubSub
        input1: Create
    - type: LogForTesting
      input: Transform
  windowing:
    type: fixed
    size: 60
options:
  streaming: true

To mitigate this issue, an intermediate MapToFields can be added to remap the PCollection into a Row schema, for example:

pipeline:
  transforms:
    - type: ReadFromPubSub
      config:
        topic: projects/daui-interpod-p2-management/topics/ryanmadden-topic
        format: json
        schema:
          type: object
          properties:
            foo: { type: number }
    - type: MapToFields
      input: ReadFromPubSub
      config:
        fields:
          foo: foo
    - type: Create
      config:
        elements:
          - bar: 1
            name: baz
    - type: Join
      name: Transform
      config:
        type: left
        equalities:
          - input0: foo
            input1: bar
      input:
        input0: MapToFields
        input1: Create
    - type: LogForTesting
      input: Transform
  windowing:
    type: fixed
    size: 60
options:
  streaming: true

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