feat(backup): file-hash incremental detection for heap tables

backup/backup.go

@@ -49,7 +49,17 @@ func DoSetup() {
 	utils.CheckGpexpandRunning(utils.BackupPreventedByGpexpandMessage)
 	timestamp := history.CurrentTimestamp()
 	createBackupLockFile(timestamp)
+	// Heap file hashes must be read before initializeConnectionPool takes the
+	// backup snapshot, and verified after it; see CollectHeapFileHashes.
+	var hashedHeapTables []heapTable
+	if MustGetFlagBool(options.HEAP_FILE_HASH) && collectsIncrementalMetadata() {
+		gplog.Info("Collecting heap table file hashes")
+		hashedHeapTables, heapFileHashes = CollectHeapFileHashes(MustGetFlagString(options.DBNAME))
+	}
 	initializeConnectionPool(timestamp)
+	if heapFileHashes != nil {
+		heapFileHashes = VerifyHeapFileHashes(MustGetFlagString(options.DBNAME), hashedHeapTables, heapFileHashes)
+	}
 	gplog.Info("Apache Cloudberry Version = %s", connectionPool.Version.VersionString)
 
 	gplog.Info("Starting backup of database %s", MustGetFlagString(options.DBNAME))
@@ -105,6 +115,11 @@ func DoSetup() {
 	}
 }
 
