dataplanev1

type CoordEnvelope ¶

type CoordEnvelope struct {

	// Types that are valid to be assigned to Msg:
	//
	//	*CoordEnvelope_Welcome
	//	*CoordEnvelope_TaskDispatch
	Msg isCoordEnvelope_Msg `protobuf_oneof:"msg"`
	// contains filtered or unexported fields
}

CoordEnvelope is the coord→worker direction. Phase C adds TaskDispatch (the load-bearing change in the migration); CancelTask stays on NATS pub/sub in Phase C and may move later.

func (*CoordEnvelope) Descriptor() ([]byte, []int)

func (x *CoordEnvelope) GetMsg() isCoordEnvelope_Msg

func (x *CoordEnvelope) GetTaskDispatch() *TaskDispatch

func (x *CoordEnvelope) GetWelcome() *Welcome

func (*CoordEnvelope) ProtoMessage()

func (x *CoordEnvelope) ProtoReflect() protoreflect.Message

func (x *CoordEnvelope) Reset()

func (x *CoordEnvelope) String() string

type CoordEnvelope_TaskDispatch ¶

type CoordEnvelope_TaskDispatch struct {
	TaskDispatch *TaskDispatch `protobuf:"bytes,2,opt,name=task_dispatch,json=taskDispatch,proto3,oneof"` // reserved: CancelTask. See phase D+ design.
}

type CoordEnvelope_Welcome ¶

type CoordEnvelope_Welcome struct {
	Welcome *Welcome `protobuf:"bytes,1,opt,name=welcome,proto3,oneof"`
}

type DataPlaneClient ¶

type DataPlaneClient interface {
	// Worker opens a long-lived bidi stream to the coordinator. The first
	// message in each direction is Hello/Welcome; subsequent messages
	// multiplex task dispatch, results, progress, and completion over
	// the same stream.
	Connect(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WorkerEnvelope, CoordEnvelope], error)
}

DataPlaneClient is the client API for DataPlane service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.

DataPlane is the worker↔coord transport for high-rate operational traffic that NATS JetStream is a poor fit for at scale: task dispatch, result batches, gather payloads, and per-task progress.

Out of scope for this service (stays on NATS): heartbeats, query cancellation broadcast, catalog/UDF KV, DLQ. See project_split_plane_design_2026-05-20.md for the full design and rationale.

Phase A defines the connection bootstrap only (Hello/Welcome). The streaming RPCs that carry tasks and results are added in Phases B–E.

func NewDataPlaneClient(cc grpc.ClientConnInterface) DataPlaneClient

type DataPlaneServer ¶

type DataPlaneServer interface {
	// Worker opens a long-lived bidi stream to the coordinator. The first
	// message in each direction is Hello/Welcome; subsequent messages
	// multiplex task dispatch, results, progress, and completion over
	// the same stream.
	Connect(grpc.BidiStreamingServer[WorkerEnvelope, CoordEnvelope]) error
	// contains filtered or unexported methods
}

DataPlaneServer is the server API for DataPlane service. All implementations must embed UnimplementedDataPlaneServer for forward compatibility.

DataPlane is the worker↔coord transport for high-rate operational traffic that NATS JetStream is a poor fit for at scale: task dispatch, result batches, gather payloads, and per-task progress.

Out of scope for this service (stays on NATS): heartbeats, query cancellation broadcast, catalog/UDF KV, DLQ. See project_split_plane_design_2026-05-20.md for the full design and rationale.

Phase A defines the connection bootstrap only (Hello/Welcome). The streaming RPCs that carry tasks and results are added in Phases B–E.

type DataPlane_ConnectClient ¶

type DataPlane_ConnectClient = grpc.BidiStreamingClient[WorkerEnvelope, CoordEnvelope]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type DataPlane_ConnectServer ¶

type DataPlane_ConnectServer = grpc.BidiStreamingServer[WorkerEnvelope, CoordEnvelope]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type FetchShuffleRequest ¶

