@michae2 reviewed 1 of 1 files at r1, 32 of 32 files at r3.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball)

pkg/sql/colexec/columnarizer.go line 235 at r4 (raw file):

		tuplesToBeSet = 1
	}
	if !c.continueBatch {

Random idea that you didn't ask for: it might be simpler to write these operators as push-iterators, and then use iter.Pull2 to convert them into a Next() function that can be called from the parent operator.

By doing that we could write the columnarizer something like:

...
var nRows int
for row, meta, ok := next(); ok; row, meta, ok = next() {
  if meta != nil {
    if meta.Err != nil {
      colexecerror.ExpectedError(meta.Err)
    }
    if !yield(nil, meta) {
      break
    }
    continue
  }
  if row == nil {
    break
  }
  EncDatumRowToColVecs(row, nRows, c.vecs, c.typs, &c.da)
  nRows++
  if c.helper.AccountForSet(nRows) {
    c.batch.SetLength(nRows)
    if !yield(c.batch, nil) {
      break
    }
    nRows = 0
    c.batch, reallocated = c.helper.ResetMaybeReallocate(c.typs, c.batch, tuplesToBeSet)
    ...
  }
}
...

Without having to manually reify control flow or local variables into state like continueBatch or nRows.

Anyway, I'll set that idea aside and keep reviewing. Just an idea.

@mgartner reviewed 1 of 1 files at r1, 13 of 32 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball and @michae2)

pkg/sql/colexec/columnarizer.go line 235 at r4 (raw file):

I'll pile on another unsolicited idea: What if we passed a callback for handling metadata down into operators so that metadata can be passed up through the operator tree outside the control flow of the loop? I don't have a good grasp of exactly where the metadata originates or terminates, so maybe this a bad idea.

There is some cost in allocating a callback which maybe you were trying to avoid? If the metadata terminates at a single point, we'd only need a single callback, I think. And the metadata would go directly to the termination point, rather than step through each operator in the tree.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball, @mgartner, and @michae2)

pkg/sql/colexec/columnarizer.go line 235 at r4 (raw file):

Thanks for the suggestions.

I'll think more about the refactor of changing pull iterators into push ones, but intuitively the current approach seems easier to follow / more understandable to me.

Re: callback idea. This actually sounds similar to how I tried to implement this initially in #65586 where I introduced a way to send metadata outside of the main data path. There are a couple of difficulties with that approach, one being that the final destination for both data (rows or batches) and metadata is DistSQLReceiver on the gateway node, and it's not thread-safe to push into it, so we need to have a synchronization mechanism between data and metadata arriving which I addressed by having separate goroutines and making some copies, but that turned out to have non-trivial perf impact (on the order of 10% slowdown on a simple workload).

Another difficulty of the callback approach is that the "true" metadata sink DistSQLReceiver could be outside of the tree of the operator that wants to emit the metadata (which also means running in different goroutines, possibly on different nodes), so we first need to push it to the root of the subtree the operator is in, and then ensure that the root pushes it towards the main subtree (so an example sequence of pushes could be something like ColBatchScan -> Outbox - network hop -> Inbox -> Synchronizer -> FlowCoordinator -> DistSQLReceiver). We either need to have a separate goroutine polling for metadata on some of these steps (which turned out to be expensive), or we'll be buffering up metadata on these intermediate "sinks" that will be flushed only by the main goroutine when there is data to push further out (but that defeats the purpose of "streaming" metadata propagation).

For a bit of historical context, back when we filed #35280 and added the current way of metadata propagation, we only had "trailing metadata" use case in the row-by-row engine, so the approach of calling DrainMeta during the plan shutdown was sufficient. Soon after we added the query progress metadata in the row-by-row engine which happened to just work there (because of explicitly propagating both data and metadata) and didn't work in the vectorized engine. Perhaps if that use case was already implemented before #35280 was addressed, we'd have just mimicked the row-by-row engine right away, but I think we didn't go that route in order to simplify the Next signature and due to performance considerations (having to check whether metadata is non-nil on every Next call), but mostly to avoid the large change that this PR ended up doing.

Looking good so far! (I've made it through topk.) Will keep reading.

@michae2 reviewed 6 of 32 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball, @mgartner, and @yuzefovich)

pkg/sql/colexec/columnarizer.go line 235 at r4 (raw file):

I'll think more about the refactor of changing pull iterators into push ones, but intuitively the current approach seems easier to follow / more understandable to me.

That's fine. I think the current approach probably also performs better than setting up a bunch of coroutines, each with their own stacks.

pkg/sql/colexec/parallel_unordered_synchronizer.go line 36 at r4 (raw file):

//
// Note that either a batch or metadata must be sent, but not both. It's also
// possible that both batch and metadata are nil (when then input has been

typo: "the" instead of "then"

pkg/sql/colexec/parallel_unordered_synchronizer.go line 341 at r4 (raw file):

					if s.metas[inputIdx] != nil {
						msg.meta = metaScratch[:]
						msg.meta[0] = *s.metas[inputIdx]

Do we need to make a copy of the metadata?

pkg/sql/colexec/parallel_unordered_synchronizer.go line 361 at r4 (raw file):

						}
						if meta := execinfra.GetTraceDataAsMetadata(s.flowCtx, span); meta != nil {
							msg.meta = append(msg.meta, *meta)

Is msg.meta ever longer than 1? I think I'm not quite understanding when it can be longer.

pkg/sql/colexec/parallel_unordered_synchronizer.go line 378 at r4 (raw file):

				}

				if sentDrainedMeta {

Instead of using sentDrainedMeta could this now be state == parallelUnorderedSynchronizerStateDraining?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball, @mgartner, and @michae2)

pkg/sql/colexec/columnarizer.go line 235 at r4 (raw file):

Ack, let's keep the current approach then.

pkg/sql/colexec/parallel_unordered_synchronizer.go line 341 at r4 (raw file):

I don't think we need to do that (in a sense that is not required), but the signature of DrainMeta is such that we return a slice of values of ProducerMetadata which is the reason we have msg.meta field defined the same way. I left a TODO on DrainMeta interface to look into changing the signature which will allow us to avoid the copy here. (We'll need to audit metadata usage carefully since we do reuse some metadata-related objects in different places.)

pkg/sql/colexec/parallel_unordered_synchronizer.go line 361 at r4 (raw file):

It could get longer if we have non-empty MetadataSources below that return some drained metadata.

pkg/sql/colexec/parallel_unordered_synchronizer.go line 378 at r4 (raw file):

No, these two are somewhat different. A single input to the PUS could send drained meta once it's fully exhausted, yet the PUS might still keep on producing more data that is needed by its output - only when the output calls DrainMeta on the PUS does it transition into the draining state (also when one of the inputs encounters an error which triggers the PUS's shutdown).

Had a few questions, but overall LGTM

Thanks for the review! Just pushed 3 more commits to review, and now the CI should be green.

Nice work! LGTM