+// This must be a full backup with --leaf-parition-data to query for incremental metadata
+func collectsIncrementalMetadata() bool {
+	return !(MustGetFlagBool(options.METADATA_ONLY) || MustGetFlagBool(options.DATA_ONLY)) && MustGetFlagBool(options.LEAF_PARTITION_DATA)
+}
+
 func DoBackup() {
 	gplog.Info("Backup Timestamp = %s", globalFPInfo.Timestamp)
 	gplog.Info("Backup Database = %s", connectionPool.DBName)
@@ -134,8 +149,7 @@ func DoBackup() {
 		gplog.Warn("No tables in backup set contain data. Performing metadata-only backup instead.")
 		backupReport.MetadataOnly = true
 	}
-	// This must be a full backup with --leaf-parition-data to query for incremental metadata
-	if !(MustGetFlagBool(options.METADATA_ONLY) || MustGetFlagBool(options.DATA_ONLY)) && MustGetFlagBool(options.LEAF_PARTITION_DATA) {
+	if collectsIncrementalMetadata() {
 		backupIncrementalMetadata()
 	} else {
 		gplog.Verbose("Skipping query for incremental metadata.")
@@ -158,7 +172,7 @@ func DoBackup() {
 
 			targetBackupTOC := toc.NewTOC(targetBackupFPInfo.GetTOCFilePath())
 			targetBackupRestorePlan = history.ReadConfigFile(targetBackupFPInfo.GetConfigFilePath()).RestorePlan
-			backupSetTables = FilterTablesForIncremental(targetBackupTOC, globalTOC, dataTables)
+			backupSetTables = FilterTablesForIncremental(targetBackupTOC, globalTOC, targetBackupRestorePlan, dataTables)
 		}
 
 		backupReport.RestorePlan = PopulateRestorePlan(backupSetTables, targetBackupRestorePlan, dataTables)

backup/global_variables.go

@@ -48,6 +48,7 @@ var (
 	filterRelationClause string
 	quotedRoleNames      map[string]string
 	backupSnapshot       string
+	heapFileHashes       map[uint32]string
 	/*
 	 * Used for synchronizing DoCleanup.  In DoInit() we increment the group
 	 * and then wait for at least one DoCleanup to finish, either in DoTeardown

backup/incremental.go

@@ -17,16 +17,27 @@ import (
  * Return the subset of tables whose data must be written to the current
  * incremental backup.  A table is skipped only when we can prove its data did
  * not change since the backup we are building on; any table we have no change
- * markers for (heap, external, or a storage format this version does not know
- * how to track) is always included.
+ * markers for (external, a storage format this version does not know how to
+ * track, or a heap table without file hashes) is always included.
  *
  * Each storage format is checked on three independent signals, and any one of
  * them differing re-reads the table.  They cover each other: the content
  * signal (AO modcount, PAX aux digest) misses pure DDL, the DDL timestamp
  * misses CLUSTER because pg_stat_last_operation does not record it, and the
  * storage identity misses nothing about a rewrite but says nothing about DML.
+ * Heap tables are checked as described in heapTableUnchanged.
  */
-func FilterTablesForIncremental(lastBackupTOC, currentTOC *toc.TOC, tables []Table) []Table {
+func FilterTablesForIncremental(lastBackupTOC, currentTOC *toc.TOC, lastBackupRestorePlan []history.RestorePlanEntry, tables []Table) []Table {
+	// A skipped heap table's data is restored from the backup the previous
+	// restore plan points it to; a table in no entry has no data to restore
+	// from, e.g. it belonged to an extension, whose tables' data is not
+	// backed up.
+	restorableFQNs := make(map[string]bool)
+	for _, entry := range lastBackupRestorePlan {
+		for _, fqn := range entry.TableFQNs {
+			restorableFQNs[fqn] = true
+		}
+	}
 	var filteredTables []Table
 	for _, table := range tables {
 		tableFQN := table.FQN()
@@ -51,12 +62,33 @@ func FilterTablesForIncremental(lastBackupTOC, currentTOC *toc.TOC, tables []Tab
 			continue
 		}
 
-		filteredTables = append(filteredTables, table)
+		if !restorableFQNs[tableFQN] || !heapTableUnchanged(lastBackupTOC, currentTOC, tableFQN) {
+			filteredTables = append(filteredTables, table)
+		}
 	}
 
 	return filteredTables
 }
 
+// heapTableUnchanged reports whether a table that is neither AO nor PAX can be
+// skipped. That needs a
+// file hash in both backups (recorded only with --heap-file-hash) and matching
+// hashes, DDL timestamps and column hashes; anything else includes the table,
+// as before.
+func heapTableUnchanged(lastBackupTOC, currentTOC *toc.TOC, fqn string) bool {
+	currentEntry, ok := currentTOC.IncrementalMetadata.Heap[fqn]
+	if !ok || currentEntry.FileHashMD5 == "" {
+		return false
+	}
+	previousEntry, ok := lastBackupTOC.IncrementalMetadata.Heap[fqn]
+	if !ok || previousEntry.FileHashMD5 == "" {
+		return false
+	}
+	return previousEntry.FileHashMD5 == currentEntry.FileHashMD5 &&
+		previousEntry.LastDDLTimestamp == currentEntry.LastDDLTimestamp &&
+		previousEntry.ColumnsMD5 == currentEntry.ColumnsMD5
+}
+
 func GetTargetBackupTimestamp() string {
 	targetTimestamp := ""
 	if fromTimestamp := MustGetFlagString(options.FROM_TIMESTAMP); fromTimestamp != "" {

backup/incremental_test.go

@@ -118,7 +118,7 @@ var _ = Describe("backup/incremental tests", func() {
 			tblPAXRewritten,
 		}
 
-		filteredTables := backup.FilterTablesForIncremental(&prevTOC, &currTOC, tables)
+		filteredTables := backup.FilterTablesForIncremental(&prevTOC, &currTOC, nil, tables)
 
 		It("Should include the heap table in the filtered list", func() {
 			Expect(filteredTables).To(ContainElement(tblHeap))
@@ -184,11 +184,89 @@ var _ = Describe("backup/incremental tests", func() {
 		tblPAX := backup.Table{Relation: backup.Relation{Schema: "public", Name: "pax_table"}}
 
 		It("Should include the PAX table", func() {
-			Expect(backup.FilterTablesForIncremental(&prevTOC, &currTOC, []backup.Table{tblPAX})).
+			Expect(backup.FilterTablesForIncremental(&prevTOC, &currTOC, nil, []backup.Table{tblPAX})).
 				To(ContainElement(tblPAX))
 		})
 	})
 
+	Describe("FilterTablesForIncremental with heap file hashes", func() {
+		heapEntry := toc.HeapEntry{FileHashMD5: "hash1", LastDDLTimestamp: "ddl1"}
+		prevTOC := toc.TOC{
+			IncrementalMetadata: toc.IncrementalEntries{
+				Heap: map[string]toc.HeapEntry{
+					"public.heap_unchanged":     heapEntry,
+					"public.heap_changed_hash":  heapEntry,
+					"public.heap_changed_ddl":   heapEntry,
+					"public.heap_changed_types": heapEntry,
+					"public.heap_no_curr_hash":  heapEntry,
+					"public.heap_empty_prev":    {LastDDLTimestamp: "ddl1"},
+					"public.heap_no_prev_data":  heapEntry,
+				},
+			},
+		}
+		currTOC := toc.TOC{
+			IncrementalMetadata: toc.IncrementalEntries{
+				Heap: map[string]toc.HeapEntry{
+					"public.heap_unchanged":     heapEntry,
+					"public.heap_changed_hash":  {FileHashMD5: "hash2", LastDDLTimestamp: "ddl1"},
+					"public.heap_changed_ddl":   {FileHashMD5: "hash1", LastDDLTimestamp: "ddl2"},
+					"public.heap_changed_types": {FileHashMD5: "hash1", LastDDLTimestamp: "ddl1", ColumnsMD5: "types2"},
+					"public.heap_no_prev":       heapEntry,
+					"public.heap_empty_prev":    heapEntry,
+					"public.heap_no_prev_data":  heapEntry,
+				},
+			},
+		}
+
+		tblUnchanged := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_unchanged"}}
+		tblChangedHash := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_changed_hash"}}
+		tblChangedDDL := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_changed_ddl"}}
+		tblChangedTypes := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_changed_types"}}
+		tblNoCurrHash := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_no_curr_hash"}}
+		tblNoPrev := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_no_prev"}}
+		tblEmptyPrev := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_empty_prev"}}
+		tblNoPrevData := backup.Table{Relation: backup.Relation{Schema: "public", Name: "heap_no_prev_data"}}
+		tables := []backup.Table{tblUnchanged, tblChangedHash, tblChangedDDL, tblChangedTypes, tblNoCurrHash, tblNoPrev, tblEmptyPrev, tblNoPrevData}
+		prevRestorePlan := []history.RestorePlanEntry{
+			{Timestamp: "ts1", TableFQNs: []string{"public.heap_unchanged", "public.heap_changed_hash"}},
+			{Timestamp: "ts2", TableFQNs: []string{"public.heap_changed_ddl", "public.heap_changed_types",
+				"public.heap_no_curr_hash", "public.heap_no_prev", "public.heap_empty_prev"}},
+		}
+
+		filteredTables := backup.FilterTablesForIncremental(&prevTOC, &currTOC, prevRestorePlan, tables)
+
+		It("Should NOT include the heap table with unchanged hash and DDL timestamp", func() {
+			Expect(filteredTables).To(Not(ContainElement(tblUnchanged)))
+		})
+
+		It("Should include the heap table having a modified file hash", func() {
+			Expect(filteredTables).To(ContainElement(tblChangedHash))
+		})
+
+		It("Should include the heap table having a modified last DDL timestamp", func() {
+			Expect(filteredTables).To(ContainElement(tblChangedDDL))
+		})
+
+		It("Should include the heap table whose data is in no entry of the previous restore plan", func() {
+			Expect(filteredTables).To(ContainElement(tblNoPrevData))
+		})
+
+		It("Should include the heap table having modified columns", func() {
+			Expect(filteredTables).To(ContainElement(tblChangedTypes))
+		})
+
+		It("Should include heap tables missing a hash in either backup", func() {
+			Expect(filteredTables).To(ContainElement(tblNoCurrHash))
+			Expect(filteredTables).To(ContainElement(tblNoPrev))
+			Expect(filteredTables).To(ContainElement(tblEmptyPrev))
+		})
+
+		It("Should include every heap table when the previous backup has no heap hashes", func() {
+			filtered := backup.FilterTablesForIncremental(&toc.TOC{}, &currTOC, prevRestorePlan, tables)
+			Expect(filtered).To(Equal(tables))
+		})
+	})
+
 	Describe("GetLatestMatchingBackupConfig", func() {
 		historyDBPath := "/tmp/hist.db"
 		contents := []history.BackupConfig{