executor: reset groupCount when Close is called in streamagg executor (#53874)

pkg/executor/internal/vecgroupchecker/BUILD.bazel

@@ -22,7 +22,7 @@ go_test(
     ],
     embed = [":vecgroupchecker"],
     flaky = True,
-    shard_count = 3,
+    shard_count = 4,
     deps = [
         "//pkg/config",
         "//pkg/expression",

pkg/executor/internal/vecgroupchecker/vec_group_checker.go

@@ -496,6 +496,7 @@ func (e *VecGroupChecker) IsExhausted() bool {
 func (e *VecGroupChecker) Reset() {
 	if e.groupOffset != nil {
 		e.groupOffset = e.groupOffset[:0]
+		e.groupCount = 0
 	}
 	if e.sameGroup != nil {
 		e.sameGroup = e.sameGroup[:0]

pkg/executor/internal/vecgroupchecker/vec_group_checker_test.go

@@ -268,3 +268,13 @@ func TestVecGroupChecker(t *testing.T) {
 	require.Equal(t, e, 3)
 	require.True(t, groupChecker.IsExhausted())
 }
+
+func TestIssue53867(t *testing.T) {
+	checker := NewVecGroupChecker(nil, true, nil)
+	checker.groupOffset = make([]int, 20)
+	checker.nextGroupID = 10
+	checker.groupCount = 15
+	require.False(t, checker.IsExhausted())
+	checker.Reset()
+	require.True(t, checker.IsExhausted())
+}

pkg/executor/test/issuetest/BUILD.bazel

@@ -8,7 +8,7 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 20,
+    shard_count = 21,
     deps = [
         "//pkg/autoid_service",
         "//pkg/config",

pkg/executor/test/issuetest/executor_issue_test.go

@@ -676,3 +676,20 @@ func TestIssue52978(t *testing.T) {
 	tk.MustQuery("select min(truncate(cast(-26340 as double), ref_11.a)) as c3 from t as ref_11;").Check(testkit.Rows("-26340"))
 	tk.MustExec("drop table if exists t")
 }
+
+func TestIssue53867(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t_xf1at0 (c_weg int ,c_u text ,c_bu double ,c__icnfdo_ tinyint ,c_micv4b_95 text ,c_ngdu3 int not null ,c_curc3h double ,c_menn1dk double ,c_pv int unique ,c_u3zry tinyint ,primary key(c_ngdu3, c_pv) NONCLUSTERED) shard_row_id_bits=4 pre_split_regions=2;")
+	tk.MustExec("create index t_kp1_idx_1 on t_xf1at0 (c_weg, c_ngdu3, c_pv);")
+	tk.MustExec("insert into t_xf1at0 (c_weg, c_ngdu3, c_pv) values (0, 0, 0), (1, 1, 1), (2, 2, 2), (3, 3, 3), (4, 4, 4), (5, 5, 5), (6, 6, 6), (7, 7, 7), (8, 8, 8), (9, 9, 9), (10, 10, 10);")
+	tk.MustExec("create table t_bhze93f (c_x393ej_ int ,c_k3kss19 double not null ,c_q6qt9_ int not null ,c_wp7o_0sstj text ,c_gus881_9 double unique ,c_wzmb0 text ,primary key(c_q6qt9_) NONCLUSTERED) shard_row_id_bits=4 pre_split_regions=2;")
+	tk.MustExec("insert into t_bhze93f (c_x393ej_, c_k3kss19, c_q6qt9_, c_wp7o_0sstj, c_gus881_9, c_wzmb0) values(-1458322912, 5.32, -811171723, cast(null as char), 96.15, 'r0exs4umz5'),(cast(null as signed), 57.87, -1624364959, 't968', 35.92, 'f'),(1237193156, 81.49, -744800718, 'pfcascv16e', cast(null as double), 'pv34_'),(1955355436, 16.26, -1560468978, cast(null as char), 65537.1, 'bps');")
+	tk.MustExec("create table t_b0t (c_g int ,c_ci4cf5ns tinyint ,c_at double ,c_xb int ,c_uyu4fop36b double ,c_zhouc text ,c_m9g6b tinyint not null unique ,c_k7rlw47ob tinyint unique ,primary key(c_xb) CLUSTERED) pre_split_regions=2;")
+
+	// Need no panic
+	tk.MustQuery("select  /*+ STREAM_AGG() */ (ref_4.c_k3kss19 / ref_4.c_k3kss19) as c2 from t_bhze93f as ref_4 where (EXISTS (select ref_5.c_wp7o_0sstj as c0 from t_bhze93f as ref_5 where (207007502 < (select distinct ref_6.c_weg as c0 from t_xf1at0 as ref_6 union all (select ref_7.c_xb as c0 from t_b0t as ref_7 where (-16090 != ref_4.c_x393ej_)) limit 1)) limit 1));")
+}