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

Add linearizable support to VSchema management #17401

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Add linearizable support to VSchema management #17401

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 7 additions & 4 deletions go/cmd/vtcombo/cli/vschema_watcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,17 +56,20 @@ func loadKeyspacesFromDir(ctx context.Context, dir string, ts *topo.Server) {
log.Fatalf("Unable to read keyspace file %v: %v", ksFile, err)
}

keyspace := &vschemapb.Keyspace{}
err = json.Unmarshal(jsonData, keyspace)
ksvs := &topo.KeyspaceVSchemaInfo{
Name: ks.Name,
Keyspace: &vschemapb.Keyspace{},
}
err = json.Unmarshal(jsonData, ksvs.Keyspace)
if err != nil {
log.Fatalf("Unable to parse keyspace file %v: %v", ksFile, err)
}

_, err = vindexes.BuildKeyspace(keyspace, env.Parser())
_, err = vindexes.BuildKeyspace(ksvs.Keyspace, env.Parser())
if err != nil {
log.Fatalf("Invalid keyspace definition: %v", err)
}
ts.SaveVSchema(ctx, ks.Name, keyspace)
ts.SaveVSchema(ctx, ksvs)
log.Infof("Loaded keyspace %v from %v\n", ks.Name, ksFile)
}
}
Expand Down
1 change: 0 additions & 1 deletion go/test/endtoend/vtgate/keyspace_watches/keyspace_watch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,6 @@ func TestRoutingWithKeyspacesToWatch(t *testing.T) {
}

