forked from cockroachdb/cockroach
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
41307: sql: fix various problematic uses of the txn in DistSQL flows r=andreimatei a=andreimatei see individual commits Release justification: Fixes bad bugs. Co-authored-by: Andrei Matei <[email protected]>
- Loading branch information
Showing
18 changed files
with
455 additions
and
117 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| // Copyright 2019 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the file licenses/BSL.txt. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0, included in the file | ||
| // licenses/APL.txt. | ||
|
|
||
| package colexec | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/col/coldata" | ||
| "github.com/cockroachdb/cockroach/pkg/col/coltypes" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/execinfra" | ||
| ) | ||
|
|
||
| // SerialUnorderedSynchronizer is an Operator that combines multiple Operator | ||
| // streams into one. It reads its inputs one by one until each one is exhausted, | ||
| // at which point it moves to the next input. See ParallelUnorderedSynchronizer | ||
| // for a parallel implementation. The serial one is used when concurrency is | ||
| // undesirable - for example when the whole query is planned on the gateway and | ||
| // we want to run it in the RootTxn. | ||
| type SerialUnorderedSynchronizer struct { | ||
| inputs []Operator | ||
| // curSerialInputIdx indicates the index of the current input being consumed. | ||
| curSerialInputIdx int | ||
| zeroBatch coldata.Batch | ||
| } | ||
|
|
||
| var _ Operator = &SerialUnorderedSynchronizer{} | ||
| var _ execinfra.OpNode = &SerialUnorderedSynchronizer{} | ||
|
|
||
| // ChildCount implements the execinfra.OpNode interface. | ||
| func (s *SerialUnorderedSynchronizer) ChildCount() int { | ||
| return len(s.inputs) | ||
| } | ||
|
|
||
| // Child implements the execinfra.OpNode interface. | ||
| func (s *SerialUnorderedSynchronizer) Child(nth int) execinfra.OpNode { | ||
| return s.inputs[nth] | ||
| } | ||
|
|
||
| // NewSerialUnorderedSynchronizer creates a new SerialUnorderedSynchronizer. | ||
| func NewSerialUnorderedSynchronizer( | ||
| inputs []Operator, typs []coltypes.T, | ||
| ) *SerialUnorderedSynchronizer { | ||
| return &SerialUnorderedSynchronizer{ | ||
| inputs: inputs, | ||
| curSerialInputIdx: 0, | ||
| zeroBatch: coldata.NewMemBatchWithSize(typs, 0), | ||
| } | ||
| } | ||
|
|
||
| // Init is part of the Operator interface. | ||
| func (s *SerialUnorderedSynchronizer) Init() { | ||
| for _, input := range s.inputs { | ||
| input.Init() | ||
| } | ||
| } | ||
|
|
||
| // Next is part of the Operator interface. | ||
| func (s *SerialUnorderedSynchronizer) Next(ctx context.Context) coldata.Batch { | ||
| for { | ||
| if s.curSerialInputIdx == len(s.inputs) { | ||
| return s.zeroBatch | ||
| } | ||
| b := s.inputs[s.curSerialInputIdx].Next(ctx) | ||
| if b.Length() == 0 { | ||
| s.curSerialInputIdx++ | ||
| } else { | ||
| return b | ||
| } | ||
| } | ||
| } |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| // Copyright 2019 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the file licenses/BSL.txt. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0, included in the file | ||
| // licenses/APL.txt. | ||
|
|
||
| package colexec | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/col/coldata" | ||
| "github.com/cockroachdb/cockroach/pkg/col/coltypes" | ||
| "github.com/cockroachdb/cockroach/pkg/util/leaktest" | ||
| "github.com/cockroachdb/cockroach/pkg/util/randutil" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestSerialUnorderedSynchronizer(t *testing.T) { | ||
| defer leaktest.AfterTest(t)() | ||
|
|
||
| ctx := context.Background() | ||
| rng, _ := randutil.NewPseudoRand() | ||
| const numInputs = 3 | ||
| const numBatches = 4 | ||
|
|
||
| typs := []coltypes.T{coltypes.Int64} | ||
| inputs := make([]Operator, numInputs) | ||
| for i := range inputs { | ||
| batch := coldata.NewMemBatchWithSize(typs, int(coldata.BatchSize())) | ||
| batch.SetLength(coldata.BatchSize()) | ||
| source := NewRepeatableBatchSource( | ||
| RandomBatch(rng, typs, int(coldata.BatchSize()), 0 /* length */, rng.Float64())) | ||
| source.ResetBatchesToReturn(numBatches) | ||
| inputs[i] = source | ||
| } | ||
| s := NewSerialUnorderedSynchronizer(inputs, typs) | ||
| resultBatches := 0 | ||
| for { | ||
| b := s.Next(ctx) | ||
| if b.Length() == 0 { | ||
| break | ||
| } | ||
| resultBatches++ | ||
| } | ||
| require.Equal(t, numInputs*numBatches, resultBatches) | ||
| } |
Oops, something went wrong.