fix(whpg-backup): verify table storage after locking and retry on a new snapshot

backup/backup.go

@@ -45,6 +45,7 @@ func DoSetup() {
 	SetLoggerVerbosity()
 	gplog.Verbose("Backup Command: %s", os.Args)
 	gplog.Info("gpbackup version = %s", GetVersion())
+	SetSnapshotAttemptsFromEnvironment()
 
 	utils.CheckGpexpandRunning(utils.BackupPreventedByGpexpandMessage)
 	timestamp := history.CurrentTimestamp()
@@ -129,6 +130,7 @@ func DoBackup() {
 
 	gplog.Info("Gathering table state information")
 	metadataTables, dataTables := RetrieveAndProcessTables()
+	backupReport.SkippedDataTables = sortedSkippedDataTables()
 	dataTables, numExtOrForeignTables := GetBackupDataSet(dataTables)
 	if len(dataTables) == 0 && !backupReport.MetadataOnly {
 		gplog.Warn("No tables in backup set contain data. Performing metadata-only backup instead.")
@@ -194,8 +196,9 @@ func DoBackup() {
 	}
 
 	printDataBackupWarnings(numExtOrForeignTables)
+	printSkippedDataTableWarnings()
 	if MustGetFlagBool(options.WITH_STATS) {
-		backupStatistics(metadataTables)
+		backupStatistics(tablesWithBackedUpData(metadataTables))
 	}
 
 	globalTOC.WriteToFileAndMakeReadOnly(globalFPInfo.GetTOCFilePath())

backup/data.go

@@ -7,6 +7,7 @@ package backup
 import (
 	"errors"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -380,6 +381,38 @@ func printDataBackupWarnings(numExtTables int64) {
 	}
 }
 
+func printSkippedDataTableWarnings() {
+	if len(skippedDataTables) > 0 {
+		gplog.Warn("Data of %d table(s) not backed up because they changed on disk after the backup snapshot was taken.", len(skippedDataTables))
+		gplog.Warn("See the backup report or %s for the list of tables.", gplog.GetLogFilePath())
+	}
+}
+
+// tablesWithBackedUpData leaves out the relations whose data changed after the
+// snapshot or was left out because of that: statistics describe data, and a
+// restore creates these tables empty.
+func tablesWithBackedUpData(tables []Table) []Table {
+	if len(changedRelations) == 0 {
+		return tables
+	}
+	kept := make([]Table, 0, len(tables))
+	for _, table := range tables {
+		if !changedRelations[table.FQN()] {
+			kept = append(kept, table)
+		}
+	}
+	return kept
+}
+
+func sortedSkippedDataTables() []string {
+	tables := make([]string, 0, len(skippedDataTables))
+	for fqn := range skippedDataTables {
+		tables = append(tables, fqn)
+	}
+	sort.Strings(tables)
+	return tables
+}
+
 // Remove external/foreign tables from the data backup set
 func GetBackupDataSet(tables []Table) ([]Table, int64) {
 	var backupDataSet []Table

backup/display_report_test.go

@@ -65,6 +65,16 @@ var _ = Describe("display-report internal tests", func() {
 			Expect(backupError).To(Equal("could not dispatch to segment seg0\nconnection refused: server closed"))
 		})
 
+		It("keeps the data-not-backed-up line out of the backup error text", func() {
+			text := "backup status:         Failure\n" +
+				"backup error:          could not dispatch to segment seg0\n\n" +
+				"data not backed up:    public.ao_t, public.heap_t\n\n" +
+				"count of database objects in backup:\ntables 1\n"
+			fields, _, backupError := parseReportText(text)
+			Expect(backupError).To(Equal("could not dispatch to segment seg0"))
+			Expect(fields).To(HaveKeyWithValue("data_not_backed_up", "public.ao_t, public.heap_t"))
+		})
+
 		It("folds a colonless continuation line into the most recently seen key", func() {
 			text := "incremental backup set:\n20260101000000\n20260102000000\n\ncount of database objects in backup:\ntables 1\n"
 			fields, objectCounts, _ := parseReportText(text)

backup/global_variables.go

@@ -33,6 +33,13 @@ const (
 /*
  * Non-flag variables
  */
+// SnapshotAttemptsEnvVar names the environment variable that tunes how many
+// snapshots a backup may take when tables change under it; see
+// SetSnapshotAttemptsFromEnvironment.
+const SnapshotAttemptsEnvVar = "WHPGBACKUP_SNAPSHOT_ATTEMPTS"
+
+const defaultSnapshotAttempts = 3
+
 var (
 	backupReport         *report.Report
 	connectionPool       *dbconn.DBConn
@@ -48,6 +55,18 @@ var (
 	filterRelationClause string
 	quotedRoleNames      map[string]string
 	backupSnapshot       string
+	// How many snapshots lockBackupSet may try before it gives up on tables
+	// that keep changing under it and skips their data.
+	maxSnapshotAttempts = defaultSnapshotAttempts
+	// The relations still changing on disk after the last snapshot attempt
+	// and the locked tables above them, by FQN. Their data cannot be read
+	// under the snapshot, and a changed relation's pg_aoseg name is stale.
+	changedRelations map[string]bool
+	// FQNs of the tables left out of the data backup set because of that.
+	skippedDataTables map[string]bool
+	// The filter options the backup started with, kept so the include and
+	// exclude lists can be resolved again under a new snapshot.
+	filterOptions *options.Options
 	/*
 	 * Used for synchronizing DoCleanup.  In DoInit() we increment the group
 	 * and then wait for at least one DoCleanup to finish, either in DoTeardown
@@ -163,6 +182,18 @@ func SetFilterRelationClause(filterClause string) {
 	filterRelationClause = filterClause
 }
 
+func SetMaxSnapshotAttempts(attempts int) {
+	maxSnapshotAttempts = attempts
+}
+
+func GetMaxSnapshotAttempts() int {
+	return maxSnapshotAttempts
+}
+
+func GetSkippedDataTables() map[string]bool {
+	return skippedDataTables
+}
+
 func SetQuotedRoleNames(quotedRoles map[string]string) {
 	quotedRoleNames = quotedRoles
 }

backup/queries_incremental.go

@@ -35,8 +35,15 @@ type aoSegTable struct {
 func getAllModCounts(connectionPool *dbconn.DBConn) map[string]int64 {
 	var segTableFQNs = getAOSegTableFQNs(connectionPool)
 	modCounts := make(map[string]int64)
-	for aoTableFQN, segTable := range segTableFQNs {
-		modCounts[aoTableFQN] = getModCount(connectionPool, segTable)
+	for aoTableFQN, segTableFQN := range segTableFQNs {
+		// A relation that changed on disk since the snapshot, or a table whose
+		// data was left out because a partition did, gets no incremental
+		// entry: its data is not backed up, and a changed relation's segment
+		// relation name comes from the snapshot and may no longer exist.
+		if changedRelations[aoTableFQN] {
+			continue
+		}
+		modCounts[aoTableFQN] = getModCount(connectionPool, segTableFQN)
 	}
 	return modCounts
 }

backup/queries_relation_test.go

@@ -61,4 +61,102 @@ var _ = Describe("backup internal tests", func() {
 			structmatcher.ExpectStructsToMatch(&expectedResult[0], &result[0])
 		})
 	})
+	Describe("GetRelationsChangedSinceSnapshot", func() {
+		tables := []backup.Relation{
+			{SchemaOid: 2200, Oid: 101, Schema: "public", Name: "unchanged"},
+			{SchemaOid: 2200, Oid: 102, Schema: "public", Name: "rewritten"},
+			{SchemaOid: 2200, Oid: 103, Schema: "public", Name: "dropped"},
+			{SchemaOid: 2200, Oid: 104, Schema: "public", Name: "parted"},
+		}
+		header := []string{"oid", "parentoid", "schema", "name", "snapshotrelfilenode", "currentrelfilenode"}
+		// The partition root has no storage of its own; its data lives in leaf 201.
+		unchangedRows := func() *sqlmock.Rows {
+			return sqlmock.NewRows(header).
+				AddRow(101, nil, "public", "unchanged", 1001, 1001).
+				AddRow(102, nil, "public", "rewritten", 1002, 1002).
+				AddRow(103, nil, "public", "dropped", 1003, 1003).
+				AddRow(104, nil, "public", "parted", 0, nil).
+				AddRow(201, 104, "public", "parted_1_prt_1", 2001, 2001)
+		}
+
+		It("returns nothing without querying when there are no tables", func() {
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, []backup.Relation{})
+			Expect(changed).To(BeEmpty())
+			Expect(mock.ExpectationsWereMet()).To(Succeed())
+		})
+		It("returns nothing when every relation still has the relfilenode the snapshot saw", func() {
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(unchangedRows())
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(BeEmpty())
+		})
+		It("reports a table whose live relfilenode differs from the snapshot as rewritten", func() {
+			rows := sqlmock.NewRows(header).
+				AddRow(101, nil, "public", "unchanged", 1001, 1001).
+				AddRow(102, nil, "public", "rewritten", 1002, 2002).
+				AddRow(103, nil, "public", "dropped", 1003, 1003)
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(rows)
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(HaveLen(1))
+			Expect(changed[0].Relation).To(Equal(tables[1]))
+			Expect(changed[0].Dropped).To(BeFalse())
+			Expect(changed[0].Ancestors).To(BeEmpty())
+		})
+		It("reports a table with no live relfilenode as dropped", func() {
+			rows := sqlmock.NewRows(header).
+				AddRow(101, nil, "public", "unchanged", 1001, 1001).
+				AddRow(102, nil, "public", "rewritten", 1002, 1002).
+				AddRow(103, nil, "public", "dropped", 1003, nil)
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(rows)
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(HaveLen(1))
+			Expect(changed[0].Relation).To(Equal(tables[2]))
+			Expect(changed[0].Dropped).To(BeTrue())
+		})
+		It("reports a changed partition below a locked table with the table as its ancestor", func() {
+			rows := sqlmock.NewRows(header).
+				AddRow(104, nil, "public", "parted", 0, nil).
+				AddRow(201, 104, "public", "parted_1_prt_1", 2001, 2002)
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(rows)
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(HaveLen(1))
+			Expect(changed[0].FQN()).To(Equal("public.parted_1_prt_1"))
+			Expect(changed[0].Dropped).To(BeFalse())
+			Expect(changed[0].Ancestors).To(Equal([]backup.Relation{tables[3]}))
+			Expect(changed[0].Describe()).To(Equal("public.parted_1_prt_1 (rewritten or truncated, partition of public.parted)"))
+		})
+		It("does not compare relations without storage", func() {
+			rows := sqlmock.NewRows(header).
+				AddRow(104, nil, "public", "parted", 0, nil)
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(rows)
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(BeEmpty())
+		})
+		It("describes a changed table with its reason", func() {
+			rewritten := backup.ChangedRelation{Relation: tables[1], Dropped: false}
+			dropped := backup.ChangedRelation{Relation: tables[2], Dropped: true}
+			Expect(rewritten.Reason()).To(Equal("rewritten or truncated"))
+			Expect(rewritten.Describe()).To(Equal("public.rewritten (rewritten or truncated)"))
+			Expect(dropped.Reason()).To(Equal("dropped and recreated"))
+			Expect(dropped.Describe()).To(Equal("public.dropped (dropped and recreated)"))
+		})
+		It("lists locked tables in input order, each followed by its changed partitions", func() {
+			rows := sqlmock.NewRows(header).
+				AddRow(201, 104, "public", "parted_1_prt_1", 2001, 2002).
+				AddRow(104, nil, "public", "parted", 0, nil).
+				AddRow(103, nil, "public", "dropped", 1003, nil).
+				AddRow(102, nil, "public", "rewritten", 1002, 2002)
+			mock.ExpectQuery(`WITH RECURSIVE (.*)`).WillReturnRows(rows)
+
+			changed := backup.GetRelationsChangedSinceSnapshot(connectionPool, tables)
+			Expect(changed).To(HaveLen(3))
+			Expect(changed[0].Name).To(Equal("rewritten"))
+			Expect(changed[1].Name).To(Equal("dropped"))
+			Expect(changed[2].Name).To(Equal("parted_1_prt_1"))
+		})
+	})
 })