engine: fix race when reading fields in Concatenate

[vmg]

Description

There's a race on this recently refactored code: you cannot use an atomic for counting which fields are set because that is inherently racy between setting the field and increasing the atomic. Using a simple Mutex and a counter is not enough either, because the same Result index may receive fields more than once, leaving a nil gap in the set of results whose types we must aggregate. One possible correct approach is using a mutex and checking all the parallel results until none of them are nil. This is implemented here.

The race was making TestSQLSelectLimit flaky.

cc @systay @dbussink

Related Issue(s)

Broken in #14245

Checklist

• "Backport to:" labels have been added if this change should be back-ported
• Tests were added or are not required
• Did the new or modified tests pass consistently locally and on the CI
• Documentation was added or is not required

Deployment Notes

[vitess-bot]

Review Checklist

Hello reviewers! 👋 Please follow this checklist when reviewing this Pull Request.

General

• Ensure that the Pull Request has a descriptive title.
• Ensure there is a link to an issue (except for internal cleanup and flaky test fixes), new features should have an RFC that documents use cases and test cases.

Tests

• Bug fixes should have at least one unit or end-to-end test, enhancement and new features should have a sufficient number of tests.

Documentation

• Apply the release notes (needs details) label if users need to know about this change.
• New features should be documented.
• There should be some code comments as to why things are implemented the way they are.
• There should be a comment at the top of each new or modified test to explain what the test does.

New flags

• Is this flag really necessary?
• Flag names must be clear and intuitive, use dashes (-), and have a clear help text.

If a workflow is added or modified:

• Each item in Jobs should be named in order to mark it as required.
• If the workflow needs to be marked as required, the maintainer team must be notified.

Backward compatibility

• Protobuf changes should be wire-compatible.
• Changes to _vt tables and RPCs need to be backward compatible.
• RPC changes should be compatible with vitess-operator
• If a flag is removed, then it should also be removed from vitess-operator and arewefastyet, if used there.
• vtctl command output order should be stable and awk-able.

[dbussink]

@vmg Looks like we have a test timeout here?

panic: test timed out after 10m0s
running tests:
	TestConcatenate_WithErrors (10m0s)

goroutine 79 [running]:
testing.(*M).startAlarm.func1()
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:2259 +0x1fc
created by time.goFunc
	/opt/hostedtoolcache/go/1.21.3/x64/src/time/sleep.go:176 +0x45

goroutine 1 [chan receive, 9 minutes]:
testing.(*T).Run(0xc000240d00, {0x1b7c6fb, 0x1a}, 0x1e4ebd0)
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:1649 +0x856
testing.runTests.func1(0x0?)
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:2054 +0x85
testing.tRunner(0xc000240d00, 0xc000141b08)
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:1595 +0x239
testing.runTests(0xc0001a8320?, {0x2ef2d80, 0xc6, 0xc6}, {0x1c?, 0x4ae0b9?, 0x2f2e1e0?})
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:2052 +0x897
testing.(*M).Run(0xc0001a8320)
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:1925 +0xb58
main.main()
	_testmain.go:443 +0x2be

goroutine 35 [chan receive]:
github.com/golang/glog.(*loggingT).flushDaemon(0x0?)
	/home/runner/go/pkg/mod/github.com/golang/[email protected]/glog.go:882 +0x8b
created by github.com/golang/glog.init.0 in goroutine 1
	/home/runner/go/pkg/mod/github.com/golang/[email protected]/glog.go:410 +0x258

goroutine 139 [semacquire, 9 minutes]:
sync.runtime_Semacquire(0xc000040b90?)
	/opt/hostedtoolcache/go/1.21.3/x64/src/runtime/sema.go:62 +0x25
sync.(*WaitGroup).Wait(0xc000040b88)
	/opt/hostedtoolcache/go/1.21.3/x64/src/sync/waitgroup.go:116 +0xa5
