flesh out vstream client example #17185
Conversation
there are already conversions for other native types, but datetime conversions were only available in the vitessdriver. This moves that implementation into sqltypes, so it can be more easily accessed elsewhere. Signed-off-by: Derek Perkins <[email protected]>
derekperkins
added
Type: Enhancement
derekperkins
requested review from
harshit-gangal,
shlomi-noach,
vmg,
rohit-nayak-ps,
frouioui,
mattlord and
GuptaManan100
as code owners
Review ChecklistHello reviewers! 👋 Please follow this checklist when reviewing this Pull Request. General
Tests
Documentation
New flags
If a workflow is added or modified:
Backward compatibility
|
vitess-bot
bot
added
NeedsBackportReason
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #17185 +/- ##
========================================
Coverage 67.33% 67.33%
========================================
Files 1569 1569
Lines 252244 252559 +315
========================================
+ Hits 169858 170072 +214
- Misses 82386 82487 +101 ☔ View full report in Codecov by Sentry. |
derekperkins
force-pushed
the
vstream-client-example
branch
from
4cb044c
to
99c0a33
Compare
|
|
||
| // alter the destination schema based on the DDL event | ||
|
|
||
| case binlogdatapb.VEventType_COPY_COMPLETED: |
There was a problem hiding this comment.
Should there be some sort of defer happening, that resets the destination tables and resets the last vgtid to empty?
| // storeLastVgtid stores the last vgtid processed by the client, so that it can resume from that position on restart. | ||
| // Storing a json blob in a database is just one way to do this, you could put it anywhere. | ||
| func storeLastVgtid(ctx context.Context, vgtid *binlogdatapb.VGtid) error { | ||
| _, err := json.Marshal(vgtid) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } |
There was a problem hiding this comment.
Is storing a full json blob the right thing to do? Do we really only want a subset if we want vstream to handle reshards?
| continue | ||
| } | ||
|
|
||
| row := sqltypes.MakeRowTrusted(fields, rc.After) |
There was a problem hiding this comment.
MakeRowTrusted has a warning that it shouldn't be used except in special circumstances, which require comments about why it is used. Is there a better way to handle this conversion?
| // TODO: I'm not exactly sure how to handle multiple rows in a single event, so I'm just going to take the last one | ||
| for _, rc := range rowEvent.RowChanges { | ||
| // ignore deletes | ||
| if rc.After == nil { | ||
| continue | ||
| } |
There was a problem hiding this comment.
I'd love any feedback on a good way to handle this
| return nil | ||
| } | ||
|
|
||
| // ******************************************************************************************************** |
There was a problem hiding this comment.
I tried to add comments to each case, but it's limited by my understanding, so please feel free to correct anything that isn't right. I think it would make sense to transfer a lot of the comments over to the proto, which has some very useful comments, but most of the type enums have none.
derekperkins
force-pushed
the
vstream-client-example
branch
from
99c0a33
to
31dc28c
Compare
Signed-off-by: Derek Perkins <[email protected]>
derekperkins
force-pushed
the
vstream-client-example
branch
from
31dc28c
to
7dd91c9
Compare
|
|
||
| // keep track of the last event time for heartbeat monitoring. We're purposefully not using the event | ||
| // timestamp, since that would cause cancellation if the stream was copying, delayed, or lagging. | ||
| lastEventReceivedAtUnix.Store(time.Now().Unix()) |
There was a problem hiding this comment.
Is this the right value to use to check the heartbeat?
| // if we haven't received an event in twice the heartbeat duration, we'll cancel the context, since | ||
| // we're likely disconnected, and exit the goroutine | ||
| if tm.Sub(time.Unix(lastEventReceivedAtUnix.Load(), 0)) > heartbeatDur*2 { |
There was a problem hiding this comment.
I made this heartbeat * 2 to give some buffer, but I don't have any idea if that's too much, not enough, etc
| // the first events will be field events, which contains schema information for any tables that are being streamed | ||
| var customerFields, corderFields []*querypb.Field |
There was a problem hiding this comment.
As I'm thinking, these are per shard, and this only works if schemas are identical across all shards. That may be fine here, but you'd have to account for it in a production setting. Maybe keep a mutex guarded map[keyspace/shard.table]*Fields?
Signed-off-by: Derek Perkins <[email protected]>
| func initStateTable(ctx context.Context, session *vtgateconn.VTGateSession, stateKeyspace, stateTable string) error { | ||
| query := fmt.Sprintf(`create table if not exists %s.%s ( | ||
| keyspace varbinary(512) not null, | ||
| table varbinary(512) not null, | ||
| vgtid json, | ||
| PRIMARY KEY (keyspace, table), | ||
| )`, stateKeyspace, stateTable) | ||
| _, err := session.Execute(ctx, query, nil) | ||
| if err != nil { | ||
| return fmt.Errorf("vstreamer: failed to create state table: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } |
There was a problem hiding this comment.
This feels like it might fit in the regular vreplication table somehow? Maybe that's shoehorning it where it doesn't go. Some of this decision hinges on whether this is a core vitess thing or a contrib framework.
| // v.shardsByKeyspace, err = getShardsByKeyspace(ctx, v.session) | ||
| // if err != nil { | ||
| // return nil, err | ||
| // } |
There was a problem hiding this comment.
Connecting through this vtgate session bypasses some routing that enables SHOW VITESS_SHARDS, returning this error, so I have it commented out for now
VT05003: unknown database '_vt' in vschema
| err = v.initTables(tables) | ||
| if err != nil { | ||
| return nil, err | ||
| } |
There was a problem hiding this comment.
I think there's a consolidation issue to deal with here. Someone could easily add a new table in code and redeploy, expecting it to catch up, but it wouldn't by default. To support that, I think we'd have to store a list of tables in state, and if a new table is added, catch it up to the same vgtid, then do a cutover to be in the same stream. That'd be a much better user experience, but a decent amount of added complexity.
go/vt/proto/vtgate/vstreamer/run.go
Outdated
| } | ||
|
|
||
| // copyRowToStruct builds a customer from a row event | ||
| func copyRowToStruct(shard shardConfig, row []sqltypes.Value, vPtr reflect.Value) error { |
There was a problem hiding this comment.
This needs to be fleshed out more, as it just does top level field matching for now, doesn't handle pointers, bytes, etc. I think it also deserves an optional scanner interface, for people to do custom unmarshaling.
go/vt/proto/vtgate/vstreamer/run.go
Outdated
| // create a new struct for the row | ||
| v := reflect.New(table.underlyingType) | ||
| table.currentBatch = reflect.Append(table.currentBatch, v) |
There was a problem hiding this comment.
While I really like the idea of returning a slice of their custom data type, it makes things like marking something as deleted tough to do. I'm considering returning a []Event something like this, so there it still does all the work of assembling a row, but still lets you dive into the raw data if needed.
type Event struct {
Data any // this is their custom unmarshaled struct
Deleted bool
RawEvent *binlogdatapb.RowEvent
RowChangeIndex // references the exact row change this was scanned from
}Also very open to suggestions about how to make the stream useful, I admittedly haven't looked much at prior art with Debezium or other systems
| if table.Keyspace == "" { | ||
| return fmt.Errorf("vstreamer: table %v has no keyspace", table) | ||
| } |
There was a problem hiding this comment.
Is there a reason to support multiple keyspaces in the same stream? At a surface level, it doesn't seem useful, but I also didn't see anything that made me believe it was unsupported at the vstream level
| // if there are no rows, or the value is null, return nil, which will start the stream from the beginning | ||
| if len(result.Rows) == 0 { | ||
| return nil, nil | ||
| } | ||
|
|
||
| if result.Rows[0][0].IsNull() { | ||
| return nil, nil | ||
| } |
There was a problem hiding this comment.
I'm not 100% sure what the minimal requirement is to bootstrap a vgtid. It seems like I'd probably have to loop through all the keyspaces + shards to make an unique, covering []*binlogdatapb.ShardGtid. Is that right?
Signed-off-by: Derek Perkins <[email protected]>
|
closing in favor of #17221 |
Description
We have decent documentation about how to start a vstream, but not much about how to best consume it. This extends the local example to show how to parse vstream events, check point vgtids, etc.
If this complicates the example too much, I'm happy to put this elsewhere. This is also my first time using vstream, so I looked at some places it is used in the codebase, but it's very possible I'm missing important cases to show. I made comment threads for the specific questions I had.
I was procrastinating something else I have to finish, so I just kept typing and typing on this PR :) At this point, I'd like for this to end up as close to production ready as possible, or at least have stubs + comments for some more complex integrations like DDL handling.
It seems like a little framework could be built around the vstream primitives, where all the looping, heartbeat monitoring, etc., happened inside, and all users would need to provide is some event handlers. I'll probably write something internally first to see how it works out, and then maybe upstream that somewhere if it makes sense.
cc @mattlord
This is based on top of another PR adding time conversion funcs to sqltypes, so once that merges, I'll rebase and this will just be the single example commit
I ran the example from Matt's CDC post, and this was the output from the local example
https://vitess.io/blog/2024-07-29-cdc-vstream/#a-look-under-the-hood-at-vstream
Checklist