sql/export: refactor processors to implement RowSource interface

We recently found an edge case where the output of EXPORT was used as an
input to a mutation. Currently, in such a setup we have two goroutines
since the export processors don't implement `execinfra.RowSource`
interface, so we cannot fuse them with the mutation processor. Presence
of concurrency forces usage of the LeafTxns, yet mutations require
access to the RootTxn, thus we have conflicting requirements. Before the
previous commit we'd get a nil pointer crash, with the previous
commit in place we now get an assertion failure.

This commit solves this issue by refactoring both processors to
implement the `RowSource` interface. In the problematic query it allows
all processors to be fused together to run in a single goroutine, so we
can use the RootTxn as required by the mutation.

Release note (bug fix): Previously, EXPORT CSV and EXPORT PARQUET stmts
could result in a node crash when their result rows were used as the
input to a mutation like an INSERT within the same SQL. This bug has
been present since before 22.1 version and has now been fixed.

Author: yuzefovich

pkg/sql/execinfra/base.go

@@ -332,45 +332,6 @@ func DrainAndClose(
 	dst.ProducerDone()
 }
 
-// NoMetadataRowSource is a wrapper on top of a RowSource that automatically
-// forwards metadata to a RowReceiver. Data rows are returned through an
-// interface similar to RowSource, except that, since metadata is taken care of,
-// only the data rows are returned.
-//
-// The point of this struct is that it'd be burdensome for some row consumers to
-// have to deal with metadata.
-type NoMetadataRowSource struct {
-	src          RowSource
-	metadataSink RowReceiver
-}
-
-// MakeNoMetadataRowSource builds a NoMetadataRowSource.
-func MakeNoMetadataRowSource(src RowSource, sink RowReceiver) NoMetadataRowSource {
-	return NoMetadataRowSource{src: src, metadataSink: sink}
-}
-
-// NextRow is analogous to RowSource.Next. If the producer sends an error, we
-// can't just forward it to metadataSink. We need to let the consumer know so
-// that it's not under the impression that everything is hunky-dory and it can
-// continue consuming rows. So, this interface returns the error. Just like with
-// a raw RowSource, the consumer should generally call ConsumerDone() and drain.
-func (rs *NoMetadataRowSource) NextRow() (rowenc.EncDatumRow, error) {
-	for {
-		row, meta := rs.src.Next()
-		if meta == nil {
-			return row, nil
-		}
-		if meta.Err != nil {
-			return nil, meta.Err
-		}
-		// We forward the metadata and ignore the returned ConsumerStatus. There's
-		// no good way to use that status here; eventually the consumer of this
-		// NoMetadataRowSource will figure out the same status and act on it as soon
-		// as a non-metadata row is received.
-		_ = rs.metadataSink.Push(nil /* row */, meta)
-	}
-}
-
 // RowChannelMsg is the message used in the channels that implement
 // local physical streams (i.e. the RowChannel's).
 type RowChannelMsg struct {

pkg/sql/export/BUILD.bazel

@@ -15,16 +15,17 @@ go_library(
         "//pkg/settings",
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execopnode",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/rowenc",
         "//pkg/sql/sem/tree",
         "//pkg/sql/types",
         "//pkg/util/encoding/csv",
+        "//pkg/util/log",
         "//pkg/util/mon",
         "//pkg/util/parquet",
-        "//pkg/util/tracing",
         "//pkg/util/unique",
         "@com_github_cockroachdb_errors//:errors",
     ],