Skip to content

Commit

Permalink
Merge pull request #45839 from dt/show-backups
Browse files Browse the repository at this point in the history 
backupccl: add appended backups to SHOW BACKUP
  • Loading branch information
@dt
dt authored Mar 8, 2020
2 parents dfa5bd5 + 7b00045 commit 2198386
Show file tree
Hide file tree
Showing 2 changed files with 162 additions and 123 deletions.
172 changes: 98 additions & 74 deletions pkg/ccl/backupccl/show.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,13 +81,14 @@ func showBackupPlanHook(
return err
}

store, err := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI(ctx, str)
if err != nil {
return errors.Wrapf(err, "make storage")
}
defer store.Close()

var encryption *roachpb.FileEncryptionOptions
if passphrase, ok := opts[backupOptEncPassphrase]; ok {
store, err := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI(ctx, str)
if err != nil {
return errors.Wrapf(err, "make storage")
}
defer store.Close()
opts, err := readEncryptionOptions(ctx, store)
if err != nil {
return err
Expand All @@ -96,21 +97,38 @@ func showBackupPlanHook(
encryption = &roachpb.FileEncryptionOptions{Key: encryptionKey}
}

desc, err := ReadBackupManifestFromURI(
ctx, str, p.ExecCfg().DistSQLSrv.ExternalStorageFromURI, encryption,
)
incPaths, err := findPriorBackups(ctx, store)
if err != nil {
return err
}

manifests := make([]BackupManifest, len(incPaths)+1)
manifests[0], err = readBackupManifestFromStore(ctx, store, encryption)
if err != nil {
return err
}

for i := range incPaths {
m, err := readBackupManifest(ctx, store, incPaths[i], encryption)
if err != nil {
return err
}
// Blank the stats to prevent memory blowup.
m.Statistics = nil
manifests[i+1] = m
}

// If we are restoring a backup with old-style foreign keys, skip over the
// FKs for which we can't resolve the cross-table references. We can't
// display them anyway, because we don't have the referenced table names,
// etc.
if err := maybeUpgradeTableDescsInBackupManifests(ctx, []BackupManifest{desc}, true /*skipFKsWithNoMatchingTable*/); err != nil {
if err := maybeUpgradeTableDescsInBackupManifests(
ctx, manifests, true, /*skipFKsWithNoMatchingTable*/
); err != nil {
return err
}

for _, row := range shower.fn(desc) {
for _, row := range shower.fn(manifests) {
select {
case <-ctx.Done():
return ctx.Err()
Expand All @@ -125,7 +143,7 @@ func showBackupPlanHook(

type backupShower struct {
header sqlbase.ResultColumns
fn func(BackupManifest) []tree.Datums
fn func([]BackupManifest) []tree.Datums
}

func backupShowerHeaders(showSchemas bool) sqlbase.ResultColumns {
Expand All @@ -146,55 +164,57 @@ func backupShowerHeaders(showSchemas bool) sqlbase.ResultColumns {
func backupShowerDefault(ctx context.Context, p sql.PlanHookState, showSchemas bool) backupShower {
return backupShower{
header: backupShowerHeaders(showSchemas),
fn: func(desc BackupManifest) []tree.Datums {
descs := make(map[sqlbase.ID]string)
for _, descriptor := range desc.Descriptors {
if database := descriptor.GetDatabase(); database != nil {
if _, ok := descs[database.ID]; !ok {
descs[database.ID] = database.Name
fn: func(manifests []BackupManifest) []tree.Datums {
var rows []tree.Datums
for _, manifest := range manifests {
descs := make(map[sqlbase.ID]string)
for _, descriptor := range manifest.Descriptors {
if database := descriptor.GetDatabase(); database != nil {
if _, ok := descs[database.ID]; !ok {
descs[database.ID] = database.Name
}
}
}
}
descSizes := make(map[sqlbase.ID]RowCount)
for _, file := range desc.Files {
// TODO(dan): This assumes each file in the backup only contains
// data from a single table, which is usually but not always
// correct. It does not account for interleaved tables or if a
// BACKUP happened to catch a newly created table that hadn't yet
// been split into its own range.
_, tableID, err := encoding.DecodeUvarintAscending(file.Span.Key)
if err != nil {
continue
}
s := descSizes[sqlbase.ID(tableID)]
s.add(file.EntryCounts)
descSizes[sqlbase.ID(tableID)] = s
}
start := tree.DNull
if desc.StartTime.WallTime != 0 {
start = tree.MakeDTimestamp(timeutil.Unix(0, desc.StartTime.WallTime), time.Nanosecond)
}
var rows []tree.Datums
var row tree.Datums
for _, descriptor := range desc.Descriptors {
if table := descriptor.Table(hlc.Timestamp{}); table != nil {
dbName := descs[table.ParentID]
row = tree.Datums{
tree.NewDString(dbName),
tree.NewDString(table.Name),
start,
tree.MakeDTimestamp(timeutil.Unix(0, desc.EndTime.WallTime), time.Nanosecond),
tree.NewDInt(tree.DInt(descSizes[table.ID].DataSize)),
tree.NewDInt(tree.DInt(descSizes[table.ID].Rows)),
descSizes := make(map[sqlbase.ID]RowCount)
for _, file := range manifest.Files {
// TODO(dan): This assumes each file in the backup only contains
// data from a single table, which is usually but not always
// correct. It does not account for interleaved tables or if a
// BACKUP happened to catch a newly created table that hadn't yet
// been split into its own range.
_, tableID, err := encoding.DecodeUvarintAscending(file.Span.Key)
if err != nil {
continue
}
if showSchemas {
schema, err := p.ShowCreate(ctx, dbName, desc.Descriptors, table, sql.OmitMissingFKClausesFromCreate)
if err != nil {
continue
s := descSizes[sqlbase.ID(tableID)]
s.add(file.EntryCounts)
descSizes[sqlbase.ID(tableID)] = s
}
start := tree.DNull
if manifest.StartTime.WallTime != 0 {
start = tree.MakeDTimestamp(timeutil.Unix(0, manifest.StartTime.WallTime), time.Nanosecond)
}
var row tree.Datums
for _, descriptor := range manifest.Descriptors {
if table := descriptor.Table(hlc.Timestamp{}); table != nil {
dbName := descs[table.ParentID]
row = tree.Datums{
tree.NewDString(dbName),
tree.NewDString(table.Name),
start,
tree.MakeDTimestamp(timeutil.Unix(0, manifest.EndTime.WallTime), time.Nanosecond),
tree.NewDInt(tree.DInt(descSizes[table.ID].DataSize)),
tree.NewDInt(tree.DInt(descSizes[table.ID].Rows)),
}
row = append(row, tree.NewDString(schema))
if showSchemas {
schema, err := p.ShowCreate(ctx, dbName, manifest.Descriptors, table, sql.OmitMissingFKClausesFromCreate)
if err != nil {
continue
}
row = append(row, tree.NewDString(schema))
}
rows = append(rows, row)
}
rows = append(rows, row)
}
}
return rows
Expand All @@ -210,14 +230,16 @@ var backupShowerRanges = backupShower{
{Name: "end_key", Typ: types.Bytes},
},

fn: func(desc BackupManifest) (rows []tree.Datums) {
for _, span := range desc.Spans {
rows = append(rows, tree.Datums{
tree.NewDString(span.Key.String()),
tree.NewDString(span.EndKey.String()),
tree.NewDBytes(tree.DBytes(span.Key)),
tree.NewDBytes(tree.DBytes(span.EndKey)),
})
fn: func(manifests []BackupManifest) (rows []tree.Datums) {
for _, manifest := range manifests {
for _, span := range manifest.Spans {
rows = append(rows, tree.Datums{
tree.NewDString(span.Key.String()),
tree.NewDString(span.EndKey.String()),
tree.NewDBytes(tree.DBytes(span.Key)),
tree.NewDBytes(tree.DBytes(span.EndKey)),
})
}
}
return rows
},
Expand All @@ -234,17 +256,19 @@ var backupShowerFiles = backupShower{
{Name: "rows", Typ: types.Int},
},

fn: func(desc BackupManifest) (rows []tree.Datums) {
for _, file := range desc.Files {
rows = append(rows, tree.Datums{
tree.NewDString(file.Path),
tree.NewDString(file.Span.Key.String()),
tree.NewDString(file.Span.EndKey.String()),
tree.NewDBytes(tree.DBytes(file.Span.Key)),
tree.NewDBytes(tree.DBytes(file.Span.EndKey)),
tree.NewDInt(tree.DInt(file.EntryCounts.DataSize)),
tree.NewDInt(tree.DInt(file.EntryCounts.Rows)),
})
fn: func(manifests []BackupManifest) (rows []tree.Datums) {
for _, manifest := range manifests {
for _, file := range manifest.Files {
rows = append(rows, tree.Datums{
tree.NewDString(file.Path),
tree.NewDString(file.Span.Key.String()),
tree.NewDString(file.Span.EndKey.String()),
tree.NewDBytes(tree.DBytes(file.Span.Key)),
tree.NewDBytes(tree.DBytes(file.Span.EndKey)),
tree.NewDInt(tree.DInt(file.EntryCounts.DataSize)),
tree.NewDInt(tree.DInt(file.EntryCounts.Rows)),
})
}
}
return rows
},
Expand Down
113 changes: 64 additions & 49 deletions pkg/ccl/backupccl/show_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,17 +9,16 @@
package backupccl_test

import (
"database/sql/driver"
"fmt"
"regexp"
"strconv"
"strings"
"testing"
"time"

"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
"github.com/cockroachdb/cockroach/pkg/util/leaktest"
"github.com/cockroachdb/cockroach/pkg/util/timeutil"
"github.com/stretchr/testify/require"
)

func TestShowBackup(t *testing.T) {
Expand All @@ -29,55 +28,70 @@ func TestShowBackup(t *testing.T) {
_, tc, sqlDB, _, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts, initNone)
defer cleanupFn()

full, inc, details := localFoo+"/full", localFoo+"/inc", localFoo+"/details"
const full, inc, inc2 = localFoo + "/full", localFoo + "/inc", localFoo + "/inc2"

beforeFull := timeutil.Now()
sqlDB.Exec(t, `BACKUP data.bank TO $1`, full)
beforeTS := sqlDB.QueryStr(t, `SELECT now()::string`)[0][0]
sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE data TO $1 AS OF SYSTEM TIME '%s'`, beforeTS), full)

var unused driver.Value
var start, end *time.Time
var dataSize, rows uint64
sqlDB.QueryRow(t, `SELECT * FROM [SHOW BACKUP $1] WHERE table_name = 'bank'`, full).Scan(
&unused, &unused, &start, &end, &dataSize, &rows,
)
if start != nil {
t.Errorf("expected null start time on full backup, got %v", *start)
}
if !(*end).After(beforeFull) {
t.Errorf("expected now (%s) to be in (%s, %s)", beforeFull, start, end)
}
if rows != numAccounts {
t.Errorf("expected %d got: %d", numAccounts, rows)
}
res := sqlDB.QueryStr(t, `SELECT table_name, start_time::string, end_time::string, rows FROM [SHOW BACKUP $1]`, full)
require.Equal(t, [][]string{{"bank", "NULL", beforeTS, strconv.Itoa(numAccounts)}}, res)

// Mess with half the rows.
affectedRows, err := sqlDB.Exec(t,
`UPDATE data.bank SET id = -1 * id WHERE id > $1`, numAccounts/2,
).RowsAffected()
if err != nil {
t.Fatal(err)
} else if affectedRows != numAccounts/2 {
t.Fatalf("expected to update %d rows, got %d", numAccounts/2, affectedRows)
}

beforeInc := timeutil.Now()
sqlDB.Exec(t, `BACKUP data.bank TO $1 INCREMENTAL FROM $2`, inc, full)

sqlDB.QueryRow(t, `SELECT * FROM [SHOW BACKUP $1] WHERE table_name = 'bank'`, inc).Scan(
&unused, &unused, &start, &end, &dataSize, &rows,
)
if start == nil {
t.Errorf("expected start time on inc backup, got %v", *start)
}
if !(*end).After(beforeInc) {
t.Errorf("expected now (%s) to be in (%s, %s)", beforeInc, start, end)
}
// We added affectedRows and removed affectedRows, so there should be 2*
// affectedRows in the backup.
if expected := affectedRows * 2; rows != uint64(expected) {
t.Errorf("expected %d got: %d", expected, rows)
}

require.NoError(t, err)
require.Equal(t, numAccounts/2, int(affectedRows))

// Backup the changes by appending to the base and by making a separate
// inc backup.
incTS := sqlDB.QueryStr(t, `SELECT now()::string`)[0][0]
sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE data TO $1 AS OF SYSTEM TIME '%s'`, incTS), full)
sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE data TO $1 AS OF SYSTEM TIME '%s' INCREMENTAL FROM $2`, incTS), inc, full)

// Check the appended base backup.
res = sqlDB.QueryStr(t, `SELECT table_name, start_time::string, end_time::string, rows FROM [SHOW BACKUP $1]`, full)
require.Equal(t, [][]string{
{"bank", "NULL", beforeTS, strconv.Itoa(numAccounts)},
{"bank", beforeTS, incTS, strconv.Itoa(int(affectedRows * 2))},
}, res)

// Check the separate inc backup.
res = sqlDB.QueryStr(t, `SELECT start_time::string, end_time::string, rows FROM [SHOW BACKUP $1]`, inc)
require.Equal(t, [][]string{
{beforeTS, incTS, strconv.Itoa(int(affectedRows * 2))},
}, res)

// Create two new tables, alphabetically on either side of bank.
sqlDB.Exec(t, `CREATE TABLE data.auth (id INT PRIMARY KEY, name STRING)`)
sqlDB.Exec(t, `CREATE TABLE data.users (id INT PRIMARY KEY, name STRING)`)
sqlDB.Exec(t, `INSERT INTO data.users VALUES (1, 'one'), (2, 'two'), (3, 'three')`)

// Backup the changes again, by appending to the base and by making a
// separate inc backup.
inc2TS := sqlDB.QueryStr(t, `SELECT now()::string`)[0][0]
sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE data TO $1 AS OF SYSTEM TIME '%s'`, inc2TS), full)
sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE data TO $1 AS OF SYSTEM TIME '%s' INCREMENTAL FROM $2, $3`, inc2TS), inc2, full, inc)

// Check the appended base backup.
res = sqlDB.QueryStr(t, `SELECT table_name, start_time::string, end_time::string, rows FROM [SHOW BACKUP $1]`, full)
require.Equal(t, [][]string{
{"bank", "NULL", beforeTS, strconv.Itoa(numAccounts)},
{"bank", beforeTS, incTS, strconv.Itoa(int(affectedRows * 2))},
{"bank", incTS, inc2TS, "0"},
{"auth", incTS, inc2TS, "0"},
{"users", incTS, inc2TS, "3"},
}, res)

// Check the separate inc backup.
res = sqlDB.QueryStr(t, `SELECT table_name, start_time::string, end_time::string, rows FROM [SHOW BACKUP $1]`, inc2)
require.Equal(t, [][]string{
{"bank", incTS, inc2TS, "0"},
{"auth", incTS, inc2TS, "0"},
{"users", incTS, inc2TS, "3"},
}, res)

const details = localFoo + "/details"
sqlDB.Exec(t, `CREATE TABLE data.details1 (c INT PRIMARY KEY)`)
sqlDB.Exec(t, `INSERT INTO data.details1 (SELECT generate_series(1, 100))`)
sqlDB.Exec(t, `ALTER TABLE data.details1 SPLIT AT VALUES (1), (42)`)
Expand All @@ -90,15 +104,15 @@ func TestShowBackup(t *testing.T) {
details2Key := roachpb.Key(sqlbase.MakeIndexKeyPrefix(details2Desc, details2Desc.PrimaryIndex.ID))

sqlDB.CheckQueryResults(t, fmt.Sprintf(`SHOW BACKUP RANGES '%s'`, details), [][]string{
{"/Table/54/1", "/Table/54/2", string(details1Key), string(details1Key.PrefixEnd())},
{"/Table/55/1", "/Table/55/2", string(details2Key), string(details2Key.PrefixEnd())},
{"/Table/56/1", "/Table/56/2", string(details1Key), string(details1Key.PrefixEnd())},
{"/Table/57/1", "/Table/57/2", string(details2Key), string(details2Key.PrefixEnd())},
})

var showFiles = fmt.Sprintf(`SELECT start_pretty, end_pretty, size_bytes, rows
FROM [SHOW BACKUP FILES '%s']`, details)
sqlDB.CheckQueryResults(t, showFiles, [][]string{
{"/Table/54/1/1", "/Table/54/1/42", "369", "41"},
{"/Table/54/1/42", "/Table/54/2", "531", "59"},
{"/Table/56/1/1", "/Table/56/1/42", "369", "41"},
{"/Table/56/1/42", "/Table/56/2", "531", "59"},
})
sstMatcher := regexp.MustCompile(`\d+\.sst`)
pathRows := sqlDB.QueryStr(t, `SELECT path FROM [SHOW BACKUP FILES $1]`, details)
Expand Down Expand Up @@ -213,6 +227,7 @@ COMMENT ON INDEX tablea_b_idx IS 'index'`
t.Fatalf("mismatched create statement: %s, want %s", createStmt, want)
}
}

}

func eqWhitespace(a, b string) bool {
Expand Down

0 comments on commit 2198386

Please sign in to comment.