Backup schema keys in incremental backups. (dgraph-io#5147)

The full schema should be backed up during incremental backups.
During restore, the schema should be cleared before processing each backup
to end up with only the final schema.

Author: martinmr

systest/backup/encryption/backup_test.go

@@ -289,12 +289,12 @@ func runRestore(t *testing.T, lastDir string, commitTs uint64) map[string]string
 	restored, err := testutil.GetPredicateValues(pdir, "movie", commitTs)
 	require.NoError(t, err)
 
-	restoredPreds, err := testutil.GetPredicateNames(pdir, commitTs)
+	restoredPreds, err := testutil.GetPredicateNames(pdir)
 	require.NoError(t, err)
 	require.ElementsMatch(t, []string{"dgraph.graphql.schema", "dgraph.graphql.xid",
 		"dgraph.type", "movie"}, restoredPreds)
 
-	restoredTypes, err := testutil.GetTypeNames(pdir, commitTs)
+	restoredTypes, err := testutil.GetTypeNames(pdir)
 	require.NoError(t, err)
 	require.ElementsMatch(t, []string{"Node", "dgraph.graphql"}, restoredTypes)
 

systest/backup/filesystem/backup_test.go

@@ -106,6 +106,12 @@ func TestBackupFilesystem(t *testing.T) {
 	_ = runBackup(t, 3, 1)
 	restored := runRestore(t, copyBackupDir, "", math.MaxUint64)
 
+	// Check the predicates and types in the schema are as expected.
+	// TODO: refactor tests so that minio and filesystem tests share most of their logic.
+	preds := []string{"dgraph.graphql.schema", "dgraph.graphql.xid", "dgraph.type", "movie"}
+	types := []string{"Node", "dgraph.graphql"}
+	testutil.CheckSchema(t, preds, types)
+
 	checks := []struct {
 		blank, expected string
 	}{
@@ -130,10 +136,27 @@ func TestBackupFilesystem(t *testing.T) {
 	t.Logf("%+v", incr1)
 	require.NoError(t, err)
 
+	// Update schema and types to make sure updates to the schema are backed up.
+	require.NoError(t, dg.Alter(ctx, &api.Operation{Schema: `
+		movie: string .
+		actor: string .
+		type Node {
+			movie
+		}
+		type NewNode {
+			actor
+		}`}))
+
 	// Perform first incremental backup.
 	_ = runBackup(t, 6, 2)
 	restored = runRestore(t, copyBackupDir, "", incr1.Txn.CommitTs)
 
+	// Check the predicates and types in the schema are as expected.
+	preds = append(preds, "actor")
+	types = append(types, "NewNode")
+	testutil.CheckSchema(t, preds, types)
+
+	// Perform some checks on the restored values.
 	checks = []struct {
 		blank, expected string
 	}{
@@ -157,6 +180,7 @@ func TestBackupFilesystem(t *testing.T) {
 	// Perform second incremental backup.
 	_ = runBackup(t, 9, 3)
 	restored = runRestore(t, copyBackupDir, "", incr2.Txn.CommitTs)
+	testutil.CheckSchema(t, preds, types)
 
 	checks = []struct {
 		blank, expected string
@@ -181,6 +205,7 @@ func TestBackupFilesystem(t *testing.T) {
 	// Perform second full backup.
 	dirs := runBackupInternal(t, true, 12, 4)
 	restored = runRestore(t, copyBackupDir, "", incr3.Txn.CommitTs)
+	testutil.CheckSchema(t, preds, types)
 
 	// Check all the values were restored to their most recent value.
 	checks = []struct {
@@ -279,16 +304,6 @@ func runRestore(t *testing.T, backupLocation, lastDir string, commitTs uint64) m
 	restored, err := testutil.GetPredicateValues(pdir, "movie", commitTs)
 	require.NoError(t, err)
 	t.Logf("--- Restored values: %+v\n", restored)
-
-	restoredPreds, err := testutil.GetPredicateNames(pdir, commitTs)
-	require.NoError(t, err)
-	require.ElementsMatch(t, []string{"dgraph.graphql.schema", "dgraph.graphql.xid", "dgraph.type",
-		"movie"}, restoredPreds)
-
-	restoredTypes, err := testutil.GetTypeNames(pdir, commitTs)
-	require.NoError(t, err)
-	require.ElementsMatch(t, []string{"Node", "dgraph.graphql"}, restoredTypes)
-
 	return restored
 }
 

systest/backup/minio/backup_test.go

@@ -112,6 +112,12 @@ func TestBackupMinio(t *testing.T) {
 	_ = runBackup(t, 3, 1)
 	restored := runRestore(t, "", math.MaxUint64)
 
+	// Check the predicates and types in the schema are as expected.
+	// TODO: refactor tests so that minio and filesystem tests share most of their logic.
+	preds := []string{"dgraph.graphql.schema", "dgraph.graphql.xid", "dgraph.type", "movie"}
+	types := []string{"Node", "dgraph.graphql"}
+	testutil.CheckSchema(t, preds, types)
+
 	checks := []struct {
 		blank, expected string
 	}{
@@ -136,10 +142,26 @@ func TestBackupMinio(t *testing.T) {
 	t.Logf("%+v", incr1)
 	require.NoError(t, err)
 
+	// Update schema and types to make sure updates to the schema are backed up.
+	require.NoError(t, dg.Alter(ctx, &api.Operation{Schema: `
+		movie: string .
+		actor: string .
+		type Node {
+			movie
+		}
+		type NewNode {
+			actor
+		}`}))
+
 	// Perform first incremental backup.
 	_ = runBackup(t, 6, 2)
 	restored = runRestore(t, "", incr1.Txn.CommitTs)
 
+	// Check the predicates and types in the schema are as expected.
+	preds = append(preds, "actor")
+	types = append(types, "NewNode")
+	testutil.CheckSchema(t, preds, types)
+
 	checks = []struct {
 		blank, expected string
 	}{
@@ -163,6 +185,7 @@ func TestBackupMinio(t *testing.T) {
 	// Perform second incremental backup.
 	_ = runBackup(t, 9, 3)
 	restored = runRestore(t, "", incr2.Txn.CommitTs)
+	testutil.CheckSchema(t, preds, types)
 
 	checks = []struct {
 		blank, expected string
@@ -187,6 +210,7 @@ func TestBackupMinio(t *testing.T) {
 	// Perform second full backup.
 	dirs := runBackupInternal(t, true, 12, 4)
 	restored = runRestore(t, "", incr3.Txn.CommitTs)
+	testutil.CheckSchema(t, preds, types)
 
 	// Check all the values were restored to their most recent value.
 	checks = []struct {
@@ -285,20 +309,8 @@ func runRestore(t *testing.T, lastDir string, commitTs uint64) map[string]string
 	}
 	pdir := "./data/restore/p1"
 	restored, err := testutil.GetPredicateValues(pdir, "movie", commitTs)
-	require.NoError(t, err)
-
-	restoredPreds, err := testutil.GetPredicateNames(pdir, commitTs)
-	require.NoError(t, err)
-	require.ElementsMatch(t, []string{"dgraph.graphql.schema", "dgraph.graphql.xid", "dgraph.type",
-		"movie"}, restoredPreds)
-
-	restoredTypes, err := testutil.GetTypeNames(pdir, commitTs)
-	require.NoError(t, err)
-	require.ElementsMatch(t, []string{"Node", "dgraph.graphql"}, restoredTypes)
-
 	require.NoError(t, err)
 	t.Logf("--- Restored values: %+v\n", restored)
-
 	return restored
 }
 

testutil/backup.go

@@ -18,6 +18,7 @@ package testutil
 
 import (
 	"fmt"
+	"testing"
 
 	"github.com/dgraph-io/badger/v2"
 	"github.com/dgraph-io/badger/v2/options"
@@ -26,6 +27,8 @@ import (
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/types"
 	"github.com/dgraph-io/dgraph/x"
+
+	"github.com/stretchr/testify/require"
 )
 
 // KEYFILE is set to the path of the file containing the key. Used for testing purposes only.
@@ -134,11 +137,32 @@ func readSchema(pdir string, dType dataType) ([]string, error) {
 }
 
 // GetPredicateNames returns the list of all the predicates stored in the restored pdir.
-func GetPredicateNames(pdir string, readTs uint64) ([]string, error) {
+func GetPredicateNames(pdir string) ([]string, error) {
 	return readSchema(pdir, schemaPredicate)
 }
 
 // GetTypeNames returns the list of all the types stored in the restored pdir.
-func GetTypeNames(pdir string, readTs uint64) ([]string, error) {
+func GetTypeNames(pdir string) ([]string, error) {
 	return readSchema(pdir, schemaType)
 }
+
+// CheckSchema checks the names of the predicates and types in the schema against the given names.
+func CheckSchema(t *testing.T, preds, types []string) {
+	pdirs := []string{
+		"./data/restore/p1",
+		"./data/restore/p2",
+		"./data/restore/p3",
+	}
+
+	restoredPreds := make([]string, 0)
+	for _, pdir := range pdirs {
+		groupPreds, err := GetPredicateNames(pdir)
+		require.NoError(t, err)
+		restoredPreds = append(restoredPreds, groupPreds...)
+
+		restoredTypes, err := GetTypeNames(pdir)
+		require.NoError(t, err)
+		require.ElementsMatch(t, types, restoredTypes)
+	}
+	require.ElementsMatch(t, preds, restoredPreds)
+}

worker/backup_processor.go

@@ -233,9 +233,12 @@ func (pr *BackupProcessor) toBackupList(key []byte, itr *badger.Iterator) (*bpb.
 	list := &bpb.KVList{}
 
 	item := itr.Item()
-	if item.Version() < pr.Request.SinceTs || item.IsDeletedOrExpired() {
+	if item.UserMeta() != posting.BitSchemaPosting && (item.Version() < pr.Request.SinceTs ||
+		item.IsDeletedOrExpired()) {
 		// Ignore versions less than given timestamp, or skip older versions of
 		// the given key by returning an empty list.
+		// Do not do this for schema and type keys. Those keys always have a
+		// version of one so they would be incorrectly rejected by above check.
 		return list, nil
 	}
 

worker/restore.go

@@ -84,6 +84,14 @@ func loadFromBackup(db *badger.DB, r io.Reader, preds predicateSet) (uint64, err
 	br := bufio.NewReaderSize(r, 16<<10)
 	unmarshalBuf := make([]byte, 1<<10)
 
+	// Delete schemas and types. Each backup file should have a complete copy of the schema.
+	if err := db.DropPrefix([]byte{x.ByteSchema}); err != nil {
+		return 0, err
+	}
+	if err := db.DropPrefix([]byte{x.ByteType}); err != nil {
+		return 0, err
+	}
+
 	loader := db.NewKVLoader(16)
 	var maxUid uint64
 	for {