[FLINK-27272] The plan for query with local sort is incorrect if adaptive batch scheduler is enabled

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.15.0
• Fix Version/s: 1.15.0, 1.16.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: godfrey he
• Reporter: godfrey he
• Created: 17/Apr/22 03:41
• Updated: 18/Apr/22 02:42
• Resolved: 18/Apr/22 02:42

Description

Add the following test case in ForwardHashExchangeTest

  @Test
    public void testRankWithHashShuffle() {
        util.verifyExecPlan(
                "SELECT * FROM (SELECT a, b, RANK() OVER(PARTITION BY a ORDER BY b) rk FROM T) WHERE rk <= 10");
    }

The result plan is:

Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a], orderBy=[b ASC], global=[true], select=[a, b, w0$o0])
+- Exchange(distribution=[forward])
   +- Sort(orderBy=[a ASC, b ASC])
      +- Exchange(distribution=[hash[a]])
         +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a], orderBy=[b ASC], global=[false], select=[a, b])
            +- Sort(orderBy=[a ASC, b ASC])
                +- TableSourceScan(table=[[default_catalog, default_database, T, project=[a, b], metadata=[]]], fields=[a, b])

There should be an additional Exchange(distribution=[forward]) node between local Rank and Sort, other wise if adaptive batch scheduler is enabled but operator chain is disabled, the result may be wrong. Because the parallelism for local Rank and Sort should be same, otherwise the adaptive batch scheduler may change their parallelism.

Local sort agg has the similar problem.

Issue Links

• links to: GitHub Pull Request #19497