Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

planner: prohibit StreamAgg with group keys for TiFlash #39547

Merged
merged 10 commits into from
Dec 1, 2022
29 changes: 29 additions & 0 deletions executor/tiflashtest/tiflash_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1207,6 +1207,35 @@ func TestAggPushDownCountStar(t *testing.T) {
tk.MustQuery("select count(*) from c, o where c.c_id=o.c_id").Check(testkit.Rows("5"))
}

func TestGroupStreamAggOnTiFlash(t *testing.T) {
store := testkit.CreateMockStore(t, withMockTiFlash(2))
tk := testkit.NewTestKit(t, store)

tk.MustExec("use test")
tk.MustExec("drop table if exists foo")
tk.MustExec("create table foo(a int, b int, c int, d int, primary key(a,b,c,d))")
tk.MustExec("alter table foo set tiflash replica 1")
tk.MustExec("insert into foo values(1,2,3,1),(1,2,3,6),(1,2,3,5)," +
"(1,2,3,2),(1,2,3,4),(1,2,3,7),(1,2,3,3),(1,2,3,0)")
tb := external.GetTableByName(t, tk, "test", "foo")
err := domain.GetDomain(tk.Session()).DDL().UpdateTableReplicaInfo(tk.Session(), tb.Meta().ID, true)
require.NoError(t, err)
tk.MustExec("set @@tidb_allow_mpp=0")
sql := "select a,b,c,count(*) from foo group by a,b,c order by a,b,c"
tk.MustQuery(sql).Check(testkit.Rows("1 2 3 8"))
rows := tk.MustQuery("explain " + sql).Rows()

for _, row := range rows {
resBuff := bytes.NewBufferString("")
fmt.Fprintf(resBuff, "%s\n", row)
res := resBuff.String()
// StreamAgg with group keys on TiFlash is not supported
if strings.Contains(res, "tiflash") {
require.NotContains(t, res, "StreamAgg")
}
}
}

func TestTiflashEmptyDynamicPruneResult(t *testing.T) {
store := testkit.CreateMockStore(t, withMockTiFlash(2))
tk := testkit.NewTestKit(t, store)
Expand Down
8 changes: 6 additions & 2 deletions planner/core/task.go
Original file line number Diff line number Diff line change
Expand Up @@ -1740,8 +1740,12 @@ func (p *PhysicalStreamAgg) attach2Task(tasks ...task) task {
t = cop.convertToRootTask(p.ctx)
attachPlan2Task(p, t)
} else {
copTaskType := cop.getStoreType()
partialAgg, finalAgg := p.newPartialAggregate(copTaskType, false)
storeType := cop.getStoreType()
// TiFlash doesn't support Stream Aggregation
if storeType == kv.TiFlash && len(p.GroupByItems) > 0 {
return invalidTask
}
partialAgg, finalAgg := p.newPartialAggregate(storeType, false)
if partialAgg != nil {
if cop.tablePlan != nil {
cop.finishIndexPlan()
Expand Down