fix bug that execution info for non root mpp task is not acurate enough (#1835)

[ti-srebot]

cherry-pick #1835 to release-5.0
You can switch your code base to this Pull Request by using git-extras:

# In tics repo:
git pr https://github.com/pingcap/tics/pull/1840

After apply modifications, you can push your change to this PR via:

git push [email protected]:ti-srebot/tics.git pr/1840:release-5.0-eff941f2e2c3

---

What problem does this PR solve?

Issue Number: close #xxx

Problem Summary:
In current implementation, the execution info for non root mpp task is merged in root mpp task, however, the merge algrithm is not good enough so sometimes the execution time of non root mpp task is very misleading, for example:

d      estRows actRows task    access object   execution info  operator info   memory  disk
Projection_10   0.00    4399    root            time:26.2s, loops:7, Concurrency:OFF    jrscfour_realtime.ods_clrg_pln_cf_inf.bfr_to_stm_txn_id, jrscfour_realtime.ods_tnar_unn_inf.txn_dt, jrscfour_realtime.ods_tnar_unn_inf.rcrd_crt_dt_tm, jrscfour_realtime.ods_tnar_unn_└─TableReader_26        0.00    4399    root            time:26.2s, loops:7, cop_task: {num: 6, max: 0s, min: 0s, avg: 0s, p95: 0s, copr_cache_hit_ratio: 0.00} data:ExchangeSender_25  N/A     N/A
  └─ExchangeSender_25   0.00    4399    cop[tiflash]            tiflash_task:{proc max:20.2s, min:766ms, p80:19.5s, p95:20.2s, iters:295, tasks:6, threads:36}  ExchangeType: PassThrough, tasks: [5, 6, 7]     N/A     N/A
    └─HashJoin_11       0.00    4399    cop[tiflash]            tiflash_task:{proc max:20.2s, min:766ms, p80:19.5s, p95:20.2s, iters:295, tasks:6, threads:36}  left outer join, equal:[eq(jrscfour_realtime.ods_clrg_pln_cf_inf.multi_tenancy_id, jrscfour_realtime.ods_std_r
      ├─ExchangeReceiver_24(Build)      30000.00        90000   cop[tiflash]            tiflash_task:{proc max:471ms, min:0s, p80:253.1ms, p95:471ms, iters:3, tasks:6, threads:60}             N/A     N/A
      │ └─ExchangeSender_23     30000.00        30000   cop[tiflash]            tiflash_task:{proc max:40.6ms, min:0s, p80:40.6ms, p95:40.6ms, iters:1, tasks:6, threads:1}     ExchangeType: Broadcast, tasks: [4]     N/A     N/A
      │   └─Selection_22        30000.00        30000   cop[tiflash]            tiflash_task:{proc max:40.6ms, min:0s, p80:40.6ms, p95:40.6ms, iters:1, tasks:6, threads:1}     eq(jrscfour_realtime.ods_std_rmtnwy_base_inf.multi_tenancy_id, "CN000"), not(isnull(jrscfour_r      │     └─TableFullScan_21  30000.00        30000   cop[tiflash]    table:T3        tiflash_task:{proc max:39.6ms, min:0s, p80:39.6ms, p95:39.6ms, iters:1, tasks:6, threads:1}     keep order:false        N/A     N/A
      └─HashJoin_14(Probe)      0.00    4399    cop[tiflash]            tiflash_task:{proc max:20.1s, min:738ms, p80:19.2s, p95:20.1s, iters:295, tasks:6, threads:36}  inner join, equal:[eq(jrscfour_realtime.ods_clrg_pln_cf_inf.oprt_pltfrm_txn_id, jrscfour_realtime.ods_
        ├─ExchangeReceiver_18(Build)    0.00    16500   cop[tiflash]            tiflash_task:{proc max:19.4s, min:0s, p80:19.2s, p95:19.4s, iters:2424, tasks:6, threads:60}            N/A     N/A
        │ └─ExchangeSender_17   0.00    5500    cop[tiflash]            tiflash_task:{proc max:2.58s, min:0s, p80:2.4s, p95:2.58s, iters:808, tasks:6, threads:46}      ExchangeType: Broadcast, tasks: [1, 2, 3]       N/A     N/A
        │   └─Selection_16      0.00    5500    cop[tiflash]            tiflash_task:{proc max:2.58s, min:0s, p80:2.4s, p95:2.58s, iters:808, tasks:6, threads:46}      eq(jrscfour_realtime.ods_clrg_pln_cf_inf.act_rcvpy_dt, 2000-01-01 00:00:00.000000), eq(jrscfour_realti
        │     └─TableFullScan_15        50000000.00     50000000        cop[tiflash]    table:T1        tiflash_task:{proc max:2.22s, min:0s, p80:1.98s, p95:2.22s, iters:859, tasks:6, threads:46}     keep order:false        N/A     N/A
        └─Selection_20(Probe)   13402171.90     13403267        cop[tiflash]            tiflash_task:{proc max:2.01s, min:418.1ms, p80:1.54s, p95:2.01s, iters:295, tasks:6, threads:36}        eq(jrscfour_realtime.ods_tnar_unn_inf.accnm, "01"), eq(jrscfour_realtime.ods_t
          └─TableFullScan_19    16666660.00     16666660        cop[tiflash]    table:T2        tiflash_task:{proc max:1.33s, min:299ms, p80:462ms, p95:1.33s, iters:295, tasks:6, threads:36}  keep order:false        N/A     N/A

The execution time of mpp task end with ExchangeSender_17 is very short, which may make the user think the data transform from ExchangeSender_17 to ExchangeReceiver_18 cost most of the time. However after a more careful debug, I found ExchangeSender_17 costs at least 18s

What is changed and how it works?

Proposal: xxx

What's Changed:

Some backgrounds:

1. According to the protocal, For each response returned by TiFlash, we need return the execution_summaries, the execution summary includes execution time, processed rows, processed iters and concurrency
2. For a unary rpc, only one response will be returned at the end of the rpc, while for a streaming rpc, multiple responses can be returned during the rpc
3. Currently when TiDB handle the execution_summaries returned by TiFlash/TiKV, it is not aware of whether the rpc call is unray or streaming, it treats all the responses as the unary call reponse, which means for processed rows, it will add all the processed rows in execution_sumamries as the total processed rows.
4. Obviously, for a streaming rpc, if we return the original execution_summaries, the data will be amplified in TiDB, so in order to make the data more acurate in TiDB side, for a streaming call, we always return a delta execution_summaries
5. A non root mpp task is always a streaming rpc, and it's parent mpp task will use TiRemoteBlockInputStream to receive the response from the non root mpp task. After decoded the response, TiRemoteBlockInputStream will merge the execution_summaries. When merge the execution_summaries for the streaming call, it merges the delta execution_summaries explicitly.

The problem is: Although TiRemoteBlockInputStream merges the delta execution_summaries explicitly, it just merge part of the streaming call's response, because when reading data from remote, we usually use multiple TiRemoteBlockInputStream to speed up the query, so every TiRemoteBlockInputStream only receive part of the execution info for the non root mpp task, which makes it much more less than the orginal ones.

This pr fix it by supporting streaming execution_summaries in TiRemoteBlockInputStream, so for a non root mpp task, it just return the orignal execution_summaries, and TiRemoteBlockInputStream will handle it correctly, however, since TiDB still not support streaming execution_summaries, TiFlash will return delta execution_summaries if the receiver side is TiDB. In the future, after TiDB support streaming execution_summaries, we can remove delta execution summaries totally.

After this pr, the above example will get the following output:

id      estRows actRows task    access object   execution info  operator info   memory  disk
Projection_10   0.00    4399    root            time:27.3s, loops:7, Concurrency:OFF    jrscfour_realtime.ods_clrg_pln_cf_inf.bfr_to_stm_txn_id, jrscfour_realtime.ods_tnar_unn_inf.txn_dt, jrscfour_realtime.ods_tnar_unn_inf.rcrd_crt_dt_tm, jrscfour_realtime.ods_tnar_unn_
└─TableReader_26        0.00    4399    root            time:27.3s, loops:7, cop_task: {num: 6, max: 0s, min: 0s, avg: 0s, p95: 0s, copr_cache_hit_ratio: 0.00} data:ExchangeSender_25  N/A     N/A
  └─ExchangeSender_25   0.00    4399    cop[tiflash]            tiflash_task:{proc max:22.1s, min:350ms, p80:20.1s, p95:22.1s, iters:295, tasks:6, threads:36}  ExchangeType: PassThrough, tasks: [5, 6, 7]     N/A     N/A
    └─HashJoin_11       0.00    4399    cop[tiflash]            tiflash_task:{proc max:22.1s, min:350ms, p80:20.1s, p95:22.1s, iters:295, tasks:6, threads:36}  left outer join, equal:[eq(jrscfour_realtime.ods_clrg_pln_cf_inf.multi_tenancy_id, jrscfour_realtime.ods_std_r
      ├─ExchangeReceiver_24(Build)      30000.00        90000   cop[tiflash]            tiflash_task:{proc max:959.9ms, min:0s, p80:518.6ms, p95:959.9ms, iters:3, tasks:6, threads:60}         N/A     N/A
      │ └─ExchangeSender_23     30000.00        30000   cop[tiflash]            tiflash_task:{proc max:74.8ms, min:0s, p80:74.8ms, p95:74.8ms, iters:1, tasks:6, threads:1}     ExchangeType: Broadcast, tasks: [4]     N/A     N/A
      │   └─Selection_22        30000.00        30000   cop[tiflash]            tiflash_task:{proc max:74.8ms, min:0s, p80:74.8ms, p95:74.8ms, iters:1, tasks:6, threads:1}     eq(jrscfour_realtime.ods_std_rmtnwy_base_inf.multi_tenancy_id, "CN000"), not(isnull(jrscfour_r
      │     └─TableFullScan_21  30000.00        30000   cop[tiflash]    table:T3        tiflash_task:{proc max:72.8ms, min:0s, p80:72.8ms, p95:72.8ms, iters:1, tasks:6, threads:1}     keep order:false        N/A     N/A
      └─HashJoin_14(Probe)      0.00    4399    cop[tiflash]            tiflash_task:{proc max:21.7s, min:348ms, p80:19.7s, p95:21.7s, iters:295, tasks:6, threads:36}  inner join, equal:[eq(jrscfour_realtime.ods_clrg_pln_cf_inf.oprt_pltfrm_txn_id, jrscfour_realtime.ods_
        ├─ExchangeReceiver_18(Build)    0.00    16500   cop[tiflash]            tiflash_task:{proc max:20.4s, min:0s, p80:20.2s, p95:20.4s, iters:2424, tasks:6, threads:60}            N/A     N/A
        │ └─ExchangeSender_17   0.00    5500    cop[tiflash]            tiflash_task:{proc max:19.7s, min:0s, p80:19.7s, p95:19.7s, iters:808, tasks:6, threads:46}     ExchangeType: Broadcast, tasks: [1, 2, 3]       N/A     N/A
        │   └─Selection_16      0.00    5500    cop[tiflash]            tiflash_task:{proc max:19.7s, min:0s, p80:19.7s, p95:19.7s, iters:808, tasks:6, threads:46}     eq(jrscfour_realtime.ods_clrg_pln_cf_inf.act_rcvpy_dt, 2000-01-01 00:00:00.000000), eq(jrscfour_realti
        │     └─TableFullScan_15        50000000.00     50000000        cop[tiflash]    table:T1        tiflash_task:{proc max:14.7s, min:0s, p80:14.7s, p95:14.7s, iters:859, tasks:6, threads:46}     keep order:false        N/A     N/A
        └─Selection_20(Probe)   13402171.90     13403267        cop[tiflash]            tiflash_task:{proc max:1.89s, min:262ms, p80:1.74s, p95:1.89s, iters:295, tasks:6, threads:36}  eq(jrscfour_realtime.ods_tnar_unn_inf.accnm, "01"), eq(jrscfour_realtime.ods_tnar_unn_
          └─TableFullScan_19    16666660.00     16666660        cop[tiflash]    table:T2        tiflash_task:{proc max:863.8ms, min:131ms, p80:778ms, p95:863.8ms, iters:295, tasks:6, threads:36}      keep order:false        N/A     N/A

As we can see, the execution time of ExchangeSender_17 is 19.7s, which is much more close to the actual execution time.

Related changes

• PR to update pingcap/docs/pingcap/docs-cn:
• Need to cherry-pick to the release branch:

Check List

Tests

• Manual test (add detailed scripts or steps below)

Side effects

Release note

• fix bug that execution info for non root mpp task is not acurate enough

[windtalker]

/run-all-tests