*: init ctx for extractWorker (#55228)

close #55217

pkg/domain/domain.go

@@ -2088,7 +2088,7 @@ func (do *Domain) SetupDumpFileGCChecker(ctx sessionctx.Context) {
 
 // SetupExtractHandle setups extract handler
 func (do *Domain) SetupExtractHandle(sctxs []sessionctx.Context) {
-	do.extractTaskHandle = NewExtractHandler(sctxs)
+	do.extractTaskHandle = newExtractHandler(do.ctx, sctxs)
 }
 
 var planReplayerHandleLease atomic.Uint64

pkg/domain/extract.go

@@ -75,10 +75,10 @@ type ExtractHandle struct {
 	worker *extractWorker
 }
 
-// NewExtractHandler new extract handler
-func NewExtractHandler(sctxs []sessionctx.Context) *ExtractHandle {
+// newExtractHandler new extract handler
+func newExtractHandler(ctx context.Context, sctxs []sessionctx.Context) *ExtractHandle {
 	h := &ExtractHandle{}
-	h.worker = newExtractWorker(sctxs[0], false)
+	h.worker = newExtractWorker(ctx, sctxs[0], false)
 	return h
 }
 
@@ -121,8 +121,13 @@ func NewExtractPlanTask(begin, end time.Time) *ExtractTask {
 	}
 }
 
-func newExtractWorker(sctx sessionctx.Context, isBackgroundWorker bool) *extractWorker {
+func newExtractWorker(
+	ctx context.Context,
+	sctx sessionctx.Context,
+	isBackgroundWorker bool,
+) *extractWorker {
 	return &extractWorker{
+		ctx:                ctx,
 		sctx:               sctx,
 		isBackgroundWorker: isBackgroundWorker,
 	}

pkg/infoschema/sieve.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"sync"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/infoschema/internal"
 )
 
@@ -150,6 +151,10 @@ func (s *Sieve[K, V]) Set(key K, value V) {
 }
 
 func (s *Sieve[K, V]) Get(key K) (value V, ok bool) {
+	failpoint.Inject("skipGet", func() {
+		var v V
+		failpoint.Return(v, false)
+	})
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if e, ok := s.items[key]; ok {

pkg/server/handler/extractorhandler/BUILD.bazel

@@ -28,6 +28,7 @@ go_test(
         ":extractorhandler",
         "//pkg/config",
         "//pkg/domain",
+        "//pkg/kv",
         "//pkg/metrics",
         "//pkg/server",
         "//pkg/server/internal/testserverclient",
@@ -36,12 +37,12 @@ go_test(
         "//pkg/session",
         "//pkg/store/mockstore/unistore",
         "//pkg/testkit",
+        "//pkg/testkit/testfailpoint",
         "//pkg/testkit/testsetup",
         "//pkg/types",
         "//pkg/util/stmtsummary/v2:stmtsummary",
         "//pkg/util/topsql/state",
         "@com_github_gorilla_mux//:mux",
-        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//tikv",
         "@org_uber_go_goleak//:goleak",

pkg/server/handler/extractorhandler/extract_test.go

@@ -24,16 +24,17 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
-	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain"
+	"github.com/pingcap/tidb/pkg/kv"
 	server2 "github.com/pingcap/tidb/pkg/server"
 	"github.com/pingcap/tidb/pkg/server/handler/extractorhandler"
 	"github.com/pingcap/tidb/pkg/server/internal/testserverclient"
 	"github.com/pingcap/tidb/pkg/server/internal/testutil"
 	"github.com/pingcap/tidb/pkg/server/internal/util"
 	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
 	"github.com/pingcap/tidb/pkg/types"
 	stmtsummaryv2 "github.com/pingcap/tidb/pkg/util/stmtsummary/v2"
 	"github.com/stretchr/testify/require"
@@ -44,7 +45,23 @@ func TestExtractHandler(t *testing.T) {
 	defer closeStmtSummary()
 
 	store := testkit.CreateMockStore(t)
+	testExtractHandler(t, store)
+}
+
+func TestExtractHandlerInfoSchemaV2(t *testing.T) {
+	setupStmtSummary()
+	defer closeStmtSummary()
+
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("set @@global.tidb_schema_cache_size = 600*1024*1024")
+	testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/infoschema/skipGet", `return()`)
 
+	testExtractHandler(t, store)
+}
+
+func testExtractHandler(t *testing.T, store kv.Storage) {
+	server2.RunInGoTestChan = make(chan struct{})
 	driver := server2.NewTiDBDriver(store)
 	client := testserverclient.NewTestServerClient()
 	cfg := util.NewTestConfig()
@@ -59,11 +76,14 @@ func TestExtractHandler(t *testing.T) {
 	dom, err := session.GetDomain(store)
 	require.NoError(t, err)
 	server.SetDomain(dom)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("set @@global.tidb_schema_cache_size = 600*1024*1024")
 	go func() {
 		err := server.Run(nil)
 		require.NoError(t, err)
 	}()
 	<-server2.RunInGoTestChan
+
 	client.Port = testutil.GetPortFromTCPAddr(server.ListenAddr())
 	client.StatusPort = testutil.GetPortFromTCPAddr(server.StatusListenerAddr())
 	client.WaitUntilServerOnline()
@@ -75,10 +95,7 @@ func TestExtractHandler(t *testing.T) {
 	eh := &extractorhandler.ExtractTaskServeHandler{ExtractHandler: dom.GetExtractHandle()}
 	router := mux.NewRouter()
 	router.Handle("/extract_task/dump", eh)
-	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/server/extractTaskServeHandler", `return(true)`))
-	defer func() {
-		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/server/extractTaskServeHandler"))
-	}()
+	testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/server/extractTaskServeHandler", `return(true)`)
 	resp0, err := client.FetchStatus(fmt.Sprintf("/extract_task/dump?type=plan&begin=%s&end=%s",
 		url.QueryEscape(startTime.Format(types.TimeFormat)), url.QueryEscape(endTime.Format(types.TimeFormat))))
 	defer os.RemoveAll(domain.GetExtractTaskDirName())

pkg/sessionctx/variable/varsutil.go

@@ -590,12 +590,21 @@ func ParseAnalyzeSkipColumnTypes(val string) map[string]struct{} {
 	return skipTypes
 }
 
+var (
+	// SchemaCacheSizeLowerBound will adjust the schema cache size to this value if
+	// it is lower than this value.
+	SchemaCacheSizeLowerBound uint64 = 512 * units.MiB
+	// SchemaCacheSizeLowerBoundStr is the string representation of
+	// SchemaCacheSizeLowerBound.
+	SchemaCacheSizeLowerBoundStr = "512MB"
+)
+
 func parseSchemaCacheSize(s *SessionVars, normalizedValue string, originalValue string) (byteSize uint64, normalizedStr string, err error) {
 	defer func() {
-		if err == nil && byteSize > 0 && byteSize < (512*units.MiB) {
+		if err == nil && byteSize > 0 && byteSize < SchemaCacheSizeLowerBound {
 			s.StmtCtx.AppendWarning(ErrTruncatedWrongValue.FastGenByArgs(TiDBSchemaCacheSize, originalValue))
-			byteSize = 512 * units.MiB
-			normalizedStr = "512MB"
+			byteSize = SchemaCacheSizeLowerBound
+			normalizedStr = SchemaCacheSizeLowerBoundStr
 		}
 		if err == nil && byteSize > math.MaxInt64 {
 			s.StmtCtx.AppendWarning(ErrTruncatedWrongValue.FastGenByArgs(TiDBSchemaCacheSize, originalValue))