Merge pull request #1049 from dolthub/james/mysql

[no-release-notes] add persist interface

Author: jycor

enginetest/enginetests.go

@@ -1014,6 +1014,7 @@ func TestUserPrivileges(t *testing.T, h Harness) {
 				Address: "localhost",
 			})
 			engine.Analyzer.Catalog.MySQLDb.AddRootAccount()
+			engine.Analyzer.Catalog.MySQLDb.SetPersister(&mysql_db.NoopPersister{})
 
 			for _, statement := range script.SetUpScript {
 				if sh, ok := harness.(SkippingHarness); ok {
@@ -1073,6 +1074,7 @@ func TestUserPrivileges(t *testing.T, h Harness) {
 			defer engine.Close()
 
 			engine.Analyzer.Catalog.MySQLDb.AddRootAccount()
+			engine.Analyzer.Catalog.MySQLDb.SetPersister(&mysql_db.NoopPersister{})
 			rootCtx := harness.NewContextWithClient(sql.Client{
 				User:    "root",
 				Address: "localhost",
@@ -1173,6 +1175,7 @@ func TestUserAuthentication(t *testing.T, h Harness) {
 			engine := mustNewEngine(t, harness)
 			defer engine.Close()
 			engine.Analyzer.Catalog.MySQLDb.AddRootAccount()
+			engine.Analyzer.Catalog.MySQLDb.SetPersister(&mysql_db.NoopPersister{})
 			if script.SetUpFunc != nil {
 				script.SetUpFunc(ctx, t, engine)
 			}
@@ -5389,6 +5392,47 @@ func TestPrepared(t *testing.T, harness Harness) {
 	}
 }
 
+type memoryPersister struct {
+	users []*mysql_db.User
+	roles []*mysql_db.RoleEdge
+}
+
+var _ mysql_db.MySQLDbPersistence = &memoryPersister{}
+
+func (p *memoryPersister) ValidateCanPersist() error {
+	return nil
+}
+
+func (p *memoryPersister) Persist(ctx *sql.Context, data []byte) error {
+	//erase everything from users and roles
+	p.users = make([]*mysql_db.User, 0)
+	p.roles = make([]*mysql_db.RoleEdge, 0)
+
+	// Deserialize the flatbuffer
+	serialMySQLDb := serial.GetRootAsMySQLDb(data, 0)
+
+	// Fill in users
+	for i := 0; i < serialMySQLDb.UserLength(); i++ {
+		serialUser := new(serial.User)
+		if !serialMySQLDb.User(serialUser, i) {
+			continue
+		}
+		user := mysql_db.LoadUser(serialUser)
+		p.users = append(p.users, user)
+	}
+
+	// Fill in roles
+	for i := 0; i < serialMySQLDb.RoleEdgesLength(); i++ {
+		serialRoleEdge := new(serial.RoleEdge)
+		if !serialMySQLDb.RoleEdges(serialRoleEdge, i) {
+			continue
+		}
+		role := mysql_db.LoadRoleEdge(serialRoleEdge)
+		p.roles = append(p.roles, role)
+	}
+	return nil
+}
+
 func TestPrivilegePersistence(t *testing.T, h Harness) {
 	harness, ok := h.(ClientHarness)
 	if !ok {
@@ -5397,103 +5441,72 @@ func TestPrivilegePersistence(t *testing.T, h Harness) {
 
 	engine := mustNewEngine(t, harness)
 	defer engine.Close()
+
+	persister := &memoryPersister{}
 	engine.Analyzer.Catalog.MySQLDb.AddRootAccount()
+	engine.Analyzer.Catalog.MySQLDb.SetPersister(persister)
 	ctx := NewContextWithClient(harness, sql.Client{
 		User:    "root",
 		Address: "localhost",
 	})
 
-	var users []*mysql_db.User
-	var roles []*mysql_db.RoleEdge
-	engine.Analyzer.Catalog.MySQLDb.SetPersistCallback(
-		func(ctx *sql.Context, buf []byte) error {
-			// erase everything from users and roles
-			users = make([]*mysql_db.User, 0)
-			roles = make([]*mysql_db.RoleEdge, 0)
-
-			// Deserialize the flatbuffer
-			serialMySQLDb := serial.GetRootAsMySQLDb(buf, 0)
-
-			// Fill in users
-			for i := 0; i < serialMySQLDb.UserLength(); i++ {
-				serialUser := new(serial.User)
-				if !serialMySQLDb.User(serialUser, i) {
-					continue
-				}
-				user := mysql_db.LoadUser(serialUser)
-				users = append(users, user)
-			}
-
-			// Fill in roles
-			for i := 0; i < serialMySQLDb.RoleEdgesLength(); i++ {
-				serialRoleEdge := new(serial.RoleEdge)
-				if !serialMySQLDb.RoleEdges(serialRoleEdge, i) {
-					continue
-				}
-				role := mysql_db.LoadRoleEdge(serialRoleEdge)
-				roles = append(roles, role)
-			}
-			return nil
-		},
-	)
-
 	RunQueryWithContext(t, engine, harness, ctx, "CREATE USER tester@localhost")
 	// If the user exists in []*mysql_db.User, then it must be NOT nil.
-	require.NotNil(t, findUser("tester", "localhost", users))
+	require.NotNil(t, findUser("tester", "localhost", persister.users))
 
 	RunQueryWithContext(t, engine, harness, ctx, "INSERT INTO mysql.user (Host, User) VALUES ('localhost', 'tester1')")
-	require.Nil(t, findUser("tester1", "localhost", users))
+	require.Nil(t, findUser("tester1", "localhost", persister.users))
 
 	RunQueryWithContext(t, engine, harness, ctx, "UPDATE mysql.user SET User = 'test_user' WHERE User = 'tester'")
-	require.NotNil(t, findUser("tester", "localhost", users))
+	require.NotNil(t, findUser("tester", "localhost", persister.users))
 
 	RunQueryWithContext(t, engine, harness, ctx, "FLUSH PRIVILEGES")
-	require.NotNil(t, findUser("tester1", "localhost", users))
-	require.Nil(t, findUser("tester", "localhost", users))
-	require.NotNil(t, findUser("test_user", "localhost", users))
+	require.NotNil(t, findUser("tester1", "localhost", persister.users))
+	require.Nil(t, findUser("tester", "localhost", persister.users))
+	require.NotNil(t, findUser("test_user", "localhost", persister.users))
 
 	RunQueryWithContext(t, engine, harness, ctx, "DELETE FROM mysql.user WHERE User = 'tester1'")
-	require.NotNil(t, findUser("tester1", "localhost", users))
+	require.NotNil(t, findUser("tester1", "localhost", persister.users))
 
 	RunQueryWithContext(t, engine, harness, ctx, "GRANT SELECT ON mydb.* TO test_user@localhost")
-	user := findUser("test_user", "localhost", users)
+	user := findUser("test_user", "localhost", persister.users)
 	require.True(t, user.PrivilegeSet.Database("mydb").Has(sql.PrivilegeType_Select))
 
 	RunQueryWithContext(t, engine, harness, ctx, "UPDATE mysql.db SET Insert_priv = 'Y' WHERE User = 'test_user'")
 	require.False(t, user.PrivilegeSet.Database("mydb").Has(sql.PrivilegeType_Insert))
 
 	RunQueryWithContext(t, engine, harness, ctx, "CREATE USER dolt@localhost")
 	RunQueryWithContext(t, engine, harness, ctx, "INSERT INTO mysql.db (Host, Db, User, Select_priv) VALUES ('localhost', 'mydb', 'dolt', 'Y')")
-	user1 := findUser("dolt", "localhost", users)
+	user1 := findUser("dolt", "localhost", persister.users)
 	require.NotNil(t, user1)
 	require.False(t, user1.PrivilegeSet.Database("mydb").Has(sql.PrivilegeType_Select))
 
 	RunQueryWithContext(t, engine, harness, ctx, "FLUSH PRIVILEGES")
-	require.Nil(t, findUser("tester1", "localhost", users))
-	user = findUser("test_user", "localhost", users)
+	require.Nil(t, findUser("tester1", "localhost", persister.users))
+	user = findUser("test_user", "localhost", persister.users)
 	require.True(t, user.PrivilegeSet.Database("mydb").Has(sql.PrivilegeType_Insert))
-	user1 = findUser("dolt", "localhost", users)
+	user1 = findUser("dolt", "localhost", persister.users)
 	require.True(t, user1.PrivilegeSet.Database("mydb").Has(sql.PrivilegeType_Select))
 
 	RunQueryWithContext(t, engine, harness, ctx, "CREATE ROLE test_role")
 	RunQueryWithContext(t, engine, harness, ctx, "GRANT SELECT ON *.* TO test_role")
-	require.Zero(t, len(roles))
+	require.Zero(t, len(persister.roles))
 	RunQueryWithContext(t, engine, harness, ctx, "GRANT test_role TO test_user@localhost")
-	require.NotZero(t, len(roles))
+	require.NotZero(t, len(persister.roles))
 
 	RunQueryWithContext(t, engine, harness, ctx, "UPDATE mysql.role_edges SET to_user = 'tester2' WHERE to_user = 'test_user'")
-	require.NotNil(t, findRole("test_user", roles))
-	require.Nil(t, findRole("tester2", roles))
+	require.NotNil(t, findRole("test_user", persister.roles))
+	require.Nil(t, findRole("tester2", persister.roles))
 
 	RunQueryWithContext(t, engine, harness, ctx, "FLUSH PRIVILEGES")
-	require.Nil(t, findRole("test_user", roles))
-	require.NotNil(t, findRole("tester2", roles))
+	require.Nil(t, findRole("test_user", persister.roles))
+	require.NotNil(t, findRole("tester2", persister.roles))
 
 	RunQueryWithContext(t, engine, harness, ctx, "INSERT INTO mysql.role_edges VALUES ('%', 'test_role', 'localhost', 'test_user', 'N')")
-	require.Nil(t, findRole("test_user", roles))
+	require.Nil(t, findRole("test_user", persister.roles))
 
 	RunQueryWithContext(t, engine, harness, ctx, "FLUSH PRIVILEGES")
-	require.NotNil(t, findRole("test_user", roles))
+	require.NotNil(t, findRole("test_user", persister.roles))
 
 	_, _, err := engine.Query(ctx, "FLUSH NO_WRITE_TO_BINLOG PRIVILEGES")
 	require.Error(t, err)

sql/mysql_db/mysql_db.go

@@ -30,9 +30,26 @@ import (
 	"github.com/dolthub/go-mysql-server/sql/mysql_db/serial"
 )
 
-// PersistCallback represents the callback that will be called when the Grant Tables have been updated and need to be
-// persisted.
-type PersistCallback func(ctx *sql.Context, data []byte) error
+// MySQLDbPersistence is used to determine the behavior of how certain tables in MySQLDb will be persisted.
+type MySQLDbPersistence interface {
+	ValidateCanPersist() error
+	Persist(ctx *sql.Context, data []byte) error
+}
+
+// NoopPersister is used when nothing in mysql db should be persisted
+type NoopPersister struct{}
+
+var _ MySQLDbPersistence = &NoopPersister{}
+
+// CanPersist implements the MySQLDbPersistence interface
+func (p *NoopPersister) ValidateCanPersist() error {
+	return nil
+}
+
+// Persist implements the MySQLDbPersistence interface
+func (p *NoopPersister) Persist(ctx *sql.Context, data []byte) error {
+	return nil
+}
 
 // MySQLDb are the collection of tables that are in the MySQL database
 type MySQLDb struct {
@@ -50,7 +67,7 @@ type MySQLDb struct {
 	//default_roles    *mysqlTable
 	//password_history *mysqlTable
 
-	persistFunc PersistCallback
+	persister MySQLDbPersistence
 }
 
 var _ sql.Database = (*MySQLDb)(nil)
@@ -136,9 +153,9 @@ func (t *MySQLDb) LoadData(ctx *sql.Context, buf []byte) error {
 	return nil
 }
 
-// SetPersistCallback sets the callback to be used when the MySQL Db tables have been updated and need to be persisted.
-func (t *MySQLDb) SetPersistCallback(persistFunc PersistCallback) {
-	t.persistFunc = persistFunc
+// SetPersister sets the custom persister to be used when the MySQL Db tables have been updated and need to be persisted.
+func (t *MySQLDb) SetPersister(persister MySQLDbPersistence) {
+	t.persister = persister
 }
 
 // AddRootAccount adds the root account to the list of accounts.
@@ -331,13 +348,13 @@ func (t *MySQLDb) Negotiate(c *mysql.Conn, user string, addr net.Addr) (mysql.Ge
 	return nil, fmt.Errorf(`the only user login interface currently supported is "mysql_native_password"`)
 }
 
+// CanPersist calls the persister's CanPersist method
+func (t *MySQLDb) ValidateCanPersist() error {
+	return t.persister.ValidateCanPersist()
+}
+
 // Persist passes along all changes to the integrator.
 func (t *MySQLDb) Persist(ctx *sql.Context) error {
-	// Do nothing if persist function is nil
-	if t.persistFunc == nil {
-		return nil
-	}
-
 	// Extract all user entries from table, and sort
 	userEntries := t.user.data.ToSlice(ctx)
 	users := make([]*User, len(userEntries))
@@ -387,7 +404,7 @@ func (t *MySQLDb) Persist(ctx *sql.Context) error {
 	b.Finish(mysqlDbOffset)
 
 	// Persist data
-	return t.persistFunc(ctx, b.FinishedBytes())
+	return t.persister.Persist(ctx, b.FinishedBytes())
 }
 
 // UserTable returns the "user" table.

sql/plan/create_role.go

@@ -106,6 +106,12 @@ func (n *CreateRole) RowIter(ctx *sql.Context, row sql.Row) (sql.RowIter, error)
 	if !ok {
 		return nil, sql.ErrDatabaseNotFound.New("mysql")
 	}
+
+	// Check if you can even persist in the first place
+	if err := mysqlDb.ValidateCanPersist(); err != nil {
+		return nil, err
+	}
+
 	userTableData := mysqlDb.UserTable().Data()
 	for _, role := range n.Roles {
 		userPk := mysql_db.UserPrimaryKey{

sql/plan/create_user.go

@@ -100,6 +100,10 @@ func (n *CreateUser) RowIter(ctx *sql.Context, row sql.Row) (sql.RowIter, error)
 	if !ok {
 		return nil, sql.ErrDatabaseNotFound.New("mysql")
 	}
+	// Check if you can even persist in the first place
+	if err := mysqlDb.ValidateCanPersist(); err != nil {
+		return nil, err
+	}
 	userTableData := mysqlDb.UserTable().Data()
 	for _, user := range n.Users {
 		userPk := mysql_db.UserPrimaryKey{

sql/plan/grant.go

@@ -201,6 +201,10 @@ func (n *Grant) RowIter(ctx *sql.Context, row sql.Row) (sql.RowIter, error) {
 	if !ok {
 		return nil, sql.ErrDatabaseNotFound.New("mysql")
 	}
+	// Check if you can even persist in the first place
+	if err := mysqlDb.ValidateCanPersist(); err != nil {
+		return nil, err
+	}
 	if n.PrivilegeLevel.Database == "*" && n.PrivilegeLevel.TableRoutine == "*" {
 		if n.ObjectType != ObjectType_Any {
 			return nil, sql.ErrGrantRevokeIllegalPrivilege.New()