Skip to content

Commit

Permalink
VStream API: validate that last PK has fields defined (#16478)
Browse files Browse the repository at this point in the history
Signed-off-by: Rohit Nayak <rohit@planetscale.com>
  • Loading branch information
rohit-nayak-ps authored Jul 26, 2024
1 parent 373cf34 commit 9aa9ab5
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 2 deletions.
6 changes: 4 additions & 2 deletions go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ type uvstreamer struct {

config *uvstreamerConfig

vs *vstreamer //last vstreamer created in uvstreamer
vs *vstreamer // last vstreamer created in uvstreamer
}

type uvstreamerConfig struct {
Expand Down Expand Up @@ -138,6 +138,9 @@ func (uvs *uvstreamer) buildTablePlan() error {
uvs.plans = make(map[string]*tablePlan)
tableLastPKs := make(map[string]*binlogdatapb.TableLastPK)
for _, tablePK := range uvs.inTablePKs {
if tablePK != nil && tablePK.Lastpk != nil && len(tablePK.Lastpk.Fields) == 0 {
return fmt.Errorf("lastpk for table %s has no fields defined", tablePK.TableName)
}
tableLastPKs[tablePK.TableName] = tablePK
}
tables := uvs.se.GetSchema()
Expand Down Expand Up @@ -313,7 +316,6 @@ func (uvs *uvstreamer) send2(evs []*binlogdatapb.VEvent) error {
}
behind := time.Now().UnixNano() - uvs.lastTimestampNs
uvs.setReplicationLagSeconds(behind / 1e9)
//log.Infof("sbm set to %d", uvs.ReplicationLagSeconds)
var evs2 []*binlogdatapb.VEvent
if len(uvs.plans) > 0 {
evs2 = uvs.filterEvents(evs)
Expand Down
27 changes: 27 additions & 0 deletions go/vt/vttablet/tabletserver/vstreamer/vstreamer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -398,6 +398,33 @@ func TestMissingTables(t *testing.T) {
runCases(t, filter, testcases, startPos, nil)
}

// TestVStreamMissingFieldsInLastPK tests that we error out if the lastpk for a table is missing the fields spec.
func TestVStreamMissingFieldsInLastPK(t *testing.T) {
ts := &TestSpec{
t: t,
ddls: []string{
"create table t1(id11 int, id12 int, primary key(id11))",
},
}
ts.Init()
defer ts.Close()
filter := &binlogdatapb.Filter{
Rules: []*binlogdatapb.Rule{{
Match: "t1",
Filter: "select * from t1",
}},
}
var tablePKs []*binlogdatapb.TableLastPK
tablePKs = append(tablePKs, getTablePK("t1", 1))
for _, tpk := range tablePKs {
tpk.Lastpk.Fields = nil
}
ctx := context.Background()
ch := make(chan []*binlogdatapb.VEvent)
err := vstream(ctx, t, "", tablePKs, filter, ch)
require.ErrorContains(t, err, "lastpk for table t1 has no fields defined")
}

func TestVStreamCopySimpleFlow(t *testing.T) {
ts := &TestSpec{
t: t,
Expand Down

0 comments on commit 9aa9ab5

Please sign in to comment.