Merge #32937

32937: distsql: move protos to distsqlpb from distsqlrun r=jordanlewis a=jordanlewis This creates a new package, distsqlpb, that contains the distsql API protobufs, generated files, and support methods. The new package permits other packages to depend on distsql protobufs without having to depend on the distsql runtime. Release note: None Co-authored-by: Jordan Lewis <[email protected]>
cockroachdb · Dec 7, 2018 · 24c4c04 · 24c4c04
2 parents 6d733ad + 3f06cf5
commit 24c4c04
Show file tree

Hide file tree

Showing 114 changed files with 3,373 additions and 3,181 deletions.
diff --git a/pkg/ccl/changefeedccl/changefeed_dist.go b/pkg/ccl/changefeedccl/changefeed_dist.go
@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/internal/client"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -108,32 +109,32 @@ func distChangefeedFlow(
 	for _, sp := range spanPartitions {
 		// TODO(dan): Merge these watches with the span-level resolved
 		// timestamps from the job progress.
-		watches := make([]distsqlrun.ChangeAggregatorSpec_Watch, len(sp.Spans))
+		watches := make([]distsqlpb.ChangeAggregatorSpec_Watch, len(sp.Spans))
 		for i, nodeSpan := range sp.Spans {
-			watches[i] = distsqlrun.ChangeAggregatorSpec_Watch{
+			watches[i] = distsqlpb.ChangeAggregatorSpec_Watch{
 				Span:            nodeSpan,
 				InitialResolved: initialHighWater,
 			}
 		}
 
 		changeAggregatorProcs = append(changeAggregatorProcs, distsqlplan.Processor{
 			Node: sp.Node,
-			Spec: distsqlrun.ProcessorSpec{
-				Core: distsqlrun.ProcessorCoreUnion{
-					ChangeAggregator: &distsqlrun.ChangeAggregatorSpec{
+			Spec: distsqlpb.ProcessorSpec{
+				Core: distsqlpb.ProcessorCoreUnion{
+					ChangeAggregator: &distsqlpb.ChangeAggregatorSpec{
 						Watches: watches,
 						Feed:    details,
 					},
 				},
-				Output: []distsqlrun.OutputRouterSpec{{Type: distsqlrun.OutputRouterSpec_PASS_THROUGH}},
+				Output: []distsqlpb.OutputRouterSpec{{Type: distsqlpb.OutputRouterSpec_PASS_THROUGH}},
 			},
 		})
 	}
 	// NB: This SpanFrontier processor depends on the set of tracked spans being
 	// static. Currently there is no way for them to change after the changefeed
 	// is created, even if it is paused and unpaused, but #28982 describes some
 	// ways that this might happen in the future.
-	changeFrontierSpec := distsqlrun.ChangeFrontierSpec{
+	changeFrontierSpec := distsqlpb.ChangeFrontierSpec{
 		TrackedSpans: trackedSpans,
 		Feed:         details,
 		JobID:        jobID,
@@ -151,8 +152,8 @@ func distChangefeedFlow(
 
 	p.AddSingleGroupStage(
 		gatewayNodeID,
-		distsqlrun.ProcessorCoreUnion{ChangeFrontier: &changeFrontierSpec},
-		distsqlrun.PostProcessSpec{},
+		distsqlpb.ProcessorCoreUnion{ChangeFrontier: &changeFrontierSpec},
+		distsqlpb.PostProcessSpec{},
 		changefeedResultTypes,
 	)
 

diff --git a/pkg/ccl/changefeedccl/changefeed_processors.go b/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
@@ -32,7 +33,7 @@ type changeAggregator struct {
 	distsqlrun.ProcessorBase
 
 	flowCtx *distsqlrun.FlowCtx
-	spec    distsqlrun.ChangeAggregatorSpec
+	spec    distsqlpb.ChangeAggregatorSpec
 	memAcc  mon.BoundAccount
 
 	// cancel shuts down the processor, both the `Next()` flow and the poller.
@@ -69,7 +70,7 @@ var _ distsqlrun.RowSource = &changeAggregator{}
 func newChangeAggregatorProcessor(
 	flowCtx *distsqlrun.FlowCtx,
 	processorID int32,
-	spec distsqlrun.ChangeAggregatorSpec,
+	spec distsqlpb.ChangeAggregatorSpec,
 	output distsqlrun.RowReceiver,
 ) (distsqlrun.Processor, error) {
 	ctx := flowCtx.EvalCtx.Ctx()
@@ -81,7 +82,7 @@ func newChangeAggregatorProcessor(
 	}
 	if err := ca.Init(
 		ca,
-		&distsqlrun.PostProcessSpec{},
+		&distsqlpb.PostProcessSpec{},
 		nil, /* types */
 		flowCtx,
 		processorID,
@@ -281,7 +282,7 @@ type changeFrontier struct {
 	distsqlrun.ProcessorBase
 
 	flowCtx *distsqlrun.FlowCtx
-	spec    distsqlrun.ChangeFrontierSpec
+	spec    distsqlpb.ChangeFrontierSpec
 	memAcc  mon.BoundAccount
 	a       sqlbase.DatumAlloc
 
@@ -327,7 +328,7 @@ var _ distsqlrun.RowSource = &changeFrontier{}
 func newChangeFrontierProcessor(
 	flowCtx *distsqlrun.FlowCtx,
 	processorID int32,
-	spec distsqlrun.ChangeFrontierSpec,
+	spec distsqlpb.ChangeFrontierSpec,
 	input distsqlrun.RowSource,
 	output distsqlrun.RowReceiver,
 ) (distsqlrun.Processor, error) {
@@ -341,7 +342,7 @@ func newChangeFrontierProcessor(
 		sf:      makeSpanFrontier(spec.TrackedSpans...),
 	}
 	if err := cf.Init(
-		cf, &distsqlrun.PostProcessSpec{},
+		cf, &distsqlpb.PostProcessSpec{},
 		input.OutputTypes(),
 		flowCtx,
 		processorID,

diff --git a/pkg/ccl/importccl/exportcsv.go b/pkg/ccl/importccl/exportcsv.go
@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -136,7 +137,7 @@ func exportPlanHook(
 			}
 		}
 
-		out := distsqlrun.ProcessorCoreUnion{CSVWriter: &distsqlrun.CSVWriterSpec{
+		out := distsqlpb.ProcessorCoreUnion{CSVWriter: &distsqlpb.CSVWriterSpec{
 			Destination: file,
 			NamePattern: exportFilePatternDefault,
 			Options:     csvOpts,
@@ -166,7 +167,7 @@ func exportPlanHook(
 func newCSVWriterProcessor(
 	flowCtx *distsqlrun.FlowCtx,
 	processorID int32,
-	spec distsqlrun.CSVWriterSpec,
+	spec distsqlpb.CSVWriterSpec,
 	input distsqlrun.RowSource,
 	output distsqlrun.RowReceiver,
 ) (distsqlrun.Processor, error) {
@@ -177,7 +178,7 @@ func newCSVWriterProcessor(
 		input:       input,
 		output:      output,
 	}
-	if err := c.out.Init(&distsqlrun.PostProcessSpec{}, sql.ExportPlanResultTypes, flowCtx.NewEvalCtx(), output); err != nil {
+	if err := c.out.Init(&distsqlpb.PostProcessSpec{}, sql.ExportPlanResultTypes, flowCtx.NewEvalCtx(), output); err != nil {
 		return nil, err
 	}
 	return c, nil
@@ -186,7 +187,7 @@ func newCSVWriterProcessor(
 type csvWriter struct {
 	flowCtx     *distsqlrun.FlowCtx
 	processorID int32
-	spec        distsqlrun.CSVWriterSpec
+	spec        distsqlpb.CSVWriterSpec
 	input       distsqlrun.RowSource
 	out         distsqlrun.ProcOutputHelper
 	output      distsqlrun.RowReceiver

diff --git a/pkg/ccl/importccl/read_import_proc.go b/pkg/ccl/importccl/read_import_proc.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/builtins"
@@ -342,7 +343,7 @@ var csvOutputTypes = []sqlbase.ColumnType{
 func newReadImportDataProcessor(
 	flowCtx *distsqlrun.FlowCtx,
 	processorID int32,
-	spec distsqlrun.ReadImportDataSpec,
+	spec distsqlpb.ReadImportDataSpec,
 	output distsqlrun.RowReceiver,
 ) (distsqlrun.Processor, error) {
 	cp := &readImportDataProcessor{
@@ -352,7 +353,7 @@ func newReadImportDataProcessor(
 		output:      output,
 	}
 
-	if err := cp.out.Init(&distsqlrun.PostProcessSpec{}, csvOutputTypes, flowCtx.NewEvalCtx(), output); err != nil {
+	if err := cp.out.Init(&distsqlpb.PostProcessSpec{}, csvOutputTypes, flowCtx.NewEvalCtx(), output); err != nil {
 		return nil, err
 	}
 	return cp, nil
@@ -369,7 +370,7 @@ type inputConverter interface {
 type readImportDataProcessor struct {
 	flowCtx     *distsqlrun.FlowCtx
 	processorID int32
-	spec        distsqlrun.ReadImportDataSpec
+	spec        distsqlpb.ReadImportDataSpec
 	out         distsqlrun.ProcOutputHelper
 	output      distsqlrun.RowReceiver
 }

diff --git a/pkg/ccl/importccl/sst_writer_proc.go b/pkg/ccl/importccl/sst_writer_proc.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
@@ -46,7 +47,7 @@ var sstOutputTypes = []sqlbase.ColumnType{
 func newSSTWriterProcessor(
 	flowCtx *distsqlrun.FlowCtx,
 	processorID int32,
-	spec distsqlrun.SSTWriterSpec,
+	spec distsqlpb.SSTWriterSpec,
 	input distsqlrun.RowSource,
 	output distsqlrun.RowReceiver,
 ) (distsqlrun.Processor, error) {
@@ -62,7 +63,7 @@ func newSSTWriterProcessor(
 		progress:    spec.Progress,
 		db:          flowCtx.EvalCtx.Txn.DB(),
 	}
-	if err := sp.out.Init(&distsqlrun.PostProcessSpec{}, sstOutputTypes, flowCtx.NewEvalCtx(), output); err != nil {
+	if err := sp.out.Init(&distsqlpb.PostProcessSpec{}, sstOutputTypes, flowCtx.NewEvalCtx(), output); err != nil {
 		return nil, err
 	}
 	return sp, nil
@@ -71,14 +72,14 @@ func newSSTWriterProcessor(
 type sstWriter struct {
 	flowCtx     *distsqlrun.FlowCtx
 	processorID int32
-	spec        distsqlrun.SSTWriterSpec
+	spec        distsqlpb.SSTWriterSpec
 	input       distsqlrun.RowSource
 	out         distsqlrun.ProcOutputHelper
 	output      distsqlrun.RowReceiver
 	tempStorage diskmap.Factory
 	settings    *cluster.Settings
 	registry    *jobs.Registry
-	progress    distsqlrun.JobProgress
+	progress    distsqlpb.JobProgress
 	db          *client.DB
 }
 

diff --git a/pkg/cli/debug.go b/pkg/cli/debug.go
@@ -41,7 +41,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/server/status/statuspb"
-	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
@@ -978,13 +978,13 @@ func parseGossipValues(gossipInfo *gossip.InfoStatus) (string, error) {
 			}
 			output = append(output, fmt.Sprintf("%q: %+v", key, healthAlert))
 		} else if strings.HasPrefix(key, gossip.KeyDistSQLNodeVersionKeyPrefix) {
-			var version distsqlrun.DistSQLVersionGossipInfo
+			var version distsqlpb.DistSQLVersionGossipInfo
 			if err := protoutil.Unmarshal(bytes, &version); err != nil {
 				return "", errors.Wrapf(err, "failed to parse value for key %q", key)
 			}
 			output = append(output, fmt.Sprintf("%q: %+v", key, version))
 		} else if strings.HasPrefix(key, gossip.KeyDistSQLDrainingPrefix) {
-			var drainingInfo distsqlrun.DistSQLDrainingInfo
+			var drainingInfo distsqlpb.DistSQLDrainingInfo
 			if err := protoutil.Unmarshal(bytes, &drainingInfo); err != nil {
 				return "", errors.Wrapf(err, "failed to parse value for key %q", key)
 			}

diff --git a/pkg/server/authentication_test.go b/pkg/server/authentication_test.go
@@ -34,7 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server/debug"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
-	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/closedts/ctpb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -709,11 +709,11 @@ func TestGRPCAuthentication(t *testing.T) {
 			return err
 		}},
 		{"distSQL", func(ctx context.Context, conn *grpc.ClientConn) error {
-			stream, err := distsqlrun.NewDistSQLClient(conn).RunSyncFlow(ctx)
+			stream, err := distsqlpb.NewDistSQLClient(conn).RunSyncFlow(ctx)
 			if err != nil {
 				return err
 			}
-			_ = stream.Send(&distsqlrun.ConsumerSignal{})
+			_ = stream.Send(&distsqlpb.ConsumerSignal{})
 			_, err = stream.Recv()
 			return err
 		}},

diff --git a/pkg/server/server.go b/pkg/server/server.go
@@ -47,6 +47,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -532,7 +533,7 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 	}
 
 	s.distSQLServer = distsqlrun.NewServer(ctx, distSQLCfg)
-	distsqlrun.RegisterDistSQLServer(s.grpc, s.distSQLServer)
+	distsqlpb.RegisterDistSQLServer(s.grpc, s.distSQLServer)
 
 	s.admin = newAdminServer(s)
 	s.status = newStatusServer(

diff --git a/pkg/sql/distinct.go b/pkg/sql/distinct.go
@@ -17,6 +17,7 @@ package sql
 import (
 	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlrun"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/types"
@@ -217,12 +218,12 @@ func (n *distinctNode) startExec(params runParams) error {
 	if err := input.startExec(params); err != nil {
 		return err
 	}
-	if err := input.InitWithOutput(&distsqlrun.PostProcessSpec{}, nil); err != nil {
+	if err := input.InitWithOutput(&distsqlpb.PostProcessSpec{}, nil); err != nil {
 		return err
 	}
 
-	post := &distsqlrun.PostProcessSpec{} // post is not used as we only use the processor for the core distinct logic.
-	var output distsqlrun.RowReceiver     // output is never used as distinct is only run as a RowSource.
+	post := &distsqlpb.PostProcessSpec{} // post is not used as we only use the processor for the core distinct logic.
+	var output distsqlrun.RowReceiver    // output is never used as distinct is only run as a RowSource.
 
 	proc, err := distsqlrun.NewDistinct(flowCtx, 0 /* processorID */, spec, input, post, output)
 	if err != nil {