Skip to content

Commit

Permalink
cherry pick pingcap#29778 to release-5.1
Browse files Browse the repository at this point in the history 
Signed-off-by: ti-srebot <ti-srebot@pingcap.com>
  • Loading branch information
@time-and-fate @ti-srebot
time-and-fate authored and ti-srebot committed Nov 15, 2021
1 parent 6be3ac5 commit 04e1fd6
Show file tree
Hide file tree
Showing 2 changed files with 267 additions and 0 deletions.
246 changes: 246 additions & 0 deletions planner/core/integration_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4087,3 +4087,249 @@ func (s *testIntegrationSuite) TestIssues27130(c *C) {
" └─IndexRangeScan 10.00 cop[tikv] table:t3, index:a(a, b, c) range:[1,1], keep order:false, stats:pseudo",
))
}
<<<<<<< HEAD
=======

func (s *testIntegrationSuite) TestIssue27242(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists UK_MU16407")
tk.MustExec("CREATE TABLE UK_MU16407 (COL3 timestamp NULL DEFAULT NULL, UNIQUE KEY U3(COL3));")
tk.MustExec(`insert into UK_MU16407 values("1985-08-31 18:03:27");`)
err := tk.ExecToErr(`SELECT COL3 FROM UK_MU16407 WHERE COL3>_utf8mb4'2039-1-19 3:14:40';`)
c.Assert(err, NotNil)
c.Assert(err.Error(), Matches, ".*Incorrect timestamp value.*")
}

func (s *testIntegrationSerialSuite) TestTemporaryTableForCte(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")

tk.MustExec("create temporary table tmp1(a int, b int, c int);")
tk.MustExec("insert into tmp1 values (1,1,1),(2,2,2),(3,3,3),(4,4,4);")
rows := tk.MustQuery("with cte1 as (with cte2 as (select * from tmp1) select * from cte2) select * from cte1 left join tmp1 on cte1.c=tmp1.c;")
rows.Check(testkit.Rows("1 1 1 1 1 1", "2 2 2 2 2 2", "3 3 3 3 3 3", "4 4 4 4 4 4"))
rows = tk.MustQuery("with cte1 as (with cte2 as (select * from tmp1) select * from cte2) select * from cte1 t1 left join cte1 t2 on t1.c=t2.c;")
rows.Check(testkit.Rows("1 1 1 1 1 1", "2 2 2 2 2 2", "3 3 3 3 3 3", "4 4 4 4 4 4"))
rows = tk.MustQuery("WITH RECURSIVE cte(a) AS (SELECT 1 UNION SELECT a+1 FROM tmp1 WHERE a < 5) SELECT * FROM cte order by a;")
rows.Check(testkit.Rows("1", "2", "3", "4", "5"))
}

func (s *testIntegrationSuite) TestGroupBySetVar(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t1")
tk.MustExec("create table t1(c1 int);")
tk.MustExec("insert into t1 values(1), (2), (3), (4), (5), (6);")
rows := tk.MustQuery("select floor(dt.rn/2) rownum, count(c1) from (select @rownum := @rownum + 1 rn, c1 from (select @rownum := -1) drn, t1) dt group by floor(dt.rn/2) order by rownum;")
rows.Check(testkit.Rows("0 2", "1 2", "2 2"))

tk.MustExec("create table ta(a int, b int);")
tk.MustExec("set sql_mode='';")

var input []string
var output []struct {
SQL string
Plan []string
}
s.testData.GetTestCases(c, &input, &output)
for i, tt := range input {
res := tk.MustQuery("explain format = 'brief' " + tt)
s.testData.OnRecord(func() {
output[i].SQL = tt
output[i].Plan = s.testData.ConvertRowsToStrings(res.Rows())
})
res.Check(testkit.Rows(output[i].Plan...))
}
}

