From 6e96794ef69e2d9b2bc1d7c4e3c1831e8a971044 Mon Sep 17 00:00:00 2001
From: zengyan1 <93656539+zengyan1@users.noreply.github.com>
Date: Fri, 18 Oct 2024 17:53:33 +0800
Subject: [PATCH] fix apply panic (#19431)

fix apply panic

Approved by: @m-schen
---
 pkg/sql/colexec/apply/apply.go | 8 ++++----
 pkg/sql/colexec/apply/types.go | 1 +
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/pkg/sql/colexec/apply/apply.go b/pkg/sql/colexec/apply/apply.go
index d2f0e94f7834..00f43d50968e 100644
--- a/pkg/sql/colexec/apply/apply.go
+++ b/pkg/sql/colexec/apply/apply.go
@@ -48,10 +48,7 @@ func (apply *Apply) Prepare(proc *process.Process) (err error) {
 	}
 
 	if apply.ctr.sels == nil {
-		apply.ctr.sels = make([]int32, colexec.DefaultBatchSize)
-		for i := range apply.ctr.sels {
-			apply.ctr.sels[i] = int32(i)
-		}
+		apply.ctr.sels = make([]int32, 0)
 	}
 
 	err = apply.TableFunction.ApplyPrepare(proc)
@@ -171,6 +168,9 @@ func (ctr *container) probe(ap *Apply, proc *process.Process, result *vm.CallRes
 				if rp.Rel == 0 {
 					err = ctr.rbat.Vecs[j].UnionMulti(ctr.inbat.Vecs[rp.Pos], int64(i), rowCountIncrease, proc.Mp())
 				} else {
+					for len(ctr.sels) < rowCountIncrease {
+						ctr.sels = append(ctr.sels, int32(len(ctr.sels)))
+					}
 					err = ctr.rbat.Vecs[j].UnionInt32(tfResult.Batch.Vecs[rp.Pos], ctr.sels[:rowCountIncrease], proc.Mp())
 				}
 				if err != nil {
diff --git a/pkg/sql/colexec/apply/types.go b/pkg/sql/colexec/apply/types.go
index 9813efdfe617..e8b1c00d5e62 100644
--- a/pkg/sql/colexec/apply/types.go
+++ b/pkg/sql/colexec/apply/types.go
@@ -110,6 +110,7 @@ func (apply *Apply) Free(proc *process.Process, pipelineFailed bool, err error)
 	ctr := &apply.ctr
 
 	ctr.cleanBatch(proc.Mp())
+	ctr.sels = nil
 
 	apply.TableFunction.Free(proc, pipelineFailed, err)