func TestVSchemaDDLWithKeyspacesToWatch(t *testing.T) {

extraVTGateArgs := []string{
"--vschema_ddl_authorized_users", "%",
}
Expand Down
7 changes: 3 additions & 4 deletions go/vt/topo/helpers/copy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ func CopyKeyspaces(ctx context.Context, fromTS, toTS *topo.Server, parser *sqlpa
}

for _, keyspace := range keyspaces {

ki, err := fromTS.GetKeyspace(ctx, keyspace)
if err != nil {
return fmt.Errorf("GetKeyspace(%v): %w", keyspace, err)
Expand All @@ -55,15 +54,15 @@ func CopyKeyspaces(ctx context.Context, fromTS, toTS *topo.Server, parser *sqlpa
}
}

vs, err := fromTS.GetVSchema(ctx, keyspace)
ksvs, err := fromTS.GetVSchema(ctx, keyspace)
switch {
case err == nil:
_, err = vindexes.BuildKeyspace(vs, parser)
_, err = vindexes.BuildKeyspace(ksvs.Keyspace, parser)
if err != nil {
log.Errorf("BuildKeyspace(%v): %v", keyspace, err)
break
}
if err := toTS.SaveVSchema(ctx, keyspace, vs); err != nil {
if err := toTS.SaveVSchema(ctx, ksvs); err != nil {
log.Errorf("SaveVSchema(%v): %v", keyspace, err)
}
case topo.IsErrType(err, topo.NoNode):
Expand Down
13 changes: 10 additions & 3 deletions go/vt/topo/srv_vschema.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,10 +171,17 @@ func (ts *Server) RebuildSrvVSchema(ctx context.Context, cells []string) error {
go func(keyspace string) {
defer wg.Done()

k, err := ts.GetVSchema(ctx, keyspace)
ksvs, err := ts.GetVSchema(ctx, keyspace)
if IsErrType(err, NoNode) {
err = nil
k = &vschemapb.Keyspace{}
ksvs = &KeyspaceVSchemaInfo{
Name: keyspace,
Keyspace: &vschemapb.Keyspace{
Sharded: false,
Vindexes: make(map[string]*vschemapb.Vindex),
Tables: make(map[string]*vschemapb.Table),
},
}
}

mu.Lock()
Expand All @@ -184,7 +191,7 @@ func (ts *Server) RebuildSrvVSchema(ctx context.Context, cells []string) error {
finalErr = err
return
}
srvVSchema.Keyspaces[keyspace] = k
srvVSchema.Keyspaces[keyspace] = ksvs.Keyspace
}(keyspace)
}
wg.Wait()
Expand Down
53 changes: 36 additions & 17 deletions go/vt/topo/vschema.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,26 +28,38 @@ import (
vschemapb "vitess.io/vitess/go/vt/proto/vschema"
)

// KeyspaceVSchemaInfo wraps a vschemapb.Keyspace and is a meta
// struct that contains metadata to give the data more context
// and convenience. This is the main way we interact with a
// keyspace's vschema.
type KeyspaceVSchemaInfo struct {
Name string
*vschemapb.Keyspace
version Version
}

// SaveVSchema saves a Vschema. A valid Vschema should be passed in. It does not verify its correctness.
// If the VSchema is empty, just remove it.
func (ts *Server) SaveVSchema(ctx context.Context, keyspace string, vschema *vschemapb.Keyspace) error {
func (ts *Server) SaveVSchema(ctx context.Context, ksvs *KeyspaceVSchemaInfo) error {
if err := ctx.Err(); err != nil {
return err
}

nodePath := path.Join(KeyspacesPath, keyspace, VSchemaFile)
data, err := vschema.MarshalVT()
nodePath := path.Join(KeyspacesPath, ksvs.Name, VSchemaFile)
data, err := ksvs.MarshalVT()
if err != nil {
return err
}

_, err = ts.globalCell.Update(ctx, nodePath, data, nil)
version, err := ts.globalCell.Update(ctx, nodePath, data, ksvs.version)
if err != nil {
log.Errorf("failed to update vschema for keyspace %s: %v", keyspace, err)
} else {
log.Infof("successfully updated vschema for keyspace %s: %+v", keyspace, vschema)
log.Errorf("failed to update vschema for keyspace %s: %v", ksvs.Name, err)
return err
}
return err
ksvs.version = version
log.Infof("successfully updated vschema for keyspace %s: %+v", ksvs.Name, ksvs.Keyspace)

return nil
}

// DeleteVSchema delete the keyspace if it exists
Expand All @@ -61,13 +73,13 @@ func (ts *Server) DeleteVSchema(ctx context.Context, keyspace string) error {
}

// GetVSchema fetches the vschema from the topo.
func (ts *Server) GetVSchema(ctx context.Context, keyspace string) (*vschemapb.Keyspace, error) {
func (ts *Server) GetVSchema(ctx context.Context, keyspace string) (*KeyspaceVSchemaInfo, error) {
if err := ctx.Err(); err != nil {
return nil, err
}

nodePath := path.Join(KeyspacesPath, keyspace, VSchemaFile)
data, _, err := ts.globalCell.Get(ctx, nodePath)
data, version, err := ts.globalCell.Get(ctx, nodePath)
if err != nil {
return nil, err
}
Expand All @@ -76,20 +88,27 @@ func (ts *Server) GetVSchema(ctx context.Context, keyspace string) (*vschemapb.K
if err != nil {
return nil, vterrors.Wrapf(err, "bad vschema data: %q", data)
}
return &vs, nil
return &KeyspaceVSchemaInfo{
Name: keyspace,
Keyspace: &vs,
version: version,
}, nil
}

// EnsureVSchema makes sure that a vschema is present for this keyspace or creates a blank one if it is missing
func (ts *Server) EnsureVSchema(ctx context.Context, keyspace string) error {
vschema, err := ts.GetVSchema(ctx, keyspace)
ksvs, err := ts.GetVSchema(ctx, keyspace)
if err != nil && !IsErrType(err, NoNode) {
log.Infof("error in getting vschema for keyspace %s: %v", keyspace, err)
}
if vschema == nil || IsErrType(err, NoNode) {
err = ts.SaveVSchema(ctx, keyspace, &vschemapb.Keyspace{
Sharded: false,
Vindexes: make(map[string]*vschemapb.Vindex),
Tables: make(map[string]*vschemapb.Table),
if ksvs == nil || ksvs.Keyspace == nil || IsErrType(err, NoNode) {
err = ts.SaveVSchema(ctx, &KeyspaceVSchemaInfo{
Name: keyspace,
Keyspace: &vschemapb.Keyspace{
Sharded: false,
Vindexes: make(map[string]*vschemapb.Vindex),
Tables: make(map[string]*vschemapb.Table),
},
})
if err != nil {
log.Errorf("could not create blank vschema: %v", err)
Expand Down
Loading
Loading