func (s *testIntegrationSerialSuite) TestPushDownGroupConcatToTiFlash(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists ts")
tk.MustExec("create table ts (col_0 char(64), col_1 varchar(64) not null, col_2 varchar(1), id int primary key);")

// Create virtual tiflash replica info.
dom := domain.GetDomain(tk.Se)
is := dom.InfoSchema()
db, exists := is.SchemaByName(model.NewCIStr("test"))
c.Assert(exists, IsTrue)
for _, tblInfo := range db.Tables {
if tblInfo.Name.L == "ts" {
tblInfo.TiFlashReplica = &model.TiFlashReplicaInfo{
Count: 1,
Available: true,
}
}
}

tk.MustExec("set @@tidb_isolation_read_engines='tiflash,tidb'; set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1;")

var input []string
var output []struct {
SQL string
Plan []string
Warning []string
}
s.testData.GetTestCases(c, &input, &output)
for i, tt := range input {
s.testData.OnRecord(func() {
output[i].SQL = tt
output[i].Plan = s.testData.ConvertRowsToStrings(tk.MustQuery(tt).Rows())
})
res := tk.MustQuery(tt)
res.Check(testkit.Rows(output[i].Plan...))

comment := Commentf("case:%v sql:%s", i, tt)
warnings := tk.Se.GetSessionVars().StmtCtx.GetWarnings()
s.testData.OnRecord(func() {
if len(warnings) > 0 {
output[i].Warning = make([]string, len(warnings))
for j, warning := range warnings {
output[i].Warning[j] = warning.Err.Error()
}
}
})
if len(output[i].Warning) == 0 {
c.Assert(len(warnings), Equals, 0, comment)
} else {
c.Assert(len(warnings), Equals, len(output[i].Warning), comment)
for j, warning := range warnings {
c.Assert(warning.Level, Equals, stmtctx.WarnLevelWarning, comment)
c.Assert(warning.Err.Error(), Equals, output[i].Warning[j], comment)
}
}
}
}

func (s *testIntegrationSuite) TestIssue27797(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t27797")
tk.MustExec("create table t27797(a int, b int, c int, d int) " +
"partition by range columns(d) (" +
"partition p0 values less than (20)," +
"partition p1 values less than(40)," +
"partition p2 values less than(60));")
tk.MustExec("insert into t27797 values(1,1,1,1), (2,2,2,2), (22,22,22,22), (44,44,44,44);")
tk.MustExec("set sql_mode='';")
result := tk.MustQuery("select count(*) from (select a, b from t27797 where d > 1 and d < 60 and b > 0 group by b, c) tt;")
result.Check(testkit.Rows("3"))

tk.MustExec("drop table if exists IDT_HP24172")
tk.MustExec("CREATE TABLE `IDT_HP24172` ( " +
"`COL1` mediumint(16) DEFAULT NULL, " +
"`COL2` varchar(20) DEFAULT NULL, " +
"`COL4` datetime DEFAULT NULL, " +
"`COL3` bigint(20) DEFAULT NULL, " +
"`COL5` float DEFAULT NULL, " +
"KEY `UM_COL` (`COL1`,`COL3`) " +
") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin " +
"PARTITION BY HASH( `COL1`+`COL3` ) " +
"PARTITIONS 8;")
tk.MustExec("insert into IDT_HP24172(col1) values(8388607);")
result = tk.MustQuery("select col2 from IDT_HP24172 where col1 = 8388607 and col1 in (select col1 from IDT_HP24172);")
result.Check(testkit.Rows("<nil>"))
}

func (s *testIntegrationSuite) TestIssue28154(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t")
defer func() {
tk.Exec("drop table if exists t")
}()
tk.MustExec("create table t(a TEXT)")
tk.MustExec("insert into t values('abc')")
result := tk.MustQuery("select * from t where from_base64('')")
result.Check(testkit.Rows())
_, err := tk.Exec("update t set a = 'def' where from_base64('')")
c.Assert(err, NotNil)
c.Assert(err.Error(), Equals, "[types:1292]Truncated incorrect DOUBLE value: ''")
result = tk.MustQuery("select * from t where from_base64('invalidbase64')")
result.Check(testkit.Rows())
tk.MustExec("update t set a = 'hig' where from_base64('invalidbase64')")
result = tk.MustQuery("select * from t where from_base64('test')")
result.Check(testkit.Rows())
_, err = tk.Exec("update t set a = 'xyz' where from_base64('test')")
c.Assert(err, NotNil)
c.Assert(err, ErrorMatches, "\\[types:1292\\]Truncated incorrect DOUBLE value.*")
result = tk.MustQuery("select * from t")
result.Check(testkit.Rows("abc"))
}

func (s *testIntegrationSerialSuite) TestRejectSortForMPP(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t")
tk.MustExec("create table t (id int, value decimal(6,3), name char(128))")
tk.MustExec("analyze table t")

// Create virtual tiflash replica info.
dom := domain.GetDomain(tk.Se)
is := dom.InfoSchema()
db, exists := is.SchemaByName(model.NewCIStr("test"))
c.Assert(exists, IsTrue)
for _, tblInfo := range db.Tables {
if tblInfo.Name.L == "t" {
tblInfo.TiFlashReplica = &model.TiFlashReplicaInfo{
Count: 1,
Available: true,
}
}
}

tk.MustExec("set @@tidb_allow_mpp=1; set @@tidb_opt_broadcast_join=0; set @@tidb_enforce_mpp=1;")

var input []string
var output []struct {
SQL string
Plan []string
}
s.testData.GetTestCases(c, &input, &output)
for i, tt := range input {
s.testData.OnRecord(func() {
output[i].SQL = tt
output[i].Plan = s.testData.ConvertRowsToStrings(tk.MustQuery(tt).Rows())
})
res := tk.MustQuery(tt)
res.Check(testkit.Rows(output[i].Plan...))
}
}