type FetchShuffleRequest struct {

	// query_id scopes the lookup — the same scoping LocalStageCache uses.
	QueryId string `protobuf:"bytes,1,opt,name=query_id,json=queryId,proto3" json:"query_id,omitempty"`
	// key is the file's S3 object key: the canonical identity shared by the
	// durable copy, the producer's local cache entry, and this fetch.
	Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
	// token is the per-query fetch token minted by the coordinator and
	// distributed via task specs to producers and consumers alike.
	Token string `protobuf:"bytes,3,opt,name=token,proto3" json:"token,omitempty"`
	// contains filtered or unexported fields
}

func (*FetchShuffleRequest) Descriptor() ([]byte, []int)

func (x *FetchShuffleRequest) GetKey() string

func (x *FetchShuffleRequest) GetQueryId() string

func (x *FetchShuffleRequest) GetToken() string

func (*FetchShuffleRequest) ProtoMessage()

func (x *FetchShuffleRequest) ProtoReflect() protoreflect.Message

func (x *FetchShuffleRequest) Reset()

func (x *FetchShuffleRequest) String() string

type Hello ¶

type Hello struct {
	WorkerId   string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"`
	BuildSha   string `protobuf:"bytes,2,opt,name=build_sha,json=buildSha,proto3" json:"build_sha,omitempty"`
	NumCpus    int32  `protobuf:"varint,3,opt,name=num_cpus,json=numCpus,proto3" json:"num_cpus,omitempty"`
	Gomemlimit int64  `protobuf:"varint,4,opt,name=gomemlimit,proto3" json:"gomemlimit,omitempty"`
	// contains filtered or unexported fields
}

Hello is the worker's introduction. Coord uses it to register the worker, bind capabilities to the scheduler, and pin a build_sha for compatibility checks.

func (*Hello) Descriptor() ([]byte, []int)

func (x *Hello) GetBuildSha() string

func (x *Hello) GetGomemlimit() int64

func (x *Hello) GetNumCpus() int32

func (x *Hello) GetWorkerId() string

func (*Hello) ProtoMessage()

func (x *Hello) ProtoReflect() protoreflect.Message

func (x *Hello) Reset()

func (x *Hello) String() string

type PeerExchangeClient ¶

type PeerExchangeClient interface {
	// FetchShuffle streams the raw bytes of one stage-output file the
	// serving worker produced and still holds locally. Rejections:
	// PermissionDenied (bad/unknown token), NotFound (key not held),
	// ResourceExhausted (serve concurrency cap held past its bound).
	FetchShuffle(ctx context.Context, in *FetchShuffleRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ShuffleChunk], error)
}

PeerExchangeClient is the client API for PeerExchange service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.

PeerExchange is the worker→worker shuffle-fetch service (streaming exchange, Phase A). Every worker hosts it when --streaming-exchange is enabled; consumers fetch stage-output files directly from the producing worker's local disk instead of round-tripping S3. The S3 object stays the durable copy — any fetch failure falls through to the S3 read path, so this service is a best-effort cache tier, never authoritative.

Deliberately NOT part of the DataPlane bidi envelope: each fetch runs on its own server-stream so HTTP/2 per-stream flow control backpressures only that fetch — no head-of-line blocking across tasks or queries sharing a peer connection. See docs/design/streaming-exchange.md §4.1.

func NewPeerExchangeClient(cc grpc.ClientConnInterface) PeerExchangeClient

type PeerExchangeServer ¶

type PeerExchangeServer interface {
	// FetchShuffle streams the raw bytes of one stage-output file the
	// serving worker produced and still holds locally. Rejections:
	// PermissionDenied (bad/unknown token), NotFound (key not held),
	// ResourceExhausted (serve concurrency cap held past its bound).
	FetchShuffle(*FetchShuffleRequest, grpc.ServerStreamingServer[ShuffleChunk]) error
	// contains filtered or unexported methods
}

