colexec: wrap DrainMeta with panic-catcher and protect columnarizer

[yuzefovich]

colexec: wrap DrainMeta with panic-catcher and protect columnarizer

Previously, in some edge cases (like when a panic is encountered during
Operator.Init) the metadata sources could have been uninitialized, so
when we tried to drain them, we'd encounter a crash. In order to avoid
that in the future, now all root components will wrap the draining with
the panic-catcher. Additionally, we now protect the columnarizer in this
case explicitly - if it wasn't initialized, it won't drain the wrapped
processor in DrainMeta.

Fixes: #62514.

Release note: None

rowexec: remove redundant implementations of MetadataSource interface

Previously, some row-by-row processors implemented
execinfra.MetadataSource interface. The idea behind that originally
was to allow for wrapped processors to return their metadata in the
vectorized flow, but nothing explicit is actually needed because every
wrapped processor has a columnarizer after it which will drain the
processor according to row-by-row model (by moving into draining state
and exhausting the trailing meta). This commit removes those redundant
implementations.

This allows us to move the interface into colexecop package where it
belongs.

Release note: None

[cockroach-teamcity]

This change is 

[yuzefovich]

Note that the first commit fixes a GA-blocker (which I'm not sure whether it deserves such status), so I intend to backport the first commit to 21.1 branch (possibly for 21.1.0).

[rytaft]

Reviewed 5 of 5 files at r1, 30 of 30 files at r2.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @michae2 and @yuzefovich)

---

pkg/sql/colflow/routers.go, line 640 at r1 (raw file):

	if inputInitialized {
		// If we encountered a panic when initializing the input, the metadata
		// sources might be uninitialized, so we drain them only otherwise.

nit: "so we drain them only otherwise" -> "so we only drain them if initialized"

---

pkg/sql/colflow/routers.go, line 643 at r1 (raw file):

		// Although the DrainMeta call contains a panic-catcher, we would rather
		// not emit errors (which are likely to occur) in this case.
		r.bufferedMeta = append(r.bufferedMeta, r.metadataSources.DrainMeta(ctx)...)

is there no way to tell from the metadataSources directly whether they are initialized? Seems like this check could go inside the DrainMeta function (maybe before the panic catcher?)

[yuzefovich]

TFTR!

bors r+

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

---

pkg/sql/colflow/routers.go, line 643 at r1 (raw file):

Previously, rytaft (Rebecca Taft) wrote…

is there no way to tell from the metadataSources directly whether they are initialized? Seems like this check could go inside the DrainMeta function (maybe before the panic catcher?)

Good point. We already had protection in most places, and I updated the code to add the check to the columnarizer (the only place that I found to be missing the initialization check), so I removed this "input initialized" business.

Actually, doing just that should be sufficient to fix the issue, but I think that it is beneficial to keep the panic-catcher around draining for stability.

[craig]

Build failed:

• GitHub CI (Cockroach)

[yuzefovich]

A flake.

bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)