Skip to content

Commit 5eedda1

Browse files
sushanbsushanb
authored
feat(bigtable): add virtual RPC descriptors (#14648)
1 parent 8968f4f commit 5eedda1

2 files changed

Lines changed: 566 additions & 0 deletions

File tree

‎bigtable/internal/transport/descriptors.go‎

Lines changed: 224 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,224 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package internal
16+
17+
import (
18+
btpb "cloud.google.com/go/bigtable/apiv2/bigtablepb"
19+
"google.golang.org/protobuf/proto"
20+
)
21+
22+
// VRpcDescriptor defines the interface for virtual RPC encoding and decoding.
23+
type VRpcDescriptor interface {
24+
Method() string
25+
Encode(req interface{}) ([]byte, error)
26+
Decode(buf []byte) (interface{}, error)
27+
}
28+
29+
// VRpcDescriptorImpl implements VRpcDescriptor using custom encoder/decoder closures.
30+
type VRpcDescriptorImpl struct {
31+
MethodName string
32+
EncodeFn func(req interface{}) ([]byte, error)
33+
DecodeFn func(buf []byte) (interface{}, error)
34+
}
35+
36+
// Method returns the virtual RPC method identifier name.
37+
func (d *VRpcDescriptorImpl) Method() string {
38+
return d.MethodName
39+
}
40+
41+
// Encode serializes standard Bigtable request payload into virtual RPC bytes.
42+
func (d *VRpcDescriptorImpl) Encode(req interface{}) ([]byte, error) {
43+
return d.EncodeFn(req)
44+
}
45+
46+
// Decode de-serializes virtual RPC bytes back into standard Bigtable response message.
47+
func (d *VRpcDescriptorImpl) Decode(buf []byte) (interface{}, error) {
48+
return d.DecodeFn(buf)
49+
}
50+
51+
// vRPC Encoder/Decoder factories
52+
53+
func createTableEncoder(subEncoder func(req interface{}, envelope *btpb.TableRequest)) func(interface{}) ([]byte, error) {
54+
return func(req interface{}) ([]byte, error) {
55+
envelope := &btpb.TableRequest{}
56+
subEncoder(req, envelope)
57+
return proto.Marshal(envelope)
58+
}
59+
}
60+
61+
func createTableDecoder(subDecoder func(envelope *btpb.TableResponse) interface{}) func([]byte) (interface{}, error) {
62+
return func(buf []byte) (interface{}, error) {
63+
envelope := &btpb.TableResponse{}
64+
if err := proto.Unmarshal(buf, envelope); err != nil {
65+
return nil, err
66+
}
67+
return subDecoder(envelope), nil
68+
}
69+
}
70+
71+
func createAuthViewEncoder(subEncoder func(req interface{}, envelope *btpb.AuthorizedViewRequest)) func(interface{}) ([]byte, error) {
72+
return func(req interface{}) ([]byte, error) {
73+
envelope := &btpb.AuthorizedViewRequest{}
74+
subEncoder(req, envelope)
75+
return proto.Marshal(envelope)
76+
}
77+
}
78+
79+
func createAuthViewDecoder(subDecoder func(envelope *btpb.AuthorizedViewResponse) interface{}) func([]byte) (interface{}, error) {
80+
return func(buf []byte) (interface{}, error) {
81+
envelope := &btpb.AuthorizedViewResponse{}
82+
if err := proto.Unmarshal(buf, envelope); err != nil {
83+
return nil, err
84+
}
85+
return subDecoder(envelope), nil
86+
}
87+
}
88+
89+
func createMatViewEncoder(subEncoder func(req interface{}, envelope *btpb.MaterializedViewRequest)) func(interface{}) ([]byte, error) {
90+
return func(req interface{}) ([]byte, error) {
91+
envelope := &btpb.MaterializedViewRequest{}
92+
subEncoder(req, envelope)
93+
return proto.Marshal(envelope)
94+
}
95+
}
96+
97+
func createMatViewDecoder(subDecoder func(envelope *btpb.MaterializedViewResponse) interface{}) func([]byte) (interface{}, error) {
98+
return func(buf []byte) (interface{}, error) {
99+
envelope := &btpb.MaterializedViewResponse{}
100+
if err := proto.Unmarshal(buf, envelope); err != nil {
101+
return nil, err
102+
}
103+
return subDecoder(envelope), nil
104+
}
105+
}
106+
107+
// ReadRowArgs contains arguments required for a virtual RPC ReadRow call.
108+
type ReadRowArgs struct {
109+
RowKey string
110+
Filter *btpb.RowFilter
111+
}
112+
113+
// ReadRowResult holds the result returned from a virtual RPC ReadRow call.
114+
type ReadRowResult struct {
115+
Row *btpb.Row
116+
}
117+
118+
// MutateRowArgs contains arguments required for a virtual RPC MutateRow call.
119+
type MutateRowArgs struct {
120+
RowKey string
121+
Mutations []*btpb.Mutation
122+
}
123+
124+
// MutateRowResult holds the result of a MutateRow virtual RPC.
125+
type MutateRowResult struct{}
126+
127+
func encodeReadRow(args ReadRowArgs) *btpb.SessionReadRowRequest {
128+
return &btpb.SessionReadRowRequest{
129+
Key: []byte(args.RowKey),
130+
Filter: args.Filter,
131+
}
132+
}
133+
134+
func decodeReadRow(resp *btpb.SessionReadRowResponse) ReadRowResult {
135+
if resp == nil {
136+
return ReadRowResult{}
137+
}
138+
return ReadRowResult{
139+
Row: resp.Row,
140+
}
141+
}
142+
143+
func encodeMutateRow(args MutateRowArgs) *btpb.SessionMutateRowRequest {
144+
return &btpb.SessionMutateRowRequest{
145+
Key: []byte(args.RowKey),
146+
Mutations: args.Mutations,
147+
}
148+
}
149+
150+
func decodeMutateRow(resp *btpb.SessionMutateRowResponse) MutateRowResult {
151+
return MutateRowResult{}
152+
}
153+
154+
var (
155+
// READ_ROW executes Point Reads on standard tables.
156+
READ_ROW = &VRpcDescriptorImpl{
157+
MethodName: "ReadRow",
158+
EncodeFn: createTableEncoder(func(req interface{}, env *btpb.TableRequest) {
159+
args := req.(ReadRowArgs)
160+
env.Payload = &btpb.TableRequest_ReadRow{
161+
ReadRow: encodeReadRow(args),
162+
}
163+
}),
164+
DecodeFn: createTableDecoder(func(env *btpb.TableResponse) interface{} {
165+
return decodeReadRow(env.GetReadRow())
166+
}),
167+
}
168+
169+
// MUTATE_ROW executes Point Mutations on standard tables.
170+
MUTATE_ROW = &VRpcDescriptorImpl{
171+
MethodName: "MutateRow",
172+
EncodeFn: createTableEncoder(func(req interface{}, env *btpb.TableRequest) {
173+
args := req.(MutateRowArgs)
174+
env.Payload = &btpb.TableRequest_MutateRow{
175+
MutateRow: encodeMutateRow(args),
176+
}
177+
}),
178+
DecodeFn: createTableDecoder(func(env *btpb.TableResponse) interface{} {
179+
return decodeMutateRow(env.GetMutateRow())
180+
}),
181+
}
182+
183+
// READ_ROW_AUTH_VIEW executes Point Reads on Authorized Views.
184+
READ_ROW_AUTH_VIEW = &VRpcDescriptorImpl{
185+
MethodName: "ReadRow",
186+
EncodeFn: createAuthViewEncoder(func(req interface{}, env *btpb.AuthorizedViewRequest) {
187+
args := req.(ReadRowArgs)
188+
env.Payload = &btpb.AuthorizedViewRequest_ReadRow{
189+
ReadRow: encodeReadRow(args),
190+
}
191+
}),
192+
DecodeFn: createAuthViewDecoder(func(env *btpb.AuthorizedViewResponse) interface{} {
193+
return decodeReadRow(env.GetReadRow())
194+
}),
195+
}
196+
197+
// MUTATE_ROW_AUTH_VIEW executes Point Mutations on Authorized Views.
198+
MUTATE_ROW_AUTH_VIEW = &VRpcDescriptorImpl{
199+
MethodName: "MutateRow",
200+
EncodeFn: createAuthViewEncoder(func(req interface{}, env *btpb.AuthorizedViewRequest) {
201+
args := req.(MutateRowArgs)
202+
env.Payload = &btpb.AuthorizedViewRequest_MutateRow{
203+
MutateRow: encodeMutateRow(args),
204+
}
205+
}),
206+
DecodeFn: createAuthViewDecoder(func(env *btpb.AuthorizedViewResponse) interface{} {
207+
return decodeMutateRow(env.GetMutateRow())
208+
}),
209+
}
210+
211+
// READ_ROW_MAT_VIEW executes Point Reads on Materialized Views.
212+
READ_ROW_MAT_VIEW = &VRpcDescriptorImpl{
213+
MethodName: "ReadRow",
214+
EncodeFn: createMatViewEncoder(func(req interface{}, env *btpb.MaterializedViewRequest) {
215+
args := req.(ReadRowArgs)
216+
env.Payload = &btpb.MaterializedViewRequest_ReadRow{
217+
ReadRow: encodeReadRow(args),
218+
}
219+
}),
220+
DecodeFn: createMatViewDecoder(func(env *btpb.MaterializedViewResponse) interface{} {
221+
return decodeReadRow(env.GetReadRow())
222+
}),
223+
}
224+
)

0 commit comments

Comments
 (0)