PeerExchangeServer is the server API for PeerExchange service. All implementations must embed UnimplementedPeerExchangeServer for forward compatibility.

PeerExchange is the worker→worker shuffle-fetch service (streaming exchange, Phase A). Every worker hosts it when --streaming-exchange is enabled; consumers fetch stage-output files directly from the producing worker's local disk instead of round-tripping S3. The S3 object stays the durable copy — any fetch failure falls through to the S3 read path, so this service is a best-effort cache tier, never authoritative.

Deliberately NOT part of the DataPlane bidi envelope: each fetch runs on its own server-stream so HTTP/2 per-stream flow control backpressures only that fetch — no head-of-line blocking across tasks or queries sharing a peer connection. See docs/design/streaming-exchange.md §4.1.

type PeerExchange_FetchShuffleClient ¶

type PeerExchange_FetchShuffleClient = grpc.ServerStreamingClient[ShuffleChunk]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type PeerExchange_FetchShuffleServer ¶

type PeerExchange_FetchShuffleServer = grpc.ServerStreamingServer[ShuffleChunk]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type ResultBatch ¶

type ResultBatch struct {
	QueryId  string `protobuf:"bytes,1,opt,name=query_id,json=queryId,proto3" json:"query_id,omitempty"`
	WorkerId string `protobuf:"bytes,2,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"`
	Terminal bool   `protobuf:"varint,3,opt,name=terminal,proto3" json:"terminal,omitempty"`
	RowCount int32  `protobuf:"varint,4,opt,name=row_count,json=rowCount,proto3" json:"row_count,omitempty"`
	Payload  []byte `protobuf:"bytes,5,opt,name=payload,proto3" json:"payload,omitempty"` // WSHF-encoded
	Err      string `protobuf:"bytes,6,opt,name=err,proto3" json:"err,omitempty"`         // non-empty on task failure (terminal=true)
	// contains filtered or unexported fields
}

ResultBatch is a chunk of a query's gather output, sent from worker to coord. Mirrors distributed.GatherBatchMsg on the NATS path so the receive-side can reuse the same WSHF decoder.

Routing: coord uses query_id to dispatch the batch to the right gatherReceiver. WSHF payload is self-contained (magic + schema + rows + nulls) — the gRPC framing is just transport.

Terminal=true marks the last batch from this worker for this query. If err is non-empty, the task failed and the partial output (if any) is discarded by the receiver.

func (*ResultBatch) Descriptor() ([]byte, []int)

func (x *ResultBatch) GetErr() string

func (x *ResultBatch) GetPayload() []byte

func (x *ResultBatch) GetQueryId() string

func (x *ResultBatch) GetRowCount() int32

func (x *ResultBatch) GetTerminal() bool

func (x *ResultBatch) GetWorkerId() string

func (*ResultBatch) ProtoMessage()

func (x *ResultBatch) ProtoReflect() protoreflect.Message

func (x *ResultBatch) Reset()

func (x *ResultBatch) String() string

type ShuffleChunk ¶

type ShuffleChunk struct {
	Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` // raw WSHF bytes, at most peerChunkBytes (256 KiB) each
	// contains filtered or unexported fields
}

func (*ShuffleChunk) Descriptor() ([]byte, []int)

func (x *ShuffleChunk) GetData() []byte

func (*ShuffleChunk) ProtoMessage()

func (x *ShuffleChunk) ProtoReflect() protoreflect.Message

func (x *ShuffleChunk) Reset()

func (x *ShuffleChunk) String() string

type TaskDispatch ¶

type TaskDispatch struct {
	TaskId           string `protobuf:"bytes,1,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`
	QueryId          string `protobuf:"bytes,2,opt,name=query_id,json=queryId,proto3" json:"query_id,omitempty"`
	StageId          string `protobuf:"bytes,3,opt,name=stage_id,json=stageId,proto3" json:"stage_id,omitempty"`
	TaskBlob         []byte `protobuf:"bytes,4,opt,name=task_blob,json=taskBlob,proto3" json:"task_blob,omitempty"`
	DeadlineUnixNano int64  `protobuf:"varint,5,opt,name=deadline_unix_nano,json=deadlineUnixNano,proto3" json:"deadline_unix_nano,omitempty"` // 0 = no explicit deadline
	// contains filtered or unexported fields
}

