Merge pull request #11697 from RaduBerinde/trivial-router

distsql: add passthrough router type

Author: RaduBerinde

pkg/sql/distsql/cluster_test.go

@@ -122,7 +122,7 @@ func TestClusterFlow(t *testing.T) {
 		Processors: []ProcessorSpec{{
 			Core: ProcessorCoreUnion{TableReader: &tr1},
 			Output: []OutputRouterSpec{{
-				Type: OutputRouterSpec_MIRROR,
+				Type: OutputRouterSpec_PASS_THROUGH,
 				Streams: []StreamEndpointSpec{
 					{Type: StreamEndpointSpec_REMOTE, StreamID: 0, TargetAddr: tc.Server(2).ServingAddr()},
 				},
@@ -136,7 +136,7 @@ func TestClusterFlow(t *testing.T) {
 		Processors: []ProcessorSpec{{
 			Core: ProcessorCoreUnion{TableReader: &tr2},
 			Output: []OutputRouterSpec{{
-				Type: OutputRouterSpec_MIRROR,
+				Type: OutputRouterSpec_PASS_THROUGH,
 				Streams: []StreamEndpointSpec{
 					{Type: StreamEndpointSpec_REMOTE, StreamID: 1, TargetAddr: tc.Server(2).ServingAddr()},
 				},
@@ -151,7 +151,7 @@ func TestClusterFlow(t *testing.T) {
 			{
 				Core: ProcessorCoreUnion{TableReader: &tr3},
 				Output: []OutputRouterSpec{{
-					Type: OutputRouterSpec_MIRROR,
+					Type: OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{
 						{Type: StreamEndpointSpec_LOCAL, StreamID: 2},
 					},
@@ -169,7 +169,7 @@ func TestClusterFlow(t *testing.T) {
 				}},
 				Core: ProcessorCoreUnion{JoinReader: &jr},
 				Output: []OutputRouterSpec{{
-					Type:    OutputRouterSpec_MIRROR,
+					Type:    OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{{Type: StreamEndpointSpec_SYNC_RESPONSE}},
 				}},
 			},

pkg/sql/distsql/data.pb.go

@@ -104,16 +104,17 @@ message InputSyncSpec {
 // it decides how to send results to multiple output streams.
 message OutputRouterSpec {
   enum Type {
-    // Each row is sent to all output streams (also used if there is a
-    // single output stream).
-    MIRROR = 0;
+    // Single output stream.
+    PASS_THROUGH = 0;
+    // Each row is sent to all output streams.
+    MIRROR = 1;
     // Each row is sent to one stream, chosen by hashing certain columns of
     // the row (specified by the hash_columns field).
-    BY_HASH = 1;
+    BY_HASH = 2;
     // Each row is sent to one stream, chosen according to preset boundaries
     // for the values of certain columns of the row. TODO(radu): an extra
     // optional structure below for the range details.
-    BY_RANGE = 2;
+    BY_RANGE = 3;
   }
   optional Type type = 1 [(gogoproto.nullable) = false];
   repeated StreamEndpointSpec streams = 2 [(gogoproto.nullable) = false];

pkg/sql/distsql/data.proto

@@ -121,6 +121,8 @@ func (is *InputSyncSpec) summary() (string, []string) {
 
 func (r *OutputRouterSpec) summary() (string, []string) {
 	switch r.Type {
+	case OutputRouterSpec_PASS_THROUGH:
+		return "", []string{}
 	case OutputRouterSpec_MIRROR:
 		return "mirror", []string{}
 	case OutputRouterSpec_BY_HASH:

pkg/sql/distsql/flow_diagram.go

@@ -67,7 +67,7 @@ func TestPlanDiagramIndexJoin(t *testing.T) {
 		Processors: []ProcessorSpec{{
 			Core: ProcessorCoreUnion{TableReader: &tr},
 			Output: []OutputRouterSpec{{
-				Type: OutputRouterSpec_MIRROR,
+				Type: OutputRouterSpec_PASS_THROUGH,
 				Streams: []StreamEndpointSpec{
 					{StreamID: 0},
 				},
@@ -79,7 +79,7 @@ func TestPlanDiagramIndexJoin(t *testing.T) {
 		Processors: []ProcessorSpec{{
 			Core: ProcessorCoreUnion{TableReader: &tr},
 			Output: []OutputRouterSpec{{
-				Type: OutputRouterSpec_MIRROR,
+				Type: OutputRouterSpec_PASS_THROUGH,
 				Streams: []StreamEndpointSpec{
 					{StreamID: 1},
 				},
@@ -92,7 +92,7 @@ func TestPlanDiagramIndexJoin(t *testing.T) {
 			{
 				Core: ProcessorCoreUnion{TableReader: &tr},
 				Output: []OutputRouterSpec{{
-					Type: OutputRouterSpec_MIRROR,
+					Type: OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{
 						{StreamID: 2},
 					},
@@ -110,7 +110,7 @@ func TestPlanDiagramIndexJoin(t *testing.T) {
 				}},
 				Core: ProcessorCoreUnion{JoinReader: &jr},
 				Output: []OutputRouterSpec{{
-					Type:    OutputRouterSpec_MIRROR,
+					Type:    OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{{Type: StreamEndpointSpec_SYNC_RESPONSE}},
 				}}},
 		},
@@ -216,7 +216,7 @@ func TestPlanDiagramJoin(t *testing.T) {
 				},
 				Core: ProcessorCoreUnion{HashJoiner: &hj},
 				Output: []OutputRouterSpec{{
-					Type: OutputRouterSpec_MIRROR,
+					Type: OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{
 						{StreamID: 101},
 					},
@@ -232,7 +232,7 @@ func TestPlanDiagramJoin(t *testing.T) {
 				}},
 				Core: ProcessorCoreUnion{Noop: &NoopCoreSpec{}},
 				Output: []OutputRouterSpec{{
-					Type:    OutputRouterSpec_MIRROR,
+					Type:    OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{{Type: StreamEndpointSpec_SYNC_RESPONSE}},
 				}}},
 		},
@@ -282,7 +282,7 @@ func TestPlanDiagramJoin(t *testing.T) {
 				},
 				Core: ProcessorCoreUnion{HashJoiner: &hj},
 				Output: []OutputRouterSpec{{
-					Type: OutputRouterSpec_MIRROR,
+					Type: OutputRouterSpec_PASS_THROUGH,
 					Streams: []StreamEndpointSpec{
 						{StreamID: 101},
 					},

pkg/sql/distsql/flow_diagram_test.go

@@ -28,19 +28,24 @@ import (
 )
 
 func makeRouter(spec *OutputRouterSpec, streams []RowReceiver) (RowReceiver, error) {
-	switch len(streams) {
-	case 0:
+	if len(streams) == 0 {
 		return nil, errors.Errorf("no streams in router")
-	case 1:
-		// Special passthrough case - no router.
-		return streams[0], nil
 	}
 
 	switch spec.Type {
+	case OutputRouterSpec_PASS_THROUGH:
+		if len(streams) != 1 {
+			return nil, errors.Errorf("expected one stream for passthrough router")
+		}
+		// No router.
+		return streams[0], nil
+
 	case OutputRouterSpec_BY_HASH:
 		return makeHashRouter(spec.HashColumns, streams)
+
 	case OutputRouterSpec_MIRROR:
 		return makeMirrorRouter(streams)
+
 	default:
 		return nil, errors.Errorf("router type %s not supported", spec.Type)
 	}
@@ -69,12 +74,18 @@ var _ RowReceiver = &mirrorRouter{}
 var crc32Table = crc32.MakeTable(crc32.Castagnoli)
 
 func makeMirrorRouter(streams []RowReceiver) (*mirrorRouter, error) {
+	if len(streams) < 2 {
+		return nil, errors.Errorf("need at least two streams for mirror router")
+	}
 	return &mirrorRouter{
 		routerBase: routerBase{streams: streams},
 	}, nil
 }
 
 func makeHashRouter(hashCols []uint32, streams []RowReceiver) (*hashRouter, error) {
+	if len(streams) < 2 {
+		return nil, errors.Errorf("need at least two streams for hash router")
+	}
 	if len(hashCols) == 0 {
 		return nil, errors.Errorf("no hash columns for BY_HASH router")
 	}

pkg/sql/distsql/routers.go

@@ -119,7 +119,7 @@ func TestMirrorRouter(t *testing.T) {
 
 	vals := sqlbase.RandEncDatumSlices(rng, numCols, numRows)
 
-	for numBuckets := 1; numBuckets <= 3; numBuckets++ {
+	for numBuckets := 2; numBuckets <= 4; numBuckets++ {
 		bufs := make([]*RowBuffer, numBuckets)
 		recvs := make([]RowReceiver, numBuckets)
 		for i := 0; i < numBuckets; i++ {

pkg/sql/distsql/routers_test.go

@@ -64,7 +64,7 @@ func TestServer(t *testing.T) {
 		Processors: []ProcessorSpec{{
 			Core: ProcessorCoreUnion{TableReader: &ts},
 			Output: []OutputRouterSpec{{
-				Type:    OutputRouterSpec_MIRROR,
+				Type:    OutputRouterSpec_PASS_THROUGH,
 				Streams: []StreamEndpointSpec{{Type: StreamEndpointSpec_SYNC_RESPONSE}},
 			}},
 		}},