golang.org/x/sync/errgroup.(*Group).Wait(0xc000040b80)
	/home/runner/go/pkg/mod/golang.org/x/[email protected]/errgroup/errgroup.go:53 +0x38
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).parallelStreamExec(0xc0001a07a0, {0x2144668, 0x2f60040}, {0x215ad98?, 0xc0009c874f}, 0x0, 0xc000057210)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:319 +0x835
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).TryStreamExecute(0x10?, {0x2144668, 0x2f60040}, {0x215ad98?, 0xc0009c874f?}, 0x176498e?, 0x0?, 0xc0009c8740?)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:237 +0xf3
vitess.io/vitess/go/vt/vtgate/engine.wrapStreamExecute({0x214d7b8, 0xc0001a07a0}, {0x215ad98, 0xc0009c874f}, 0x1b5d89f?, 0x6?)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/fake_primitive_test.go:140 +0x12c
vitess.io/vitess/go/vt/vtgate/engine.TestConcatenate_WithErrors(0x0?)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate_test.go:156 +0x810
testing.tRunner(0xc0009109c0, 0x1e4ebd0)
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:1595 +0x239
created by testing.(*T).Run in goroutine 1
	/opt/hostedtoolcache/go/1.21.3/x64/src/testing/testing.go:1648 +0x82b

goroutine 143 [sync.Cond.Wait, 9 minutes]:
sync.runtime_notifyListWait(0xc0000407d0, 0x2)
	/opt/hostedtoolcache/go/1.21.3/x64/src/runtime/sema.go:527 +0x159
sync.(*Cond).Wait(0xc0000407c0)
	/opt/hostedtoolcache/go/1.21.3/x64/src/sync/cond.go:70 +0x75
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).parallelStreamExec.func2.1(0xc00034d340)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:301 +0x4f0
vitess.io/vitess/go/vt/vtgate/engine.(*fakePrimitive).TryStreamExecute(0xc000204370, {0x60?, 0x7fa9ddd42030?}, {0x7fa9deadc3c8?, 0x60?}, 0xc000508000?, 0x1, 0xc000516f60)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/fake_primitive_test.go:98 +0x4eb
vitess.io/vitess/go/vt/vtgate/engine.(*noopVCursor).StreamExecutePrimitive(0xc0009c86e0?, {0x2144928, 0xc0002044b0}, {0x214d2d8, 0xc000204370}, 0x0?, 0xe0?, 0xc0005a8eb8?)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/fake_vcursor_test.go:144 +0x95
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).parallelStreamExec.func2()
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:279 +0x376
golang.org/x/sync/errgroup.(*Group).Go.func1()
	/home/runner/go/pkg/mod/golang.org/x/[email protected]/errgroup/errgroup.go:75 +0x77
created by golang.org/x/sync/errgroup.(*Group).Go in goroutine 139
	/home/runner/go/pkg/mod/golang.org/x/[email protected]/errgroup/errgroup.go:72 +0x125

goroutine 145 [sync.Cond.Wait, 9 minutes]:
sync.runtime_notifyListWait(0xc0000407d0, 0x3)
	/opt/hostedtoolcache/go/1.21.3/x64/src/runtime/sema.go:527 +0x159
sync.(*Cond).Wait(0xc0000407c0)
	/opt/hostedtoolcache/go/1.21.3/x64/src/sync/cond.go:70 +0x75
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).parallelStreamExec.func2.1(0xc00034d2d0)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:301 +0x4f0
vitess.io/vitess/go/vt/vtgate/engine.(*fakePrimitive).TryStreamExecute(0xc000204410, {0x60?, 0x7fa9ddd42030?}, {0x7fa9deadc3c8?, 0x60?}, 0xc000508000?, 0x1, 0xc000516f00)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/fake_primitive_test.go:98 +0x4eb
vitess.io/vitess/go/vt/vtgate/engine.(*noopVCursor).StreamExecutePrimitive(0x4bf946?, {0x2144928, 0xc0002044b0}, {0x214d2d8, 0xc000204410}, 0x0?, 0x89?, 0xc0005abea8?)
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/fake_vcursor_test.go:144 +0x95
vitess.io/vitess/go/vt/vtgate/engine.(*Concatenate).parallelStreamExec.func2()
	/home/runner/work/vitess/vitess/go/vt/vtgate/engine/concatenate.go:279 +0x376
golang.org/x/sync/errgroup.(*Group).Go.func1()
	/home/runner/go/pkg/mod/golang.org/x/[email protected]/errgroup/errgroup.go:75 +0x77
created by golang.org/x/sync/errgroup.(*Group).Go in goroutine 139
	/home/runner/go/pkg/mod/golang.org/x/[email protected]/errgroup/errgroup.go:72 +0x125
FAIL	vitess.io/vitess/go/vt/vtgate/engine	600.065s

Is it hanging indefinitely?