MPP shared CTE may wait for missing local sinks with UNION ALL producer #68837

Bug Report

Parent tracking issue: #61241

MPP shared CTE currently assumes that CteSinkNum/CteSourceNum can be counted globally from the split MPP DAG. This is not safe when a CTE producer contains UNION ALL and the union branches are scheduled to different TiFlash workers.

1. Minimal reproduce step (Required)

Use a TiFlash MPP setup with at least two TiFlash workers and enable MPP shared CTE execution:

set @@tidb_enforce_mpp = 'on';
set @@tidb_opt_enable_mpp_shared_cte_execution = 'on';

create table t(a int, b int, c int);
create table s(a int, b int, c int);
alter table t set tiflash replica 1;
alter table s set tiflash replica 1;

with c as (
  select * from t
  union all
  select * from s
)
select *
from c c1 join c c2 on c1.a = c2.b;

The issue requires the two table scan branches of the CTE producer to be assigned to different TiFlash workers. This can happen because MPP table scan tasks are generated from the actual scan ranges/regions and are not padded to all TiFlash workers.

One observed fragment layout from a planner probe:

FRAG 00: PhysicalCTESink workers=[tiflash0] tableScan=t cte=sink(id=0 sinkNum=2 sourceNum=2)
FRAG 01: PhysicalCTESink workers=[tiflash1] tableScan=s cte=sink(id=0 sinkNum=2 sourceNum=2)
FRAG 02: ExchangeSender workers=[tiflash0 tiflash1] cte=source(id=0 sinkNum=2 sourceNum=2)
FRAG 03: ExchangeSender workers=[tiflash0 tiflash1] cte=source(id=0 sinkNum=2 sourceNum=2)
FRAG 04: ExchangeSender workers=[tiflash0 tiflash1] receivers=2

2. What did you expect to see? (Required)

The query should either execute correctly with colocated MPP shared CTE, or the planner should avoid MPP shared CTE for this shape.

For a colocated n:n CTE execution model, every TiFlash worker that runs a CTESource should see the local number of CTESink executors that matches the CteSinkNum sent to TiFlash.

3. What did you see instead (Required)

TiDB generates two PhysicalCTESink fragments for the same CTE because UNION ALL is untwisted into multiple fragments. It then writes CteSinkNum=2 globally to both CTESink and CTESource executors.

However, in the fragment layout above:

• tiflash0 has only FRAG 00 as a local CTESink, but its local CTESource expects CteSinkNum=2.
• tiflash1 has only FRAG 01 as a local CTESink, but its local CTESource expects CteSinkNum=2.

TiFlash registers CTE sinks in the node-local CTEManager and checks registered_sink_num == expected_sink_num before a source can read. With the layout above, each node can register only one local sink while expecting two, so the source can wait until timeout.

The same risk applies to equivalent shapes where the CTE producer's probe-side path is split by UNION ALL, for example Projection/Selection over UNION ALL or a hash join whose probe side is UNION ALL.

4. What is your TiDB version? (Required)

latest master

SELECT tidb_version() output is not attached because this report is based on planner fragment generation plus TiFlash CTE runtime contract inspection, not an end-to-end cluster run.