[Bug]: Flink might ignore Reshuffle in batch mode

[je-ik]

What happened?

Translation for Reshuffle in batch uses optimizer hint, which can be ignored. This results in scalability issues, for instance in batch SDF.

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

[kennknowles]

But it is by design that it could be ignored. Is it extremely likely to be ignored? Like you saw real performance issues?

IMO forcing a groupby and then expansion is probably excessive cost on top of what reshuffle really needs to do

[je-ik]

Yes, I saw performance issues on batch SDF. What is even worse is that this was not deterministic. We either have to (somehow) force redistribution on Reshuffle or use GBK-unGBK in the implementation of SDF.

Prior to the change to hints, there was call to rebalance, which would be good enoughz although it was changed in order to not ignore chain of Reshuffles. I'm not sure why this matters, I will try to find more info.