[release-24.0] vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402) - #20749
Draft
vitess-bot[bot] wants to merge 1 commit into
Draft
[release-24.0] vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)#20749vitess-bot[bot] wants to merge 1 commit into
vitess-bot[bot] wants to merge 1 commit into
Conversation
vitess-bot
Bot
added
Type: Bug
Component: Query Serving
Backport
Contributor
Author
|
Hello @arthurschreiber, there are conflicts in this backport. Please address them in order to merge this Pull Request. You can execute the snippet below to reset your branch and resolve the conflict manually. Make sure you replace |
vitess-bot
Bot
requested review from
arthurschreiber,
frouioui,
harshit-gangal and
systay
July 29, 2026 16:28
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot could not run the full agentic suite for this review because it was automatically requested on a bot-authored pull request. Request a review from Copilot under Reviewers to retry with the full agentic suite. Improved support for bot-authored pull requests is coming soon.
This PR improves Vitess’ streaming execution behavior so that OK-packet metadata (RowsAffected/InsertID/SessionStateChanges/etc.) and execution statistics are propagated/recorded consistently with the buffered execution path.
Changes:
- Propagate OK-packet data through vtgate’s
StreamExecuteand dbconnpool’sExecuteStreamFetch. - Record per-plan query execution stats for streaming queries on success and several error paths.
- Add unit and end-to-end tests covering streaming OK-packet propagation and streaming stats behavior.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| go/vt/vtgate/executor_stream_test.go | Adds streaming tests for OK-packet propagation and per-plan stats recording. |
| go/vt/vtgate/executor_stats_test.go | Adds streaming stats tests (but currently contains unresolved merge conflicts). |
| go/vt/vtgate/executor.go | Updates StreamExecute to accumulate OK-packet data and to record stats on streaming errors/success (but currently contains unresolved merge conflicts). |
| go/vt/dbconnpool/connection.go | Ensures streaming callbacks receive OK-packet metadata when no resultset is produced. |
| go/vt/dbconnpool/connection_test.go | Adds a unit test ensuring ExecuteStreamFetch forwards OK-packet fields. |
| go/test/endtoend/vtgate/unsharded/streaming_okpacket_test.go | Adds an end-to-end test validating affected rows for streamed CALL on OLAP path. |
| go/mysql/streaming_query_test.go | Extends streaming OK-packet test to include session-state changes via session tracking. |
Comment on lines
+82
to
+118
| <<<<<<< HEAD | ||
| ||||||| parent of 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) | ||
| func TestSlowQueriesCounter(t *testing.T) { | ||
| executor, sbc1, _, _, ctx := createExecutorEnv(t) | ||
|
|
||
| oldThreshold := slowQueryThreshold | ||
| slowQueryThreshold = time.Hour | ||
| t.Cleanup(func() { | ||
| slowQueryThreshold = oldThreshold | ||
| sbc1.ExecDelayResponse = 0 | ||
| }) | ||
|
|
||
| sbc1.SetResults([]*sqltypes.Result{ | ||
| sqltypes.MakeTestResult(sqltypes.MakeTestFields("id", "int64"), "1"), | ||
| sqltypes.MakeTestResult(sqltypes.MakeTestFields("id", "int64"), "1"), | ||
| sqltypes.MakeTestResult(sqltypes.MakeTestFields("id", "int64"), "1"), | ||
| }) | ||
|
|
||
| session := econtext.NewSafeSession(&vtgatepb.Session{TargetString: KsTestSharded}) | ||
| initialCount := getTotalSlowQueryCount() | ||
|
|
||
| _, err := executorExecSession(ctx, executor, session, "select id from user where id = 1", nil) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, initialCount, getTotalSlowQueryCount(), "fast query should not increment slow query count") | ||
|
|
||
| sbc1.ExecDelayResponse = 20 * time.Millisecond | ||
| slowQueryThreshold = 5 * time.Millisecond | ||
| _, err = executorExecSession(ctx, executor, session, "select id from user where id = 1", nil) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, initialCount+1, getTotalSlowQueryCount(), "slow query should increment slow query count") | ||
|
|
||
| sbc1.ExecDelayResponse = 20 * time.Millisecond | ||
| slowQueryThreshold = 0 | ||
| _, err = executorExecSession(ctx, executor, session, "select id from user where id = 1", nil) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, initialCount+1, getTotalSlowQueryCount(), "disabled slow query threshold should not increment slow query count") | ||
| } |
Comment on lines
+251
to
+286
| <<<<<<< HEAD | ||
| ||||||| parent of 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) | ||
|
|
||
| func getTotalSlowQueryCount() int64 { | ||
| var total int64 | ||
| for _, count := range slowQueries.Counts() { | ||
| total += count | ||
| } | ||
| return total | ||
| } | ||
| ======= | ||
|
|
||
| func getTotalQueryExecutionsCount() int64 { | ||
| var total int64 | ||
| for _, count := range queryExecutions.Counts() { | ||
| total += count | ||
| } | ||
| return total | ||
| } | ||
|
|
||
| func getTotalQueryExecutionsByTableCount() int64 { | ||
| var total int64 | ||
| for _, count := range queryExecutionsByTable.Counts() { | ||
| total += count | ||
| } | ||
| return total | ||
| } | ||
|
|
||
| func getTotalSlowQueryCount() int64 { | ||
| var total int64 | ||
| for _, count := range slowQueries.Counts() { | ||
| total += count | ||
| } | ||
| return total | ||
| } | ||
| >>>>>>> 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) |
Comment on lines
+394
to
+447
| <<<<<<< HEAD | ||
| ||||||| parent of 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) | ||
|
|
||
| updateLogStats := func() { | ||
| logStats.StmtType = plan.QueryType.String() | ||
| logStats.PlanType = plan.Type.String() | ||
| logStats.TablesUsed = plan.TablesUsed | ||
| executedRoot := vc.ExecutedPrimitive() | ||
| if executedRoot == nil { | ||
| executedRoot = plan.Instructions | ||
| } | ||
| logStats.RoutingIndexesUsed = engine.GetRoutingIndexes(executedRoot) | ||
| logStats.TabletType = vc.TabletType().String() | ||
| logStats.ExecuteTime = time.Since(execStart) | ||
| logStats.ActiveKeyspace = vc.GetKeyspace() | ||
|
|
||
| e.updateQueryStats(plan.QueryType.String(), plan.Type.String(), vc.TabletType().String(), int64(logStats.ShardQueries), plan.TablesUsed) | ||
| } | ||
|
|
||
| ======= | ||
|
|
||
| updateLogStats := func(err error) { | ||
| logStats.StmtType = plan.QueryType.String() | ||
| logStats.PlanType = plan.Type.String() | ||
| logStats.TabletType = vc.TabletType().String() | ||
| logStats.ExecuteTime = time.Since(execStart) | ||
| logStats.ActiveKeyspace = vc.GetKeyspace() | ||
|
|
||
| // On error, leave the tables, routing indexes and row counts unset so the | ||
| // per-table counters are not incremented, matching the buffered Execute path. | ||
| var tablesUsed []string | ||
| var errCount uint64 | ||
| if err != nil { | ||
| logStats.Error = err | ||
| errCount = 1 | ||
| } else { | ||
| srr.mu.Lock() | ||
| logStats.RowsAffected = srr.rowsAffected | ||
| logStats.RowsReturned = uint64(srr.rowsReturned) | ||
| srr.mu.Unlock() | ||
| logStats.TablesUsed = plan.TablesUsed | ||
| tablesUsed = plan.TablesUsed | ||
| executedRoot := vc.ExecutedPrimitive() | ||
| if executedRoot == nil { | ||
| executedRoot = plan.Instructions | ||
| } | ||
| logStats.RoutingIndexesUsed = engine.GetRoutingIndexes(executedRoot) | ||
| } | ||
|
|
||
| e.updateQueryStats(plan.QueryType.String(), plan.Type.String(), vc.TabletType().String(), int64(logStats.ShardQueries), tablesUsed) | ||
| plan.AddStats(1, time.Since(logStats.StartTime), logStats.ShardQueries, logStats.RowsAffected, logStats.RowsReturned, errCount) | ||
| } | ||
|
|
||
| >>>>>>> 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) |
Comment on lines
+492
to
+503
| <<<<<<< HEAD | ||
| logStats.TablesUsed = plan.TablesUsed | ||
| logStats.TabletType = vc.TabletType().String() | ||
| logStats.ExecuteTime = time.Since(execStart) | ||
| logStats.ActiveKeyspace = vc.GetKeyspace() | ||
|
|
||
| e.updateQueryStats(plan.QueryType.String(), plan.Type.String(), vc.TabletType().String(), int64(logStats.ShardQueries), plan.TablesUsed) | ||
| ||||||| parent of 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) | ||
| updateLogStats() | ||
| ======= | ||
| updateLogStats(nil) | ||
| >>>>>>> 8055e6946d (vtgate: report RowsAffected for stored-procedure calls over the streaming path (#20402)) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This is a backport of #20402