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.

Sign up for GitHub

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

Jump to bottom

store/copr: partly revert pr/35975, do it correctly this time #55196

Merged
merged 7 commits into from
Aug 6, 2024
Merged

store/copr: partly revert pr/35975, do it correctly this time #55196

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1d5138a
store/copr: revert pr/35975, do it correctly this time
tiancaiamao Aug 5, 2024
b8ad4a6
partly revert
tiancaiamao Aug 5, 2024
0e89103
Merge branch 'master' into re
tiancaiamao Aug 5, 2024
85d4b1d
fix CI
tiancaiamao Aug 5, 2024
3cd26e4
fix TestCoprocessorOOMAction
tiancaiamao Aug 6, 2024
8635649
Merge branch 'master' into re
tiancaiamao Aug 6, 2024
4e9dfc4
make integration test stable
tiancaiamao Aug 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 38 additions & 11 deletions pkg/distsql/request_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,9 @@ type RequestBuilder struct {
kv.Request
is infoschema.MetaOnlyInfoSchema
err error

// When SetDAGRequest is called, builder will also this field.
dag *tipb.DAGRequest
}

// Build builds a "kv.Request".
Expand Down Expand Up @@ -75,6 +78,29 @@ func (builder *RequestBuilder) Build() (*kv.Request, error) {
if builder.Request.KeyRanges == nil {
builder.Request.KeyRanges = kv.NewNonPartitionedKeyRanges(nil)
}

if dag := builder.dag; dag != nil {
if execCnt := len(dag.Executors); execCnt == 1 {
oldConcurrency := builder.Request.Concurrency
// select * from t order by id
if builder.Request.KeepOrder {
// When the DAG is just simple scan and keep order, set concurrency to 2.
// If a lot data are returned to client, mysql protocol is the bottleneck so concurrency 2 is enough.
// If very few data are returned to client, the speed is not optimal but good enough.
switch dag.Executors[0].Tp {
case tipb.ExecType_TypeTableScan, tipb.ExecType_TypeIndexScan, tipb.ExecType_TypePartitionTableScan:
builder.Request.Concurrency = 2
failpoint.Inject("testRateLimitActionMockConsumeAndAssert", func(val failpoint.Value) {
if val.(bool) {
// When the concurrency is too small, test case tests/realtikvtest/sessiontest.TestCoprocessorOOMAction can't trigger OOM condition
builder.Request.Concurrency = oldConcurrency
}
})
}
}
}
}

return &builder.Request, builder.err
}

