planner: add an unique key into mysql.bind_info (#60296)

ref #60148

Author: qw4990

br/pkg/restore/snap_client/systable_restore_test.go

@@ -117,5 +117,5 @@ func TestCheckSysTableCompatibility(t *testing.T) {
 
 // The above variables are in the file br/pkg/restore/systable_restore.go
 func TestMonitorTheSystemTableIncremental(t *testing.T) {
-	require.Equal(t, int64(245), session.CurrentBootstrapVersion)
+	require.Equal(t, int64(246), session.CurrentBootstrapVersion)
 }

pkg/bindinfo/binding_operator.go

@@ -97,6 +97,13 @@ func (op *bindingOperator) CreateBinding(sctx sessionctx.Context, bindings []*Bi
 			// overflow directly.
 
 			// Insert the Bindings to the storage.
+			var sqlDigest, planDigest any // null by default
+			if binding.SQLDigest != "" {
+				sqlDigest = binding.SQLDigest
+			}
+			if binding.PlanDigest != "" {
+				planDigest = binding.PlanDigest
+			}
 			_, err = exec(
 				sctx,
 				`INSERT INTO mysql.bind_info VALUES (%?,%?, %?, %?, %?, %?, %?, %?, %?, %?, %?)`,
@@ -109,8 +116,8 @@ func (op *bindingOperator) CreateBinding(sctx sessionctx.Context, bindings []*Bi
 				binding.Charset,
 				binding.Collation,
 				binding.Source,
-				binding.SQLDigest,
-				binding.PlanDigest,
+				sqlDigest,
+				planDigest,
 			)
 			failpoint.Inject("CreateGlobalBindingNthFail", func(val failpoint.Value) {
 				n := val.(int)

pkg/bindinfo/binding_operator_test.go

@@ -260,8 +260,6 @@ func TestSetBindingStatusWithoutBindingInCache(t *testing.T) {
 
 	// Simulate creating bindings on other machines
 	_, sqlDigest := parser.NormalizeDigestForBinding("select * from `test` . `t` where `a` > ?")
-	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT /*+ USE_INDEX(`t` `idx_a`)*/ * FROM `test`.`t` WHERE `a` > 10', 'test', 'deleted', '2000-01-01 09:00:00', '2000-01-01 09:00:00', '', '','" +
-		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT /*+ USE_INDEX(`t` `idx_a`)*/ * FROM `test`.`t` WHERE `a` > 10', 'test', 'enabled', '2000-01-02 09:00:00', '2000-01-02 09:00:00', '', '','" +
 		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("set binding disabled for select * from t where a > 10")
@@ -274,8 +272,6 @@ func TestSetBindingStatusWithoutBindingInCache(t *testing.T) {
 	utilCleanBindingEnv(tk)
 
 	// Simulate creating bindings on other machines
-	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT * FROM `test`.`t` WHERE `a` > 10', 'test', 'deleted', '2000-01-01 09:00:00', '2000-01-01 09:00:00', '', '','" +
-		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT * FROM `test`.`t` WHERE `a` > 10', 'test', 'disabled', '2000-01-02 09:00:00', '2000-01-02 09:00:00', '', '','" +
 		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("set binding enabled for select * from t where a > 10")

pkg/session/BUILD.bazel

@@ -155,7 +155,6 @@ go_test(
     shard_count = 50,
     deps = [
         "//pkg/autoid_service",
-        "//pkg/bindinfo",
         "//pkg/config",
         "//pkg/ddl",
         "//pkg/domain",
@@ -165,7 +164,6 @@ go_test(
         "//pkg/keyspace",
         "//pkg/kv",
         "//pkg/meta",
-        "//pkg/parser",
         "//pkg/parser/ast",
         "//pkg/parser/auth",
         "//pkg/session/types",

pkg/session/bootstrap.go

@@ -304,10 +304,11 @@ const (
 		charset TEXT NOT NULL,
 		collation TEXT NOT NULL,
 		source VARCHAR(10) NOT NULL DEFAULT 'unknown',
-		sql_digest varchar(64),
-		plan_digest varchar(64),
+		sql_digest varchar(64) DEFAULT NULL,
+		plan_digest varchar(64) DEFAULT NULL,
 		INDEX sql_index(original_sql(700),default_db(68)) COMMENT "accelerate the speed when add global binding query",
-		INDEX time_index(update_time) COMMENT "accelerate the speed when querying with last update time"
+		INDEX time_index(update_time) COMMENT "accelerate the speed when querying with last update time",
+		UNIQUE INDEX digest_index(plan_digest, sql_digest) COMMENT "avoid duplicated records"
 	) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;`
 
 	// CreateRoleEdgesTable stores the role and user relationship information.
@@ -1275,11 +1276,14 @@ const (
 
 	// version245 updates column types of mysql.bind_info.
 	version245 = 245
+
+	// version246 adds new unique index for mysql.bind_info.
+	version246 = 246
 )
 
 // currentBootstrapVersion is defined as a variable, so we can modify its value for testing.
 // please make sure this is the largest version
-var currentBootstrapVersion int64 = version245
+var currentBootstrapVersion int64 = version246
 
 // DDL owner key's expired time is ManagerSessionTTL seconds, we should wait the time and give more time to have a chance to finish it.
 var internalSQLTimeout = owner.ManagerSessionTTL + 15
@@ -1460,6 +1464,7 @@ var (
 		upgradeToVer243,
 		upgradeToVer244,
 		upgradeToVer245,
+		upgradeToVer246,
 	}
 )
 
@@ -3406,6 +3411,60 @@ func upgradeToVer245(s sessiontypes.Session, ver int64) {
 	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN bind_sql LONGTEXT NOT NULL")
 }
 
+func upgradeToVer246(s sessiontypes.Session, ver int64) {
+	if ver >= version246 {
+		return
+	}
+
+	// log duplicated digests that will be set to null.
+	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
+	rs, err := s.ExecuteInternal(ctx,
+		`select plan_digest, sql_digest from mysql.bind_info group by plan_digest, sql_digest having count(1) > 1`)
+	if err != nil {
+		logutil.BgLogger().Fatal("failed to get duplicated plan and sql digests", zap.Error(err))
+		return
+	}
+	req := rs.NewChunk(nil)
+	duplicatedDigests := make(map[string]struct{})
+	for {
+		err = rs.Next(ctx, req)
+		if err != nil {
+			logutil.BgLogger().Fatal("failed to get duplicated plan and sql digests", zap.Error(err))
+			return
+		}
+		if req.NumRows() == 0 {
+			break
+		}
+		for i := 0; i < req.NumRows(); i++ {
+			planDigest, sqlDigest := req.GetRow(i).GetString(0), req.GetRow(i).GetString(1)
+			duplicatedDigests[sqlDigest+", "+planDigest] = struct{}{}
+		}
+		req.Reset()
+	}
+	if err := rs.Close(); err != nil {
+		logutil.BgLogger().Warn("failed to close record set", zap.Error(err))
+	}
+	if len(duplicatedDigests) > 0 {
+		digestList := make([]string, 0, len(duplicatedDigests))
+		for k := range duplicatedDigests {
+			digestList = append(digestList, "("+k+")")
+		}
+		logutil.BgLogger().Warn("set the following (plan digest, sql digest) in mysql.bind_info to null " +
+			"for adding new unique index: " + strings.Join(digestList, ", "))
+	}
+
+	// to avoid the failure of adding the unique index, remove duplicated rows on these 2 digest columns first.
+	// in most cases, there should be no duplicated rows, since now we only store one binding for each sql_digest.
+	// compared with upgrading failure, it's OK to set these 2 columns to null.
+	doReentrantDDL(s, `UPDATE mysql.bind_info SET plan_digest=null, sql_digest=null
+                       WHERE (plan_digest, sql_digest) in (
+                         select plan_digest, sql_digest from mysql.bind_info
+                         group by plan_digest, sql_digest having count(1) > 1)`)
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN sql_digest VARCHAR(64) DEFAULT NULL")
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN plan_digest VARCHAR(64) DEFAULT NULL")
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info ADD UNIQUE INDEX digest_index(plan_digest, sql_digest)", dbterror.ErrDupKeyName)
+}
+
 // initGlobalVariableIfNotExists initialize a global variable with specific val if it does not exist.
 func initGlobalVariableIfNotExists(s sessiontypes.Session, name string, val any) {
 	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)