Merge pull request #6949 from planetscale/fp-6940

[Forward Port] #6940 - Fix error handling in olap mode

go/test/endtoend/vtgate/misc_test.go

@@ -350,6 +350,7 @@ func TestOffsetAndLimitWithOLAP(t *testing.T) {
 	conn, err := mysql.Connect(ctx, &vtParams)
 	require.NoError(t, err)
 	defer conn.Close()
+	defer exec(t, conn, "set workload=oltp;delete from t1")
 
 	exec(t, conn, "insert into t1(id1, id2) values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)")
 	assertMatches(t, conn, "select id1 from t1 order by id1 limit 3 offset 2", "[[INT64(3)] [INT64(4)] [INT64(5)]]")
@@ -404,6 +405,19 @@ func TestInformationSchemaWithSubquery(t *testing.T) {
 	assert.Empty(t, result.Rows)
 }
 
+func TestInsertStmtInOLAP(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	exec(t, conn, `set workload='olap'`)
+	_, err = conn.ExecuteFetch(`insert into t1(id1, id2) values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)`, 1000, true)
+	require.Error(t, err)
+	assertMatches(t, conn, `select id1 from t1 order by id1`, `[]`)
+}
+
 func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
 	t.Helper()
 	qr := exec(t, conn, query)

go/vt/vtgate/executor.go

@@ -1229,22 +1229,23 @@ func (e *Executor) StreamExecute(ctx context.Context, method string, safeSession
 		return nil
 	})
 
-	// Send left-over rows.
-	if len(result.Rows) > 0 || !seenResults {
-		if err := callback(result); err != nil {
-			return err
+	// Send left-over rows if there is no error on execution.
+	if err == nil {
+		if len(result.Rows) > 0 || !seenResults {
+			if err := callback(result); err != nil {
+				return err
+			}
 		}
+		// save session stats for future queries
+		if !safeSession.foundRowsHandled {
+			safeSession.FoundRows = foundRows
+		}
+		safeSession.RowCount = -1
 	}
 
 	logStats.ExecuteTime = time.Since(execStart)
 	e.updateQueryCounts(plan.Instructions.RouteType(), plan.Instructions.GetKeyspaceName(), plan.Instructions.GetTableName(), int64(logStats.ShardQueries))
 
-	// save session stats for future queries
-	if !safeSession.foundRowsHandled {
-		safeSession.FoundRows = foundRows
-	}
-	safeSession.RowCount = -1
-
 	return err
 }