sql: fix various problematic uses of the txn in DistSQL flows

[andreimatei]

see individual commits

Release justification: Fixes bad bugs.

[cockroach-teamcity]

This change is 

[andreimatei]

I haven't made the vectorized UnorderedSync serial yet. Working on that now.

[andreimatei]

handled the vectorized synchronizer

[RaduBerinde]

This looks good to me, I'm comfortable with this going into 19.2.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei, @jordanlewis, and @RaduBerinde)

---

pkg/sql/distsql_running.go, line 156 at r3 (raw file):

			for nodeID, spec := range flows {
				fuseOpt := flowinfra.Normal
				if nodeID == thisNodeID && localState.IsLocal {

[nit] Doesn't the latter imply the former? Perhaps we should set this outside of the loop and assert that there is a single flow. There are a few similar instances below. Maybe we should move this into a helper function that gets the list of flows and the localstate.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 23 at r3 (raw file):

// 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.

[nit] Add some context as to why it exists, maybe point at the relevant issue #s.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 30 at r3 (raw file):

	curSerialInputIdx int
	zeroBatch         coldata.Batch
}

[nit] add type assertions so it's clear what interface(s) it's supposed to implement

---

pkg/sql/colflow/vectorized_flow.go, line 407 at r3 (raw file):

			memUsed += op.(colexec.StaticMemoryOperator).EstimateStaticMemoryUsage()
		} else {
			if opt == flowinfra.FuseAggressively {

Add a comment here, maybe point to the issue #s.

---

pkg/sql/distsql/server.go, line 361 at r4 (raw file):

		return nil, nil, err
	}
	if meta := req.TxnCoordMeta; meta != nil {

This block is used 3 times, should be a helper

---

pkg/sql/rowflow/input_sync.go, line 368 at r3 (raw file):

// makeOrderedSync creates an orderedSynchronizer. ordering dictates how rows
// are to be compared. Use sqlbase.NoOrdering to indicate that the row ordering
// doesn't matter and sources should be consumed in index order.

Add more context as to why you would want that, maybe refer to FuseAggressively.

---

pkg/sql/rowflow/row_based_flow.go, line 129 at r3 (raw file):

							return true
						}
					} else {

[nit] Would be more readable if we put this first under a if len(in.Streams) == 1 check (we won't need the else).

---

pkg/sql/rowflow/row_based_flow.go, line 154 at r3 (raw file):

	spec *execinfrapb.FlowSpec, streamID execinfrapb.StreamID,
) *execinfrapb.ProcessorSpec {
	for _, pspec := range spec.Processors {

[nit] don't copy each spec, do pspec := &spec.Processors[i]

---

pkg/sql/rowflow/row_based_flow.go, line 156 at r3 (raw file):

	for _, pspec := range spec.Processors {
		if len(pspec.Output) > 1 {
			return nil

Shouldn't this be continue ? (same below)

---

pkg/sql/rowflow/row_based_flow.go, line 276 at r3 (raw file):

				// dropped).
				streams := make([]execinfra.RowSource, len(is.Streams))
				numLocal := 0

what's numLocal for?

---

pkg/sql/rowflow/row_based_flow.go, line 289 at r3 (raw file):

				}
				var err error
				var ordering sqlbase.ColumnOrdering

[nit] you can do ordering := sqlbase.NoOrdering and flip theif

[jordanlewis]

Reviewed 3 of 3 files at r1, 4 of 4 files at r2, 13 of 13 files at r3, 6 of 6 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei)

---

pkg/sql/distsql_running.go, line 353 at r4 (raw file):

	// For such flows, we were supposed to have fused everything.
	if txn != nil && planCtx.isLocal && flow.ConcurrentExecution() {
		recv.SetError(errors.Errorf(

Can you make this an "internal error"? use AssertionErrorf.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 62 at r3 (raw file):

// Next is part of the Operator interface.
func (s *SerialUnorderedSynchronizer) Next(ctx context.Context) coldata.Batch {
	if ctx.Err() != nil {

This is unnecessary (I think and hope) - it's handled by the CancelChecker which is always instantiated below the scan operators.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 71 at r3 (raw file):

		b := s.inputs[s.curSerialInputIdx].Next(ctx)
		if b.Length() == 0 {
			s.curSerialInputIdx++

I won't make a stink about it but you could consider removing inputs when they're exhausted right? But I won't make a stink about it.

---

pkg/sql/rowflow/input_sync.go, line 114 at r3 (raw file):

func (s *orderedSynchronizer) Less(i, j int) bool {
	// If we're not enforcing any ordering between rows, let's consume sources in
	// their index order.

nit: extra conditional on every row is kinda sadface - can we instead swap out the Less implementation at constructor time? or just make another copy of the synchronizer like you did for the col one?

[yuzefovich]

Looks good to me as well. You will also add the tests for the bugs that are being fixed, right?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei and @jordanlewis)

---

pkg/sql/distsql_running.go, line 352 at r4 (raw file):

	// since they might have mutations), and the RootTxn does not permit concurrency.
	// For such flows, we were supposed to have fused everything.
	if txn != nil && planCtx.isLocal && flow.ConcurrentExecution() {

nit: it is unusual to see a single opening parenthesis at the end of the line (several lines above).

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 62 at r3 (raw file):

Previously, jordanlewis (Jordan Lewis) wrote…

This is unnecessary (I think and hope) - it's handled by the CancelChecker which is always instantiated below the scan operators.

Yeah, I also think this is unnecessary.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 28 at r4 (raw file):

	// curSerialInputIdx is used if parallel is false. It indicates the index of
	// the current input being consumed.
	curSerialInputIdx int

nit: the comment above seems to be out-of-date.

---

pkg/sql/colflow/vectorized_flow.go, line 85 at r4 (raw file):

// ConcurrentExecution is part of the Flow interface.
func (f *vectorizedFlow) ConcurrentExecution() bool {
	return f.operatorConcurrency || f.FlowBase.ConcurrentExecution()

The second part of this conditional is always (at least right now) false for the vectorized flows, but I guess it doesn't matter.

---

pkg/sql/distsql/server.go, line 393 at r4 (raw file):

	// This will directly by the flow if the flow has no concurrency. If there is
	// concurrency, a LeafTxn will be created.
	Txn *client.Txn

nit: a few words are missing, maybe "This will be used directly ..." (above)

---

pkg/sql/physicalplan/physical_plan.go, line 1132 at r4 (raw file):

	for i := 0; i < len(p.ResultRouters); i++ {
		pIdx := p.ResultRouters[i]
		node := p.Processors[p.ResultRouters[i]].Node

---

pkg/sql/rowflow/row_based_flow.go, line 111 at r4 (raw file):

						// If it's an orderedSynchronizer, the we look inside it to see if
						// the processor we're trying to fuse feeds into it.
						orderedSync, ok := inputSyncs[pIdx][inIdx].(*orderedSynchronizer)

nit: s/the we/then we/ (above).

[andreimatei]

I've isolated a commit (r6) for clarifying the various isLocal fields and for unifying the handling of queries that were forced to be planned locally with ones that happened to not have any remote flows (for whatever reason).
I've reworded the commit message on r9 to remove any references to remote flows. That was FUD; I had failed to read the code properly.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei, @jordanlewis, @RaduBerinde, and @yuzefovich)

---

pkg/sql/distsql_running.go, line 156 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] Doesn't the latter imply the former? Perhaps we should set this outside of the loop and assert that there is a single flow. There are a few similar instances below. Maybe we should move this into a helper function that gets the list of flows and the localstate.

done
I didn't add a helper. The other guys besides this one are randos that don't really matter, and also not completely analogous to this case.

---

pkg/sql/distsql_running.go, line 352 at r4 (raw file):

Previously, yuzefovich wrote…

[nit]: it is unusual to see a single opening parenthesis at the end of the line (several lines above).

Done.

---

pkg/sql/distsql_running.go, line 353 at r4 (raw file):

Previously, jordanlewis (Jordan Lewis) wrote…

Can you make this an "internal error"? use AssertionErrorf.

Done.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 23 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] Add some context as to why it exists, maybe point at the relevant issue #s.

added some words

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 30 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] add type assertions so it's clear what interface(s) it's supposed to implement

done

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 62 at r3 (raw file):

Previously, yuzefovich wrote…

Yeah, I also think this is unnecessary.

Done.

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 71 at r3 (raw file):

Previously, jordanlewis (Jordan Lewis) wrote…

I won't make a stink about it but you could consider removing inputs when they're exhausted right? But I won't make a stink about it.

meh, I like it more this way (with immutable inputs)

---

pkg/sql/colexec/serial_unordered_synchronizer.go, line 28 at r4 (raw file):

Previously, yuzefovich wrote…

[nit]: the comment above seems to be out-of-date.

Done.

---

pkg/sql/colflow/vectorized_flow.go, line 407 at r3 (raw file):

Previously, RaduBerinde wrote…

Add a comment here, maybe point to the issue #s.

done

---

pkg/sql/colflow/vectorized_flow.go, line 85 at r4 (raw file):

Previously, yuzefovich wrote…

The second part of this conditional is always (at least right now) false for the vectorized flows, but I guess it doesn't matter.

yeah, I went back and forth, but I'd leave it like this

---

pkg/sql/distsql/server.go, line 361 at r4 (raw file):

Previously, RaduBerinde wrote…

This block is used 3 times, should be a helper

I've moved this stuff back to Server.setupFlow().

---

pkg/sql/distsql/server.go, line 393 at r4 (raw file):

Previously, yuzefovich wrote…

[nit]: a few words are missing, maybe "This will be used directly ..." (above)

done

---

pkg/sql/rowflow/input_sync.go, line 368 at r3 (raw file):

Previously, RaduBerinde wrote…

Add more context as to why you would want that, maybe refer to FuseAggressively.

done

---

pkg/sql/rowflow/row_based_flow.go, line 129 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] Would be more readable if we put this first under a if len(in.Streams) == 1 check (we won't need the else).

done

---

pkg/sql/rowflow/row_based_flow.go, line 154 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] don't copy each spec, do pspec := &spec.Processors[i]

done

---

pkg/sql/rowflow/row_based_flow.go, line 156 at r3 (raw file):

Previously, RaduBerinde wrote…

Shouldn't this be continue ? (same below)

well this should never happen (cause we don't have procs with multiple outputs, right?). But changed.

---

pkg/sql/rowflow/row_based_flow.go, line 276 at r3 (raw file):

Previously, RaduBerinde wrote…

what's numLocal for?

nothing; leftover. removed.

---

pkg/sql/rowflow/row_based_flow.go, line 289 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] you can do ordering := sqlbase.NoOrdering and flip theif

done

---

pkg/sql/rowflow/row_based_flow.go, line 162 at r4 (raw file):

			// The output is not pass-through and thus is being sent through a
			// router.
			return nil

this one should be a continue, yeah

[andreimatei]

You will also add the tests for the bugs that are being fixed, right?

I wasn't planning to. The only test that makes sense is the test that there's no concurrency on a RootTxn, and that's what we have with the assertion in DistSQLPlanner.Run().
Since we've eliminated this concurrency, there's no timings I can play with any more to force a bad scenario. So the type of test I would have written below to demonstrate problems (e.g. have a union and block one side until the other one encounters a retriable error) would now be fairly non-sensical to write.

What's generally needed is an extension of Jepsen to more complex SQL queries, but that's big ask :)
So I'm thinking about it, but not sure what to do about it. Other than talk to Radu about stressing by hand the FK checks that showed problems initially, but I'm unsure about the current status of the FK checks.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei, @jordanlewis, @RaduBerinde, and @yuzefovich)

[andreimatei]

merging so @knz can start a beta in the morning
bors r+

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei, @jordanlewis, @RaduBerinde, and @yuzefovich)

---

pkg/sql/rowflow/input_sync.go, line 114 at r3 (raw file):

Previously, jordanlewis (Jordan Lewis) wrote…

nit: extra conditional on every row is kinda sadface - can we instead swap out the Less implementation at constructor time? or just make another copy of the synchronizer like you did for the col one?

I don't think that's worth it... This input sync is a fat cat anyway, this extra comparison will be lost in the noise. In the lean and mean vectorized world, that's a separate story...

[andreimatei]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andreimatei, @jordanlewis, @RaduBerinde, and @yuzefovich)

---

pkg/sql/rowflow/row_based_flow.go, line 111 at r4 (raw file):

Previously, yuzefovich wrote…

[nit]: s/the we/then we/ (above).

done

[craig]

Build succeeded

• GitHub CI (Cockroach)