Merge branch 'master' into try_to_fix_pool

.github/licenserc.yml

@@ -40,4 +40,5 @@ header:
     - "tidb-binlog/proto/go-binlog/secondary_binlog.pb.go"
     - "**/*.sql"
     - ".bazelversion"
+    - "build/image/.ci_bazel"
   comment: on-failure

bindinfo/capture_test.go

@@ -434,8 +434,9 @@ func TestUpdateSubqueryCapture(t *testing.T) {
 	rows := tk.MustQuery("show global bindings").Rows()
 	require.Len(t, rows, 1)
 	bindSQL := "UPDATE /*+ hash_join(@`upd_1` `test`.`t1`), use_index(@`upd_1` `test`.`t1` `idx_b`), use_index(@`sel_1` `test`.`t2` ), use_index(@`sel_2` `test`.`t2` )*/ `test`.`t1` SET `b`=1 WHERE `b` = 2 AND (`a` IN (SELECT `a` FROM `test`.`t2` WHERE `b` = 1) OR `c` IN (SELECT `a` FROM `test`.`t2` WHERE `b` = 1))"
+	originSQL := "UPDATE `test`.`t1` SET `b`=1 WHERE `b` = 2 AND (`a` IN (SELECT `a` FROM `test`.`t2` WHERE `b` = 1) OR `c` IN (SELECT `a` FROM `test`.`t2` WHERE `b` = 1))"
 	require.Equal(t, bindSQL, rows[0][1])
-	tk.MustExec(bindSQL)
+	tk.MustExec(originSQL)
 	require.Len(t, tk.Session().GetSessionVars().StmtCtx.GetWarnings(), 0)
 }
 

br/pkg/gluetidb/glue.go

@@ -288,6 +288,8 @@ func (gs *tidbSession) CreateTable(ctx context.Context, dbName model.CIStr, tabl
 		return errors.Trace(err)
 	}
 	gs.se.SetValue(sessionctx.QueryString, query)
