Use VStream() #120

notfelineit · 2025-02-04T02:13:14Z

Summary of changes

This PR transitions to use VTGate's VStream client instead of the specialized psdbconnectStream client. Some of the main changes during this transition are:

Processing individual VEvents instead of batched VEvents (Stream would batch them first, and then send this batched response to the connector)
Handling VStream COPY ("full sync") and RUNNING ("incremental sync") phases separately:
- COPY
  - Explicitly wait for a COPY COMPLETED event
  - Return first VGTID that is after the stop VGTID
- RUNNING
  - Wait for a VGTID that is after the stop VGTID
  - Return VGTID that is after stop VGTID
Explicitly handle "no new VGTIDs". This can happen if there was no new activity on the database after the stop position.

Overview

A single Airbyte sync is either a "full sync" or an "incremental sync".

"Full sync": just sync all the records up to the current stop position; uses VStream COPY
"Incremental sync": sync from a start position up to the current stop position; uses VStream binlog streaming (RUNNING)

In total, a single Airbyte sync will call VStream() at least once: once to fetch the stop position, and the rest to stream rows up to the stop position.

A sync session calls the Read() method, which does:

Call the VStream() api with VGTID set to "current" in order to a stop position
Loop this until we reach the first VGTID after the stop position:
- If there is a start position (incremental sync):
  - Call VStream() with the VGTID set to start position
  - Process VStream() response events:
    - If VGTID event: update the current position to VGTID event's VGTID
    - If FIELD event: save fields for use later
    - If ROW event: save rows to for queueing later
  - While processing VStream() response events:
    - If VGTID has advanced past the stop position, flush the queued Airbyte records to Airbyte, and return the current VGTID as the start position for the next incremental sync
    - If context.DeadlineExceeded before we have advanced past the stop position, still flush records because this indicates there are no new VGTIDs
    - If we came across any rows, queue them as Airbyte records
- If there is no start position (full sync):
  - Call VStream() with VGTID set to empty (""). This specifies VStream COPY
  - Process VStream() response events like in incremental sync, with one addition:
    - If COPY_COMPLETED event: note that we have completed the copy phase
  - While processing VStream() response events:
    - If we've completed the copy phase, and VGTID has advanced past the stop position, flush the queued Airbyte records to Airbyte, and return the current VGTID as the start position for the next incremental sync
    - If context.DeadlineExceeded before we have advanced past the stop position, still flush records because this indicates there are no new VGTIDs
    - If we came across any rows, queue them as Airbyte records

proto/vtgateservice.proto

Makefile

…nto use-vstream

notfelineit · 2025-02-04T21:25:02Z

Logs will look something like:

