stmtsummary: fix issue of concurrent map read and write (#35367)

infoschema/tables_test.go

@@ -21,7 +21,9 @@ import (
 	"net/http/httptest"
 	"os"
 	"runtime"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -1870,3 +1872,31 @@ func (s *testClusterTableSuite) TestDataLockWaitsPrivilege(c *C) {
 	}, nil, nil), IsTrue)
 	_ = tk.MustQuery("select * from information_schema.DATA_LOCK_WAITS")
 }
+
+func (s *testTableSuite) TestStmtSummaryIssue35340(c *C) {
+	tk := s.newTestKitWithRoot(c)
+	tk.MustExec("set global tidb_stmt_summary_refresh_interval=1800")
+	tk.MustExec("set global tidb_stmt_summary_max_stmt_count = 3000")
+	for i := 0; i < 100; i++ {
+		user := "user" + strconv.Itoa(i)
+		tk.MustExec(fmt.Sprintf("create user '%v'@'localhost'", user))
+	}
+	tk.MustExec("flush privileges")
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			tk := s.newTestKitWithRoot(c)
+			for j := 0; j < 100; j++ {
+				user := "user" + strconv.Itoa(j)
+				c.Assert(tk.Se.Auth(&auth.UserIdentity{
+					Username: user,
+					Hostname: "localhost",
+				}, nil, nil), IsTrue)
+				tk.MustQuery("select count(*) from information_schema.statements_summary;")
+			}
+		}()
+	}
+	wg.Wait()
+}

util/stmtsummary/reader.go

@@ -118,11 +118,7 @@ func (ssr *stmtSummaryReader) getStmtByDigestRow(ssbd *stmtSummaryByDigest, begi
 
 	// `ssElement` is lazy expired, so expired elements could also be read.
 	// `beginTime` won't change since `ssElement` is created, so locking is not needed here.
-	isAuthed := true
-	if ssr.user != nil && !ssr.isSuper {
-		_, isAuthed = ssElement.authUsers[ssr.user.Username]
-	}
-	if ssElement == nil || ssElement.beginTime < beginTimeForCurInterval || !isAuthed {
+	if ssElement == nil || ssElement.beginTime < beginTimeForCurInterval {
 		return nil
 	}
 	return ssr.getStmtByDigestElementRow(ssElement, ssbd)
@@ -131,6 +127,13 @@ func (ssr *stmtSummaryReader) getStmtByDigestRow(ssbd *stmtSummaryByDigest, begi
 func (ssr *stmtSummaryReader) getStmtByDigestElementRow(ssElement *stmtSummaryByDigestElement, ssbd *stmtSummaryByDigest) []types.Datum {
 	ssElement.Lock()
 	defer ssElement.Unlock()
+	isAuthed := true
+	if ssr.user != nil && !ssr.isSuper {
+		_, isAuthed = ssElement.authUsers[ssr.user.Username]
+	}
+	if !isAuthed {
+		return nil
+	}
 	datums := make([]types.Datum, len(ssr.columnValueFactories))
 	for i, factory := range ssr.columnValueFactories {
 		datums[i] = types.NewDatum(factory(ssElement, ssbd))
@@ -144,12 +147,9 @@ func (ssr *stmtSummaryReader) getStmtByDigestHistoryRow(ssbd *stmtSummaryByDiges
 
 	rows := make([][]types.Datum, 0, len(ssElements))
 	for _, ssElement := range ssElements {
-		isAuthed := true
-		if ssr.user != nil && !ssr.isSuper {
-			_, isAuthed = ssElement.authUsers[ssr.user.Username]
-		}
-		if isAuthed {
-			rows = append(rows, ssr.getStmtByDigestElementRow(ssElement, ssbd))
+		row := ssr.getStmtByDigestElementRow(ssElement, ssbd)
+		if len(row) > 0 {
+			rows = append(rows, row)
 		}
 	}
 	return rows