func (s *testIntegrationSuite) TestIssues29711(c *C) {
tk := testkit.NewTestKit(c, s.store)
tk.MustExec("use test")

tk.MustExec("drop table if exists tbl_29711")
tk.MustExec("CREATE TABLE `tbl_29711` (" +
"`col_250` text COLLATE utf8_unicode_ci NOT NULL," +
"`col_251` enum('Alice','Bob','Charlie','David') COLLATE utf8_unicode_ci NOT NULL DEFAULT 'Charlie'," +
"PRIMARY KEY (`col_251`,`col_250`(1)) NONCLUSTERED);")
tk.MustQuery("explain format=brief " +
"select col_250,col_251 from tbl_29711 where col_251 between 'Bob' and 'David' order by col_250,col_251 limit 6;").
Check(testkit.Rows(
"TopN 6.00 root test.tbl_29711.col_250, test.tbl_29711.col_251, offset:0, count:6",
"└─IndexLookUp 6.00 root ",
" ├─IndexRangeScan(Build) 30.00 cop[tikv] table:tbl_29711, index:PRIMARY(col_251, col_250) range:[\"Bob\",\"Bob\"], [\"Charlie\",\"Charlie\"], [\"David\",\"David\"], keep order:false, stats:pseudo",
" └─TopN(Probe) 6.00 cop[tikv] test.tbl_29711.col_250, test.tbl_29711.col_251, offset:0, count:6",
" └─TableRowIDScan 30.00 cop[tikv] table:tbl_29711 keep order:false, stats:pseudo",
))

tk.MustExec("drop table if exists t29711")
tk.MustExec("CREATE TABLE `t29711` (" +
"`a` varchar(10) DEFAULT NULL," +
"`b` int(11) DEFAULT NULL," +
"`c` int(11) DEFAULT NULL," +
"KEY `ia` (`a`(2)))")
tk.MustQuery("explain format=brief select * from t29711 use index (ia) order by a limit 10;").
Check(testkit.Rows(
"TopN 10.00 root test.t29711.a, offset:0, count:10",
"└─IndexLookUp 10.00 root ",
" ├─IndexFullScan(Build) 10000.00 cop[tikv] table:t29711, index:ia(a) keep order:false, stats:pseudo",
" └─TopN(Probe) 10.00 cop[tikv] test.t29711.a, offset:0, count:10",
" └─TableRowIDScan 10000.00 cop[tikv] table:t29711 keep order:false, stats:pseudo",
))

}
>>>>>>> a0d3f4fe5... planner: fix topn wrongly pushed to index scan side when it's a prefix index (#29778)
21 changes: 21 additions & 0 deletions planner/core/task.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1280,14 +1280,35 @@ func (p *PhysicalTopN) getPushedDownTopN(childPlan PhysicalPlan) *PhysicalTopN {
return topN
}

// canPushToIndexPlan checks if this TopN can be pushed to the index side of copTask.
// It can be pushed to the index side when all columns used by ByItems are available from the index side and
// there's no prefix index column.
func (p *PhysicalTopN) canPushToIndexPlan(indexPlan PhysicalPlan, byItemCols []*expression.Column) bool {
schema := indexPlan.Schema()
for _, col := range byItemCols {
pos := schema.ColumnIndex(col)
if pos == -1 {
return false
}
if schema.Columns[pos].IsPrefix {
return false
}
}
return true
}

func (p *PhysicalTopN) attach2Task(tasks ...task) task {
t := tasks[0].copy()
inputCount := t.count()
if copTask, ok := t.(*copTask); ok && p.canPushDown(copTask.getStoreType()) && len(copTask.rootTaskConds) == 0 {
// If all columns in topN are from index plan, we push it to index plan, otherwise we finish the index plan and
// push it to table plan.
var pushedDownTopN *PhysicalTopN
<<<<<<< HEAD
if !copTask.indexPlanFinished && p.allColsFromSchema(copTask.indexPlan.Schema()) {
=======
if !copTask.indexPlanFinished && p.canPushToIndexPlan(copTask.indexPlan, cols) {
>>>>>>> a0d3f4fe5... planner: fix topn wrongly pushed to index scan side when it's a prefix index (#29778)
pushedDownTopN = p.getPushedDownTopN(copTask.indexPlan)
copTask.indexPlan = pushedDownTopN
} else {
Expand Down

0 comments on commit 04e1fd6

Please sign in to comment.