sql: Add a BulkRowWriter processor to be initially used by CTAS.

This change introduces a BulkRowWriter processor which uses the
AddSSTable method to write rows from a RowSource to the target
table. The processor is required to make the CTAS statement
scalable.

Release note: None

pkg/sql/distsql_plan_ctas.go

@@ -0,0 +1,48 @@
+package sql
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/internal/client"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/pkg/errors"
+)
+
+// CTASPlanResultTypes is the result types for EXPORT plans.
+var CTASPlanResultTypes = []types.T{
+	*types.Bytes, // rows
+}
+
+// PlanAndRunCTAS plans and runs the CREATE TABLE AS command.
+func PlanAndRunCTAS(
+	ctx context.Context,
+	dsp *DistSQLPlanner,
+	evalCtx *extendedEvalContext,
+	txn *client.Txn,
+	canDistribute bool,
+	isLocal bool,
+	in planNode,
+	out distsqlpb.ProcessorCoreUnion,
+	recv *DistSQLReceiver,
+) {
+	planCtx := dsp.NewPlanningCtx(ctx, evalCtx, txn)
+	planCtx.isLocal = isLocal
+
+	p, err := dsp.createPlanForNode(planCtx, in)
+	if err != nil {
+		recv.SetError(errors.Wrapf(err, "constructing distSQL plan"))
+		return
+	}
+
+	p.AddNoGroupingStage(
+		out, distsqlpb.PostProcessSpec{}, CTASPlanResultTypes, distsqlpb.Ordering{},
+	)
+
+	// The bulk row writers will emit a binary encoded BulkOpSummary.
+	p.PlanToStreamColMap = []int{0}
+	p.ResultTypes = CTASPlanResultTypes
+
+	dsp.FinalizePlan(planCtx, &p)
+	dsp.Run(planCtx, txn, &p, recv, evalCtx, nil /* finishedSetupFn */)
+}

pkg/sql/distsqlpb/flow_diagram.go

@@ -458,6 +458,11 @@ func (s *CSVWriterSpec) summary() (string, []string) {
 	return "CSVWriter", []string{s.Destination}
 }
 
+// summary implements the diagramCellType interface.
+func (s *BulkRowWriterSpec) summary() (string, []string) {
+	return "BulkRowWriterSpec", []string{}
+}
+
 // summary implements the diagramCellType interface.
 func (w *WindowerSpec) summary() (string, []string) {
 	details := make([]string, 0, len(w.WindowFns))

pkg/sql/distsqlpb/processors.proto

@@ -102,6 +102,7 @@ message ProcessorCoreUnion {
   optional ChangeAggregatorSpec changeAggregator = 25;
   optional ChangeFrontierSpec changeFrontier = 26;
   optional OrdinalitySpec ordinality = 27;
+  optional BulkRowWriterSpec bulkRowWriter = 28;
 
   reserved 6, 12;
 }