Skip to content

Commit

Permalink
This is an automated cherry-pick of pingcap#41370
Browse files Browse the repository at this point in the history 
Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
  • Loading branch information
@Dousir9 @ti-chi-bot
Dousir9 authored and ti-chi-bot committed Feb 15, 2023
1 parent 5ccc10b commit 185f9ae
Show file tree
Hide file tree
Showing 2 changed files with 340 additions and 0 deletions.
275 changes: 275 additions & 0 deletions planner/core/integration_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4992,3 +4992,278 @@ func (s *testIntegrationSuite) TestExplainAnalyzeDMLCommit(c *C) {
c.Assert(err, IsNil)
tk.MustQuery("select * from t").Check(testkit.Rows())
}
<<<<<<< HEAD
=======

func TestPlanCacheForIndexJoinRangeFallback(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec(`set @@tidb_enable_prepared_plan_cache=1`)
tk.MustExec("set @@tidb_enable_collect_execution_info=0")
tk.MustExec("use test")
tk.MustExec("drop table if exists t1, t2")
tk.MustExec("create table t1(a int, b varchar(10), c varchar(10), index idx_a_b(a, b))")
tk.MustExec("create table t2(d int)")
tk.MustExec("set @@tidb_opt_range_max_size=1275")
// 1275 is enough for [? a,? a], [? b,? b], [? c,? c] but is not enough for [? aaaaaa,? aaaaaa], [? bbbbbb,? bbbbbb], [? cccccc,? cccccc].
rows := tk.MustQuery("explain format='brief' select /*+ inl_join(t1) */ * from t1 join t2 on t1.a = t2.d where t1.b in ('a', 'b', 'c')").Rows()
require.True(t, strings.Contains(rows[6][4].(string), "range: decided by [eq(test.t1.a, test.t2.d) in(test.t1.b, a, b, c)]"))
tk.MustQuery("show warnings").Check(testkit.Rows())
rows = tk.MustQuery("explain format='brief' select /*+ inl_join(t1) */ * from t1 join t2 on t1.a = t2.d where t1.b in ('aaaaaa', 'bbbbbb', 'cccccc');").Rows()
require.True(t, strings.Contains(rows[6][4].(string), "range: decided by [eq(test.t1.a, test.t2.d)]"))
tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 Memory capacity of 1275 bytes for 'tidb_opt_range_max_size' exceeded when building ranges. Less accurate ranges such as full range are chosen"))

tk.MustExec("prepare stmt1 from 'select /*+ inl_join(t1) */ * from t1 join t2 on t1.a = t2.d where t1.b in (?, ?, ?)'")
tk.MustExec("set @a='a', @b='b', @c='c'")
tk.MustExec("execute stmt1 using @a, @b, @c")
tk.MustQuery("show warnings").Check(testkit.Rows())
tk.MustExec("set @a='aaaaaa', @b='bbbbbb', @c='cccccc'")
tk.MustExec("execute stmt1 using @a, @b, @c")
tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("1"))
tk.MustExec("execute stmt1 using @a, @b, @c")
tkProcess := tk.Session().ShowProcess()
ps := []*util.ProcessInfo{tkProcess}
tk.Session().SetSessionManager(&testkit.MockSessionManager{PS: ps})
rows = tk.MustQuery(fmt.Sprintf("explain for connection %d", tkProcess.ID)).Rows()
// We don't limit range mem usage when rebuilding index join ranges for the cached plan. So [? aaaaaa,? aaaaaa], [? bbbbbb,? bbbbbb], [? cccccc,? cccccc] can be built.
require.True(t, strings.Contains(rows[6][4].(string), "range: decided by [eq(test.t1.a, test.t2.d) in(test.t1.b, aaaaaa, bbbbbb, cccccc)]"))

