Hi all, I have a Flink SQL job currently operational with the

table.exec.resource.default-parallelism=16

. It’s performing well generally, but there’s a join operation at the end of the job graph which seems to be causing a bottleneck. Upon retrieving the job

INFO

, I noticed that it’s using the

GLOBAL

ship strategy and it seems to be forcing the operator to maintain a parallelism of 1.

{
  "id": 96,
  "type": "Join[93]",
  "pact": "Operator",
  "contents": "...",
  "parallelism": 1,
  "predecessors": [
    {
      "id": 88,
      "ship_strategy": "GLOBAL",
      "side": "second"
    },
    {
      "id": 94,
      "ship_strategy": "GLOBAL",
      "side": "second"
    }
  ]
}

I’ve tried searching through the documentation for more information on this, but I haven’t found anything yet. I would like to know why is it doing this, and if there is a way to configure it. Thank you!

What is the nature of the join constraint? Is it an equi-join?

Hmm, that’s a good question. It is something in the form of

ON tbl1.id = tbl2.id_1 OR tbl1.id = tbl2.id_2

Facing a similar issue. I have 2 sources where one is a kafka source and other one is ‘upsert-kafka’. Not able to do a temporal join among these two and ended up in equi join. I’m planning to use broadcast hint to use broadcast join.

/*+ BROADCAST(t1) */

🍿

I'm just guessing here, but I wonder if the planner gave up on being able to execute this join in parallel because the join condition is looser than an equi-join. I would try rewriting the query as the union of two equi-joins and see if that makes a difference.

Hi David, Yes, I made this change yesterday and now it is looking better. Now it is not reutilizing the

left

table and reprocessing it for each part of the union, but at least it is not bottlenecking the application Thank you!

Hi @Hygor Knust - Is your join condition on different columns, for example t1.id1 = t2.id1 and t1.id2 = t2.id2?

Yes, correct

Okay. Just curious, wouldn’t that give duplicates?