Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

engine: fix race in concatenate #15454

Merged
merged 2 commits into from
Mar 12, 2024
Merged

engine: fix race in concatenate #15454

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c9dafd6
engine: fix race in concatenate
vmg Mar 12, 2024
7cab524
endtoend: run test in a loop
vmg Mar 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions go/test/endtoend/vtgate/queries/union/union_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import (
"testing"

"vitess.io/vitess/go/test/endtoend/cluster"

"vitess.io/vitess/go/test/endtoend/utils"

"github.com/stretchr/testify/assert"
Expand Down Expand Up @@ -111,9 +110,15 @@ func TestUnionAll(t *testing.T) {
mcmp.AssertMatchesNoOrder("select tbl2.id1 FROM ((select id1 from t1 order by id1 limit 5) union all (select id1 from t1 order by id1 desc limit 5)) as tbl1 INNER JOIN t1 as tbl2 ON tbl1.id1 = tbl2.id1",
"[[INT64(1)] [INT64(2)] [INT64(2)] [INT64(1)]]")

// union all between two select unique in tables
mcmp.AssertMatchesNoOrder("select id1 from t1 where id1 in (1, 2, 3, 4, 5, 6, 7, 8) union all select id1 from t1 where id1 in (1, 2, 3, 4, 5, 6, 7, 8)",
"[[INT64(1)] [INT64(2)] [INT64(1)] [INT64(2)]]")
// this test is quite good at uncovering races in the Concatenate engine primitive. make it run many times
// see: https://github.com/vitessio/vitess/issues/15434
if utils.BinaryIsAtLeastAtVersion(20, "vtgate") {
for i := 0; i < 100; i++ {
// union all between two select unique in tables
mcmp.AssertMatchesNoOrder("select id1 from t1 where id1 in (1, 2, 3, 4, 5, 6, 7, 8) union all select id1 from t1 where id1 in (1, 2, 3, 4, 5, 6, 7, 8)",
"[[INT64(1)] [INT64(2)] [INT64(1)] [INT64(2)]]")
}
}

// 4 tables union all
mcmp.AssertMatchesNoOrder("select id1, id2 from t1 where id1 = 1 union all select id3,id4 from t2 where id3 = 3 union all select id1, id2 from t1 where id1 = 2 union all select id3,id4 from t2 where id3 = 4",
Expand Down
45 changes: 22 additions & 23 deletions go/vt/vtgate/engine/concatenate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,36 +285,35 @@ func (c *Concatenate) parallelStreamExec(inCtx context.Context, vcursor VCursor,
currIndex, currSource := i, source
wg.Go(func() error {
err := vcursor.StreamExecutePrimitive(ctx, currSource, bindVars, true, func(resultChunk *sqltypes.Result) error {
// Process fields when they arrive; coordinate field agreement across sources.
if resultChunk.Fields != nil {
muFields.Lock()
muFields.Lock()

// Process fields when they arrive; coordinate field agreement across sources.
if resultChunk.Fields != nil && rest[currIndex] == nil {
// Capture the initial result chunk to determine field types later.
if rest[currIndex] == nil {
rest[currIndex] = resultChunk

// If this was the last source to report its fields, derive the final output fields.
if !slices.Contains(rest, nil) {
rest[currIndex] = resultChunk

// If this was the last source to report its fields, derive the final output fields.
if !slices.Contains(rest, nil) {
// We have received fields from all sources. We can now calculate the output types
var err error
resultChunk.Fields, fieldTypes, err = c.getFieldTypes(vcursor, rest)
if err != nil {
muFields.Unlock()

// We have received fields from all sources. We can now calculate the output types
var err error
resultChunk.Fields, fieldTypes, err = c.getFieldTypes(vcursor, rest)
if err != nil {
return err
}

defer condFields.Broadcast()
return callback(resultChunk, currIndex)
return err
}

muFields.Unlock()
defer condFields.Broadcast()
return callback(resultChunk, currIndex)
}
// Wait for fields from all sources.
for slices.Contains(rest, nil) {
condFields.Wait()
}
muFields.Unlock()
}

// Wait for fields from all sources.
for slices.Contains(rest, nil) {
condFields.Wait()
}
muFields.Unlock()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why didn't y'all use the lock ... defer unlock pattern? That way you call unlock once at line 289 instead of 3 times in various places. And it's future-proof.
Nice work, btw 👍

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That wouldn't be correct. The lock needs to be released before we call callback or we'd deadlock, so using a defer doesn't work in 2 of the cases.


// Context check to avoid extra work.
if ctx.Err() != nil {
return nil
Expand Down
Loading