+	// Disable foreign key check when batch create tables.
+	gs.se.GetSessionVars().ForeignKeyChecks = false
 	// Clone() does not clone partitions yet :(
 	table = table.Clone()
 	if table.Partition != nil {

build/image/.ci_bazel

@@ -0,0 +1 @@
+build:ci --remote_cache=http://bazel-cache.pingcap.net:8080/tidb --remote_timeout="15s"

build/image/README.md

@@ -0,0 +1,7 @@
+## CI Image
+
+Here is the Dockerfile for the CI image.
+
+- ```base``` is the base image with golang, development tools and so on.
+- ```centos7_jenkins``` is the production image with CI environment tool in tidb repo. it is based on ```base```.
+- ```.ci_bazel``` is the global default bazel config. it tell bazel where to get cache.

build/image/base

@@ -0,0 +1,44 @@
+# Copyright 2023 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM hub.pingcap.net/jenkins/centos7_jenkins
+
+USER root
+WORKDIR /root
+
+ENV GOLANG_VERSION 1.19.5
+ENV GOLANG_DOWNLOAD_URL https://dl.google.com/go/go$GOLANG_VERSION.linux-amd64.tar.gz
+ENV GOLANG_DOWNLOAD_SHA256 36519702ae2fd573c9869461990ae550c8c0d955cd28d2827a6b159fda81ff95
+ENV GOPATH /go
+ENV GOROOT /usr/local/go
+ENV PATH $GOPATH/bin:$GOROOT/bin:$PATH
+ADD https://github.com/bazelbuild/bazel/releases/download/5.3.2/bazel-5.3.2-linux-x86_64 /usr/bin/bazel
+ADD https://uploader.codecov.io/latest/linux/codecov /usr/bin/codecov
+RUN curl https://setup.ius.io | sh || true && \
+    chmod u+x /usr/bin/bazel && yum update -y && \
+    yum install -y supervisor tree libcurl-devel gettext autoconf python-pip python3-pip patch git wget gcc python autoconf make curl-devel expat-devel gettext-devel openssl-devel zlib-devel perl-CPAN perl-devel && \
+    git clone --depth=1 --branch=v2.37.2 https://github.com/git/git && cd git && yum remove -y git && make configure && ./configure --prefix=/usr/local &&  make -j16 install && cd .. && rm -rf git && \
+    pip3 install s3cmd requests && pip3 install requests && \
+    curl -fsSL "$GOLANG_DOWNLOAD_URL" -o golang.tar.gz \
+	&& echo "$GOLANG_DOWNLOAD_SHA256  golang.tar.gz" | sha256sum -c - \
+	&& tar -C /usr/local -xzf golang.tar.gz \
+	&& rm golang.tar.gz && \
+    mkdir /go && chown jenkins:jenkins /go && \
+    curl -fsSL "http://pingcap-dev.hk.ufileos.com/jenkins/jenkins-slave-docker-sqllogictest.tar.gz" | tar xz -C "/git" \
+        && chown -R jenkins:jenkins /git && \
+    chown jenkins:jenkins /usr/bin/bazel && \
+    chown jenkins:jenkins /usr/bin/codecov && \
+    chmod +x /usr/bin/codecov
+USER jenkins
+WORKDIR /home/jenkins

build/image/centos7_jenkins

@@ -0,0 +1,24 @@
+# Copyright 2023 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM hub.pingcap.net/wangweizhen/base_image:go11920230111
+USER root
+WORKDIR /root
+COPY .ci_bazel /data/bazel
+RUN mkdir -p /data/tikv1 /data/tikv2 /data/tikv3 /data/pd && \ 
+    chown -R jenkins:jenkins /data
+USER jenkins
+WORKDIR /home/jenkins
+RUN go install github.com/hawkingrei/bazel_collect@latest && \
+    go install github.com/bazelbuild/bazel-gazelle/cmd/gazelle@latest

distsql/distsql.go

@@ -42,7 +42,7 @@ func DispatchMPPTasks(ctx context.Context, sctx sessionctx.Context, tasks []*kv.
 	ctx = WithSQLKvExecCounterInterceptor(ctx, sctx.GetSessionVars().StmtCtx)
 	_, allowTiFlashFallback := sctx.GetSessionVars().AllowFallbackToTiKV[kv.TiFlash]
 	ctx = SetTiFlashMaxThreadsInContext(ctx, sctx)
-	resp := sctx.GetMPPClient().DispatchMPPTasks(ctx, sctx.GetSessionVars().KVVars, tasks, allowTiFlashFallback, startTs, mppQueryID)
+	resp := sctx.GetMPPClient().DispatchMPPTasks(ctx, sctx.GetSessionVars().KVVars, tasks, allowTiFlashFallback, startTs, mppQueryID, sctx.GetSessionVars().ChooseMppVersion())
 	if resp == nil {
 		return nil, errors.New("client returns nil response")
 	}

executor/brie.go

@@ -524,6 +524,8 @@ func (gs *tidbGlueSession) CreateTable(ctx context.Context, dbName model.CIStr,
 		return err
 	}
 	gs.se.SetValue(sessionctx.QueryString, result.String())
+	// Disable foreign key check when batch create tables.
+	gs.se.GetSessionVars().ForeignKeyChecks = false
 
 	// Clone() does not clone partitions yet :(
 	table = table.Clone()

executor/builder.go

@@ -3401,6 +3401,7 @@ func (b *executorBuilder) buildMPPGather(v *plannercore.PhysicalTableReader) Exe
 		b.err = err
 		return nil
 	}
+
 	gather := &MPPGather{
 		baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ID()),
 		is:           b.is,

executor/executor_test.go

@@ -2114,6 +2114,8 @@ func TestIncorrectLimitArg(t *testing.T) {
 
 	tk.MustGetErrMsg(`execute stmt1 using @a;`, `[planner:1210]Incorrect arguments to LIMIT`)
 	tk.MustGetErrMsg(`execute stmt2 using @b, @a;`, `[planner:1210]Incorrect arguments to LIMIT`)
+	tk.MustGetErrMsg(`execute stmt2 using @a, @b;`, `[planner:1210]Incorrect arguments to LIMIT`)
+	tk.MustGetErrMsg(`execute stmt2 using @a, @a;`, `[planner:1210]Incorrect arguments to LIMIT`)
 }
 
 func TestExecutorLimit(t *testing.T) {

executor/mpp_gather.go

@@ -93,10 +93,14 @@ func (e *MPPGather) appendMPPDispatchReq(pf *plannercore.Fragment) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+
 		logutil.BgLogger().Info("Dispatch mpp task", zap.Uint64("timestamp", mppTask.StartTs),
 			zap.Int64("ID", mppTask.ID), zap.Uint64("QueryTs", mppTask.MppQueryID.QueryTs), zap.Uint64("LocalQueryId", mppTask.MppQueryID.LocalQueryID),
 			zap.Uint64("ServerID", mppTask.MppQueryID.ServerID), zap.String("address", mppTask.Meta.GetAddress()),
-			zap.String("plan", plannercore.ToString(pf.ExchangeSender)))
+			zap.String("plan", plannercore.ToString(pf.ExchangeSender)),
+			zap.Int64("mpp-version", mppTask.MppVersion.ToInt64()),
+			zap.String("exchange-compression-mode", pf.ExchangeSender.CompressionMode.Name()),
+		)
 		req := &kv.MPPDispatchRequest{
 			Data:       pbData,
 			Meta:       mppTask.Meta,

executor/seqtest/prepared_test.go

@@ -826,6 +826,12 @@ func TestSetPlanCacheLimitSwitch(t *testing.T) {
 
 	tk.MustExec("set @@global.tidb_enable_plan_cache_for_param_limit = ON;")
 	tk.MustQuery("select @@global.tidb_enable_plan_cache_for_param_limit").Check(testkit.Rows("1"))
+
+	tk.MustGetErrMsg("set @@global.tidb_enable_plan_cache_for_param_limit = '';", "[variable:1231]Variable 'tidb_enable_plan_cache_for_param_limit' can't be set to the value of ''")
+	tk.MustGetErrMsg("set @@global.tidb_enable_plan_cache_for_param_limit = 11;", "[variable:1231]Variable 'tidb_enable_plan_cache_for_param_limit' can't be set to the value of '11'")
+	tk.MustGetErrMsg("set @@global.tidb_enable_plan_cache_for_param_limit = enabled;", "[variable:1231]Variable 'tidb_enable_plan_cache_for_param_limit' can't be set to the value of 'enabled'")
+	tk.MustGetErrMsg("set @@global.tidb_enable_plan_cache_for_param_limit = disabled;", "[variable:1231]Variable 'tidb_enable_plan_cache_for_param_limit' can't be set to the value of 'disabled'")
+	tk.MustGetErrMsg("set @@global.tidb_enable_plan_cache_for_param_limit = open;", "[variable:1231]Variable 'tidb_enable_plan_cache_for_param_limit' can't be set to the value of 'open'")
 }
 
 func TestPlanCacheLimitSwitchEffective(t *testing.T) {

executor/set_test.go

@@ -2071,3 +2071,59 @@ func TestSetChunkReuseVariable(t *testing.T) {
 	// error value
 	tk.MustGetErrCode("set @@tidb_enable_reuse_chunk=s;", errno.ErrWrongValueForVar)
 }
+
+func TestSetMppVersionVariable(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustQuery("select @@session.mpp_version").Check(testkit.Rows("UNSPECIFIED"))
+	tk.MustExec("SET SESSION mpp_version = -1")
+	tk.MustQuery("select @@session.mpp_version").Check(testkit.Rows("-1"))
+	tk.MustExec("SET SESSION mpp_version = 0")
+	tk.MustQuery("select @@session.mpp_version").Check(testkit.Rows("0"))
+	tk.MustExec("SET SESSION mpp_version = 1")
+	tk.MustQuery("select @@session.mpp_version").Check(testkit.Rows("1"))
+	tk.MustExec("SET SESSION mpp_version = unspecified")
+	tk.MustQuery("select @@session.mpp_version").Check(testkit.Rows("unspecified"))
+	{
+		tk.MustGetErrMsg("SET SESSION mpp_version = 2", "incorrect value: 2. mpp_version options: -1 (unspecified), 0, 1")
+	}
+	{
+		tk.MustExec("SET GLOBAL mpp_version = 1")
+		tk.MustQuery("select @@global.mpp_version").Check(testkit.Rows("1"))
+		tk.MustExec("SET GLOBAL mpp_version = -1")
+		tk.MustQuery("select @@global.mpp_version").Check(testkit.Rows("-1"))
+	}
+}
+
+func TestSetMppExchangeCompressionModeVariable(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustGetErrMsg(
+		"SET SESSION mpp_exchange_compression_mode = 123",
+		"incorrect value: `123`. mpp_exchange_compression_mode options: NONE, FAST, HIGH_COMPRESSION, UNSPECIFIED")
+	tk.MustQuery("select @@session.mpp_exchange_compression_mode").Check(testkit.Rows("UNSPECIFIED"))
+
+	tk.MustExec("SET SESSION mpp_exchange_compression_mode = none")
+	tk.MustQuery("select @@session.mpp_exchange_compression_mode").Check(testkit.Rows("none"))
+	tk.MustExec("SET SESSION mpp_exchange_compression_mode = fast")
+	tk.MustQuery("select @@session.mpp_exchange_compression_mode").Check(testkit.Rows("fast"))
+	tk.MustExec("SET SESSION mpp_exchange_compression_mode = HIGH_COMPRESSION")
+	tk.MustQuery("select @@session.mpp_exchange_compression_mode").Check(testkit.Rows("HIGH_COMPRESSION"))
+
+	{
+		tk.MustExec("SET GLOBAL mpp_exchange_compression_mode = none")
+		tk.MustQuery("select @@global.mpp_exchange_compression_mode").Check(testkit.Rows("none"))
+	}
+	{
+		tk.MustExec("SET mpp_version = 0")
+		tk.MustExec("SET mpp_exchange_compression_mode = unspecified")
+		require.Equal(t, len(tk.Session().GetSessionVars().StmtCtx.GetWarnings()), 0)
+	}
+	{
+		tk.MustExec("SET mpp_version = 0")
+		tk.MustExec("SET mpp_exchange_compression_mode = HIGH_COMPRESSION")
+		warnings := tk.Session().GetSessionVars().StmtCtx.GetWarnings()
+		require.Equal(t, len(warnings), 1)
+		require.Equal(t, warnings[0].Err.Error(), "mpp exchange compression won't work under current mpp version 0")
+	}
+}

executor/tiflashtest/BUILD.bazel

@@ -14,6 +14,7 @@ go_test(
         "//config",
         "//domain",
         "//executor",
+        "//kv",
         "//meta/autoid",
         "//parser/terror",
         "//planner/core",

executor/tiflashtest/tiflash_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/executor"
+	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/terror"
 	plannercore "github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/store/mockstore"
@@ -629,10 +630,41 @@ func TestDispatchTaskRetry(t *testing.T) {
 	require.NoError(t, err)
 	tk.MustExec("set @@session.tidb_enforce_mpp=ON")
 	require.Nil(t, failpoint.Enable("github.com/pingcap/tidb/store/mockstore/unistore/mppDispatchTimeout", "3*return(true)"))
-	tk.MustQuery("select count(*) from t").Check(testkit.Rows("4"))
+	tk.MustQuery("select count(*) from t group by b").Check(testkit.Rows("4"))
 	require.Nil(t, failpoint.Disable("github.com/pingcap/tidb/store/mockstore/unistore/mppDispatchTimeout"))
 }
 
+func TestMppVersionError(t *testing.T) {
+	store := testkit.CreateMockStore(t, withMockTiFlash(2))
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int not null primary key, b int not null)")
+	tk.MustExec("alter table t set tiflash replica 1")
+	tk.MustExec("insert into t values(1,0),(2,0),(3,0),(4,0)")
+	tb := external.GetTableByName(t, tk, "test", "t")
+	err := domain.GetDomain(tk.Session()).DDL().UpdateTableReplicaInfo(tk.Session(), tb.Meta().ID, true)
+	require.NoError(t, err)
+	tk.MustExec("set @@session.tidb_enforce_mpp=ON")
+	{
+		item := fmt.Sprintf("return(%d)", kv.GetNewestMppVersion()+1)
+		require.Nil(t, failpoint.Enable("github.com/pingcap/tidb/store/mockstore/unistore/MppVersionError", item))
+	}
+	{
+		err := tk.QueryToErr("select count(*) from t group by b")
+		require.Error(t, err)
+	}
+	require.Nil(t, failpoint.Disable("github.com/pingcap/tidb/store/mockstore/unistore/MppVersionError"))
+	{
+		item := fmt.Sprintf("return(%d)", kv.GetNewestMppVersion())
+		require.Nil(t, failpoint.Enable("github.com/pingcap/tidb/store/mockstore/unistore/MppVersionError", item))
+	}
+	{
+		tk.MustQuery("select count(*) from t group by b").Check(testkit.Rows("4"))
+	}
+	require.Nil(t, failpoint.Disable("github.com/pingcap/tidb/store/mockstore/unistore/MppVersionError"))
+}
+
 func TestCancelMppTasks(t *testing.T) {
 	var hang = "github.com/pingcap/tidb/store/mockstore/unistore/mppRecvHang"
 	store := testkit.CreateMockStore(t, withMockTiFlash(2))
@@ -1325,15 +1357,15 @@ func TestDisaggregatedTiFlashQuery(t *testing.T) {
 	require.NoError(t, err)
 	tk.MustQuery("explain select * from t1 where c1 < 2").Check(testkit.Rows(
 		"PartitionUnion_10 9970.00 root  ",
-		"├─TableReader_15 3323.33 root  data:ExchangeSender_14",
+		"├─TableReader_15 3323.33 root  MppVersion: 1, data:ExchangeSender_14",
 		"│ └─ExchangeSender_14 3323.33 mpp[tiflash]  ExchangeType: PassThrough",
 		"│   └─Selection_13 3323.33 mpp[tiflash]  lt(test.t1.c1, 2)",
 		"│     └─TableFullScan_12 10000.00 mpp[tiflash] table:t1, partition:p0 keep order:false, stats:pseudo",
-		"├─TableReader_19 3323.33 root  data:ExchangeSender_18",
+		"├─TableReader_19 3323.33 root  MppVersion: 1, data:ExchangeSender_18",
 		"│ └─ExchangeSender_18 3323.33 mpp[tiflash]  ExchangeType: PassThrough",
 		"│   └─Selection_17 3323.33 mpp[tiflash]  lt(test.t1.c1, 2)",
 		"│     └─TableFullScan_16 10000.00 mpp[tiflash] table:t1, partition:p1 keep order:false, stats:pseudo",
-		"└─TableReader_23 3323.33 root  data:ExchangeSender_22",
+		"└─TableReader_23 3323.33 root  MppVersion: 1, data:ExchangeSender_22",
 		"  └─ExchangeSender_22 3323.33 mpp[tiflash]  ExchangeType: PassThrough",
 		"    └─Selection_21 3323.33 mpp[tiflash]  lt(test.t1.c1, 2)",
 		"      └─TableFullScan_20 10000.00 mpp[tiflash] table:t1, partition:p2 keep order:false, stats:pseudo"))

kv/BUILD.bazel

@@ -84,6 +84,7 @@ go_test(
         "@com_github_pingcap_kvproto//pkg/coprocessor",
         "@com_github_pingcap_kvproto//pkg/deadlock",
         "@com_github_pingcap_kvproto//pkg/kvrpcpb",
+        "@com_github_pingcap_tipb//go-tipb",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//oracle",