statistics: replace separate TopN merge with combined TopN+histogram merge for global stats (#68147)

ref #66220

Author: mjonss

pkg/executor/analyze_test.go

@@ -164,15 +164,6 @@ func TestAnalyzePartitionTableByConcurrencyInDynamic(t *testing.T) {
 		{
 			concurrency: "1",
 		},
-		{
-			concurrency: "2",
-		},
-		{
-			concurrency: "3",
-		},
-		{
-			concurrency: "4",
-		},
 		{
 			concurrency: "5",
 		},
@@ -181,8 +172,6 @@ func TestAnalyzePartitionTableByConcurrencyInDynamic(t *testing.T) {
 	for _, tc := range testcases {
 		concurrency := tc.concurrency
 		fmt.Println("testcase ", concurrency)
-		tk.MustExec(fmt.Sprintf("set @@global.tidb_merge_partition_stats_concurrency=%v", concurrency))
-		tk.MustQuery("select @@global.tidb_merge_partition_stats_concurrency").Check(testkit.Rows(concurrency))
 		tk.MustExec(fmt.Sprintf("set @@tidb_analyze_partition_concurrency=%v", concurrency))
 		tk.MustQuery("select @@tidb_analyze_partition_concurrency").Check(testkit.Rows(concurrency))
 
@@ -201,30 +190,9 @@ func TestAnalyzePartitionTableByConcurrencyInDynamic(t *testing.T) {
 			strconv.FormatInt(int64(i), 10), "500",
 		})
 	}
-	testcases = []struct {
-		concurrency string
-	}{
-		{
-			concurrency: "1",
-		},
-		{
-			concurrency: "2",
-		},
-		{
-			concurrency: "3",
-		},
-		{
-			concurrency: "4",
-		},
-		{
-			concurrency: "5",
-		},
-	}
 	for _, tc := range testcases {
 		concurrency := tc.concurrency
 		fmt.Println("testcase ", concurrency)
-		tk.MustExec(fmt.Sprintf("set @@tidb_merge_partition_stats_concurrency=%v", concurrency))
-		tk.MustQuery("select @@tidb_merge_partition_stats_concurrency").Check(testkit.Rows(concurrency))
 		tk.MustExec("analyze table t")
 		tk.MustQuery("show stats_topn where partition_name = 'global' and table_name = 't'").CheckAt([]int{5, 6}, expected)
 	}

pkg/executor/set_test.go

@@ -1620,6 +1620,16 @@ func TestSetConcurrency(t *testing.T) {
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1287 The 'tidb_index_serial_scan_concurrency' variable is deprecated. Sequential scans follow 'tidb_executor_concurrency', and index statistics collection uses 'tidb_analyze_distsql_scan_concurrency'."))
 	tk.MustQuery("select @@tidb_index_serial_scan_concurrency;").Check(testkit.Rows("4"))
 
+	// tidb_merge_partition_stats_concurrency is deprecated: setting to 1 is silent, other values warn, value always stays 1.
+	tk.MustExec("set @@tidb_merge_partition_stats_concurrency=1")
+	tk.MustQuery("show warnings").Check(testkit.Rows())
+	tk.MustQuery("select @@tidb_merge_partition_stats_concurrency").Check(testkit.Rows("1"))
+	tk.MustExec("set @@tidb_merge_partition_stats_concurrency=4")
+	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1287 tidb_merge_partition_stats_concurrency is deprecated: the merge no longer runs concurrently, so this setting has no effect. Kept for backward compatibility."))
+	tk.MustQuery("select @@tidb_merge_partition_stats_concurrency").Check(testkit.Rows("1"))
+	// Global getter is overridden too, a stale persisted non-1 value must not leak through.
+	tk.MustQuery("select @@global.tidb_merge_partition_stats_concurrency").Check(testkit.Rows("1"))
+
 	// test setting deprecated value unset
 	tk.MustExec("set @@tidb_index_lookup_concurrency=-1;")
 	tk.MustExec("set @@tidb_index_lookup_join_concurrency=-1;")

pkg/planner/cardinality/row_count_column.go

@@ -106,8 +106,14 @@ func equalRowCountOnColumn(sctx planctx.PlanContext, c *statistics.Column, val t
 	histCnt, matched := c.Histogram.EqualRowCount(sctx, val, true)
 	// Calculate histNDV here as it's needed for both the underrepresented check and later calculations
 	histNDV := float64(c.Histogram.NDV - int64(c.TopN.Num()))
+	// A bucket's upper bound is a value observed in the data, so a zero
+	// Repeat is not a count of zero rows: it means no point frequency was
+	// recorded for it. Merged global histograms produce such buckets when
+	// an upper falls on a merge cut, and the sampled builder produces them
+	// when the estimated NDV exceeds the histogram's row count. Fall
+	// through to the uniform estimate rather than report an exact zero.
 	// also check if this last bucket end value is underrepresented
-	if matched && !IsLastBucketEndValueUnderrepresented(sctx,
+	if matched && histCnt > 0 && !IsLastBucketEndValueUnderrepresented(sctx,
 		&c.Histogram, val, histCnt, histNDV, realtimeRowCount, modifyCount) {
 		return statistics.DefaultRowEst(histCnt), nil
 	}

pkg/planner/cardinality/row_count_index.go

@@ -433,8 +433,11 @@ func equalRowCountOnIndex(sctx planctx.PlanContext, idx *statistics.Index, b []b
 	histCnt, matched := idx.Histogram.EqualRowCount(sctx, val, true)
 	// Calculate histNDV here as it's needed for both the underrepresented check and later calculations
 	histNDV := float64(idx.Histogram.NDV - int64(idx.TopN.Num()))
+	// A zero Repeat means no point frequency was recorded for this upper
+	// bound, not that the value has no rows. See equalRowCount in
+	// row_count_column.go.
 	// also check if this last bucket end value is underrepresented
-	if matched && !IsLastBucketEndValueUnderrepresented(sctx,
+	if matched && histCnt > 0 && !IsLastBucketEndValueUnderrepresented(sctx,
 		&idx.Histogram, val, histCnt, histNDV, realtimeRowCount, modifyCount) {
 		return statistics.DefaultRowEst(histCnt)
 	}

pkg/planner/cardinality/selectivity_test.go

@@ -3013,3 +3013,37 @@ func TestUninitializedStats(t *testing.T) {
 	tk.MustQuery("show stats_histograms").CheckNotContain("allEvicted")
 	tk.MustQuery("explain analyze format = 'brief' select /*+ use_index(t1, idx_expr) */ * from t1 where (cast(json_unquote(json_extract(`c2`, _utf8mb4'$.location_id')) as char(255)) collate utf8mb4_bin) > '100'  and c2 > 'abc';").CheckNotContain("unInitialized")
 }
+
+// TestEqualEstimateOnZeroRepeatBucketUpper covers equality estimation on a
+// bucket upper that carries no point frequency. Merged global histograms
+// produce such buckets when an upper falls on a merge cut, and the sampled
+// builder produces them when the estimated NDV exceeds the histogram's row
+// count. A bucket upper is a value observed in the data, so a zero Repeat
+// means "not recorded" rather than "no rows", and the estimate must fall
+// back to the uniform average instead of reporting an exact zero.
+func TestEqualEstimateOnZeroRepeatBucketUpper(t *testing.T) {
+	tp := types.NewFieldType(mysql.TypeLonglong)
+	colInfo := &model.ColumnInfo{ID: 1, FieldType: *tp}
+	// 200 rows over an NDV of 100, so the uniform average is 2 per value.
+	hg := statistics.NewHistogram(colInfo.ID, 100, 0, 0, tp, 2, 0)
+	lo1, up1 := types.NewIntDatum(1), types.NewIntDatum(50)
+	lo2, up2 := types.NewIntDatum(51), types.NewIntDatum(100)
+	hg.AppendBucket(&lo1, &up1, 100, 0) // no frequency recorded for 50
+	hg.AppendBucket(&lo2, &up2, 200, 5) // 100 was observed 5 times
+	col := &statistics.Column{
+		Histogram:         *hg,
+		Info:              colInfo,
+		StatsLoadedStatus: statistics.NewStatsFullLoadStatus(),
+		StatsVer:          2,
+	}
+	sctx := mock.NewContext()
+
+	est, err := getColumnRowCount(sctx, col, getRange(50, 50), 200, 0, false)
+	require.NoError(t, err)
+	require.Equal(t, 2.0, est.Est,
+		"a zero Repeat must fall back to the uniform average, not report zero rows")
+
+	est, err = getColumnRowCount(sctx, col, getRange(100, 100), 200, 0, false)
+	require.NoError(t, err)
+	require.Equal(t, 5.0, est.Est, "an observed Repeat must still be used as is")
+}

pkg/sessionctx/vardef/tidb_vars.go

@@ -1026,7 +1026,8 @@ const (
 	TiDBOptUseInvisibleIndexes = "tidb_opt_use_invisible_indexes"
 	// TiDBAnalyzePartitionConcurrency is the number of concurrent workers to save statistics to the system tables.
 	TiDBAnalyzePartitionConcurrency = "tidb_analyze_partition_concurrency"
-	// TiDBMergePartitionStatsConcurrency indicates the concurrency when merge partition stats into global stats
+	// TiDBMergePartitionStatsConcurrency is deprecated. It is kept for backward compatibility
+	// but no longer affects behavior. Global stats always use the combined merge algorithm.
 	TiDBMergePartitionStatsConcurrency = "tidb_merge_partition_stats_concurrency"
 	// TiDBEnableAsyncMergeGlobalStats indicates whether to enable async merge global stats
 	TiDBEnableAsyncMergeGlobalStats = "tidb_enable_async_merge_global_stats"
@@ -1758,23 +1759,22 @@ const (
 	MinTiDBInstancePlanCacheMemSize                   = 100 * size.MB
 	DefTiDBInstancePlanCacheReservedPercentage        = 0.1
 	// MaxDDLReorgBatchSize is exported for testing.
-	MaxDDLReorgBatchSize                  int32  = 10240
-	MinDDLReorgBatchSize                  int32  = 32
-	MinExpensiveQueryTimeThreshold        uint64 = 10 // 10s
-	MinExpensiveTxnTimeThreshold          uint64 = 60 // 60s
-	DefTiDBAutoBuildStatsConcurrency             = DefBuildStatsConcurrency
-	DefTiDBSysProcScanConcurrency                = DefAnalyzeDistSQLScanConcurrency
-	DefTiDBRcWriteCheckTs                        = false
-	DefTiDBForeignKeyChecks                      = true
-	DefTiDBForeignKeyCheckInSharedLock           = false
-	DefTiDBOptAdvancedJoinHint                   = true
-	DefTiDBAnalyzePartitionConcurrency           = 2
-	DefTiDBOptRangeMaxSize                       = 64 * int64(size.MB) // 64 MB
-	DefTiDBCostModelVer                          = 2
-	DefTiDBServerMemoryLimitSessMinSize          = 128 << 20
-	DefTiDBMergePartitionStatsConcurrency        = 1
-	DefTiDBServerMemoryLimitGCTrigger            = 0.7
-	DefTiDBEnableGOGCTuner                       = true
+	MaxDDLReorgBatchSize                int32  = 10240
+	MinDDLReorgBatchSize                int32  = 32
+	MinExpensiveQueryTimeThreshold      uint64 = 10 // 10s
+	MinExpensiveTxnTimeThreshold        uint64 = 60 // 60s
+	DefTiDBAutoBuildStatsConcurrency           = DefBuildStatsConcurrency
+	DefTiDBSysProcScanConcurrency              = DefAnalyzeDistSQLScanConcurrency
+	DefTiDBRcWriteCheckTs                      = false
+	DefTiDBForeignKeyChecks                    = true
+	DefTiDBForeignKeyCheckInSharedLock         = false
+	DefTiDBOptAdvancedJoinHint                 = true
+	DefTiDBAnalyzePartitionConcurrency         = 2
+	DefTiDBOptRangeMaxSize                     = 64 * int64(size.MB) // 64 MB
+	DefTiDBCostModelVer                        = 2
+	DefTiDBServerMemoryLimitSessMinSize        = 128 << 20
+	DefTiDBServerMemoryLimitGCTrigger          = 0.7
+	DefTiDBEnableGOGCTuner                     = true
 	// DefTiDBGOGCTunerThreshold is to limit TiDBGOGCTunerThreshold.
 	DefTiDBGOGCTunerThreshold                 float64 = 0.6
 	DefTiDBGOGCMaxValue                               = 500

pkg/sessionctx/variable/session.go

@@ -1708,8 +1708,6 @@ type SessionVars struct {
 
 	// AnalyzePartitionConcurrency indicates concurrency for partitions in Analyze
 	AnalyzePartitionConcurrency int
-	// AnalyzePartitionMergeConcurrency indicates concurrency for merging partition stats
-	AnalyzePartitionMergeConcurrency int
 
 	// EnableAsyncMergeGlobalStats indicates whether to enable async merge global stats
 	EnableAsyncMergeGlobalStats bool

pkg/sessionctx/variable/sysvar.go

@@ -3230,11 +3230,28 @@ var defaultSysVars = []*SysVar{
 		},
 	},
 	{
-		Scope: vardef.ScopeGlobal | vardef.ScopeSession, Name: vardef.TiDBMergePartitionStatsConcurrency, Value: strconv.FormatInt(vardef.DefTiDBMergePartitionStatsConcurrency, 10), Type: vardef.TypeInt, MinValue: 1, MaxValue: vardef.MaxConfigurableConcurrency,
-		SetSession: func(s *SessionVars, val string) error {
-			s.AnalyzePartitionMergeConcurrency = TidbOptInt(val, vardef.DefTiDBMergePartitionStatsConcurrency)
+		Scope: vardef.ScopeGlobal | vardef.ScopeSession, Name: vardef.TiDBMergePartitionStatsConcurrency, Value: "1", Type: vardef.TypeInt, MinValue: 1, MaxValue: vardef.MaxConfigurableConcurrency,
+		SetSession: func(_ *SessionVars, _ string) error {
+			// Deprecated: do nothing.
 			return nil
 		},
+		// Both read paths return "1" unconditionally. Validation alone is
+		// not enough: session.GetGlobalSysVar() applies only
+		// ValidateFromType on the persisted value (skipping the
+		// Validation callback), so a cluster upgraded from an older
+		// TiDB with a non-1 value persisted in mysql.global_variables
+		// would otherwise read that stale value.
+		GetSession: func(_ *SessionVars) (string, error) { return "1", nil },
+		GetGlobal:  func(_ context.Context, _ *SessionVars) (string, error) { return "1", nil },
+		Validation: func(vars *SessionVars, normalizedValue string, _ string, _ vardef.ScopeFlag) (string, error) {
+			if normalizedValue != "1" {
+				// Use errWarnDeprecatedSyntax (MySQL code 1287) for
+				// consistency with other deprecated sysvar warnings
+				// such as tidb_index_serial_scan_concurrency.
+				vars.StmtCtx.AppendWarning(errWarnDeprecatedSyntax.FastGen("tidb_merge_partition_stats_concurrency is deprecated: the merge no longer runs concurrently, so this setting has no effect. Kept for backward compatibility."))
+			}
+			return "1", nil
+		},
 	},
 	{
 		Scope: vardef.ScopeGlobal | vardef.ScopeSession,

pkg/statistics/BUILD.bazel

@@ -51,6 +51,7 @@ go_library(
         "//pkg/util/memory",
         "//pkg/util/ranger",
         "//pkg/util/sqlexec",
+        "//pkg/util/sqlkiller",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_tipb//go-tipb",
@@ -69,10 +70,13 @@ go_test(
         "cmsketch_test.go",
         "estimate_test.go",
         "fmsketch_test.go",
-        "histogram_bench_test.go",
+        "histogram_fuzz_test.go",
         "histogram_test.go",
         "integration_test.go",
         "main_test.go",
+        "merge_global_cases_test.go",
+        "merge_global_test.go",
+        "merge_global_types_test.go",
         "sample_test.go",
         "scalar_test.go",
         "statistics_test.go",
@@ -81,13 +85,16 @@ go_test(
     data = glob(["testdata/**"]),
     embed = [":statistics"],
     flaky = True,
-    shard_count = 45,
+    shard_count = 50,
     deps = [
         "//pkg/config",
         "//pkg/meta/model",
         "//pkg/parser/ast",
+        "//pkg/parser/charset",
         "//pkg/parser/mysql",
+        "//pkg/planner/cardinality",
         "//pkg/planner/core/resolve",
+        "//pkg/planner/planctx",
         "//pkg/sessionctx",
         "//pkg/sessionctx/stmtctx",
         "//pkg/sessionctx/vardef",
@@ -107,6 +114,7 @@ go_test(
         "//pkg/util/mock",
         "//pkg/util/ranger",
         "//pkg/util/sqlexec",
+        "//pkg/util/sqlkiller",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",

pkg/statistics/cmsketch_util.go

@@ -17,58 +17,24 @@ package statistics
 import (
 	"time"
 
+	"github.com/pingcap/tidb/pkg/tablecodec"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/codec"
-	"github.com/pingcap/tidb/pkg/util/hack"
 )
 
-// DatumMapCache is used to store the mapping from the string type to datum type.
-// The datum is used to find the value in the histogram.
-type DatumMapCache struct {
-	datumMap map[hack.MutableString]types.Datum
-}
-
-// NewDatumMapCache creates a new DatumMapCache.
-func NewDatumMapCache() *DatumMapCache {
-	return &DatumMapCache{
-		datumMap: make(map[hack.MutableString]types.Datum),
-	}
-}
-
-// Get gets the datum from the cache.
-func (d *DatumMapCache) Get(key hack.MutableString) (val types.Datum, ok bool) {
-	val, ok = d.datumMap[key]
-	return
-}
-
-// Put puts the datum into the cache.
-func (d *DatumMapCache) Put(val TopNMeta, encodedVal hack.MutableString,
-	tp byte, isIndex bool, loc *time.Location) (dat types.Datum, err error) {
-	dat, err = topNMetaToDatum(val, tp, isIndex, loc)
-	if err != nil {
-		return dat, err
-	}
-	d.datumMap[encodedVal] = dat
-	return dat, nil
-}
-
 func topNMetaToDatum(val TopNMeta,
-	tp byte, isIndex bool, loc *time.Location) (dat types.Datum, err error) {
+	ft *types.FieldType, isIndex bool, loc *time.Location) (dat types.Datum, err error) {
 	if isIndex {
 		dat.SetBytes(val.Encoded)
-	} else {
-		var err error
-		if types.IsTypeTime(tp) {
-			// Handle date time values specially since they are encoded to int and we'll get int values if using DecodeOne.
-			_, dat, err = codec.DecodeAsDateTime(val.Encoded, tp, loc)
-		} else if types.IsTypeFloat(tp) {
-			_, dat, err = codec.DecodeAsFloat32(val.Encoded, tp)
-		} else {
-			_, dat, err = codec.DecodeOne(val.Encoded)
-		}
-		if err != nil {
-			return dat, err
-		}
+		return dat, nil
+	}
+	if _, dat, err = codec.DecodeOne(val.Encoded); err != nil {
+		return dat, err
 	}
-	return dat, err
+	// The key encodes a value in its flattened form: ENUM, SET and BIT
+	// as their numeric value, times as a packed integer, TypeFloat as a
+	// float64. Unflatten restores the kind the column's own values
+	// carry, which matters because Datum.Compare dispatches on kind and
+	// because a histogram's chunk column is typed.
+	return tablecodec.Unflatten(dat, ft, loc)
 }