TaskDispatch is one unit of work coord pushes to a worker. The body is an opaque blob — distributed.Marshal(Task) on the coord side, distributed.Unmarshal on the worker side — because the Task struct has too many fields and too much churn to express in proto. gRPC just carries the bytes.

task_id, query_id, stage_id are duplicated outside the blob so the dispatcher and progress paths can route without unmarshaling. They MUST match the values inside task_blob; worker-side code uses the blob for execution and the outer fields for routing/logging only.

Idempotency: workers de-dupe on task_id. Coord may redispatch the same task_id after a TCP disconnect; duplicate completions are dropped by the result handler.

func (*TaskDispatch) Descriptor() ([]byte, []int)

func (x *TaskDispatch) GetDeadlineUnixNano() int64

func (x *TaskDispatch) GetQueryId() string

func (x *TaskDispatch) GetStageId() string

func (x *TaskDispatch) GetTaskBlob() []byte

func (x *TaskDispatch) GetTaskId() string

func (*TaskDispatch) ProtoMessage()

func (x *TaskDispatch) ProtoReflect() protoreflect.Message

func (x *TaskDispatch) Reset()

func (x *TaskDispatch) String() string

type TaskProgress ¶

type TaskProgress struct {
	QueryId           string `protobuf:"bytes,1,opt,name=query_id,json=queryId,proto3" json:"query_id,omitempty"`
	StageId           string `protobuf:"bytes,2,opt,name=stage_id,json=stageId,proto3" json:"stage_id,omitempty"`
	TaskId            string `protobuf:"bytes,3,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`
	WorkerId          string `protobuf:"bytes,4,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"`
	RowsProcessed     int64  `protobuf:"varint,5,opt,name=rows_processed,json=rowsProcessed,proto3" json:"rows_processed,omitempty"`
	BytesProcessed    int64  `protobuf:"varint,6,opt,name=bytes_processed,json=bytesProcessed,proto3" json:"bytes_processed,omitempty"`
	TimestampUnixNano int64  `protobuf:"varint,7,opt,name=timestamp_unix_nano,json=timestampUnixNano,proto3" json:"timestamp_unix_nano,omitempty"`
	// contains filtered or unexported fields
}

TaskProgress is the per-task liveness + throughput signal workers emit from their hot loops (~2 s cadence, advisory). Coord consumes it for two purposes:

1. WorkerRegistry liveness — any TaskProgress proves the worker goroutine is alive even if the global heartbeat is starved by GC mark-assist (Q03 SF10 2026-04-29 motivation for the multi-signal liveness work).
2. Per-stage awaitStageProgress — bridge fans messages out to the stage's progress channel for "no row progress for N seconds" stall detection.

rows_processed / bytes_processed are monotonic over the task's lifetime; coord computes throughput as deltas across consecutive messages from the same task_id.

func (*TaskProgress) Descriptor() ([]byte, []int)

func (x *TaskProgress) GetBytesProcessed() int64

func (x *TaskProgress) GetQueryId() string

func (x *TaskProgress) GetRowsProcessed() int64

func (x *TaskProgress) GetStageId() string

func (x *TaskProgress) GetTaskId() string

func (x *TaskProgress) GetTimestampUnixNano() int64

func (x *TaskProgress) GetWorkerId() string

func (*TaskProgress) ProtoMessage()

func (x *TaskProgress) ProtoReflect() protoreflect.Message

func (x *TaskProgress) Reset()

func (x *TaskProgress) String() string

type UnimplementedDataPlaneServer ¶

type UnimplementedDataPlaneServer struct{}

UnimplementedDataPlaneServer must be embedded to have forward compatible implementations.