{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: Checking connection"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: State file detected, parsing provided file state.json"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: Syncing stream t1 with sync mode incremental"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: Using serialized cursor for stream test-connectors:t1"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Peeking to see if there's any new rows"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] New rows found, syncing rows for 1m0s"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Syncing rows from cursor [shard:\"-\"  keyspace:\"test-connectors\"  position:\"MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-15610,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96\"]"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Will be waiting for COPY_COMPLETED event? false"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Requesting to sync from cursor position [MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-15610,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96] to stop cursor position [MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-17634,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96] in cells planetscale_operator_default; using last known PK: false"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] FIELD event found, setting fields to [name:\"id\"  type:INT64  table:\"t1\"  org_table:\"t1\"  database:\"test-connectors\"  org_name:\"id\"  column_length:20  charset:63  flags:49667  column_type:\"bigint\" name:\"uuid\"  type:VARCHAR  table:\"t1\"  org_table:\"t1\"  database:\"test-connectors\"  org_name:\"uuid\"  column_length:1024  charset:255  column_type:\"varchar(256)\"]"}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Stop position [MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-17634,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96] found. Waiting for next VGTID after stop position."}}
{"type":"LOG","log":{"level":"INFO","message":"PlanetScale Source :: [test-connectors:primary:t1 shard : -] Exiting sync and flushing records because current position MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-17635,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96 has passed stop position MySQL56/0d5afdd6-da80-11ef-844c-26dc1854a614:1-74,e1e896df-dae3-11ef-895b-626e6780cb50:1-17634,e50c022a-dade-11ef-8083-d2b0b749d1bb:1-96"}}

… incremental (running) phase with no new rows

notfelineit · 2025-02-05T21:00:54Z

cmd/internal/planetscale_edge_database.go

@@ -189,9 +195,12 @@ func (p PlanetScaleEdgeDatabase) Read(ctx context.Context, w io.Writer, ps Plane
 		if lcErr != nil {
 			return currentSerializedCursor, errors.Wrap(err, "Unable to get latest cursor position")
 		}
+		if latestCursorPosition == "" {


Necessary because if we get an empty stop position, the sync will run indefinitely.

how would we get an empty stop position? from the user setting that or from VStream sending it somehow?

When I tested locally (e2e), there was one case where I got an empty string in an error case. I forgot the exact conditions but thought it would be good to handle.

notfelineit · 2025-02-05T21:01:17Z

cmd/internal/planetscale_edge_database.go


 		// the last synced VGTID is not at least, or after the current VGTID
-		if currentPosition.Position != "" && !positionAtLeast(latestCursorPosition, currentPosition.Position) {
+		if currentPosition.Position != "" && !positionAfter(latestCursorPosition, currentPosition.Position) {


If the stop position is behind the start position, exit the sync.

is !positionAfter compatible with the two positions being identical? reads that way.

Yeah - if they are equal, !positionAfter will return true (positionAfter will be false)

notfelineit · 2025-02-05T21:02:08Z

cmd/internal/planetscale_edge_database.go

+	if err != nil {
+		return tc, 0, err
+	}
+	if conn != nil {


For reviewers: will this correctly defer closing the connection?

looks like it would.

when err != nil will it always be the case that conn == nil?

Yeah, although conn should not be nil if err was not nil. So you could shorten it to:

vtgateClient, conn, err := p.initializeVTGateClient(ctx, ps) if err != nil { return tc, 0, err } defer conn.Close()

Thank you both! I'll simplify.

Realized we don't mock conn in testing, and that's likely the reason for that behavior 🤔 If alright, I'd like to handle mocking properly so we don't need the nil check in a separate PR.

notfelineit · 2025-02-05T21:03:34Z

cmd/internal/planetscale_edge_database.go

+			p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sCopy phase completed. Waiting for next VGTID after stop position.", preamble))
+			watchForVgGtidChange = true
+		}
+		if !waitForCopyCompleted && positionEqual(tc.Position, stopPosition) {


During an incremental sync, we'll have one VStream response for every VGTID from the start position to the stop position, so it's safe to stop looking for rows at the exact stop position.

so it's safe to stop looking for rows at the exact stop position.
does setting watchForVgGtidChange = true make it stop looking for exact stop position?

I updated the var name to "canFinishSync" as I think that better represents the purpose of the variable..essentially it's saying "Have we finished queueing all the records we need to given the STOP POSITION or copy completed"?

notfelineit · 2025-02-05T21:04:15Z

cmd/internal/planetscale_edge_database.go

-			// the current VGTID may have already advanced past the stop position.
-			watchForVgGtidChange = watchForVgGtidChange || positionAtLeast(tc.Position, stopPosition)
-			for _, result := range res.Result {
+		if waitForCopyCompleted && copyCompletedSeen {


During a full sync, we do not care about finding the exact stop position, we only care about looking for rows up to completing the copy phase.

notfelineit · 2025-02-05T21:04:42Z

cmd/internal/planetscale_edge_database.go

+		}
+
+		// Exit sync and flush records once the VGTID position is past the desired stop position, and we're no longer waiting for COPY phase to complete
+		if watchForVgGtidChange && positionAfter(tc.Position, stopPosition) {


Once we know we can stop looking for rows, and the current position has advanced past the stop position, we can exit syncing.

based on what this watchForVgGtidChange does, maybe a better variable name would be watchForStopPosition?

notfelineit · 2025-02-05T21:05:44Z

cmd/internal/planetscale_edge_database.go

 				qr := sqltypes.Proto3ToResult(result)
 				for _, row := range qr.Rows {
 					resultCount += 1
 					sqlResult := &sqltypes.Result{
-						Fields: result.Fields,
+						Fields: fields,


These are the same fields we found from the FIELD VEvent.

notfelineit · 2025-02-05T21:06:17Z

cmd/internal/planetscale_edge_database.go

+					return gtid, nil
+				}
+			}
+			return "", errors.New("unable to find VEvent of VGTID type to use as stop cursor")


Only find a stop position from the VGTID VEvent.

@mattlord is that a safe assumption? that there will always be a VGTID event to get the stop position from?

Yes, you should get the snapshot's GTID position as the second event in the first batch, for example:

[type:VGTID vgtid:{shard_gtids:{keyspace:"customer" shard:"-80"} shard_gtids:{keyspace:"customer" shard:"80-" gtid:"MySQL56/17e6f390-e4a7-11ef-928d-554558eb9218:1-57"}} keyspace:"customer" shard:"80-"]

Thank you! 🫶

notfelineit · 2025-02-05T21:06:27Z

cmd/internal/planetscale_edge_database.go

+	}
+}
+
+func toTopoTabletType(tabletType psdbconnect.TabletType) topodata.TabletType {


This mapping is taken from edge-gateway: https://github.com/planetscale/edge-gateway/blob/046a1b95e88d249fd423f3bfad8f531915b2c777/cmd/psdb-api/internal/psdbconnect/v1alpha1/handlers.go#L78

maxenglander

i didn't review the tests yet because it's a lot, but reviewed most everything else. looking pretty good to me, just left a few comments/questions for now.

maxenglander · 2025-02-06T15:00:37Z

cmd/internal/mock_types.go

-func (x *connectSyncClientMock) Recv() (*psdbconnect.SyncResponse, error) {
-	if x.lastResponseSent >= len(x.syncResponses) {
+func (x *vtgateVStreamClientMock) Recv() (*vtgate.VStreamResponse, error) {
+	if x.lastResponseSent >= len(x.vstreamResponses) {


a little unclear to me what lastResponseSent is tracking.

So I am pretty sure last response sent keeps track of which index of mock sync responses should be sent next!

maxenglander · 2025-02-06T15:02:42Z

cmd/internal/planetscale_edge_database.go

@@ -189,9 +195,12 @@ func (p PlanetScaleEdgeDatabase) Read(ctx context.Context, w io.Writer, ps Plane
 		if lcErr != nil {
 			return currentSerializedCursor, errors.Wrap(err, "Unable to get latest cursor position")
 		}
+		if latestCursorPosition == "" {


how would we get an empty stop position? from the user setting that or from VStream sending it somehow?

maxenglander · 2025-02-06T15:03:55Z

cmd/internal/planetscale_edge_database.go


 		// the last synced VGTID is not at least, or after the current VGTID
-		if currentPosition.Position != "" && !positionAtLeast(latestCursorPosition, currentPosition.Position) {
+		if currentPosition.Position != "" && !positionAfter(latestCursorPosition, currentPosition.Position) {


is !positionAfter compatible with the two positions being identical? reads that way.

maxenglander · 2025-02-06T15:05:56Z

cmd/internal/planetscale_edge_database.go

+	if err != nil {
+		return tc, 0, err
+	}
+	if conn != nil {


looks like it would.

when err != nil will it always be the case that conn == nil?

maxenglander · 2025-02-06T15:12:24Z

cmd/internal/planetscale_edge_database.go

+
+		if waitForCopyCompleted && copyCompletedSeen {
+			p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sCopy phase completed. Waiting for next VGTID after stop position.", preamble))
+			watchForVgGtidChange = true


is this a typo? should it just be VGtid instead of VgGtid?

maxenglander · 2025-02-06T15:15:21Z

cmd/internal/planetscale_edge_database.go

+			p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sCopy phase completed. Waiting for next VGTID after stop position.", preamble))
+			watchForVgGtidChange = true
+		}
+		if !waitForCopyCompleted && positionEqual(tc.Position, stopPosition) {


so it's safe to stop looking for rows at the exact stop position.
does setting watchForVgGtidChange = true make it stop looking for exact stop position?

maxenglander · 2025-02-06T15:17:33Z

cmd/internal/planetscale_edge_database.go

+		}
+
+		// Exit sync and flush records once the VGTID position is past the desired stop position, and we're no longer waiting for COPY phase to complete
+		if watchForVgGtidChange && positionAfter(tc.Position, stopPosition) {


based on what this watchForVgGtidChange does, maybe a better variable name would be watchForStopPosition?

maxenglander · 2025-02-06T15:19:38Z

cmd/internal/planetscale_edge_database.go

+}
+
+func (p PlanetScaleEdgeDatabase) initializeVTGateClient(ctx context.Context, ps PlanetScaleSource) (vtgateservice.VitessClient, grpcclient.ConnPool, error) {
+	if p.vtgateClientFn == nil {


do we need to support both the p.vtgateClientFn == nil path as well as the != nil path?

Yeah, unfortunately this is a remnant of the old testing style. I think there's opportunity for improvements on the tests and mocking in a next PR 🙏

mattlord · 2025-02-06T15:56:32Z

cmd/internal/mock_types.go

+
+	"google.golang.org/grpc"
+	"vitess.io/vitess/go/vt/proto/vtgate"
+	vtgateservice "vitess.io/vitess/go/vt/proto/vtgateservice"


Just noting that this alias is not needed since it matches the package name.

Thank you! 😄

mattlord · 2025-02-06T16:00:19Z

cmd/internal/planetscale_edge_database.go

+	if err != nil {
+		return tc, 0, err
+	}
+	if conn != nil {


Yeah, although conn should not be nil if err was not nil. So you could shorten it to:

vtgateClient, conn, err := p.initializeVTGateClient(ctx, ps) if err != nil { return tc, 0, err } defer conn.Close()

mattlord · 2025-02-06T16:30:14Z

cmd/internal/planetscale_edge_database.go

+		for _, event := range res.Events {
+			switch event.Type {
+			case binlogdata.VEventType_VGTID:
+				vgtid := event.GetVgtid().ShardGtids[0]


This assumes there's only a single shard. The position will not be the same for all shards. Here's an example event:

type:VGTID vgtid:{shard_gtids:{keyspace:"customer" shard:"-80" gtid:"MySQL56/0f2b79a6-e4a7-11ef-a8a5-e8942912de27:1-58"} shard_gtids:{keyspace:"customer" shard:"80-" gtid:"MySQL56/17e6f390-e4a7-11ef-928d-554558eb9218:1-57"}} keyspace:"customer" shard:"-80" type:COMMIT timestamp:1738859305 current_time:1738859305605332000 keyspace:"customer" shard:"-80"]

We should discuss this. Perhaps I'm missing something here. Also seems odd to me that it's a table cursor. Are the syncs always for a single table?

I think that we want loop over the ShardGtids and join them. We have code in vitess for this to get the union of two GTID sets: replication.GTIDSet.Union

Or is this somehow per table per shard here?

Your confusion is valid - the code here runs per table per shard. There's likely some optimizations we can make, as I don't 100% understand why the connector was built to do one VStream per table per shard originally.

For reference, the code here is executed within two for loops: for table in tables, then for shard in shards: https://github.com/planetscale/airbyte-source/blob/main/cmd/airbyte-source/read.go#L98-L110

mattlord · 2025-02-06T17:10:51Z

cmd/internal/planetscale_edge_database.go

+					return gtid, nil
+				}
+			}
+			return "", errors.New("unable to find VEvent of VGTID type to use as stop cursor")


Yes, you should get the snapshot's GTID position as the second event in the first batch, for example:

[type:VGTID vgtid:{shard_gtids:{keyspace:"customer" shard:"-80"} shard_gtids:{keyspace:"customer" shard:"80-" gtid:"MySQL56/17e6f390-e4a7-11ef-928d-554558eb9218:1-57"}} keyspace:"customer" shard:"80-"]

mattlord · 2025-02-06T17:13:42Z

cmd/internal/planetscale_edge_database.go

@@ -400,8 +457,33 @@ func (p PlanetScaleEdgeDatabase) printQueryResult(qr *sqltypes.Result, tableName
 	}
 }

-// positionAtLeast returns true if position `a` is equal to or after position `b`
-func positionAtLeast(a string, b string) bool {
+func buildVStreamRequest(tabletType psdbconnect.TabletType, table string, shard string, keyspace string, gtid string) *vtgate.VStreamRequest {


OK, so I see that this is a single table and a single shard. So perhaps this answers my question above...

Yes! Exactly 😄

notfelineit · 2025-02-10T21:06:39Z

cmd/internal/planetscale_edge_database.go

-			p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sExiting sync and flushing records due to error: %+v", preamble, err))
+			if s, ok := status.FromError(err); ok && s.Code() == codes.DeadlineExceeded && canFinishSync {
+				// No next VGTID found, but we previously found stop position or finished VStream COPY phase
+				p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sExiting sync and flushing records because no new VGTID found after stop position %+v", preamble, stopPosition))


FYI, this is the line that addresses:

Explicitly handle "no new VGTIDs". This can happen if there was no new activity on the database after the stop position.

mattlord · 2025-02-10T22:31:07Z

Nice work on this! I approved, for what that's worth. 🙂

notfelineit added 3 commits February 3, 2025 17:58

Adding VStream

b1853c3

Split out

f885974

Add vtgate service

7a6e7a4

notfelineit commented Feb 4, 2025

View reviewed changes

proto/vtgateservice.proto Outdated Show resolved Hide resolved

Update proto/vtgateservice.proto

309f337

notfelineit commented Feb 4, 2025

View reviewed changes

Makefile Outdated Show resolved Hide resolved

notfelineit and others added 10 commits February 3, 2025 18:13

Update Makefile

b2120e3

Run tidy

3f0fede

Merge branch 'use-vstream' of github.com:planetscale/airbyte-source i…

21918b2

…nto use-vstream

Try modifying getLatestCursorPosition to use vtgate

c47c820

Reset protos

d77e959

Get getLatestCursor working locally

06f5d6d

Get local sync working

00a549f

Undo changes to makefile

772b9ae

Refactor and reuse some logic, add helpful logs

a873826

Clean up logs more

0baf971

notfelineit force-pushed the use-vstream branch from ed34d3e to 0baf971 Compare February 4, 2025 21:23

re run proto

3c744d6

notfelineit added 7 commits February 4, 2025 17:08

Update mock types and modify first few tests

5b061ec

Convert more tests

2238dcc

Handle cases of copy phase and first VGTID is past stop position, and…

45f45d5

… incremental (running) phase with no new rows

Remove some noisy logs

6c4e381

Update test testing stopping at next VGtid after stop position

4b8c976

Finish tests

375dbc8

Remove unused mock types

317637e

notfelineit commented Feb 5, 2025

View reviewed changes

Remove unused method

2c22667

notfelineit marked this pull request as ready for review February 5, 2025 21:29

mattlord requested review from mattrobenolt, maxenglander and mattlord February 5, 2025 21:45

notfelineit force-pushed the use-vstream branch from 5f3dcbd to 2c22667 Compare February 5, 2025 23:30

maxenglander reviewed Feb 6, 2025

View reviewed changes

mattlord reviewed Feb 6, 2025

View reviewed changes

notfelineit added 7 commits February 6, 2025 12:45

Simplify var names and logic

3fffdc9

Clean up some more logs

0a6aed6

Handle case of no new VGTID found

2f4559e

Remove unused method

7026de9

Remove extra spaces

cbf56ba

Address some PR comments

787467c

Clean up protos

02057d4

notfelineit commented Feb 10, 2025

View reviewed changes

mattlord approved these changes Feb 10, 2025

View reviewed changes

Add test for copy catchup

6689561

notfelineit merged commit e84e4ba into main Feb 11, 2025
3 checks passed

notfelineit deleted the use-vstream branch February 11, 2025 01:27

Use VStream() #120

Use VStream() #120

Conversation

notfelineit commented Feb 4, 2025 • edited Loading

Summary of changes

Overview

notfelineit commented Feb 4, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

notfelineit Feb 5, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

notfelineit Feb 5, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

notfelineit Feb 5, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

notfelineit Feb 5, 2025 • edited Loading

Choose a reason for hiding this comment

maxenglander left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

notfelineit Feb 10, 2025 • edited Loading

Choose a reason for hiding this comment

mattlord commented Feb 10, 2025

notfelineit commented Feb 4, 2025 •

edited

Loading

notfelineit commented Feb 4, 2025 •

edited

Loading

notfelineit Feb 5, 2025 •

edited

Loading

notfelineit Feb 5, 2025 •

edited

Loading

notfelineit Feb 5, 2025 •

edited

Loading

notfelineit Feb 5, 2025 •

edited

Loading

notfelineit Feb 10, 2025 •

edited

Loading