planner: use logical cores as default fine grained stream count (#55544) (#55575)

close #55543

Author: ti-chi-bot

pkg/planner/core/optimizer.go

@@ -19,7 +19,6 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"runtime"
 	"slices"
 	"strconv"
 	"time"
@@ -121,6 +120,8 @@ var optRuleList = []logicalOptRule{
 	&resolveExpand{},
 }
 
+const initialMaxCores uint64 = 10000
+
 // Interaction Rule List
 /* The interaction rule will be trigger when it satisfies following conditions:
 1. The related rule has been trigger and changed the plan
@@ -797,9 +798,7 @@ func (h *fineGrainedShuffleHelper) updateTarget(t shuffleTarget, p *basePhysical
 	h.plans = append(h.plans, p)
 }
 
-// calculateTiFlashStreamCountUsingMinLogicalCores uses minimal logical cpu cores among tiflash servers, and divide by 2
-// return false, 0 if any err happens
-func calculateTiFlashStreamCountUsingMinLogicalCores(ctx context.Context, sctx PlanContext, serversInfo []infoschema.ServerInfo) (bool, uint64) {
+func getTiFlashServerMinLogicalCores(ctx context.Context, sctx PlanContext, serversInfo []infoschema.ServerInfo) (bool, uint64) {
 	failpoint.Inject("mockTiFlashStreamCountUsingMinLogicalCores", func(val failpoint.Value) {
 		intVal, err := strconv.Atoi(val.(string))
 		if err == nil {
@@ -812,7 +811,6 @@ func calculateTiFlashStreamCountUsingMinLogicalCores(ctx context.Context, sctx P
 	if err != nil {
 		return false, 0
 	}
-	var initialMaxCores uint64 = 10000
 	var minLogicalCores = initialMaxCores // set to a large enough value here
 	for _, row := range rows {
 		if row[4].GetString() == "cpu-logical-cores" {
@@ -823,14 +821,19 @@ func calculateTiFlashStreamCountUsingMinLogicalCores(ctx context.Context, sctx P
 		}
 	}
 	// No need to check len(serersInfo) == serverCount here, since missing some servers' info won't affect the correctness
-	if minLogicalCores > 1 && minLogicalCores != initialMaxCores {
-		if runtime.GOARCH == "amd64" {
-			// In most x86-64 platforms, `Thread(s) per core` is 2
-			return true, minLogicalCores / 2
-		}
-		// ARM cpus don't implement Hyper-threading.
+	return true, minLogicalCores
+}
+
+// calculateTiFlashStreamCountUsingMinLogicalCores uses minimal logical cpu cores among tiflash servers
+// return false, 0 if any err happens
+func calculateTiFlashStreamCountUsingMinLogicalCores(ctx context.Context, sctx PlanContext, serversInfo []infoschema.ServerInfo) (bool, uint64) {
+	valid, minLogicalCores := getTiFlashServerMinLogicalCores(ctx, sctx, serversInfo)
+	if !valid {
+		return false, 0
+	}
+	if minLogicalCores != initialMaxCores {
+		// use logical core number as the stream count, the same as TiFlash's default max_threads: https://github.com/pingcap/tiflash/blob/v7.5.0/dbms/src/Interpreters/SettingsCommon.h#L166
 		return true, minLogicalCores
-		// Other platforms are too rare to consider
 	}
 
 	return false, 0

pkg/planner/core/optimizer_test.go

@@ -304,7 +304,7 @@ func TestHandleFineGrainedShuffle(t *testing.T) {
 	require.NoError(t, failpoint.Enable(fpName, fpExpr))
 	defer func() { require.NoError(t, failpoint.Disable(fpName)) }()
 	fpName2 := "github.com/pingcap/tidb/pkg/planner/core/mockTiFlashStreamCountUsingMinLogicalCores"
-	require.NoError(t, failpoint.Enable(fpName2, `return("8")`))
+	require.NoError(t, failpoint.Enable(fpName2, `return("16")`))
 	sctx.GetSessionVars().TiFlashFineGrainedShuffleStreamCount = 0
 
 	col0 := &expression.Column{
@@ -329,7 +329,7 @@ func TestHandleFineGrainedShuffle(t *testing.T) {
 	recv.children = []PhysicalPlan{hashSender}
 	hashSender.children = []PhysicalPlan{tableScan}
 	tableScan.Schema().Columns = append(tableScan.Schema().Columns, col0)
-	start(hashAgg, 8, 3, 0)
+	start(hashAgg, 16, 3, 0)
 
 	// Join(x) <- ExchangeReceiver <- ExchangeSender
 	//                   <- ExchangeReceiver <- ExchangeSender
@@ -357,9 +357,9 @@ func TestHandleFineGrainedShuffle(t *testing.T) {
 	hashSender1.HashCols = partitionCols
 	tableScan1.Schema().Columns = append(tableScan1.Schema().Columns, col0)
 	handleFineGrainedShuffle(nil, sctx.GetPlanCtx(), tableReader)
-	require.Equal(t, uint64(8), hashJoin.TiFlashFineGrainedShuffleStreamCount)
-	require.Equal(t, uint64(8), recv1.TiFlashFineGrainedShuffleStreamCount)
-	require.Equal(t, uint64(8), hashSender1.TiFlashFineGrainedShuffleStreamCount)
+	require.Equal(t, uint64(16), hashJoin.TiFlashFineGrainedShuffleStreamCount)
+	require.Equal(t, uint64(16), recv1.TiFlashFineGrainedShuffleStreamCount)
+	require.Equal(t, uint64(16), hashSender1.TiFlashFineGrainedShuffleStreamCount)
 	require.Equal(t, uint64(0), recv.TiFlashFineGrainedShuffleStreamCount)
 	require.Equal(t, uint64(0), hashSender.TiFlashFineGrainedShuffleStreamCount)
 	clear(plans)