Expand Down Expand Up @@ -154,17 +180,18 @@ func (builder *RequestBuilder) SetDAGRequest(dag *tipb.DAGRequest) *RequestBuild
builder.Request.Tp = kv.ReqTypeDAG
builder.Request.Cacheable = true
builder.Request.Data, builder.err = dag.Marshal()
}
if execCnt := len(dag.Executors); execCnt != 0 && dag.Executors[execCnt-1].GetLimit() != nil {
limit := dag.Executors[execCnt-1].GetLimit()
builder.Request.LimitSize = limit.GetLimit()
// When the DAG is just simple scan and small limit, set concurrency to 1 would be sufficient.
if execCnt == 2 {
if limit.Limit < estimatedRegionRowCount {
if kr := builder.Request.KeyRanges; kr != nil {
builder.Request.Concurrency = kr.PartitionNum()
} else {
builder.Request.Concurrency = 1
builder.dag = dag
if execCnt := len(dag.Executors); execCnt != 0 && dag.Executors[execCnt-1].GetLimit() != nil {
limit := dag.Executors[execCnt-1].GetLimit()
builder.Request.LimitSize = limit.GetLimit()
// When the DAG is just simple scan and small limit, set concurrency to 1 would be sufficient.
if execCnt == 2 {
if limit.Limit < estimatedRegionRowCount {
if kr := builder.Request.KeyRanges; kr != nil {
builder.Request.Concurrency = kr.PartitionNum()
} else {
builder.Request.Concurrency = 1
}
}
}
}
Expand Down
5 changes: 5 additions & 0 deletions pkg/session/test/variable/variable_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,11 @@ func TestCoprocessorOOMAction(t *testing.T) {
sql: "select id from t5",
},
}

require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/distsql/testRateLimitActionMockConsumeAndAssert", `return(true)`))
defer func() {
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/distsql/testRateLimitActionMockConsumeAndAssert"))
}()
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/store/copr/testRateLimitActionMockConsumeAndAssert", `return(true)`))
defer func() {
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/store/copr/testRateLimitActionMockConsumeAndAssert"))
Expand Down
11 changes: 0 additions & 11 deletions pkg/store/copr/coprocessor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,17 +220,6 @@ func (c *CopClient) BuildCopIterator(ctx context.Context, req *kv.Request, vars
}

if it.req.KeepOrder {
oldConcurrency := it.concurrency
partitionNum := req.KeyRanges.PartitionNum()
if partitionNum > 0 && partitionNum < it.concurrency {
it.concurrency = partitionNum
}
failpoint.Inject("testRateLimitActionMockConsumeAndAssert", func(val failpoint.Value) {
if val.(bool) {
// When the concurrency is too small, test case tests/realtikvtest/sessiontest.TestCoprocessorOOMAction can't trigger OOM condition
it.concurrency = oldConcurrency
}
})
if it.smallTaskConcurrency > 20 {
it.smallTaskConcurrency = 20
}
Expand Down
134 changes: 134 additions & 0 deletions tests/integrationtest/r/executor/issues.result
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -868,3 +868,137 @@ set @@tidb_max_chunk_size = default;
select tan(9021874879467600608071521900001091070693729763119983979);
tan(9021874879467600608071521900001091070693729763119983979)
8.068627196084492
drop table if exists t;
create table t (id int auto_increment, c char(120), primary key(id));
create table pt (id int primary key auto_increment, val int) partition by range (id)
(PARTITION p1 VALUES LESS THAN (100),
PARTITION p2 VALUES LESS THAN (200),
PARTITION p3 VALUES LESS THAN (300),
PARTITION p4 VALUES LESS THAN (400),
PARTITION p5 VALUES LESS THAN (500),
PARTITION p6 VALUES LESS THAN (600),
PARTITION p7 VALUES LESS THAN (700));
insert into t (c) values ('abc'), ('def'), ('ghi'), ('jkl');
insert into t (c) select (c) from t;
insert into t (c) select (c) from t;
insert into t (c) select (c) from t;
insert into t (c) select (c) from t;
insert into t (c) select (c) from t;
insert into t (c) select (c) from t;
split table t between (0) and (40960) regions 30;
TOTAL_SPLIT_REGION SCATTER_FINISH_RATIO
29 1
analyze table t;
insert into pt (val) values (123),(456),(789),(1112);
insert into pt (val) select (val) from pt;
insert into pt (val) select (val) from pt;
insert into pt (val) select (val) from pt;
insert into pt (val) select (val) from pt;
insert into pt (val) select (val) from pt;
insert into pt (val) select (val) from pt;
split table pt between (0) and (40960) regions 30;
TOTAL_SPLIT_REGION SCATTER_FINISH_RATIO
203 1
analyze table pt;
set @@tidb_distsql_scan_concurrency = default;
explain analyze select * from t order by id; # expected distsql concurrency 2
id estRows actRows task access object execution info operator info memory disk
TableReader_11 256.00 <actRows> root NULL max_distsql_concurrency: 2 NULL <memory> <disk>
└─TableFullScan_10 256.00 <actRows> cop[tikv] table:t NULL keep order:true <memory> <disk>
explain analyze select * from t limit 100; # expected distsql concurrency 1
id estRows actRows task access object execution info operator info memory disk
Limit_7 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_11 100.00 <actRows> root NULL max_distsql_concurrency: 1 NULL <memory> <disk>
└─Limit_10 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─TableFullScan_9 100.00 <actRows> cop[tikv] table:t NULL keep order:false <memory> <disk>
explain analyze select * from t limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_7 256.00 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_11 256.00 <actRows> root NULL max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_10 256.00 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─TableFullScan_9 256.00 <actRows> cop[tikv] table:t NULL keep order:false <memory> <disk>
explain analyze select * from t where c = 'abc' limit 100; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_8 0.26 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_13 0.26 <actRows> root NULL max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_12 0.26 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─Selection_11 0.26 <actRows> cop[tikv] NULL NULL eq(executor__issues.t.c, "abc") <memory> <disk>
└─TableFullScan_10 256.00 <actRows> cop[tikv] table:t NULL keep order:false, stats:partial[c:unInitialized] <memory> <disk>
explain analyze select * from t where c = 'abc' limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_8 0.26 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_13 0.26 <actRows> root NULL max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_12 0.26 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─Selection_11 0.26 <actRows> cop[tikv] NULL NULL eq(executor__issues.t.c, "abc") <memory> <disk>
└─TableFullScan_10 256.00 <actRows> cop[tikv] table:t NULL keep order:false, stats:partial[c:unInitialized] <memory> <disk>
explain analyze select * from t order by id limit 100; # expected distsql concurrency 1
id estRows actRows task access object execution info operator info memory disk
Limit_10 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_17 100.00 <actRows> root NULL max_distsql_concurrency: 1 NULL <memory> <disk>
└─Limit_16 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─TableFullScan_15 100.00 <actRows> cop[tikv] table:t NULL keep order:true <memory> <disk>
explain analyze select * from t order by id limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_11 256.00 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_21 256.00 <actRows> root NULL max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_20 256.00 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─TableFullScan_19 256.00 <actRows> cop[tikv] table:t NULL keep order:true <memory> <disk>
explain analyze select * from t where c = 'abd' order by id limit 100;
id estRows actRows task access object execution info operator info memory disk
Limit_11 0.26 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_20 0.26 <actRows> root NULL max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_19 0.26 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─Selection_18 0.26 <actRows> cop[tikv] NULL NULL eq(executor__issues.t.c, "abd") <memory> <disk>
└─TableFullScan_17 256.00 <actRows> cop[tikv] table:t NULL keep order:true, stats:partial[c:unInitialized] <memory> <disk>
select @@tidb_partition_prune_mode;
@@tidb_partition_prune_mode
dynamic
explain analyze select * from pt order by id; # expected distsql concurrency 2
id estRows actRows task access object execution info operator info memory disk
TableReader_11 256.00 <actRows> root partition:all max_distsql_concurrency: 2 NULL <memory> <disk>
└─TableFullScan_10 256.00 <actRows> cop[tikv] table:pt NULL keep order:true <memory> <disk>
explain analyze select * from pt limit 100; # expected distsql concurrency 7
id estRows actRows task access object execution info operator info memory disk
Limit_7 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_11 100.00 <actRows> root partition:all max_distsql_concurrency: 7 NULL <memory> <disk>
└─Limit_10 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─TableFullScan_9 100.00 <actRows> cop[tikv] table:pt NULL keep order:false <memory> <disk>
explain analyze select * from pt limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_7 256.00 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_11 256.00 <actRows> root partition:all max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_10 256.00 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─TableFullScan_9 256.00 <actRows> cop[tikv] table:pt NULL keep order:false <memory> <disk>
explain analyze select * from pt where val = 125 limit 100; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_8 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_13 100.00 <actRows> root partition:all max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_12 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─Selection_11 100.00 <actRows> cop[tikv] NULL NULL eq(executor__issues.pt.val, 125) <memory> <disk>
└─TableFullScan_10 125.00 <actRows> cop[tikv] table:pt NULL keep order:false, stats:partial[val:missing] <memory> <disk>
explain analyze select * from pt where val = 125 limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_8 204.80 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_13 204.80 <actRows> root partition:all max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_12 204.80 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─Selection_11 204.80 <actRows> cop[tikv] NULL NULL eq(executor__issues.pt.val, 125) <memory> <disk>
└─TableFullScan_10 256.00 <actRows> cop[tikv] table:pt NULL keep order:false, stats:partial[val:missing] <memory> <disk>
explain analyze select * from pt order by id limit 100; # expected distsql concurrency 7, but currently get 1, see issue #55190
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Known issue #55190
Will not be fixed in this PR.

id estRows actRows task access object execution info operator info memory disk
Limit_10 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_17 100.00 <actRows> root partition:all max_distsql_concurrency: 1 NULL <memory> <disk>
└─Limit_16 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─TableFullScan_15 100.00 <actRows> cop[tikv] table:pt NULL keep order:true <memory> <disk>
explain analyze select * from pt order by id limit 100000; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_11 256.00 <actRows> root NULL NULL offset:0, count:100000 <memory> <disk>
└─TableReader_21 256.00 <actRows> root partition:all max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_20 256.00 <actRows> cop[tikv] NULL NULL offset:0, count:100000 <memory> <disk>
└─TableFullScan_19 256.00 <actRows> cop[tikv] table:pt NULL keep order:true <memory> <disk>
explain analyze select * from pt where val = 126 order by id limit 100; # expected distsql concurrency 15
id estRows actRows task access object execution info operator info memory disk
Limit_11 100.00 <actRows> root NULL NULL offset:0, count:100 <memory> <disk>
└─TableReader_20 100.00 <actRows> root partition:all max_distsql_concurrency: 15 NULL <memory> <disk>
└─Limit_19 100.00 <actRows> cop[tikv] NULL NULL offset:0, count:100 <memory> <disk>
└─Selection_18 100.00 <actRows> cop[tikv] NULL NULL eq(executor__issues.pt.val, 126) <memory> <disk>
└─TableFullScan_17 125.00 <actRows> cop[tikv] table:pt NULL keep order:true, stats:partial[val:missing] <memory> <disk>
Loading