// Test the plan with range fallback would not be put into cache.
tk.MustExec("prepare stmt2 from 'select /*+ inl_join(t1) */ * from t1 join t2 on t1.a = t2.d where t1.b in (?, ?, ?, ?, ?)'")
tk.MustExec("set @a='a', @b='b', @c='c', @d='d', @e='e'")
tk.MustExec("execute stmt2 using @a, @b, @c, @d, @e")
tk.MustQuery("show warnings").Sort().Check(testkit.Rows("Warning 1105 Memory capacity of 1275 bytes for 'tidb_opt_range_max_size' exceeded when building ranges. Less accurate ranges such as full range are chosen",
"Warning 1105 skip plan-cache: in-list is too long"))
tk.MustExec("execute stmt2 using @a, @b, @c, @d, @e")
tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
}

// https://github.com/pingcap/tidb/issues/38295.
func TestIssue38295(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("CREATE TABLE t0(c0 BLOB(298) , c1 BLOB(182) , c2 NUMERIC);")
tk.MustExec("CREATE VIEW v0(c0) AS SELECT t0.c1 FROM t0;")
tk.MustExec("INSERT INTO t0 VALUES (-1, 'a', '2046549365');")
tk.MustExec("CREATE INDEX i0 ON t0(c2);")
tk.MustGetErrCode("SELECT t0.c1, t0.c2 FROM t0 GROUP BY MOD(t0.c0, DEFAULT(t0.c2));", errno.ErrFieldNotInGroupBy)
tk.MustExec("UPDATE t0 SET c2=1413;")
}

func TestOuterJoinEliminationForIssue18216(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t1, t2;")
tk.MustExec("create table t1 (a int, c int);")
tk.MustExec("insert into t1 values (1, 1), (1, 2), (2, 3), (2, 4);")
tk.MustExec("create table t2 (a int, c int);")
tk.MustExec("insert into t2 values (1, 1), (1, 2), (2, 3), (2, 4);")
// The output might be unstable.
tk.MustExec("select group_concat(c order by (select group_concat(c order by a) from t2 where a=t1.a)) from t1; ")
tk.MustQuery("select group_concat(c order by (select group_concat(c order by c) from t2 where a=t1.a), c desc) from t1;").Check(testkit.Rows("2,1,4,3"))
}

func TestAutoIncrementCheckWithCheckConstraint(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec(`CREATE TABLE t (
id INTEGER NOT NULL AUTO_INCREMENT,
CHECK (id IN (0, 1)),
KEY idx_autoinc_id (id)
)`)
}

// https://github.com/pingcap/tidb/issues/36888.
func TestIssue36888(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("CREATE TABLE t0(c0 INT);")
tk.MustExec("CREATE TABLE t1(c0 INT);")

tk.MustExec("INSERT INTO t0 VALUES (NULL);")
tk.MustQuery("SELECT t0.c0 FROM t0 LEFT JOIN t1 ON t0.c0>=t1.c0 WHERE (CONCAT_WS(t0.c0, t1.c0) IS NULL);").Check(testkit.Rows("<nil>"))
}

// https://github.com/pingcap/tidb/issues/40285.
func TestIssue40285(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("CREATE TABLE t(col1 enum('p5', '9a33x') NOT NULL DEFAULT 'p5',col2 tinyblob DEFAULT NULL) ENGINE = InnoDB DEFAULT CHARSET = latin1 COLLATE = latin1_bin;")
tk.MustQuery("(select last_value(col1) over () as r0 from t) union all (select col2 as r0 from t);")
}

