statistics: fix wrong result in the MergeGlobalStatsTopNByConcurrency (#48757)

ref #48713

Author: hawkingrei

pkg/statistics/handle/globalstats/BUILD.bazel

@@ -44,7 +44,7 @@ go_test(
     ],
     embed = [":globalstats"],
     flaky = True,
-    shard_count = 21,
+    shard_count = 23,
     deps = [
         "//pkg/config",
         "//pkg/parser/model",

pkg/statistics/handle/globalstats/globalstats_test.go

@@ -1240,3 +1240,38 @@ func TestGlobalIndexStatistics(t *testing.T) {
 				"└─IndexRangeScan_11 4.00 cop[tikv] table:t, index:idx(b) range:[-inf,16), keep order:true"))
 	}
 }
+
+func TestIssues24349(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("set @@tidb_partition_prune_mode='dynamic'")
+	testKit.MustExec("set @@tidb_analyze_version=2")
+	defer testKit.MustExec("set @@tidb_analyze_version=1")
+	defer testKit.MustExec("set @@tidb_partition_prune_mode='static'")
+	testIssues24349(testKit)
+}
+
+func TestIssues24349WithConcurrency(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("set @@tidb_partition_prune_mode='dynamic'")
+	testKit.MustExec("set @@tidb_analyze_version=2")
+	testKit.MustExec("set global tidb_merge_partition_stats_concurrency=2")
+	defer testKit.MustExec("set @@tidb_analyze_version=1")
+	defer testKit.MustExec("set @@tidb_partition_prune_mode='static'")
+	defer testKit.MustExec("set global tidb_merge_partition_stats_concurrency=1")
+	testIssues24349(testKit)
+}
+
+func testIssues24349(testKit *testkit.TestKit) {
+	testKit.MustExec("create table t (a int, b int) partition by hash(a) partitions 3")
+	testKit.MustExec("insert into t values (0, 3), (0, 3), (0, 3), (0, 2), (1, 1), (1, 2), (1, 2), (1, 2), (1, 3), (1, 4), (2, 1), (2, 1)")
+	testKit.MustExec("analyze table t with 1 topn, 3 buckets")
+	testKit.MustQuery("show stats_buckets where partition_name='global'").Check(testkit.Rows(
+		"test t global a 0 0 2 2 0 2 0",
+		"test t global b 0 0 3 1 1 2 0",
+		"test t global b 0 1 10 1 4 4 0",
+	))
+}

pkg/statistics/handle/globalstats/merge_worker.go

@@ -107,6 +107,7 @@ func (worker *topnStatsMergeWorker) Run(timeZone *time.Location, isIndex bool,
 		datumMap := statistics.NewDatumMapCache()
 
 		for i, topN := range checkTopNs {
+			i = i + start
 			if err := worker.killer.HandleSignal(); err != nil {
 				resp.Err = err
 				worker.respCh <- resp

pkg/statistics/handle/handletest/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 37,
+    shard_count = 36,
     deps = [
         "//pkg/config",
         "//pkg/domain",

pkg/statistics/handle/handletest/handle_test.go

@@ -1361,24 +1361,6 @@ func TestStatsCacheUpdateSkip(t *testing.T) {
 	require.Equal(t, statsTbl2, statsTbl1)
 }
 
-func TestIssues24349(t *testing.T) {
-	store := testkit.CreateMockStore(t)
-	testKit := testkit.NewTestKit(t, store)
-	testKit.MustExec("use test")
-	testKit.MustExec("set @@tidb_partition_prune_mode='dynamic'")
-	testKit.MustExec("set @@tidb_analyze_version=2")
-	defer testKit.MustExec("set @@tidb_analyze_version=1")
-	defer testKit.MustExec("set @@tidb_partition_prune_mode='static'")
-	testKit.MustExec("create table t (a int, b int) partition by hash(a) partitions 3")
-	testKit.MustExec("insert into t values (0, 3), (0, 3), (0, 3), (0, 2), (1, 1), (1, 2), (1, 2), (1, 2), (1, 3), (1, 4), (2, 1), (2, 1)")
-	testKit.MustExec("analyze table t with 1 topn, 3 buckets")
-	testKit.MustQuery("show stats_buckets where partition_name='global'").Check(testkit.Rows(
-		"test t global a 0 0 2 2 0 2 0",
-		"test t global b 0 0 3 1 1 2 0",
-		"test t global b 0 1 10 1 4 4 0",
-	))
-}
-
 func testIncrementalModifyCountUpdateHelper(analyzeSnapshot bool) func(*testing.T) {
 	return func(t *testing.T) {
 		store, dom := testkit.CreateMockStoreAndDomain(t)

pkg/statistics/histogram.go

@@ -1188,6 +1188,7 @@ func newbucket4MergingForRecycle() *bucket4Merging {
 }
 
 func releasebucket4MergingForRecycle(b *bucket4Merging) {
+	b.disjointNDV = 0
 	bucket4MergingPool.Put(b)
 }