support idempotence for batch's Clean method (#17838)

Supports idempotence for batch's Clean method,
which helps to avoid race conditions caused by the bug of calling it more times than required.

Approved by: @XuPeng-SH, @aunjgr

pkg/container/batch/batch.go

@@ -193,29 +193,27 @@ func (bat *Batch) GetSubBatch(cols []string) *Batch {
 }
 
 func (bat *Batch) Clean(m *mpool.MPool) {
-	if bat == EmptyBatch {
-		return
-	}
-	if atomic.LoadInt64(&bat.Cnt) == 0 {
-		// panic("batch is already cleaned")
-		return
-	}
-	if atomic.AddInt64(&bat.Cnt, -1) > 0 {
+	// situations that batch was still in use.
+	// we use `!= 0` but not `>0` to avoid the situation that the batch was cleaned more than required.
+	if bat == EmptyBatch || atomic.AddInt64(&bat.Cnt, -1) != 0 {
 		return
 	}
+
 	for _, vec := range bat.Vecs {
 		if vec != nil {
 			vec.Free(m)
+			bat.ReplaceVector(vec, nil)
 		}
 	}
 	for _, agg := range bat.Aggs {
 		if agg != nil {
 			agg.Free()
 		}
 	}
-	bat.Attrs = nil
-	bat.rowCount = 0
+	bat.Aggs = nil
 	bat.Vecs = nil
+	bat.Attrs = nil
+	bat.SetRowCount(0)
 }
 
 func (bat *Batch) Last() bool {

pkg/vm/process/process2.go

@@ -26,6 +26,7 @@ import (
 // when this batch is no longer in use, places all vectors into the pool.
 func (proc *Process) PutBatch(bat *batch.Batch) {
 	// situations that batch was still in use.
+	// we use `!= 0` but not `>0` to avoid the situation that the batch was cleaned more than required.
 	if bat == batch.EmptyBatch || atomic.AddInt64(&bat.Cnt, -1) != 0 {
 		return
 	}