// https://github.com/pingcap/tidb/issues/41273
func TestIssue41273(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec(`CREATE TABLE t (
a set('nwbk','r5','1ad3u','van','ir1z','y','9m','f1','z','e6yd','wfev') NOT NULL DEFAULT 'ir1z,f1,e6yd',
b enum('soo2','4s4j','qi9om','8ue','i71o','qon','3','3feh','6o1i','5yebx','d') NOT NULL DEFAULT '8ue',
c varchar(66) DEFAULT '13mdezixgcn',
PRIMARY KEY (a,b) /*T![clustered_index] CLUSTERED */,
UNIQUE KEY ib(b),
KEY ia(a)
)ENGINE=InnoDB DEFAULT CHARSET=ascii COLLATE=ascii_bin;`)
tk.MustExec("INSERT INTO t VALUES('ir1z,f1,e6yd','i71o','13mdezixgcn'),('ir1z,f1,e6yd','d','13mdezixgcn'),('nwbk','8ue','13mdezixgcn');")
expectedRes := []string{"ir1z,f1,e6yd d 13mdezixgcn", "ir1z,f1,e6yd i71o 13mdezixgcn", "nwbk 8ue 13mdezixgcn"}
tk.MustQuery("select * from t where a between 'e6yd' and 'z' or b <> '8ue';").Sort().Check(testkit.Rows(expectedRes...))
tk.MustQuery("select /*+ use_index_merge(t) */ * from t where a between 'e6yd' and 'z' or b <> '8ue';").Sort().Check(testkit.Rows(expectedRes...))
// For now tidb doesn't support push set type to TiKV, and column a is a set type, so we shouldn't generate a IndexMerge path.
require.False(t, tk.HasPlanForLastExecution("IndexMerge"))
}

