Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[release-20.0] VStream API: validate that last PK has fields defined (#16478) #16487

Merged
merged 2 commits into from
Jul 30, 2024
Merged

[release-20.0] VStream API: validate that last PK has fields defined (#16478) #16487

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
926eabe
VStream API: validate that last PK has fields defined (#16478)
vitess-bot[bot] Jul 26, 2024
5229a10
Merge remote-tracking branch 'origin/release-20.0' into backport-1647…
frouioui Jul 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go
View file
Open in desktop
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
View file
Open in desktop
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
Loading