exec: reset internal unsafeBatch in orderedAggregator

[asubiotto]

Not doing so could lead to correctness results. This was not caught by
tests, so runTests has been extended to check for operators that can be
initialized with a variable output size. This increases
verifySelAndNullResets's test coverage, since it doesn't do anything if
only a single batch is output, which is the case for most unit tests.

Release note: None

Release justification: This commit fixes correctness bugs and increases
test coverage (Category 2).

Fixes #40641

[cockroach-teamcity]

This change is 

[asubiotto]

Modified the variable size interface so that the merge joiner can also be tested and the tests are failing. Will investigate.

[asubiotto]

Not too sure what's up. Pushed up the changes to test the mergejoiner. @yuzefovich would you mind taking a quick look to see if you see anything obvious? make test PKG=./pkg/sql/exec TESTS=TestMergeJoiner/verifySelAndNullResets should fail on this branch.

[yuzefovich]

Nice find! This was a bug with the merge joiner. Here is the diff that fixes it:

--- a/pkg/sql/exec/mergejoiner_tmpl.go
+++ b/pkg/sql/exec/mergejoiner_tmpl.go
@@ -1238,13 +1238,13 @@ func (o *mergeJoin_JOIN_TYPE_STRING_FILTER_INFO_STRINGOp) calculateOutputCount(
 }
 
 func (o *mergeJoin_JOIN_TYPE_STRING_FILTER_INFO_STRINGOp) Next(ctx context.Context) coldata.Batch {
+       if o.needToResetOutput {
+               o.needToResetOutput = false
+               o.output.ResetInternalBatch()
+       }
        for {
                switch o.state {
                case mjEntry:
-                       if o.needToResetOutput {
-                               o.needToResetOutput = false
-                               o.output.ResetInternalBatch()
-                       }
                        o.initProberState(ctx)
 
                        if o.nonEmptyBufferedGroup() {

The incorrect assumption was that we would always get into mjEntry state before returning an output batch. However, this is not the case when the output cross product spans several batches - in such a case, on the second, third, etc output batches we go straight to mjBuild state, so we would never actually reset the internal output batch.

[yuzefovich]

Thanks!

Reviewed 3 of 3 files at r1.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @asubiotto)

---

pkg/sql/exec/utils_test.go, line 81 at r1 (raw file):

// initialized with variable output size batches. This allows runTests to
// increase test coverage of these operators.
type variableBatchSizeInitializer interface {

[nit]: also would call this variableOutputBatchSizeInitializer.

I added a similar mjTestInitializer interface in mergejoiner_test.go, I think we should remove that in favor of the one you're introducing.

---

pkg/sql/exec/utils_test.go, line 82 at r1 (raw file):

// increase test coverage of these operators.
type variableBatchSizeInitializer interface {
	initWithBatchSize(outputSize uint16)

[nit]: I'd call this initWithOutputBatchSize and rename the argument to just size.

[yuzefovich]

Here is a better fix:

--- a/pkg/sql/exec/mergejoiner.go
+++ b/pkg/sql/exec/mergejoiner.go
@@ -351,9 +351,8 @@ type mergeJoinBase struct {
        right mergeJoinInput
 
        // Output buffer definition.
-       output            coldata.Batch
-       needToResetOutput bool
-       outputBatchSize   uint16
+       output          coldata.Batch
+       outputBatchSize uint16
        // outputReady is a flag to indicate that merge joiner is ready to emit an
        // output batch.
        outputReady bool
diff --git a/pkg/sql/exec/mergejoiner_tmpl.go b/pkg/sql/exec/mergejoiner_tmpl.go
index b40213cb7d..5fc228e389 100644
--- a/pkg/sql/exec/mergejoiner_tmpl.go
+++ b/pkg/sql/exec/mergejoiner_tmpl.go
@@ -1238,13 +1238,10 @@ func (o *mergeJoin_JOIN_TYPE_STRING_FILTER_INFO_STRINGOp) calculateOutputCount(
 }
 
 func (o *mergeJoin_JOIN_TYPE_STRING_FILTER_INFO_STRINGOp) Next(ctx context.Context) coldata.Batch {
+       o.output.ResetInternalBatch()
        for {
                switch o.state {
                case mjEntry:
-                       if o.needToResetOutput {
-                               o.needToResetOutput = false
-                               o.output.ResetInternalBatch()
-                       }
                        o.initProberState(ctx)
 
                        if o.nonEmptyBufferedGroup() {
@@ -1281,7 +1278,6 @@ func (o *mergeJoin_JOIN_TYPE_STRING_FILTER_INFO_STRINGOp) Next(ctx context.Conte
                                o.output.SetLength(o.builderState.outCount)
                                // Reset builder out count.
                                o.builderState.outCount = uint16(0)
-                               o.needToResetOutput = true
                                o.outputReady = false
                                return o.output
                        }

[asubiotto]

Applied the mergejoiner patch

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @yuzefovich)

---

pkg/sql/exec/utils_test.go, line 81 at r1 (raw file):

Previously, yuzefovich wrote…

[nit]: also would call this variableOutputBatchSizeInitializer.

I added a similar mjTestInitializer interface in mergejoiner_test.go, I think we should remove that in favor of the one you're introducing.

Done.

---

pkg/sql/exec/utils_test.go, line 82 at r1 (raw file):

Previously, yuzefovich wrote…

[nit]: I'd call this initWithOutputBatchSize and rename the argument to just size.

Done.

[asubiotto]

bors r=yuzefovich

[asubiotto]

bors r-

[asubiotto]

bors r=yuzefovich

[craig]

Build failed

• GitHub CI (Cockroach)

[asubiotto]

Release justification missing from PR description

bors r=yuzefovich

[craig]

Build succeeded

• GitHub CI (Cockroach)