func TestIsIPv4ToTiFlash(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t")
tk.MustExec("create table t(v4 varchar(100), v6 varchar(100))")
tk.MustExec("insert into t values('123.123.123.123', 'F746:C349:48E3:22F2:81E0:0EA8:E7B6:8286')")
tk.MustExec("insert into t values('0.0.0.0', '0000:0000:0000:0000:0000:0000:0000:0000')")
tk.MustExec("insert into t values('127.0.0.1', '2001:0:2851:b9f0:6d:2326:9036:f37a')")

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

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

rows := [][]interface{}{
{"TableReader_9", "root", "MppVersion: 1, data:ExchangeSender_8"},
{"└─ExchangeSender_8", "mpp[tiflash]", "ExchangeType: PassThrough"},
{" └─Projection_4", "mpp[tiflash]", "is_ipv4(test.t.v4)->Column#4"},
{" └─TableFullScan_7", "mpp[tiflash]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select is_ipv4(v4) from t;").CheckAt([]int{0, 2, 4}, rows)
}

func TestIsIPv6ToTiFlash(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t")
tk.MustExec("create table t(v4 varchar(100), v6 varchar(100))")
tk.MustExec("insert into t values('123.123.123.123', 'F746:C349:48E3:22F2:81E0:0EA8:E7B6:8286')")
tk.MustExec("insert into t values('0.0.0.0', '0000:0000:0000:0000:0000:0000:0000:0000')")
tk.MustExec("insert into t values('127.0.0.1', '2001:0:2851:b9f0:6d:2326:9036:f37a')")

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

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

rows := [][]interface{}{
{"TableReader_9", "root", "MppVersion: 1, data:ExchangeSender_8"},
{"└─ExchangeSender_8", "mpp[tiflash]", "ExchangeType: PassThrough"},
{" └─Projection_4", "mpp[tiflash]", "is_ipv6(test.t.v6)->Column#4"},
{" └─TableFullScan_7", "mpp[tiflash]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select is_ipv6(v6) from t;").CheckAt([]int{0, 2, 4}, rows)
}

// https://github.com/pingcap/tidb/issues/41355
// The "virtual generated column" push down is not supported now.
// This test covers: TopN, Projection, Selection.
func TestVirtualExprPushDown(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test;")
tk.MustExec("drop table if exists t;")
tk.MustExec("CREATE TABLE t (c1 int DEFAULT 0, c2 int GENERATED ALWAYS AS (abs(c1)) VIRTUAL);")
tk.MustExec("insert into t(c1) values(1), (-1), (2), (-2), (99), (-99);")
tk.MustExec("set @@tidb_isolation_read_engines = 'tikv'")

// TopN to tikv.
rows := [][]interface{}{
{"TopN_7", "root", "test.t.c2, offset:0, count:2"},
{"└─TableReader_13", "root", "data:TableFullScan_12"},
{" └─TableFullScan_12", "cop[tikv]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select * from t order by c2 limit 2;").CheckAt([]int{0, 2, 4}, rows)

// Projection to tikv.
rows = [][]interface{}{
{"Projection_3", "root", "plus(test.t.c1, test.t.c2)->Column#4"},
{"└─TableReader_5", "root", "data:TableFullScan_4"},
{" └─TableFullScan_4", "cop[tikv]", "keep order:false, stats:pseudo"},
}
tk.MustExec("set session tidb_opt_projection_push_down='ON';")
tk.MustQuery("explain select c1 + c2 from t;").CheckAt([]int{0, 2, 4}, rows)
tk.MustExec("set session tidb_opt_projection_push_down='OFF';")

// Selection to tikv.
rows = [][]interface{}{
{"Selection_7", "root", "gt(test.t.c2, 1)"},
{"└─TableReader_6", "root", "data:TableFullScan_5"},
{" └─TableFullScan_5", "cop[tikv]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select * from t where c2 > 1;").CheckAt([]int{0, 2, 4}, rows)

tk.MustExec("set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1")
tk.MustExec("set @@tidb_isolation_read_engines = 'tiflash'")
is := dom.InfoSchema()
db, exists := is.SchemaByName(model.NewCIStr("test"))
require.True(t, exists)
for _, tblInfo := range db.Tables {
if tblInfo.Name.L == "t" {
tblInfo.TiFlashReplica = &model.TiFlashReplicaInfo{
Count: 1,
Available: true,
}
}
}

// TopN to tiflash.
rows = [][]interface{}{
{"TopN_7", "root", "test.t.c2, offset:0, count:2"},
{"└─TableReader_15", "root", "data:TableFullScan_14"},
{" └─TableFullScan_14", "cop[tiflash]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select * from t order by c2 limit 2;").CheckAt([]int{0, 2, 4}, rows)

// Projection to tiflash.
rows = [][]interface{}{
{"Projection_3", "root", "plus(test.t.c1, test.t.c2)->Column#4"},
{"└─TableReader_6", "root", "data:TableFullScan_5"},
{" └─TableFullScan_5", "cop[tiflash]", "keep order:false, stats:pseudo"},
}
tk.MustExec("set session tidb_opt_projection_push_down='ON';")
tk.MustQuery("explain select c1 + c2 from t;").CheckAt([]int{0, 2, 4}, rows)
tk.MustExec("set session tidb_opt_projection_push_down='OFF';")

// Selection to tiflash.
rows = [][]interface{}{
{"Selection_8", "root", "gt(test.t.c2, 1)"},
{"└─TableReader_7", "root", "data:TableFullScan_6"},
{" └─TableFullScan_6", "cop[tiflash]", "keep order:false, stats:pseudo"},
}
tk.MustQuery("explain select * from t where c2 > 1;").CheckAt([]int{0, 2, 4}, rows)
}
>>>>>>> d2d91b5d969 (planner: add more checks when pushing TopN down (#41370))
65 changes: 65 additions & 0 deletions planner/core/task.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1206,6 +1206,7 @@ func (p *PhysicalLimit) sinkIntoIndexLookUp(t task) bool {
return true
}

<<<<<<< HEAD
// GetCost computes cost of TopN operator itself.
func (p *PhysicalTopN) GetCost(count float64, isRoot bool) float64 {
heapSize := float64(p.Offset + p.Count)
Expand Down Expand Up @@ -1260,6 +1261,8 @@ func (p *PhysicalSort) GetCost(count float64, schema *expression.Schema) float64
return cpuCost + memoryCost + diskCost
}

=======
>>>>>>> d2d91b5d969 (planner: add more checks when pushing TopN down (#41370))
func (p *PhysicalSort) attach2Task(tasks ...task) task {
t := tasks[0].copy()
t = attachPlan2Task(p, t)
Expand Down Expand Up @@ -1312,6 +1315,56 @@ func (p *PhysicalTopN) canPushToIndexPlan(indexPlan PhysicalPlan, byItemCols []*
return true
}

// canExpressionConvertedToPB checks whether each of the the expression in TopN can be converted to pb.
func (p *PhysicalTopN) canExpressionConvertedToPB(storeTp kv.StoreType) bool {
exprs := make([]expression.Expression, 0, len(p.ByItems))
for _, item := range p.ByItems {
exprs = append(exprs, item.Expr)
}
return expression.CanExprsPushDown(p.ctx.GetSessionVars().StmtCtx, exprs, p.ctx.GetClient(), storeTp)
}

// containVirtualColumn checks whether TopN.ByItems contains virtual generated columns.
func (p *PhysicalTopN) containVirtualColumn(tCols []*expression.Column) bool {
for _, by := range p.ByItems {
cols := expression.ExtractColumns(by.Expr)
for _, col := range cols {
for _, tCol := range tCols {
// A column with ID > 0 indicates that the column can be resolved by data source.
if tCol.ID > 0 && tCol.ID == col.ID && tCol.VirtualExpr != nil {
return true
}
}
}
}
return false
}

// canPushDownToTiKV checks whether this topN can be pushed down to TiKV.
func (p *PhysicalTopN) canPushDownToTiKV(copTask *copTask) bool {
if !p.canExpressionConvertedToPB(kv.TiKV) {
return false
}
if len(copTask.rootTaskConds) != 0 {
return false
}
if p.containVirtualColumn(copTask.plan().Schema().Columns) {
return false
}
return true
}

// canPushDownToTiFlash checks whether this topN can be pushed down to TiFlash.
func (p *PhysicalTopN) canPushDownToTiFlash(mppTask *mppTask) bool {
if !p.canExpressionConvertedToPB(kv.TiFlash) {
return false
}
if p.containVirtualColumn(mppTask.plan().Schema().Columns) {
return false
}
return true
}

func (p *PhysicalTopN) attach2Task(tasks ...task) task {
t := tasks[0].copy()
inputCount := t.count()
Expand All @@ -1320,7 +1373,15 @@ func (p *PhysicalTopN) attach2Task(tasks ...task) task {
cols = append(cols, expression.ExtractColumns(item.Expr)...)
}
needPushDown := len(cols) > 0
<<<<<<< HEAD
if copTask, ok := t.(*copTask); ok && needPushDown && p.canPushDown(copTask.getStoreType()) && len(copTask.rootTaskConds) == 0 {
=======
if copTask, ok := t.(*copTask); ok && needPushDown && p.canPushDownToTiKV(copTask) {
newTask, changed := p.pushTopNDownToDynamicPartition(copTask)
if changed {
return newTask
}
>>>>>>> d2d91b5d969 (planner: add more checks when pushing TopN down (#41370))
// 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
Expand All @@ -1332,8 +1393,12 @@ func (p *PhysicalTopN) attach2Task(tasks ...task) task {
pushedDownTopN = p.getPushedDownTopN(copTask.tablePlan)
copTask.tablePlan = pushedDownTopN
}
<<<<<<< HEAD
copTask.addCost(pushedDownTopN.GetCost(inputCount, false))
} else if mppTask, ok := t.(*mppTask); ok && needPushDown && p.canPushDown(kv.TiFlash) {
=======
} else if mppTask, ok := t.(*mppTask); ok && needPushDown && p.canPushDownToTiFlash(mppTask) {
>>>>>>> d2d91b5d969 (planner: add more checks when pushing TopN down (#41370))
pushedDownTopN := p.getPushedDownTopN(mppTask.p)
mppTask.p = pushedDownTopN
}
Expand Down

0 comments on commit 185f9ae

Please sign in to comment.