Is your feature request related to a problem? Please describe.
Most of Pregel-based algorithms have actually only one message. But in the current implementation even if the message is only one, structs are still generated internally:

    val sendMsgsColList = sendMsgs.toList.map { case (id, msg) =>
      struct(id.as(ID), msg.as("msg"))
    }

Describe the solution you would like
Add a separate implementation of Pregel engine for a single message case. In the top-level run check the length of the messages and choose the correct implementation. While it is possible to update the existing one it will create a mess of nested if-else everywhere.

I think we can add a lot of additional optimizations for the single-message case in the future, so it is better to have a clear standalone Pregel engine for this case. For users it will be hidden under the top-level Pregel API.

Component

• Scala Core Internal
• Scala API
• Spark Connect Plugin
• Infrastructure
• PySpark Classic
• PySpark Connect

Additional context
No API changes. It should work explicitly.

Are you planning on creating a PR?

• I'm willing to make a pull-request