fix(spill): stop at max level instead of erroring, fix memory leaks

pkg/sql/colexec/group/helper.go

@@ -200,13 +200,13 @@ func (ctr *container) spillDataToDisk(proc *process.Process, parentBkt *spillBuc
 
 	// if current spill bucket is not created, create a new one.
 	if ctr.currentSpillBkt == nil {
-		// check parent level, if it is too deep, return error.
+		// check parent level, if it is too deep, stop spilling and keep data in memory.
 		// we only allow to spill up to spillMaxPass passes.
 		// each pass we take spillMaskBits bits from the hashCode, and use them as the index
 		// to select the spill bucket.  Default params, 32^3 = 32768 spill buckets -- if this
-		// is still not enough, probably we cannot do much anyway, just fail the query.
+		// is still not enough, probably we cannot do much anyway, keep the remaining data in memory.
 		if parentLv >= spillMaxPass {
-			return 0, 0, moerr.NewInternalError(proc.Ctx, "spill level too deep")
+			return 0, 0, nil
 		}
 
 		var parentName string
@@ -420,6 +420,7 @@ func (ctr *container) loadSpilledData(proc *process.Process, opAnalyzer process.
 		if err := bkt.free(); err != nil && retErr == nil {
 			retErr = err
 		}
+		ctr.freeSpillAggList()
 	}()
 
 	// reposition to the start of the file.

pkg/sql/colexec/group/types2.go

@@ -238,6 +238,13 @@ func (ctr *container) free() {
 		ctr.spillGbBatch.Clean(ctr.mp)
 		ctr.spillGbBatch = nil
 	}
+	ctr.spillBuf = nil
+	ctr.spillReader = nil
+	ctr.spillHashCodes = nil
+	ctr.spillChunkFlags = nil
+	ctr.spillFlagFlat = nil
+	ctr.spillNonEmptyBuckets = nil
+	ctr.spillBucketRowIds = nil
 
 	mpool.DeleteMPool(ctr.mp)
 	ctr.mp = nil

pkg/sql/colexec/hashbuild/types.go

@@ -144,6 +144,11 @@ func (hashBuild *HashBuild) Free(proc *process.Process, pipelineFailed bool, err
 	hashBuild.cleanupSpillFiles(proc)
 	hashBuild.ctr.hashmapBuilder.Free(proc)
 	hashBuild.ctr.cleanSpillBufferPool(proc)
+	hashBuild.ctr.freeSpillExprExecs()
+	hashBuild.ctr.spillKeyVecs = nil
+	hashBuild.ctr.spillHashValues = nil
+	hashBuild.ctr.spillBucketRowIds = nil
+	hashBuild.ctr.spillNonEmptyBuckets = nil
 }
 
 func (hashBuild *HashBuild) cleanupSpillFiles(proc *process.Process) {

pkg/sql/colexec/hashjoin/types.go

@@ -222,6 +222,9 @@ func (hashJoin *HashJoin) Free(proc *process.Process, pipelineFailed bool, err e
 	ctr.cleanNonEqCondExecutor()
 	ctr.cleanEqCondExecutors()
 	ctr.cleanupSpillFiles(proc)
+	ctr.freeSpillBuildExprExecs()
+	ctr.spillBucketRowIds = nil
+	ctr.spillNonEmptyBuckets = nil
 }
 
 func (ctr *container) cleanupSpillFiles(proc *process.Process) {