diff --git a/go-tipb/analyze.pb.go b/go-tipb/analyze.pb.go
index bfccc3ad8..497f00634 100644
--- a/go-tipb/analyze.pb.go
+++ b/go-tipb/analyze.pb.go
@@ -31,6 +31,8 @@
 		ChecksumRequest
 		ChecksumResponse
 		Executor
+		ExchangeSender
+		ExchangeReceiver
 		TableScan
 		Join
 		IndexScan
diff --git a/go-tipb/executor.pb.go b/go-tipb/executor.pb.go
index 91c82d18d..0098aad59 100644
--- a/go-tipb/executor.pb.go
+++ b/go-tipb/executor.pb.go
@@ -21,38 +21,44 @@ var _ = math.Inf
 type ExecType int32
 
 const (
-	ExecType_TypeTableScan   ExecType = 0
-	ExecType_TypeIndexScan   ExecType = 1
-	ExecType_TypeSelection   ExecType = 2
-	ExecType_TypeAggregation ExecType = 3
-	ExecType_TypeTopN        ExecType = 4
-	ExecType_TypeLimit       ExecType = 5
-	ExecType_TypeStreamAgg   ExecType = 6
-	ExecType_TypeJoin        ExecType = 7
-	ExecType_TypeKill        ExecType = 8
+	ExecType_TypeTableScan        ExecType = 0
+	ExecType_TypeIndexScan        ExecType = 1
+	ExecType_TypeSelection        ExecType = 2
+	ExecType_TypeAggregation      ExecType = 3
+	ExecType_TypeTopN             ExecType = 4
+	ExecType_TypeLimit            ExecType = 5
+	ExecType_TypeStreamAgg        ExecType = 6
+	ExecType_TypeJoin             ExecType = 7
+	ExecType_TypeKill             ExecType = 8
+	ExecType_TypeExchangeSender   ExecType = 9
+	ExecType_TypeExchangeReceiver ExecType = 10
 )
 
 var ExecType_name = map[int32]string{
-	0: "TypeTableScan",
-	1: "TypeIndexScan",
-	2: "TypeSelection",
-	3: "TypeAggregation",
-	4: "TypeTopN",
-	5: "TypeLimit",
-	6: "TypeStreamAgg",
-	7: "TypeJoin",
-	8: "TypeKill",
+	0:  "TypeTableScan",
+	1:  "TypeIndexScan",
+	2:  "TypeSelection",
+	3:  "TypeAggregation",
+	4:  "TypeTopN",
+	5:  "TypeLimit",
+	6:  "TypeStreamAgg",
+	7:  "TypeJoin",
+	8:  "TypeKill",
+	9:  "TypeExchangeSender",
+	10: "TypeExchangeReceiver",
 }
 var ExecType_value = map[string]int32{
-	"TypeTableScan":   0,
-	"TypeIndexScan":   1,
-	"TypeSelection":   2,
-	"TypeAggregation": 3,
-	"TypeTopN":        4,
-	"TypeLimit":       5,
-	"TypeStreamAgg":   6,
-	"TypeJoin":        7,
-	"TypeKill":        8,
+	"TypeTableScan":        0,
+	"TypeIndexScan":        1,
+	"TypeSelection":        2,
+	"TypeAggregation":      3,
+	"TypeTopN":             4,
+	"TypeLimit":            5,
+	"TypeStreamAgg":        6,
+	"TypeJoin":             7,
+	"TypeKill":             8,
+	"TypeExchangeSender":   9,
+	"TypeExchangeReceiver": 10,
 }
 
 func (x ExecType) Enum() *ExecType {
@@ -73,6 +79,43 @@ func (x *ExecType) UnmarshalJSON(data []byte) error {
 }
 func (ExecType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{0} }
 
+type ExchangeType int32
+
+const (
+	ExchangeType_PassThrough ExchangeType = 0
+	ExchangeType_Broadcast   ExchangeType = 1
+	ExchangeType_Hash        ExchangeType = 2
+)
+
+var ExchangeType_name = map[int32]string{
+	0: "PassThrough",
+	1: "Broadcast",
+	2: "Hash",
+}
+var ExchangeType_value = map[string]int32{
+	"PassThrough": 0,
+	"Broadcast":   1,
+	"Hash":        2,
+}
+
+func (x ExchangeType) Enum() *ExchangeType {
+	p := new(ExchangeType)
+	*p = x
+	return p
+}
+func (x ExchangeType) String() string {
+	return proto.EnumName(ExchangeType_name, int32(x))
+}
+func (x *ExchangeType) UnmarshalJSON(data []byte) error {
+	value, err := proto.UnmarshalJSONEnum(ExchangeType_value, data, "ExchangeType")
+	if err != nil {
+		return err
+	}
+	*x = ExchangeType(value)
+	return nil
+}
+func (ExchangeType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{1} }
+
 type EngineType int32
 
 const (
@@ -108,7 +151,7 @@ func (x *EngineType) UnmarshalJSON(data []byte) error {
 	*x = EngineType(value)
 	return nil
 }
-func (EngineType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{1} }
+func (EngineType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{2} }
 
 type JoinType int32
 
@@ -157,7 +200,7 @@ func (x *JoinType) UnmarshalJSON(data []byte) error {
 	*x = JoinType(value)
 	return nil
 }
-func (JoinType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{2} }
+func (JoinType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{3} }
 
 type JoinExecType int32
 
@@ -188,22 +231,23 @@ func (x *JoinExecType) UnmarshalJSON(data []byte) error {
 	*x = JoinExecType(value)
 	return nil
 }
-func (JoinExecType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{3} }
+func (JoinExecType) EnumDescriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{4} }
 
 // It represents a Executor.
 type Executor struct {
-	Tp               ExecType     `protobuf:"varint,1,opt,name=tp,enum=tipb.ExecType" json:"tp"`
-	TblScan          *TableScan   `protobuf:"bytes,2,opt,name=tbl_scan,json=tblScan" json:"tbl_scan,omitempty"`
-	IdxScan          *IndexScan   `protobuf:"bytes,3,opt,name=idx_scan,json=idxScan" json:"idx_scan,omitempty"`
-	Selection        *Selection   `protobuf:"bytes,4,opt,name=selection" json:"selection,omitempty"`
-	Aggregation      *Aggregation `protobuf:"bytes,5,opt,name=aggregation" json:"aggregation,omitempty"`
-	TopN             *TopN        `protobuf:"bytes,6,opt,name=topN" json:"topN,omitempty"`
-	Limit            *Limit       `protobuf:"bytes,7,opt,name=limit" json:"limit,omitempty"`
-	StreamAgg        *Aggregation `protobuf:"bytes,8,opt,name=stream_agg,json=streamAgg" json:"stream_agg,omitempty"`
-	Join             *Join        `protobuf:"bytes,9,opt,name=join" json:"join,omitempty"`
-	ExecutorId       *string      `protobuf:"bytes,10,opt,name=executor_id,json=executorId" json:"executor_id,omitempty"`
-	Kill             *Kill        `protobuf:"bytes,11,opt,name=kill" json:"kill,omitempty"`
-	XXX_unrecognized []byte       `json:"-"`
+	Tp               ExecType          `protobuf:"varint,1,opt,name=tp,enum=tipb.ExecType" json:"tp"`
+	TblScan          *TableScan        `protobuf:"bytes,2,opt,name=tbl_scan,json=tblScan" json:"tbl_scan,omitempty"`
+	IdxScan          *IndexScan        `protobuf:"bytes,3,opt,name=idx_scan,json=idxScan" json:"idx_scan,omitempty"`
+	Selection        *Selection        `protobuf:"bytes,4,opt,name=selection" json:"selection,omitempty"`
+	Aggregation      *Aggregation      `protobuf:"bytes,5,opt,name=aggregation" json:"aggregation,omitempty"`
+	TopN             *TopN             `protobuf:"bytes,6,opt,name=topN" json:"topN,omitempty"`
+	Limit            *Limit            `protobuf:"bytes,7,opt,name=limit" json:"limit,omitempty"`
+	ExchangeReceiver *ExchangeReceiver `protobuf:"bytes,8,opt,name=exchange_receiver,json=exchangeReceiver" json:"exchange_receiver,omitempty"`
+	Join             *Join             `protobuf:"bytes,9,opt,name=join" json:"join,omitempty"`
+	ExecutorId       *string           `protobuf:"bytes,10,opt,name=executor_id,json=executorId" json:"executor_id,omitempty"`
+	Kill             *Kill             `protobuf:"bytes,11,opt,name=kill" json:"kill,omitempty"`
+	ExchangeSender   *ExchangeSender   `protobuf:"bytes,12,opt,name=exchange_sender,json=exchangeSender" json:"exchange_sender,omitempty"`
+	XXX_unrecognized []byte            `json:"-"`
 }
 
 func (m *Executor) Reset()                    { *m = Executor{} }
@@ -260,9 +304,9 @@ func (m *Executor) GetLimit() *Limit {
 	return nil
 }
 
-func (m *Executor) GetStreamAgg() *Aggregation {
+func (m *Executor) GetExchangeReceiver() *ExchangeReceiver {
 	if m != nil {
-		return m.StreamAgg
+		return m.ExchangeReceiver
 	}
 	return nil
 }
@@ -288,6 +332,81 @@ func (m *Executor) GetKill() *Kill {
 	return nil
 }
 
+func (m *Executor) GetExchangeSender() *ExchangeSender {
+	if m != nil {
+		return m.ExchangeSender
+	}
+	return nil
+}
+
+// ExchangeSender will build connection with ExchangeReceiver.
+type ExchangeSender struct {
+	Tp               ExchangeType `protobuf:"varint,1,opt,name=tp,enum=tipb.ExchangeType" json:"tp"`
+	EncodedTaskMeta  [][]byte     `protobuf:"bytes,2,rep,name=encoded_task_meta,json=encodedTaskMeta" json:"encoded_task_meta,omitempty"`
+	PartitionKeys    []*Expr      `protobuf:"bytes,3,rep,name=partition_keys,json=partitionKeys" json:"partition_keys,omitempty"`
+	Child            *Executor    `protobuf:"bytes,4,opt,name=child" json:"child,omitempty"`
+	XXX_unrecognized []byte       `json:"-"`
+}
+
+func (m *ExchangeSender) Reset()                    { *m = ExchangeSender{} }
+func (m *ExchangeSender) String() string            { return proto.CompactTextString(m) }
+func (*ExchangeSender) ProtoMessage()               {}
+func (*ExchangeSender) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{1} }
+
+func (m *ExchangeSender) GetTp() ExchangeType {
+	if m != nil {
+		return m.Tp
+	}
+	return ExchangeType_PassThrough
+}
+
+func (m *ExchangeSender) GetEncodedTaskMeta() [][]byte {
+	if m != nil {
+		return m.EncodedTaskMeta
+	}
+	return nil
+}
+
+func (m *ExchangeSender) GetPartitionKeys() []*Expr {
+	if m != nil {
+		return m.PartitionKeys
+	}
+	return nil
+}
+
+func (m *ExchangeSender) GetChild() *Executor {
+	if m != nil {
+		return m.Child
+	}
+	return nil
+}
+
+// ExchangeReceiver accept connection and receiver data from ExchangeSender.
+type ExchangeReceiver struct {
+	EncodedTaskMeta  [][]byte     `protobuf:"bytes,1,rep,name=encoded_task_meta,json=encodedTaskMeta" json:"encoded_task_meta,omitempty"`
+	FieldTypes       []*FieldType `protobuf:"bytes,2,rep,name=field_types,json=fieldTypes" json:"field_types,omitempty"`
+	XXX_unrecognized []byte       `json:"-"`
+}
+
+func (m *ExchangeReceiver) Reset()                    { *m = ExchangeReceiver{} }
+func (m *ExchangeReceiver) String() string            { return proto.CompactTextString(m) }
+func (*ExchangeReceiver) ProtoMessage()               {}
+func (*ExchangeReceiver) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{2} }
+
+func (m *ExchangeReceiver) GetEncodedTaskMeta() [][]byte {
+	if m != nil {
+		return m.EncodedTaskMeta
+	}
+	return nil
+}
+
+func (m *ExchangeReceiver) GetFieldTypes() []*FieldType {
+	if m != nil {
+		return m.FieldTypes
+	}
+	return nil
+}
+
 type TableScan struct {
 	TableId          int64         `protobuf:"varint,1,opt,name=table_id,json=tableId" json:"table_id"`
 	Columns          []*ColumnInfo `protobuf:"bytes,2,rep,name=columns" json:"columns,omitempty"`
@@ -301,7 +420,7 @@ type TableScan struct {
 func (m *TableScan) Reset()                    { *m = TableScan{} }
 func (m *TableScan) String() string            { return proto.CompactTextString(m) }
 func (*TableScan) ProtoMessage()               {}
-func (*TableScan) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{1} }
+func (*TableScan) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{3} }
 
 func (m *TableScan) GetTableId() int64 {
 	if m != nil {
@@ -364,7 +483,7 @@ type Join struct {
 func (m *Join) Reset()                    { *m = Join{} }
 func (m *Join) String() string            { return proto.CompactTextString(m) }
 func (*Join) ProtoMessage()               {}
-func (*Join) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{2} }
+func (*Join) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{4} }
 
 func (m *Join) GetJoinType() JoinType {
 	if m != nil {
@@ -456,7 +575,7 @@ type IndexScan struct {
 func (m *IndexScan) Reset()                    { *m = IndexScan{} }
 func (m *IndexScan) String() string            { return proto.CompactTextString(m) }
 func (*IndexScan) ProtoMessage()               {}
-func (*IndexScan) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{3} }
+func (*IndexScan) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{5} }
 
 func (m *IndexScan) GetTableId() int64 {
 	if m != nil {
@@ -511,7 +630,7 @@ type Selection struct {
 func (m *Selection) Reset()                    { *m = Selection{} }
 func (m *Selection) String() string            { return proto.CompactTextString(m) }
 func (*Selection) ProtoMessage()               {}
-func (*Selection) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{4} }
+func (*Selection) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{6} }
 
 func (m *Selection) GetConditions() []*Expr {
 	if m != nil {
@@ -545,7 +664,7 @@ type Projection struct {
 func (m *Projection) Reset()                    { *m = Projection{} }
 func (m *Projection) String() string            { return proto.CompactTextString(m) }
 func (*Projection) ProtoMessage()               {}
-func (*Projection) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{5} }
+func (*Projection) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{7} }
 
 func (m *Projection) GetExprs() []*Expr {
 	if m != nil {
@@ -584,7 +703,7 @@ type Aggregation struct {
 func (m *Aggregation) Reset()                    { *m = Aggregation{} }
 func (m *Aggregation) String() string            { return proto.CompactTextString(m) }
 func (*Aggregation) ProtoMessage()               {}
-func (*Aggregation) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{6} }
+func (*Aggregation) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{8} }
 
 func (m *Aggregation) GetGroupBy() []*Expr {
 	if m != nil {
@@ -639,7 +758,7 @@ type TopN struct {
 func (m *TopN) Reset()                    { *m = TopN{} }
 func (m *TopN) String() string            { return proto.CompactTextString(m) }
 func (*TopN) ProtoMessage()               {}
-func (*TopN) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{7} }
+func (*TopN) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{9} }
 
 func (m *TopN) GetOrderBy() []*ByItem {
 	if m != nil {
@@ -672,7 +791,7 @@ type Limit struct {
 func (m *Limit) Reset()                    { *m = Limit{} }
 func (m *Limit) String() string            { return proto.CompactTextString(m) }
 func (*Limit) ProtoMessage()               {}
-func (*Limit) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{8} }
+func (*Limit) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{10} }
 
 func (m *Limit) GetLimit() uint64 {
 	if m != nil {
@@ -701,7 +820,7 @@ type Kill struct {
 func (m *Kill) Reset()                    { *m = Kill{} }
 func (m *Kill) String() string            { return proto.CompactTextString(m) }
 func (*Kill) ProtoMessage()               {}
-func (*Kill) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{9} }
+func (*Kill) Descriptor() ([]byte, []int) { return fileDescriptorExecutor, []int{11} }
 
 func (m *Kill) GetConnID() uint64 {
 	if m != nil {
@@ -733,7 +852,7 @@ func (m *ExecutorExecutionSummary) Reset()         { *m = ExecutorExecutionSumma
 func (m *ExecutorExecutionSummary) String() string { return proto.CompactTextString(m) }
 func (*ExecutorExecutionSummary) ProtoMessage()    {}
 func (*ExecutorExecutionSummary) Descriptor() ([]byte, []int) {
-	return fileDescriptorExecutor, []int{10}
+	return fileDescriptorExecutor, []int{12}
 }
 
 func (m *ExecutorExecutionSummary) GetTimeProcessedNs() uint64 {
@@ -766,6 +885,8 @@ func (m *ExecutorExecutionSummary) GetExecutorId() string {
 
 func init() {
 	proto.RegisterType((*Executor)(nil), "tipb.Executor")
+	proto.RegisterType((*ExchangeSender)(nil), "tipb.ExchangeSender")
+	proto.RegisterType((*ExchangeReceiver)(nil), "tipb.ExchangeReceiver")
 	proto.RegisterType((*TableScan)(nil), "tipb.TableScan")
 	proto.RegisterType((*Join)(nil), "tipb.Join")
 	proto.RegisterType((*IndexScan)(nil), "tipb.IndexScan")
@@ -777,6 +898,7 @@ func init() {
 	proto.RegisterType((*Kill)(nil), "tipb.Kill")
 	proto.RegisterType((*ExecutorExecutionSummary)(nil), "tipb.ExecutorExecutionSummary")
 	proto.RegisterEnum("tipb.ExecType", ExecType_name, ExecType_value)
+	proto.RegisterEnum("tipb.ExchangeType", ExchangeType_name, ExchangeType_value)
 	proto.RegisterEnum("tipb.EngineType", EngineType_name, EngineType_value)
 	proto.RegisterEnum("tipb.JoinType", JoinType_name, JoinType_value)
 	proto.RegisterEnum("tipb.JoinExecType", JoinExecType_name, JoinExecType_value)
@@ -859,11 +981,11 @@ func (m *Executor) MarshalTo(dAtA []byte) (int, error) {
 		}
 		i += n6
 	}
-	if m.StreamAgg != nil {
+	if m.ExchangeReceiver != nil {
 		dAtA[i] = 0x42
 		i++
-		i = encodeVarintExecutor(dAtA, i, uint64(m.StreamAgg.Size()))
-		n7, err := m.StreamAgg.MarshalTo(dAtA[i:])
+		i = encodeVarintExecutor(dAtA, i, uint64(m.ExchangeReceiver.Size()))
+		n7, err := m.ExchangeReceiver.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
@@ -895,6 +1017,111 @@ func (m *Executor) MarshalTo(dAtA []byte) (int, error) {
 		}
 		i += n9
 	}
+	if m.ExchangeSender != nil {
+		dAtA[i] = 0x62
+		i++
+		i = encodeVarintExecutor(dAtA, i, uint64(m.ExchangeSender.Size()))
+		n10, err := m.ExchangeSender.MarshalTo(dAtA[i:])
+		if err != nil {
+			return 0, err
+		}
+		i += n10
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
+}
+
+func (m *ExchangeSender) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalTo(dAtA)
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ExchangeSender) MarshalTo(dAtA []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	dAtA[i] = 0x8
+	i++
+	i = encodeVarintExecutor(dAtA, i, uint64(m.Tp))
+	if len(m.EncodedTaskMeta) > 0 {
+		for _, b := range m.EncodedTaskMeta {
+			dAtA[i] = 0x12
+			i++
+			i = encodeVarintExecutor(dAtA, i, uint64(len(b)))
+			i += copy(dAtA[i:], b)
+		}
+	}
+	if len(m.PartitionKeys) > 0 {
+		for _, msg := range m.PartitionKeys {
+			dAtA[i] = 0x1a
+			i++
+			i = encodeVarintExecutor(dAtA, i, uint64(msg.Size()))
+			n, err := msg.MarshalTo(dAtA[i:])
+			if err != nil {
+				return 0, err
+			}
+			i += n
+		}
+	}
+	if m.Child != nil {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
+		n11, err := m.Child.MarshalTo(dAtA[i:])
+		if err != nil {
+			return 0, err
+		}
+		i += n11
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
+}
+
+func (m *ExchangeReceiver) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalTo(dAtA)
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ExchangeReceiver) MarshalTo(dAtA []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	if len(m.EncodedTaskMeta) > 0 {
+		for _, b := range m.EncodedTaskMeta {
+			dAtA[i] = 0xa
+			i++
+			i = encodeVarintExecutor(dAtA, i, uint64(len(b)))
+			i += copy(dAtA[i:], b)
+		}
+	}
+	if len(m.FieldTypes) > 0 {
+		for _, msg := range m.FieldTypes {
+			dAtA[i] = 0x12
+			i++
+			i = encodeVarintExecutor(dAtA, i, uint64(msg.Size()))
+			n, err := msg.MarshalTo(dAtA[i:])
+			if err != nil {
+				return 0, err
+			}
+			i += n
+		}
+	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
@@ -1200,11 +1427,11 @@ func (m *Selection) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0x1a
 		i++
 		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
-		n10, err := m.Child.MarshalTo(dAtA[i:])
+		n12, err := m.Child.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
-		i += n10
+		i += n12
 	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
@@ -1255,11 +1482,11 @@ func (m *Projection) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0x1a
 		i++
 		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
-		n11, err := m.Child.MarshalTo(dAtA[i:])
+		n13, err := m.Child.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
-		i += n11
+		i += n13
 	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
@@ -1342,11 +1569,11 @@ func (m *Aggregation) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0x32
 		i++
 		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
-		n12, err := m.Child.MarshalTo(dAtA[i:])
+		n14, err := m.Child.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
-		i += n12
+		i += n14
 	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
@@ -1388,11 +1615,11 @@ func (m *TopN) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0x1a
 		i++
 		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
-		n13, err := m.Child.MarshalTo(dAtA[i:])
+		n15, err := m.Child.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
-		i += n13
+		i += n15
 	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
@@ -1422,11 +1649,11 @@ func (m *Limit) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0x12
 		i++
 		i = encodeVarintExecutor(dAtA, i, uint64(m.Child.Size()))
-		n14, err := m.Child.MarshalTo(dAtA[i:])
+		n16, err := m.Child.MarshalTo(dAtA[i:])
 		if err != nil {
 			return 0, err
 		}
-		i += n14
+		i += n16
 	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
@@ -1545,8 +1772,8 @@ func (m *Executor) Size() (n int) {
 		l = m.Limit.Size()
 		n += 1 + l + sovExecutor(uint64(l))
 	}
-	if m.StreamAgg != nil {
-		l = m.StreamAgg.Size()
+	if m.ExchangeReceiver != nil {
+		l = m.ExchangeReceiver.Size()
 		n += 1 + l + sovExecutor(uint64(l))
 	}
 	if m.Join != nil {
@@ -1561,6 +1788,57 @@ func (m *Executor) Size() (n int) {
 		l = m.Kill.Size()
 		n += 1 + l + sovExecutor(uint64(l))
 	}
+	if m.ExchangeSender != nil {
+		l = m.ExchangeSender.Size()
+		n += 1 + l + sovExecutor(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ExchangeSender) Size() (n int) {
+	var l int
+	_ = l
+	n += 1 + sovExecutor(uint64(m.Tp))
+	if len(m.EncodedTaskMeta) > 0 {
+		for _, b := range m.EncodedTaskMeta {
+			l = len(b)
+			n += 1 + l + sovExecutor(uint64(l))
+		}
+	}
+	if len(m.PartitionKeys) > 0 {
+		for _, e := range m.PartitionKeys {
+			l = e.Size()
+			n += 1 + l + sovExecutor(uint64(l))
+		}
+	}
+	if m.Child != nil {
+		l = m.Child.Size()
+		n += 1 + l + sovExecutor(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ExchangeReceiver) Size() (n int) {
+	var l int
+	_ = l
+	if len(m.EncodedTaskMeta) > 0 {
+		for _, b := range m.EncodedTaskMeta {
+			l = len(b)
+			n += 1 + l + sovExecutor(uint64(l))
+		}
+	}
+	if len(m.FieldTypes) > 0 {
+		for _, e := range m.FieldTypes {
+			l = e.Size()
+			n += 1 + l + sovExecutor(uint64(l))
+		}
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2098,7 +2376,7 @@ func (m *Executor) Unmarshal(dAtA []byte) error {
 			iNdEx = postIndex
 		case 8:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field StreamAgg", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field ExchangeReceiver", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -2122,10 +2400,10 @@ func (m *Executor) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			if m.StreamAgg == nil {
-				m.StreamAgg = &Aggregation{}
+			if m.ExchangeReceiver == nil {
+				m.ExchangeReceiver = &ExchangeReceiver{}
 			}
-			if err := m.StreamAgg.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if err := m.ExchangeReceiver.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
@@ -2225,6 +2503,313 @@ func (m *Executor) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 12:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ExchangeSender", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.ExchangeSender == nil {
+				m.ExchangeSender = &ExchangeSender{}
+			}
+			if err := m.ExchangeSender.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipExecutor(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *ExchangeSender) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowExecutor
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ExchangeSender: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ExchangeSender: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Tp", wireType)
+			}
+			m.Tp = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Tp |= (ExchangeType(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field EncodedTaskMeta", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.EncodedTaskMeta = append(m.EncodedTaskMeta, make([]byte, postIndex-iNdEx))
+			copy(m.EncodedTaskMeta[len(m.EncodedTaskMeta)-1], dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field PartitionKeys", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.PartitionKeys = append(m.PartitionKeys, &Expr{})
+			if err := m.PartitionKeys[len(m.PartitionKeys)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Child", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Child == nil {
+				m.Child = &Executor{}
+			}
+			if err := m.Child.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipExecutor(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *ExchangeReceiver) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowExecutor
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ExchangeReceiver: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ExchangeReceiver: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field EncodedTaskMeta", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.EncodedTaskMeta = append(m.EncodedTaskMeta, make([]byte, postIndex-iNdEx))
+			copy(m.EncodedTaskMeta[len(m.EncodedTaskMeta)-1], dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field FieldTypes", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.FieldTypes = append(m.FieldTypes, &FieldType{})
+			if err := m.FieldTypes[len(m.FieldTypes)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipExecutor(dAtA[iNdEx:])
@@ -4155,87 +4740,98 @@ var (
 func init() { proto.RegisterFile("executor.proto", fileDescriptorExecutor) }
 
 var fileDescriptorExecutor = []byte{
-	// 1307 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0x4f, 0x6f, 0xdb, 0xc6,
-	0x13, 0x35, 0x25, 0x4a, 0x22, 0x47, 0xb2, 0x4c, 0xef, 0x2f, 0x09, 0x98, 0xe0, 0x57, 0x5b, 0x31,
-	0x12, 0xd4, 0x15, 0x52, 0x27, 0x75, 0xda, 0x6b, 0x91, 0x38, 0x4d, 0x5a, 0xd5, 0x81, 0x6b, 0x30,
-	0x46, 0xaf, 0x04, 0x45, 0xae, 0xe9, 0x75, 0xc8, 0x5d, 0x66, 0x49, 0x22, 0xd2, 0xad, 0x97, 0xde,
-	0x0b, 0xf4, 0xd2, 0x0f, 0xd0, 0x73, 0x6f, 0xbd, 0xf4, 0xd8, 0x53, 0x8e, 0xbd, 0xf6, 0x52, 0x14,
-	0xe9, 0x17, 0x29, 0xf6, 0x0f, 0x29, 0xca, 0xb2, 0xd3, 0x9c, 0x48, 0xbe, 0x79, 0x6f, 0x76, 0x66,
-	0x67, 0x67, 0x87, 0x30, 0xc4, 0x33, 0x1c, 0x96, 0x05, 0xe3, 0x7b, 0x19, 0x67, 0x05, 0x43, 0x66,
-	0x41, 0xb2, 0xe9, 0x2d, 0x07, 0xcf, 0x32, 0x8e, 0xf3, 0x9c, 0x30, 0xaa, 0xf0, 0x5b, 0x83, 0x3c,
-	0x3c, 0xc3, 0x69, 0xa0, 0xbf, 0xae, 0xc5, 0x2c, 0x66, 0xf2, 0xf5, 0xbe, 0x78, 0x53, 0xe8, 0xce,
-	0xef, 0x6d, 0xb0, 0x9e, 0x6a, 0x77, 0xe8, 0x0e, 0xb4, 0x8a, 0xcc, 0x35, 0x46, 0xc6, 0xee, 0x70,
-	0x7f, 0xb8, 0x27, 0xbc, 0xee, 0x09, 0xdb, 0xc9, 0x3c, 0xc3, 0x07, 0xe6, 0x9b, 0xbf, 0xb6, 0xd7,
-	0xbc, 0x56, 0x91, 0xa1, 0x31, 0x58, 0xc5, 0x34, 0xf1, 0xf3, 0x30, 0xa0, 0x6e, 0x6b, 0x64, 0xec,
-	0xf6, 0xf7, 0x37, 0x14, 0xf7, 0x24, 0x98, 0x26, 0xf8, 0x45, 0x18, 0x50, 0xaf, 0x57, 0x4c, 0x13,
-	0xf1, 0x22, 0xb8, 0x24, 0x9a, 0x29, 0x6e, 0xbb, 0xc9, 0x9d, 0xd0, 0x08, 0xcf, 0x14, 0x97, 0x44,
-	0xf2, 0x05, 0x7d, 0x0c, 0x76, 0x8e, 0x13, 0x1c, 0x16, 0x84, 0x51, 0xd7, 0x6c, 0x92, 0x5f, 0x54,
-	0xb0, 0xb7, 0x60, 0xa0, 0x87, 0xd0, 0x0f, 0xe2, 0x98, 0xe3, 0x38, 0x90, 0x82, 0x8e, 0x14, 0x6c,
-	0x2a, 0xc1, 0xe3, 0x85, 0xc1, 0x6b, 0xb2, 0xd0, 0x16, 0x98, 0x05, 0xcb, 0x8e, 0xdc, 0xae, 0x64,
-	0x83, 0x8e, 0x9b, 0x65, 0x47, 0x9e, 0xc4, 0xd1, 0x6d, 0xe8, 0x24, 0x24, 0x25, 0x85, 0xdb, 0x93,
-	0x84, 0xbe, 0x22, 0x3c, 0x17, 0x90, 0xa7, 0x2c, 0xe8, 0x01, 0x40, 0x5e, 0x70, 0x1c, 0xa4, 0x7e,
-	0x10, 0xc7, 0xae, 0x75, 0xd5, 0xb2, 0xb6, 0x22, 0x3d, 0x8e, 0x63, 0xb1, 0xe8, 0x39, 0x23, 0xd4,
-	0xb5, 0x9b, 0x8b, 0x7e, 0xcd, 0x08, 0xf5, 0x24, 0x8e, 0xb6, 0xa1, 0x5f, 0x55, 0xd4, 0x27, 0x91,
-	0x0b, 0x23, 0x63, 0xd7, 0xf6, 0xa0, 0x82, 0x26, 0x91, 0x70, 0xf0, 0x92, 0x24, 0x89, 0xdb, 0x6f,
-	0x3a, 0x38, 0x24, 0x49, 0xe2, 0x49, 0x7c, 0xe7, 0x87, 0x16, 0xd8, 0xf5, 0xe6, 0xa3, 0x6d, 0xb0,
-	0x0a, 0xf1, 0x21, 0x7c, 0x89, 0x5a, 0xb6, 0x75, 0xed, 0x7a, 0x12, 0x9d, 0x44, 0x68, 0x0c, 0xbd,
-	0x90, 0x25, 0x65, 0x4a, 0x73, 0xb7, 0x35, 0x6a, 0xef, 0xf6, 0xf7, 0x1d, 0xe5, 0xf1, 0x89, 0x04,
-	0x27, 0xf4, 0x94, 0x79, 0x15, 0x01, 0xb9, 0x60, 0x46, 0x38, 0x0f, 0x65, 0xf1, 0x2c, 0xed, 0x48,
-	0x22, 0xe8, 0x1e, 0xa0, 0x8c, 0x93, 0x34, 0xe0, 0x73, 0x5f, 0x91, 0x7d, 0x12, 0xe5, 0xae, 0x39,
-	0x6a, 0xef, 0xb6, 0x3d, 0x47, 0x5b, 0xb4, 0xc7, 0x28, 0x47, 0x8f, 0xc0, 0xa1, 0x78, 0x56, 0xf8,
-	0x1c, 0x07, 0x91, 0x8f, 0x69, 0x4c, 0x28, 0x96, 0x25, 0x1b, 0x56, 0x8b, 0x3f, 0x95, 0x58, 0xe3,
-	0xa8, 0x0d, 0x05, 0xdf, 0xc3, 0x41, 0xa4, 0x2c, 0xe8, 0x1e, 0x74, 0x79, 0x40, 0x63, 0x9c, 0xbb,
-	0x5d, 0x19, 0xb4, 0x3e, 0xa0, 0x87, 0x78, 0xee, 0x09, 0x58, 0xab, 0x34, 0x67, 0xe7, 0x57, 0x13,
-	0x4c, 0xb1, 0xc5, 0xe8, 0x13, 0xb0, 0xc5, 0x26, 0xfb, 0xc5, 0x3c, 0xc3, 0xcb, 0x47, 0x5b, 0x98,
-	0x1b, 0xeb, 0x59, 0xe7, 0xfa, 0x1b, 0x7d, 0x0e, 0x43, 0x29, 0x11, 0x15, 0x50, 0xba, 0x96, 0xd4,
-	0xa1, 0x85, 0xee, 0x42, 0x5b, 0x0c, 0xce, 0x1b, 0x98, 0x38, 0xf4, 0xe1, 0x19, 0x49, 0x22, 0x8e,
-	0xc5, 0xa1, 0x6f, 0xc4, 0x5a, 0x35, 0x9a, 0x57, 0xdb, 0xd1, 0x6d, 0xb0, 0x09, 0xa5, 0x58, 0x14,
-	0x7e, 0x26, 0x0f, 0x7d, 0x55, 0x2d, 0x4b, 0xc2, 0x93, 0x68, 0x86, 0x1e, 0xc0, 0x30, 0xc1, 0xa7,
-	0x85, 0x2f, 0x63, 0x7a, 0x89, 0xe7, 0xb9, 0xdb, 0x91, 0x4e, 0xa1, 0x72, 0x9a, 0x71, 0x6f, 0x20,
-	0x18, 0x22, 0xb0, 0x43, 0x3c, 0xcf, 0xd1, 0x3e, 0x6c, 0x70, 0x12, 0x9f, 0x35, 0x25, 0xdd, 0x15,
-	0xc9, 0xba, 0xa4, 0xd4, 0x9a, 0x07, 0xd0, 0xcf, 0x38, 0x9b, 0x62, 0x99, 0x70, 0xee, 0xf6, 0x24,
-	0x5f, 0xf7, 0xdf, 0x33, 0x82, 0x93, 0x48, 0xa4, 0xe6, 0x81, 0xe4, 0x88, 0x57, 0xa9, 0x98, 0x96,
-	0x24, 0x89, 0xb4, 0xc2, 0xba, 0x42, 0x21, 0x39, 0x4a, 0xf1, 0x10, 0x36, 0x64, 0x26, 0x21, 0xa3,
-	0x11, 0x11, 0x5d, 0x92, 0xbb, 0xf6, 0x4a, 0x5c, 0x32, 0xd9, 0x27, 0x35, 0x03, 0x7d, 0x06, 0x8e,
-	0x4a, 0xa6, 0xa1, 0x82, 0x15, 0x95, 0x4a, 0x78, 0x59, 0xc6, 0x8a, 0x33, 0xcc, 0x9b, 0xb2, 0xfe,
-	0xaa, 0x4c, 0x72, 0x16, 0xb2, 0x9d, 0x3f, 0x0d, 0xb0, 0xeb, 0xbb, 0xe9, 0xbf, 0x5b, 0x69, 0x1b,
-	0x2c, 0x22, 0xd8, 0x82, 0xd0, 0x6a, 0x12, 0x24, 0xba, 0xdc, 0x6b, 0xed, 0xf7, 0xed, 0x35, 0x73,
-	0xa5, 0xd7, 0x6e, 0x40, 0xb7, 0xa4, 0xe4, 0x55, 0xa9, 0x7a, 0xc6, 0xf2, 0xf4, 0xd7, 0x15, 0x3d,
-	0xd8, 0xbd, 0xbc, 0x07, 0x77, 0x7e, 0x34, 0xc0, 0xae, 0xaf, 0x52, 0x34, 0x06, 0x68, 0x6c, 0x8d,
-	0xb1, 0xb2, 0x35, 0x0d, 0x2b, 0xfa, 0x14, 0x86, 0x3c, 0xa3, 0xcd, 0xad, 0x54, 0x17, 0xc7, 0xba,
-	0xe2, 0x7b, 0x19, 0xd5, 0x47, 0x2a, 0xa3, 0x8d, 0x12, 0xdc, 0x81, 0x8e, 0x3c, 0xe7, 0xfa, 0xe6,
-	0xbf, 0xd8, 0x04, 0xca, 0xb8, 0xf3, 0x9d, 0x01, 0x70, 0xcc, 0xd9, 0xb9, 0x0e, 0x6b, 0x04, 0x1d,
-	0x31, 0xc8, 0x2e, 0x8b, 0x48, 0x19, 0xd0, 0x18, 0x6c, 0x11, 0x8c, 0x62, 0x5d, 0x1a, 0x87, 0xc5,
-	0xd5, 0xcb, 0xfb, 0x86, 0xf0, 0x7d, 0x0b, 0xfa, 0x8d, 0xbb, 0x1b, 0xdd, 0x05, 0x2b, 0xe6, 0xac,
-	0xcc, 0xfc, 0xe9, 0xfc, 0x92, 0x30, 0x7a, 0xd2, 0x76, 0x30, 0x17, 0xb4, 0x20, 0x8e, 0xfd, 0xd3,
-	0x92, 0x86, 0x3a, 0x8e, 0x25, 0x5a, 0x10, 0xc7, 0xcf, 0x4a, 0x1a, 0xa2, 0x11, 0x58, 0x6a, 0x16,
-	0xe0, 0x68, 0xe9, 0x1a, 0xad, 0x51, 0x74, 0x1f, 0x06, 0x22, 0xa3, 0x7a, 0x4d, 0xf3, 0xb2, 0xa4,
-	0x80, 0x67, 0xf4, 0x4b, 0xbd, 0xb2, 0x16, 0xd4, 0xab, 0x77, 0xae, 0x12, 0x3c, 0xd6, 0x31, 0xd4,
-	0xfb, 0xd0, 0x7d, 0xd7, 0x3e, 0xbc, 0x02, 0x53, 0xcc, 0x42, 0xf4, 0x21, 0x58, 0x8c, 0x47, 0x98,
-	0x2f, 0xf2, 0x1f, 0x28, 0xc1, 0xc1, 0x7c, 0x52, 0xe0, 0xd4, 0xeb, 0x49, 0xeb, 0xc1, 0x1c, 0xdd,
-	0xaa, 0xc6, 0xa5, 0x38, 0xfb, 0xa6, 0xce, 0x4b, 0xcf, 0xc9, 0xf7, 0xdb, 0xfa, 0x09, 0x74, 0xe4,
-	0x74, 0x5d, 0xb8, 0x32, 0xde, 0xe1, 0xaa, 0xf5, 0x2e, 0x57, 0x8f, 0xc0, 0x14, 0x33, 0x11, 0xfd,
-	0x1f, 0xba, 0x21, 0xa3, 0x74, 0xf2, 0xc5, 0x92, 0x2b, 0x8d, 0x89, 0x75, 0x5e, 0x95, 0x98, 0xcf,
-	0xa5, 0xaf, 0xaa, 0x14, 0x0a, 0xda, 0xf9, 0xcd, 0x00, 0xb7, 0xf2, 0xaa, 0x9e, 0x84, 0xd1, 0x17,
-	0x65, 0x2a, 0xda, 0x08, 0x8d, 0x61, 0xb3, 0x20, 0x29, 0xf6, 0x33, 0xce, 0x42, 0x9c, 0xe7, 0x38,
-	0xf2, 0x65, 0xdb, 0x18, 0xbb, 0xa6, 0xb7, 0x21, 0x0c, 0xc7, 0x15, 0x7e, 0x24, 0x8e, 0xe8, 0x26,
-	0x2d, 0x53, 0x41, 0x8d, 0xca, 0x10, 0x47, 0x3e, 0x67, 0xaf, 0x73, 0xb5, 0x47, 0xde, 0x06, 0x2d,
-	0xd3, 0x63, 0x8d, 0x7b, 0xec, 0x75, 0x8e, 0xee, 0xc2, 0x50, 0x70, 0x49, 0x81, 0x79, 0xa0, 0x7a,
-	0xab, 0x2d, 0x89, 0xeb, 0xb4, 0x4c, 0x27, 0x35, 0x78, 0xf1, 0x27, 0xc1, 0xbc, 0xf8, 0x93, 0x30,
-	0xfe, 0xd9, 0x50, 0x7f, 0x72, 0x72, 0x04, 0x6d, 0xc2, 0xba, 0x78, 0xd6, 0x3f, 0x05, 0xce, 0x5a,
-	0x05, 0xd5, 0x97, 0x9b, 0x63, 0x54, 0x50, 0x7d, 0x27, 0x38, 0x2d, 0xf4, 0x3f, 0xd8, 0x10, 0x50,
-	0xa3, 0x1b, 0x9c, 0x36, 0x1a, 0x80, 0x25, 0xbd, 0xb1, 0xec, 0xc8, 0x31, 0xd1, 0x3a, 0xd8, 0xe2,
-	0x4b, 0x96, 0xcd, 0xe9, 0xd4, 0x4e, 0xaa, 0xdf, 0x1d, 0xa7, 0x5b, 0xf1, 0xc5, 0x6c, 0x71, 0x7a,
-	0xd5, 0x97, 0xa8, 0x8d, 0x63, 0x8d, 0xf7, 0x00, 0x16, 0xa3, 0x1e, 0xd9, 0xd0, 0x79, 0xce, 0xc2,
-	0x20, 0x71, 0xd6, 0x90, 0x05, 0xe6, 0x09, 0x39, 0xfc, 0xd6, 0x31, 0x50, 0x1f, 0x7a, 0x27, 0xe4,
-	0x59, 0x12, 0xe4, 0x67, 0x4e, 0x6b, 0xfc, 0x8b, 0x01, 0x56, 0x35, 0xa9, 0x17, 0x39, 0x50, 0xcc,
-	0xa5, 0xf7, 0x35, 0x74, 0x1d, 0x36, 0x65, 0x34, 0xf8, 0xb4, 0xf8, 0xa6, 0x2c, 0x34, 0x6c, 0xa0,
-	0x1b, 0x80, 0xe4, 0xf8, 0x11, 0x53, 0x61, 0x81, 0xb7, 0x90, 0x03, 0x03, 0x95, 0x72, 0x4a, 0x24,
-	0xd2, 0x46, 0xd7, 0xc0, 0x91, 0x19, 0xd3, 0x82, 0xd4, 0xa8, 0x89, 0x6e, 0xc2, 0xf5, 0x25, 0xb7,
-	0xb5, 0xa9, 0x83, 0x3e, 0x80, 0x9b, 0x95, 0x60, 0xd5, 0xdc, 0x1d, 0x8f, 0x60, 0xd0, 0xfc, 0x43,
-	0xa8, 0x56, 0xfc, 0x2a, 0xc8, 0xcf, 0x54, 0xc8, 0x07, 0x1f, 0xbd, 0x79, 0xbb, 0x65, 0xfc, 0xf1,
-	0x76, 0xcb, 0xf8, 0xfb, 0xed, 0x96, 0xf1, 0xd3, 0x3f, 0x5b, 0x6b, 0x70, 0x3d, 0x64, 0xe9, 0x5e,
-	0x46, 0x68, 0x1c, 0x06, 0xd9, 0x5e, 0x41, 0xa2, 0xa9, 0x3c, 0xe1, 0xc7, 0xc6, 0xbf, 0x01, 0x00,
-	0x00, 0xff, 0xff, 0x30, 0x63, 0x3f, 0x17, 0xeb, 0x0b, 0x00, 0x00,
+	// 1484 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xcf, 0x73, 0xdb, 0xc4,
+	0x17, 0x8f, 0x6c, 0xd9, 0x96, 0x9f, 0x1d, 0x5b, 0xd9, 0x6f, 0xda, 0x51, 0x3b, 0x5f, 0x12, 0xd7,
+	0xd3, 0x0e, 0xc1, 0x53, 0xd2, 0x5f, 0x30, 0xc3, 0x05, 0xa6, 0x4d, 0x69, 0xc1, 0xa4, 0x84, 0x8c,
+	0x92, 0xe1, 0xaa, 0x91, 0xa5, 0x8d, 0xbc, 0x89, 0xb4, 0xab, 0xae, 0x24, 0x6a, 0xdf, 0xb8, 0x70,
+	0x67, 0x06, 0x0e, 0xfc, 0x15, 0xdc, 0xb8, 0x70, 0xe1, 0xda, 0x23, 0x27, 0x66, 0xb8, 0x30, 0x4c,
+	0xf9, 0x47, 0x98, 0xdd, 0x95, 0x64, 0x39, 0x4e, 0x42, 0x4f, 0xd2, 0x7e, 0xde, 0xe7, 0xbd, 0x7d,
+	0xef, 0xed, 0x7b, 0xbb, 0x0f, 0x7a, 0x78, 0x86, 0xbd, 0x2c, 0x65, 0x7c, 0x37, 0xe6, 0x2c, 0x65,
+	0x48, 0x4f, 0x49, 0x3c, 0xb9, 0x69, 0xe2, 0x59, 0xcc, 0x71, 0x92, 0x10, 0x46, 0x15, 0x7e, 0xb3,
+	0x9b, 0x78, 0x53, 0x1c, 0xb9, 0xf9, 0x6a, 0x33, 0x60, 0x01, 0x93, 0xbf, 0xf7, 0xc4, 0x9f, 0x42,
+	0x87, 0x3f, 0xea, 0x60, 0x3c, 0xcb, 0xcd, 0xa1, 0xdb, 0x50, 0x4b, 0x63, 0x4b, 0x1b, 0x68, 0x3b,
+	0xbd, 0x87, 0xbd, 0x5d, 0x61, 0x75, 0x57, 0xc8, 0x8e, 0xe7, 0x31, 0xde, 0xd3, 0x5f, 0xff, 0xb5,
+	0xbd, 0x66, 0xd7, 0xd2, 0x18, 0x8d, 0xc0, 0x48, 0x27, 0xa1, 0x93, 0x78, 0x2e, 0xb5, 0x6a, 0x03,
+	0x6d, 0xa7, 0xf3, 0xb0, 0xaf, 0xb8, 0xc7, 0xee, 0x24, 0xc4, 0x47, 0x9e, 0x4b, 0xed, 0x56, 0x3a,
+	0x09, 0xc5, 0x8f, 0xe0, 0x12, 0x7f, 0xa6, 0xb8, 0xf5, 0x2a, 0x77, 0x4c, 0x7d, 0x3c, 0x53, 0x5c,
+	0xe2, 0xcb, 0x1f, 0xf4, 0x3e, 0xb4, 0x13, 0x1c, 0x62, 0x2f, 0x25, 0x8c, 0x5a, 0x7a, 0x95, 0x7c,
+	0x54, 0xc0, 0xf6, 0x82, 0x81, 0x1e, 0x41, 0xc7, 0x0d, 0x02, 0x8e, 0x03, 0x57, 0x2a, 0x34, 0xa4,
+	0xc2, 0x86, 0x52, 0x78, 0xb2, 0x10, 0xd8, 0x55, 0x16, 0xda, 0x02, 0x3d, 0x65, 0xf1, 0x81, 0xd5,
+	0x94, 0x6c, 0xc8, 0xfd, 0x66, 0xf1, 0x81, 0x2d, 0x71, 0x74, 0x0b, 0x1a, 0x21, 0x89, 0x48, 0x6a,
+	0xb5, 0x24, 0xa1, 0xa3, 0x08, 0x2f, 0x04, 0x64, 0x2b, 0x09, 0x7a, 0x0a, 0x1b, 0x78, 0xe6, 0x4d,
+	0x5d, 0x1a, 0x60, 0x87, 0x63, 0x0f, 0x93, 0x6f, 0x30, 0xb7, 0x0c, 0x49, 0xbf, 0x5e, 0xe4, 0x4c,
+	0x89, 0xed, 0x5c, 0x6a, 0x9b, 0xf8, 0x1c, 0x22, 0xfc, 0x38, 0x65, 0x84, 0x5a, 0xed, 0xaa, 0x1f,
+	0x5f, 0x30, 0x42, 0x6d, 0x89, 0xa3, 0x6d, 0xe8, 0x14, 0x87, 0xec, 0x10, 0xdf, 0x82, 0x81, 0xb6,
+	0xd3, 0xb6, 0xa1, 0x80, 0xc6, 0xbe, 0x30, 0x70, 0x46, 0xc2, 0xd0, 0xea, 0x54, 0x0d, 0xec, 0x93,
+	0x30, 0xb4, 0x25, 0x8e, 0x3e, 0x86, 0x7e, 0xe9, 0x65, 0x82, 0xa9, 0x8f, 0xb9, 0xd5, 0x95, 0xd4,
+	0xcd, 0x65, 0x1f, 0x8f, 0xa4, 0xcc, 0xee, 0xe1, 0xa5, 0xf5, 0xf0, 0x37, 0x0d, 0x7a, 0xcb, 0x14,
+	0xb4, 0x53, 0x29, 0x0e, 0xb4, 0x6c, 0x64, 0xa5, 0x40, 0x36, 0x30, 0xf5, 0x98, 0x8f, 0x7d, 0x27,
+	0x75, 0x93, 0x33, 0x27, 0xc2, 0xa9, 0x6b, 0xd5, 0x06, 0xf5, 0x9d, 0xae, 0xdd, 0xcf, 0x05, 0xc7,
+	0x6e, 0x72, 0xf6, 0x25, 0x4e, 0x5d, 0xf4, 0x00, 0x7a, 0xb1, 0xcb, 0x53, 0x22, 0x4e, 0xc7, 0x39,
+	0xc3, 0xf3, 0xc4, 0xaa, 0x0f, 0xea, 0x8b, 0x88, 0x9e, 0xcd, 0x62, 0x6e, 0xaf, 0x97, 0x8c, 0x7d,
+	0x3c, 0x4f, 0xd0, 0x6d, 0x68, 0x78, 0x53, 0x12, 0xfa, 0x79, 0x8d, 0x54, 0x0a, 0x55, 0xe4, 0xc6,
+	0x56, 0xc2, 0x61, 0x0c, 0xe6, 0xf9, 0x73, 0xb8, 0xd8, 0x31, 0xed, 0x62, 0xc7, 0xee, 0x43, 0xe7,
+	0x84, 0xe0, 0xd0, 0x77, 0xd2, 0x79, 0x8c, 0x13, 0xe9, 0x7e, 0x59, 0x8f, 0xcf, 0x85, 0x40, 0x04,
+	0x6d, 0xc3, 0x49, 0xf1, 0x9b, 0x0c, 0xbf, 0xaf, 0x41, 0xbb, 0x6c, 0x01, 0xb4, 0x0d, 0x46, 0x2a,
+	0x16, 0xe2, 0xf8, 0x44, 0xd2, 0xea, 0x79, 0x82, 0x5a, 0x12, 0x1d, 0xfb, 0x68, 0x04, 0x2d, 0x8f,
+	0x85, 0x59, 0x44, 0x0b, 0xe3, 0xa6, 0x32, 0xfe, 0x54, 0x82, 0x63, 0x7a, 0xc2, 0xec, 0x82, 0x80,
+	0x2c, 0xd0, 0x7d, 0x9c, 0x78, 0xb2, 0x85, 0x8c, 0xdc, 0x90, 0x44, 0xd0, 0x5d, 0x40, 0x31, 0x27,
+	0x91, 0xcb, 0xe7, 0x8e, 0x22, 0x3b, 0xc4, 0x4f, 0x2c, 0x7d, 0x50, 0xdf, 0xa9, 0xdb, 0x66, 0x2e,
+	0xc9, 0x2d, 0xfa, 0x09, 0x7a, 0x0c, 0x26, 0xc5, 0xb3, 0xd4, 0xe1, 0xd8, 0xf5, 0x1d, 0x4c, 0x03,
+	0x42, 0xb1, 0x6c, 0x9c, 0x5e, 0xb1, 0xf9, 0x33, 0x89, 0x55, 0xce, 0xb3, 0x27, 0xf8, 0x36, 0x76,
+	0x7d, 0x25, 0x41, 0x77, 0xa1, 0xc9, 0x45, 0x4e, 0x13, 0xab, 0x29, 0x9d, 0xce, 0xb3, 0xbf, 0x8f,
+	0xe7, 0xb6, 0x80, 0x73, 0xad, 0x9c, 0x33, 0xfc, 0x45, 0x07, 0x5d, 0x54, 0x35, 0x7a, 0x00, 0x6d,
+	0x51, 0xd7, 0x32, 0x99, 0xcb, 0x17, 0x8c, 0x10, 0x57, 0xf6, 0x33, 0x4e, 0xf3, 0x35, 0xfa, 0x04,
+	0x7a, 0x52, 0x45, 0x14, 0xbd, 0xd2, 0xab, 0x55, 0x6b, 0x4f, 0xe8, 0x9d, 0xbb, 0x9c, 0xba, 0xa7,
+	0x15, 0x4c, 0x5c, 0x3d, 0xb2, 0x12, 0x38, 0xa6, 0x79, 0x4d, 0x9d, 0xaf, 0x94, 0x52, 0x8e, 0x6e,
+	0x41, 0x9b, 0x50, 0x8a, 0x45, 0xaf, 0xcd, 0x64, 0x59, 0x15, 0xa7, 0x65, 0x48, 0x78, 0xec, 0xcf,
+	0xd0, 0x7d, 0xe8, 0x85, 0xf8, 0x24, 0x75, 0xa4, 0x4f, 0xb2, 0x50, 0x1b, 0x2b, 0x85, 0xda, 0x15,
+	0x0c, 0xe1, 0x98, 0xac, 0xd3, 0x87, 0xd0, 0xe7, 0x24, 0x98, 0x56, 0x55, 0x9a, 0xab, 0xb5, 0x2d,
+	0x29, 0xa5, 0xce, 0x7d, 0xe8, 0xc4, 0x9c, 0x4d, 0x70, 0x5e, 0x75, 0xad, 0x4b, 0xaa, 0x4e, 0x72,
+	0x64, 0xd5, 0x09, 0x8d, 0x49, 0x46, 0xca, 0x3a, 0x35, 0x2e, 0xd1, 0x90, 0x1c, 0xa5, 0xf1, 0x08,
+	0xfa, 0x32, 0x12, 0x8f, 0x51, 0x5f, 0x76, 0x55, 0x62, 0xb5, 0x57, 0xfc, 0x92, 0xc1, 0x3e, 0x2d,
+	0x19, 0xe8, 0x43, 0x30, 0x55, 0x30, 0x15, 0x2d, 0x58, 0xd1, 0x52, 0x01, 0x2f, 0xab, 0xb1, 0x74,
+	0x8a, 0x79, 0x55, 0xad, 0xb3, 0xaa, 0x26, 0x39, 0x0b, 0xb5, 0xe1, 0x9f, 0x1a, 0xb4, 0xcb, 0x17,
+	0xe2, 0xbf, 0x5b, 0x69, 0x1b, 0x0c, 0x22, 0xd8, 0x82, 0x50, 0xab, 0x12, 0x24, 0xba, 0xdc, 0x6b,
+	0xf5, 0xb7, 0xed, 0x35, 0x7d, 0xa5, 0xd7, 0xae, 0x43, 0x33, 0xa3, 0xe4, 0x65, 0xa6, 0x7a, 0xc6,
+	0xb0, 0xf3, 0xd5, 0x25, 0x3d, 0xd8, 0xbc, 0xb8, 0x07, 0x87, 0x3f, 0x68, 0xd0, 0x2e, 0x1f, 0x34,
+	0x34, 0x02, 0xa8, 0xa4, 0x46, 0x5b, 0x49, 0x4d, 0x45, 0x8a, 0x3e, 0x80, 0x1e, 0x8f, 0x69, 0x35,
+	0x95, 0xea, 0xe2, 0x58, 0x57, 0x7c, 0x3b, 0xa6, 0x79, 0x49, 0xc5, 0xb4, 0x72, 0x04, 0xe5, 0x75,
+	0x59, 0xbf, 0xea, 0xba, 0xfc, 0x56, 0x03, 0x38, 0xe4, 0xec, 0x34, 0x77, 0x6b, 0x00, 0x0d, 0x31,
+	0x4e, 0x5c, 0xe4, 0x91, 0x12, 0xa0, 0x11, 0xb4, 0x85, 0x33, 0x8a, 0x75, 0xa1, 0x1f, 0x06, 0x57,
+	0x3f, 0x6f, 0xeb, 0xc2, 0x77, 0x35, 0xe8, 0x54, 0x1e, 0x6e, 0x74, 0x07, 0x8c, 0x80, 0xb3, 0x2c,
+	0x76, 0x26, 0xf3, 0x0b, 0xdc, 0x68, 0x49, 0xd9, 0xde, 0x5c, 0xd0, 0xdc, 0x20, 0x70, 0x4e, 0x32,
+	0xea, 0xe5, 0x7e, 0x2c, 0xd1, 0xdc, 0x20, 0x78, 0x9e, 0x51, 0x0f, 0x0d, 0xc0, 0x48, 0x52, 0x8e,
+	0xdd, 0x08, 0xfb, 0x4b, 0xd7, 0x68, 0x89, 0xa2, 0x7b, 0xd0, 0x15, 0x11, 0x95, 0x7b, 0xea, 0x17,
+	0x05, 0x05, 0x3c, 0xa6, 0x9f, 0xe5, 0x3b, 0xe7, 0x0a, 0xe5, 0xee, 0x8d, 0xcb, 0x14, 0x9e, 0xe4,
+	0x3e, 0x94, 0x79, 0x68, 0x5e, 0x95, 0x87, 0x97, 0xa0, 0x8b, 0x89, 0x04, 0xbd, 0x0b, 0x06, 0xe3,
+	0x3e, 0xe6, 0x8b, 0xf8, 0xbb, 0x4a, 0x61, 0x6f, 0x3e, 0x4e, 0x71, 0x64, 0xb7, 0xa4, 0x74, 0x6f,
+	0x8e, 0x6e, 0x16, 0x43, 0x8b, 0xa8, 0x7d, 0x3d, 0x8f, 0x2b, 0x9f, 0x56, 0xde, 0x2e, 0xf5, 0x63,
+	0x68, 0xc8, 0x19, 0x67, 0x61, 0x4a, 0xbb, 0xc2, 0x54, 0xed, 0x2a, 0x53, 0x8f, 0x41, 0x17, 0x63,
+	0x08, 0xfa, 0x3f, 0x34, 0x3d, 0x46, 0xe9, 0xf8, 0xd3, 0x25, 0x53, 0x39, 0x26, 0xf6, 0x79, 0x99,
+	0x61, 0x3e, 0x97, 0xb6, 0x8a, 0xa3, 0x50, 0xd0, 0xf0, 0x57, 0x0d, 0xac, 0xc2, 0xaa, 0xfa, 0x12,
+	0x46, 0x8f, 0xb2, 0x48, 0xb4, 0x91, 0x78, 0xc2, 0x53, 0x12, 0x61, 0x27, 0xe6, 0xcc, 0xc3, 0x49,
+	0x82, 0x7d, 0x47, 0xb6, 0x8d, 0xb6, 0xa3, 0xdb, 0x7d, 0x21, 0x38, 0x2c, 0xf0, 0x03, 0x51, 0xa2,
+	0x1b, 0x34, 0x8b, 0x04, 0xd5, 0xcf, 0x3c, 0xec, 0x3b, 0x9c, 0xbd, 0x4a, 0x54, 0x8e, 0xec, 0x3e,
+	0xcd, 0xa2, 0xc3, 0x1c, 0xb7, 0xd9, 0xab, 0x04, 0xdd, 0x81, 0x9e, 0xe0, 0x92, 0x14, 0x73, 0x57,
+	0xf5, 0x56, 0x5d, 0x12, 0xd7, 0x69, 0x16, 0x8d, 0x4b, 0xf0, 0xfc, 0x5c, 0xa6, 0x9f, 0x9f, 0xcb,
+	0x46, 0x7f, 0x68, 0x6a, 0x9e, 0x96, 0x4f, 0xd0, 0x06, 0xac, 0x8b, 0x6f, 0x39, 0x14, 0x98, 0x6b,
+	0x05, 0x54, 0x5e, 0x6e, 0xa6, 0x56, 0x40, 0xe5, 0x9d, 0x60, 0xd6, 0xd0, 0xff, 0xa0, 0x2f, 0xa0,
+	0x4a, 0x37, 0x98, 0x75, 0xd4, 0x05, 0x43, 0x5a, 0x63, 0xf1, 0x81, 0xa9, 0xa3, 0x75, 0x68, 0x8b,
+	0x95, 0x3c, 0x36, 0xb3, 0x51, 0x1a, 0x91, 0xc5, 0xfc, 0x24, 0x08, 0xcc, 0x66, 0xc1, 0x17, 0x6f,
+	0x8b, 0xd9, 0x2a, 0x56, 0xe2, 0x6c, 0x4c, 0x03, 0x5d, 0x07, 0x24, 0x56, 0xcb, 0x23, 0x9e, 0xd9,
+	0x46, 0x16, 0x6c, 0x56, 0xf1, 0x62, 0x72, 0x32, 0x61, 0xf4, 0x11, 0x74, 0xab, 0xe3, 0x1e, 0xea,
+	0x43, 0xe7, 0xd0, 0x4d, 0x92, 0xe3, 0x29, 0x67, 0x59, 0x30, 0x35, 0xd7, 0x84, 0x43, 0x7b, 0x9c,
+	0xb9, 0xbe, 0xe7, 0x26, 0xa9, 0xa9, 0x21, 0x03, 0xf4, 0xcf, 0xdd, 0x64, 0x6a, 0xd6, 0x46, 0xbb,
+	0x00, 0x8b, 0xb1, 0x02, 0xb5, 0xa1, 0xf1, 0x82, 0x79, 0x6e, 0x68, 0xae, 0x09, 0xca, 0x31, 0xd9,
+	0xff, 0xda, 0xd4, 0x50, 0x07, 0x5a, 0xc7, 0xe4, 0x79, 0xa8, 0xf8, 0x3f, 0x6b, 0x60, 0x14, 0x53,
+	0xc1, 0x22, 0x5f, 0x14, 0x73, 0x19, 0xc9, 0x1a, 0xba, 0x06, 0x1b, 0x32, 0x72, 0x7c, 0x92, 0x7e,
+	0x95, 0xa5, 0x39, 0xac, 0x15, 0x21, 0xd9, 0xe2, 0x05, 0x5a, 0xe0, 0x35, 0x64, 0x42, 0x57, 0xa5,
+	0x37, 0x22, 0x12, 0xa9, 0xa3, 0x4d, 0x30, 0x65, 0x76, 0x69, 0x4a, 0x4a, 0x54, 0x47, 0x37, 0xe0,
+	0xda, 0x92, 0xd9, 0x52, 0xd4, 0x40, 0xef, 0xc0, 0x8d, 0x42, 0x61, 0x55, 0xdc, 0x1c, 0x0d, 0xa0,
+	0x5b, 0x9d, 0x46, 0x8a, 0x1d, 0x45, 0xf8, 0xca, 0xe5, 0xbd, 0xf7, 0x5e, 0xbf, 0xd9, 0xd2, 0x7e,
+	0x7f, 0xb3, 0xa5, 0xfd, 0xfd, 0x66, 0x4b, 0xfb, 0xe9, 0x9f, 0xad, 0x35, 0xb8, 0xe6, 0xb1, 0x68,
+	0x37, 0x26, 0x34, 0xf0, 0xdc, 0x78, 0x37, 0x25, 0xfe, 0x44, 0x76, 0xd3, 0xa1, 0xf6, 0x6f, 0x00,
+	0x00, 0x00, 0xff, 0xff, 0x82, 0xc7, 0x48, 0xab, 0xdd, 0x0d, 0x00, 0x00,
 }
diff --git a/proto/executor.proto b/proto/executor.proto
index 8f0acbf42..df85d704b 100644
--- a/proto/executor.proto
+++ b/proto/executor.proto
@@ -23,6 +23,8 @@ enum ExecType {
 	TypeStreamAgg = 6;
 	TypeJoin = 7;
 	TypeKill = 8;
+	TypeExchangeSender = 9;
+	TypeExchangeReceiver = 10;
 }
 
 // It represents a Executor.
@@ -34,10 +36,31 @@ message Executor {
 	optional Aggregation aggregation = 5;
 	optional TopN topN = 6;
 	optional Limit limit = 7;
-	optional Aggregation stream_agg= 8;
+	optional ExchangeReceiver exchange_receiver = 8;
 	optional Join join = 9;
 	optional string executor_id = 10;
 	optional Kill kill = 11;
+	optional ExchangeSender exchange_sender = 12;
+}
+
+enum ExchangeType {
+	PassThrough = 0;
+	Broadcast = 1;
+        Hash = 2;
+}
+
+// ExchangeSender will build connection with ExchangeReceiver.
+message ExchangeSender {
+	optional ExchangeType tp = 1 [(gogoproto.nullable) = false];
+	repeated bytes encoded_task_meta = 2;
+	repeated Expr partition_keys = 3; 
+	optional Executor child = 4;
+}
+
+// ExchangeReceiver accept connection and receiver data from ExchangeSender.
+message ExchangeReceiver {
+	repeated bytes encoded_task_meta = 1;
+	repeated FieldType field_types = 2;
 }
 
 enum EngineType {