executor: fix pipelined window invalid memory address (#30418) (#30459)

close #30326

executor/pipelined_window.go

@@ -178,7 +178,7 @@ func (e *PipelinedWindowExec) getRowsInPartition(ctx context.Context) (err error
 		var drained, samePartition bool
 		drained, err = e.fetchChild(ctx)
 		if err != nil {
-			err = errors.Trace(err)
+			return errors.Trace(err)
 		}
 		// we return immediately to use a combination of true newPartition but 0 in e.rowToConsume to indicate the data source is drained,
 		if drained {

expression/integration_test.go

@@ -10199,6 +10199,18 @@ func (s *testIntegrationSuite) TestIssue28643(c *C) {
 	tk.MustQuery("select hour(a) from t;").Check(testkit.Rows("838", "838"))
 }
 
+func (s *testIntegrationSuite) TestIssue30326(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t(a int);")
+	tk.MustExec("insert into t values(1),(1),(2),(2);")
+	tk.MustExec("set tidb_window_concurrency = 1;")
+	err := tk.QueryToErr("select (FIRST_VALUE(1) over (partition by v.a)) as c3 from (select a from t where t.a = (select a from t t2 where t.a = t2.a)) as v;")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[executor:1242]Subquery returns more than 1 row")
+}
+
 func (s *testIntegrationSuite) TestIssue29244(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")