NOTE: this should be embedded by value instead of pointer to avoid a nil pointer dereference when methods are called.

func (UnimplementedDataPlaneServer) Connect(grpc.BidiStreamingServer[WorkerEnvelope, CoordEnvelope]) error

type UnimplementedPeerExchangeServer ¶

type UnimplementedPeerExchangeServer struct{}

UnimplementedPeerExchangeServer must be embedded to have forward compatible implementations.

NOTE: this should be embedded by value instead of pointer to avoid a nil pointer dereference when methods are called.

func (UnimplementedPeerExchangeServer) FetchShuffle(*FetchShuffleRequest, grpc.ServerStreamingServer[ShuffleChunk]) error

type UnsafeDataPlaneServer ¶

type UnsafeDataPlaneServer interface {
	// contains filtered or unexported methods
}

UnsafeDataPlaneServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to DataPlaneServer will result in compilation errors.

type UnsafePeerExchangeServer ¶

type UnsafePeerExchangeServer interface {
	// contains filtered or unexported methods
}

UnsafePeerExchangeServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to PeerExchangeServer will result in compilation errors.

type Welcome ¶

type Welcome struct {
	ClusterId  string `protobuf:"bytes,1,opt,name=cluster_id,json=clusterId,proto3" json:"cluster_id,omitempty"`
	ServerTime int64  `protobuf:"varint,2,opt,name=server_time,json=serverTime,proto3" json:"server_time,omitempty"` // unix nanos
	// contains filtered or unexported fields
}

Welcome is the coord's reply. Carries cluster identity and any initial config the worker needs.

func (*Welcome) Descriptor() ([]byte, []int)

func (x *Welcome) GetClusterId() string

func (x *Welcome) GetServerTime() int64

func (*Welcome) ProtoMessage()

func (x *Welcome) ProtoReflect() protoreflect.Message

func (x *Welcome) Reset()

func (x *Welcome) String() string

type WorkerEnvelope ¶

type WorkerEnvelope struct {

	// Types that are valid to be assigned to Msg:
	//
	//	*WorkerEnvelope_Hello
	//	*WorkerEnvelope_ResultBatch
	//	*WorkerEnvelope_TaskProgress
	Msg isWorkerEnvelope_Msg `protobuf_oneof:"msg"`
	// contains filtered or unexported fields
}

WorkerEnvelope is the worker→coord direction of the bidi stream. Phase B added ResultBatch. Phase E added TaskProgress so the liveness signal shares fate with results (heap-pinned worker that can't deliver results also can't deliver progress, so coord redispatches).

func (*WorkerEnvelope) Descriptor() ([]byte, []int)

func (x *WorkerEnvelope) GetHello() *Hello

func (x *WorkerEnvelope) GetMsg() isWorkerEnvelope_Msg

func (x *WorkerEnvelope) GetResultBatch() *ResultBatch

func (x *WorkerEnvelope) GetTaskProgress() *TaskProgress

func (*WorkerEnvelope) ProtoMessage()

func (x *WorkerEnvelope) ProtoReflect() protoreflect.Message

func (x *WorkerEnvelope) Reset()

func (x *WorkerEnvelope) String() string

type WorkerEnvelope_Hello ¶

type WorkerEnvelope_Hello struct {
	Hello *Hello `protobuf:"bytes,1,opt,name=hello,proto3,oneof"`
}

type WorkerEnvelope_ResultBatch ¶

type WorkerEnvelope_ResultBatch struct {
	ResultBatch *ResultBatch `protobuf:"bytes,2,opt,name=result_batch,json=resultBatch,proto3,oneof"`
}

type WorkerEnvelope_TaskProgress ¶

type WorkerEnvelope_TaskProgress struct {
	TaskProgress *TaskProgress `protobuf:"bytes,3,opt,name=task_progress,json=taskProgress,proto3,oneof"` // reserved: TaskComplete, TaskFailed (future).
}