sql/colexec: fix spill threshold (#23862)

- Auto spill threshold now uses (MemoryTotal - fileCacheMemory) / GoMaxProcs / 8 for both hashbuild and group operators, avoiding over-commitment when filecache is large
- Add CanSpill bool to HashBuild; set true only for hashjoin so dedupjoin never triggers spill

Approved by: @ouyuanning

Author: aunjgr

pkg/sql/colexec/group/types2.go

@@ -26,6 +26,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/aggexec"
@@ -130,7 +131,8 @@ func (ctr *container) setSpillMem(m int64, aggs []aggexec.AggFuncExecExpression)
 
 	if m == 0 {
 		// 0 means auto config.   Here the formula is made up on the fly.
-		mem := int64(system.MemoryTotal()) / int64(system.GoMaxProcs()) / 8
+		fileCacheMem := fileservice.GlobalMemoryCacheSizeHint.Load()
+		mem := (int64(system.MemoryTotal()) - fileCacheMem) / int64(system.GoMaxProcs()) / 8
 		// min 128MB
 		if mem < common.MiB*128 {
 			mem = common.MiB * 128

pkg/sql/colexec/hashbuild/build.go

@@ -173,7 +173,7 @@ func (ctr *container) build(hashBuild *HashBuild, proc *process.Process, analyze
 		}
 
 		// Check if we should enter spill mode based on batch memory size
-		if hashBuild.NeedHashMap && hashBuild.shouldSpillBatches() {
+		if hashBuild.shouldSpillBatches() {
 			spillMode = true
 			// Create spill files once
 			spilledBuckets, spillFiles, err = createSpillFiles(proc)

pkg/sql/colexec/hashbuild/spill.go

@@ -206,8 +206,11 @@ func (ctr *container) rowCnt() int64 {
 }
 
 func (hashBuild *HashBuild) shouldSpillBatches() bool {
+	if !hashBuild.CanSpill || !hashBuild.IsShuffle || !hashBuild.NeedHashMap {
+		return false
+	}
 	ctr := &hashBuild.ctr
-	if !hashBuild.IsShuffle || ctr.spillThreshold <= 0 {
+	if ctr.spillThreshold <= 0 {
 		return false
 	}
 	if ctr.spillThreshold <= 100000 {

pkg/sql/colexec/hashbuild/spill_test.go

@@ -150,6 +150,8 @@ func TestShouldSpillBatches(t *testing.T) {
 		hb := &HashBuild{
 			IsShuffle:      true,
 			SpillThreshold: 1, // 1 byte
+			CanSpill:       true,
+			NeedHashMap:    true,
 		}
 		hb.ctr.setSpillThreshold(1)
 		bat := batch.NewWithSize(1)
@@ -528,6 +530,8 @@ func TestShouldSpillBatchesRowThreshold(t *testing.T) {
 	hb := &HashBuild{
 		IsShuffle:      true,
 		SpillThreshold: 10, // Small row threshold
+		CanSpill:       true,
+		NeedHashMap:    true,
 	}
 	hb.ctr.setSpillThreshold(10)
 

pkg/sql/colexec/hashbuild/types.go

@@ -21,6 +21,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/reuse"
 	"github.com/matrixorigin/matrixone/pkg/common/system"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/vm"
 	"github.com/matrixorigin/matrixone/pkg/vm/message"
@@ -56,6 +57,7 @@ type HashBuild struct {
 	NeedBatches       bool
 	NeedAllocateSels  bool
 	IsShuffle         bool
+	CanSpill          bool
 	Conditions        []*plan.Expr
 	JoinMapTag        int32
 	JoinMapRefCnt     int32
@@ -145,7 +147,8 @@ func (hashBuild *HashBuild) ExecProjection(proc *process.Process, input *batch.B
 func (ctr *container) setSpillThreshold(threshold int64) {
 	if threshold == 0 {
 		// 0 means auto config
-		mem := int64(system.MemoryTotal()) / int64(system.GoMaxProcs()) / 8
+		fileCacheMem := fileservice.GlobalMemoryCacheSizeHint.Load()
+		mem := (int64(system.MemoryTotal()) - fileCacheMem) / int64(system.GoMaxProcs()) / 8
 		// min 128MB
 		if mem < common.MiB*128 {
 			mem = common.MiB * 128

pkg/sql/compile/operator.go

@@ -134,6 +134,7 @@ func dupOperator(sourceOp vm.Operator, index int, maxParallel int) vm.Operator {
 		op.NeedBatches = t.NeedBatches
 		op.NeedAllocateSels = t.NeedAllocateSels
 		op.IsShuffle = t.IsShuffle
+		op.CanSpill = t.CanSpill
 		op.Conditions = t.Conditions
 		op.JoinMapTag = t.JoinMapTag
 		op.JoinMapRefCnt = t.JoinMapRefCnt
@@ -1565,6 +1566,7 @@ func constructBroadcastHashBuild(op vm.Operator, proc *process.Process, mcpu int
 
 		ret.HashOnPK = arg.HashOnPK
 		ret.NeedAllocateSels = !arg.HashOnPK
+		ret.CanSpill = true
 		if len(arg.RuntimeFilterSpecs) > 0 {
 			ret.RuntimeFilterSpec = arg.RuntimeFilterSpecs[0]
 		}
@@ -1657,6 +1659,7 @@ func constructShuffleHashBuild(node *plan.Node, op vm.Operator, proc *process.Pr
 
 		ret.HashOnPK = arg.HashOnPK
 		ret.NeedAllocateSels = !arg.HashOnPK
+		ret.CanSpill = true
 		if len(arg.RuntimeFilterSpecs) > 0 {
 			ret.RuntimeFilterSpec = plan2.DeepCopyRuntimeFilterSpec(arg.RuntimeFilterSpecs[0])
 		}