diff --git a/backup/backup.go b/backup/backup.go index 07c2d253..ddbcc197 100644 --- a/backup/backup.go +++ b/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) diff --git a/backup/global_variables.go b/backup/global_variables.go index 3397c6b3..01233617 100644 --- a/backup/global_variables.go +++ b/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 diff --git a/backup/incremental.go b/backup/incremental.go index bd64eeb3..95c92576 100644 --- a/backup/incremental.go +++ b/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 != "" { diff --git a/backup/incremental_test.go b/backup/incremental_test.go index 532d0c89..14a17eba 100644 --- a/backup/incremental_test.go +++ b/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{ diff --git a/backup/queries_incremental.go b/backup/queries_incremental.go index 332895e1..a484bb1a 100644 --- a/backup/queries_incremental.go +++ b/backup/queries_incremental.go @@ -2,6 +2,9 @@ package backup import ( "fmt" + "strconv" + "strings" + "time" "github.com/apache/cloudberry-backup/toc" "github.com/apache/cloudberry-go-libs/dbconn" @@ -198,6 +201,508 @@ func selectLastDDLTimestamps(connectionPool *dbconn.DBConn, query string) map[st return resultMap } +// heapTable carries the catalog identity of a heap table eligible for file-hash +// incremental backup. Hashes are keyed by Oid, which is also what we pass into +// the file-hash plpgsql function (pg_relation_filepath resolves it locally on +// each segment); FQN is only for log messages, since a table may be renamed +// before the backup snapshot is taken. +type heapTable struct { + Oid uint32 + FQN string +} + +// heapFileHashBatchSize bounds the number of oids hashed per query. +const heapFileHashBatchSize = 1000 + +// CollectHeapFileHashes returns a per-table hash of the on-disk state of every +// heap table in the database, along with the tables it looked at. It must run +// before the backup snapshot is taken, and VerifyHeapFileHashes after it: a +// change committed after the hash was read either is flushed by the next +// backup's CHECKPOINT, moving the data files on, or, if committed before the +// snapshot, is caught by VerifyHeapFileHashes. +// +// A nil map means the hashes could not be collected; every heap table is +// then included in the next incremental, as without --heap-file-hash. +func CollectHeapFileHashes(dbName string) ([]heapTable, map[uint32]string) { + hashConn := dbconn.NewDBConnFromEnvironment(dbName) + hashConn.MustConnect(1) + defer hashConn.Close() + + // The column hashes (see getHeapColumnHashes) need a recursive query. + if hashConn.Version.IsGPDB() && hashConn.Version.Before("6") { + gplog.Warn("--heap-file-hash requires Greenplum 6 or later, skipping heap file hash collection") + return nil, nil + } + + if !ensureFileStatFunction(hashConn) { + return nil, nil + } + defer dropFileStatFunction(hashConn) + + // CHECKPOINT flushes dirty pages so pg_stat_file sees up-to-date mtime/size. + // Without it a committed change may still sit in shared buffers, and the + // hash would match the previous backup's. + gplog.Verbose("Executing CHECKPOINT before heap file hash collection") + if _, err := hashConn.Exec("CHECKPOINT;", 0); err != nil { + gplog.Warn("CHECKPOINT failed, skipping heap file hash collection: %v", err) + return nil, nil + } + // pg_stat_file reports mtime in whole seconds, so gpbackup_file_info refuses + // to hash a file modified in the current second. Wait out the second the + // CHECKPOINT wrote in, so only files written during collection are refused. + time.Sleep(time.Second) + + heapTables, err := getHeapTables(hashConn, SchemaFilterClause("n")) + if err != nil { + gplog.Warn("Could not list heap tables, skipping heap file hash collection: %v", err) + return nil, nil + } + hashes := getFileHashesForTables(hashConn, heapTables) + + // The data files also hold rows of transactions still in progress, and + // their COMMIT only touches the commit log, not the files. Such a table + // must not keep its hash, or the commit would never show up as a change. + // Checking after the hashes are read and before the snapshot is enough: + // a writer that committed in between is visible to the backup, and one + // still in progress still holds its lock here. + writtenOids, err := getTablesWithActiveWriters(hashConn) + if err != nil { + gplog.Warn("Could not check for in-progress writers, skipping heap file hash collection: %v", err) + return nil, nil + } + for _, t := range heapTables { + if _, ok := hashes[t.Oid]; ok && writtenOids[t.Oid] { + gplog.Verbose("Not recording a file hash for %s: a transaction writing to it is in progress", t.FQN) + delete(hashes, t.Oid) + } + } + return heapTables, hashes +} + +// VerifyHeapFileHashes runs after the backup snapshot is taken and keeps only +// the hashes of tables whose files did not change since CollectHeapFileHashes +// read them. A change committed after the first CHECKPOINT may still sit in +// shared buffers, invisible to that hash but visible to the snapshot; a +// second CHECKPOINT flushes it, so the table's files no longer match and its +// hash is dropped. +func VerifyHeapFileHashes(dbName string, heapTables []heapTable, hashes map[uint32]string) map[uint32]string { + hashConn := dbconn.NewDBConnFromEnvironment(dbName) + hashConn.MustConnect(1) + defer hashConn.Close() + + if !ensureFileStatFunction(hashConn) { + return nil + } + defer dropFileStatFunction(hashConn) + + gplog.Verbose("Executing CHECKPOINT to verify heap file hashes") + if _, err := hashConn.Exec("CHECKPOINT;", 0); err != nil { + gplog.Warn("CHECKPOINT failed, skipping heap file hash collection: %v", err) + return nil + } + + hashedTables := make([]heapTable, 0, len(hashes)) + for _, t := range heapTables { + if _, ok := hashes[t.Oid]; ok { + hashedTables = append(hashedTables, t) + } + } + return keepUnchangedHashes(hashedTables, hashes, getFileHashesForTables(hashConn, hashedTables)) +} + +// keepUnchangedHashes returns the hashes of tables that current still reports +// unchanged. A table missing from current could not be re-read, so its hash +// is dropped too. +func keepUnchangedHashes(tables []heapTable, hashes, current map[uint32]string) map[uint32]string { + verified := make(map[uint32]string, len(hashes)) + for _, t := range tables { + hash, ok := hashes[t.Oid] + if !ok { + continue + } + if current[t.Oid] == hash { + verified[t.Oid] = hash + } else { + gplog.Verbose("Not recording a file hash for %s: its files changed while the backup started", t.FQN) + } + } + return verified +} + +// getTablesWithActiveWriters returns the oids of relations on which another +// transaction holds a lock taken by data-modifying commands (INSERT, UPDATE, +// DELETE, COPY FROM, TRUNCATE, ...). Such a lock is held until the writing +// transaction ends. +func getTablesWithActiveWriters(hashConn *dbconn.DBConn) (map[uint32]bool, error) { + query := ` + SELECT DISTINCT l.relation AS oid + FROM pg_catalog.pg_locks l + WHERE l.locktype = 'relation' + AND l.granted + AND l.pid IS DISTINCT FROM pg_catalog.pg_backend_pid() + AND l.database = (SELECT oid FROM pg_catalog.pg_database WHERE datname = pg_catalog.current_database()) + AND l.mode IN ('RowExclusiveLock', 'ShareRowExclusiveLock', 'ExclusiveLock', 'AccessExclusiveLock')` + + var results []struct{ Oid uint32 } + if err := hashConn.Select(&results, query); err != nil { + return nil, err + } + oids := make(map[uint32]bool, len(results)) + for _, r := range results { + oids[r.Oid] = true + } + return oids, nil +} + +// ensureFileStatFunction (re)creates gp_toolkit.gpbackup_file_info, a plpgsql +// function that describes the main fork of a relation on the local segment: +// its path and the mtime+size of each of its segment files. Pure SQL, no +// plpython/shell. +// +// pg_relation_filepath only returns the first segment file; a relation larger +// than RELSEG_SIZE (1GB by default) continues in path.1, path.2, ... (see +// md.c), so the function walks them until one is missing. The path itself +// carries the relfilenode, which catches rewrites (TRUNCATE, VACUUM FULL, +// CLUSTER). The function returns an empty string when a file was modified in +// the current second, since a later write in that same second would not move +// the mtime; the caller then records no hash, forcing the table into the next +// incremental. +// +// The function is created for each phase and dropped when it ends, so a body +// left by an older gpbackup is never reused and no object is left behind in +// gp_toolkit (an extension on GPDB 7 and Cloudberry). +func ensureFileStatFunction(hashConn *dbconn.DBConn) bool { + gplog.Verbose("Creating gp_toolkit.gpbackup_file_info function") + + // Drop the (text, text) signature from earlier versions of this feature. + _, _ = hashConn.Exec("DROP FUNCTION IF EXISTS gp_toolkit.gpbackup_file_info(text, text);", 0) + + createSQL := ` +CREATE OR REPLACE FUNCTION gp_toolkit.gpbackup_file_info(p_oid oid) +RETURNS text AS $BODY$ +DECLARE + v_base text; + v_path text; + v_mod timestamptz; + v_size bigint; + v_now timestamptz; + v_segno int := 0; + v_result text; +BEGIN + v_base := pg_relation_filepath(p_oid); + IF v_base IS NULL THEN + RETURN ''; + END IF; + v_result := v_base; + v_now := date_trunc('second', clock_timestamp()); + + LOOP + IF v_segno = 0 THEN + v_path := v_base; + ELSE + v_path := v_base || '.' || v_segno; + END IF; + + BEGIN + SELECT s.modification, s.size INTO v_mod, v_size FROM pg_stat_file(v_path) s; + EXCEPTION WHEN undefined_file THEN + EXIT; + END; + + IF v_mod >= v_now THEN + RETURN ''; + END IF; + v_result := v_result || ';' || v_segno || ':' || v_mod::text || '|' || v_size; + v_segno := v_segno + 1; + END LOOP; + + IF v_segno = 0 THEN + RETURN ''; + END IF; + RETURN v_result; +END; +$BODY$ LANGUAGE plpgsql;` + + if _, err := hashConn.Exec(createSQL, 0); err != nil { + gplog.Warn("Could not create gpbackup_file_info function, skipping heap file hash collection: %v", err) + return false + } + return true +} + +func dropFileStatFunction(hashConn *dbconn.DBConn) { + if _, err := hashConn.Exec("DROP FUNCTION IF EXISTS gp_toolkit.gpbackup_file_info(oid);", 0); err != nil { + gplog.Warn("Could not drop gpbackup_file_info function: %v", err) + } +} + +// getHeapTables returns (oid, FQN) pairs of heap tables, leaf partitions included. +func getHeapTables(conn *dbconn.DBConn, filterClause string) ([]heapTable, error) { + amJoin, heapFilter := heapTableFilter(conn) + query := fmt.Sprintf(` + SELECT c.oid, + quote_ident(n.nspname) || '.' || quote_ident(c.relname) AS tablefqn + FROM pg_class c + JOIN pg_namespace n ON c.relnamespace = n.oid + %s + WHERE %s + AND c.relkind = 'r' + AND %s`, amJoin, heapFilter, filterClause) + + var results []struct { + Oid uint32 + TableFQN string + } + if err := conn.Select(&results, query); err != nil { + return nil, err + } + out := make([]heapTable, len(results)) + for i, r := range results { + out[i] = heapTable{Oid: r.Oid, FQN: r.TableFQN} + } + return out, nil +} + +// getFileHashesForTables hashes gpbackup_file_info over every segment for each +// table, in batches. A table gets a hash only if every segment described its +// files; otherwise it is left out and the next incremental includes it. +func getFileHashesForTables(hashConn *dbconn.DBConn, tables []heapTable) map[uint32]string { + result := make(map[uint32]string) + for start := 0; start < len(tables); start += heapFileHashBatchSize { + end := start + heapFileHashBatchSize + if end > len(tables) { + end = len(tables) + } + batch := tables[start:end] + oids := make([]string, len(batch)) + for i, t := range batch { + oids[i] = strconv.FormatUint(uint64(t.Oid), 10) + } + + // gp_dist_random runs gpbackup_file_info on each segment, where + // pg_relation_filepath resolves that segment's own copy of the table. + query := fmt.Sprintf(` + SELECT oid, + md5(string_agg(gp_segment_id::text || ',' || info, chr(10) ORDER BY gp_segment_id)) AS filehash + FROM ( + SELECT gp_segment_id, c.oid, gp_toolkit.gpbackup_file_info(c.oid) AS info + FROM gp_dist_random('pg_class') c + WHERE c.oid IN (%s) + ) x + GROUP BY oid + HAVING bool_and(COALESCE(info, '') <> '')`, strings.Join(oids, ",")) + + var results []struct { + Oid uint32 + FileHash string + } + if err := hashConn.Select(&results, query); err != nil { + gplog.Warn("Could not get file hashes for %d heap tables: %v", len(batch), err) + continue + } + for _, r := range results { + result[r.Oid] = r.FileHash + } + } + return result +} + +// heapIncrementalEntry is a heap table's TOC entry, without its file hash, +// along with the FQN it has in the backup snapshot. +type heapIncrementalEntry struct { + FQN string + Entry toc.HeapEntry +} + +// getHeapIncrementalEntries returns, for every heap table in the backup set, +// keyed by oid, what besides the file hash decides whether the table changed: +// its last DDL timestamp and a hash of its columns and of the types they use +// (see getHeapColumnHashes). A nil result means the column hashes could not +// be read and no heap table may be skipped. +func getHeapIncrementalEntries(connectionPool *dbconn.DBConn) map[uint32]heapIncrementalEntry { + tables := getHeapLastDDLTimestamps(connectionPool) + columnHashes, err := getHeapColumnHashes(connectionPool) + if err != nil { + gplog.Warn("Could not read the columns of heap tables, skipping heap file hash collection: %v", err) + return nil + } + for oid, table := range tables { + columnsMD5, ok := columnHashes[oid] + if !ok { + delete(tables, oid) + continue + } + table.Entry.ColumnsMD5 = columnsMD5 + tables[oid] = table + } + return tables +} + +// heapTableFilter returns the join and the condition that select the heap +// tables eligible for file hashes from pg_class c. Unlogged tables are left +// out: a CHECKPOINT other than at shutdown does not write their buffers, so +// their files may not move when they change. +func heapTableFilter(conn *dbconn.DBConn) (amJoin string, heapFilter string) { + if conn.Version.IsGPDB() && conn.Version.Before("6") { + return "", "c.relstorage = 'h'" + } + if conn.Version.IsGPDB() && conn.Version.Before("7") { + return "", "c.relstorage = 'h' AND c.relpersistence = 'p'" + } + return "JOIN pg_am a ON c.relam = a.oid", "a.amname = 'heap' AND c.relpersistence = 'p'" +} + +// getHeapLastDDLTimestamps returns every heap table in the backup set with its +// last DDL timestamp (empty if none is recorded). A DDL that does not rewrite +// the table, e.g. DROP COLUMN, leaves the data files untouched, so the file +// hash alone would reuse data dumped with the old column list. +func getHeapLastDDLTimestamps(connectionPool *dbconn.DBConn) map[uint32]heapIncrementalEntry { + amJoin, heapFilter := heapTableFilter(connectionPool) + query := fmt.Sprintf(` + SELECT c.oid, + quote_ident(n.nspname) || '.' || quote_ident(c.relname) AS tablefqn, + COALESCE(lastop.lastddltimestamp::text, '') AS lastddltimestamp + FROM pg_class c + JOIN pg_namespace n ON c.relnamespace = n.oid + %s + LEFT JOIN ( SELECT lo.objid, + MAX(lo.statime) AS lastddltimestamp + FROM pg_stat_last_operation lo + WHERE lo.staactionname IN ('CREATE', 'ALTER', 'TRUNCATE') + GROUP BY lo.objid + ) lastop ON c.oid = lastop.objid + WHERE %s + AND c.relkind = 'r' + AND %s`, amJoin, heapFilter, relationAndSchemaFilterClause()) + + var results []struct { + Oid uint32 + TableFQN string + LastDDLTimestamp string + } + err := connectionPool.Select(&results, query) + gplog.FatalOnError(err) + resultMap := make(map[uint32]heapIncrementalEntry, len(results)) + for _, result := range results { + resultMap[result.Oid] = heapIncrementalEntry{ + FQN: result.TableFQN, + Entry: toc.HeapEntry{LastDDLTimestamp: result.LastDDLTimestamp}, + } + } + return resultMap +} + +// getHeapColumnHashes returns, keyed by oid, a hash of every heap table's +// columns (number, name, type, typmod, dropped) and of the definitions of the +// enum and composite types they use, directly or through domains, arrays, +// composite, range and multirange types: the labels of each enum and the +// attributes of each composite type. Restore loads a skipped table's old data +// file into the newest column list, so neither may change. Some changes touch +// neither the data files nor the table's DDL timestamp: ALTER TYPE ... RENAME +// VALUE or ADD/DROP ATTRIBUTE, or DROP DOMAIN ... CASCADE dropping a column. +// +// Tables that use, anywhere in that closure, a type whose values are dumped as +// the names of catalog objects they reference by oid (regclass and the other +// reg* types, aclitem) get no hash, and so are never skipped: renaming or +// dropping the referenced object changes what COPY writes, or makes the old +// value fail to load, and nothing here records it. Tables without columns get +// no hash either. +// +// The query runs in the backup transaction under a savepoint, so a failure +// leaves the transaction usable and the caller can fall back to not skipping +// heap tables. +func getHeapColumnHashes(connectionPool *dbconn.DBConn) (map[uint32]string, error) { + resultMap := make(map[uint32]string) + amJoin, heapFilter := heapTableFilter(connectionPool) + // Multirange types (PostgreSQL 14) point to their range type only through + // pg_range.rngmultitypid. + multirangeDeps := "" + if connectionPool.Version.IsCBDB() { + multirangeDeps = ` + UNION ALL + SELECT rngmultitypid, rngtypid FROM pg_range` + } + query := fmt.Sprintf(` + WITH RECURSIVE type_deps(typid, deptypid) AS ( + SELECT oid, typbasetype FROM pg_type WHERE typtype = 'd' + UNION ALL + SELECT oid, typelem FROM pg_type WHERE typelem <> 0 + UNION ALL + SELECT t.oid, ta.atttypid + FROM pg_type t + JOIN pg_attribute ta ON ta.attrelid = t.typrelid + WHERE t.typtype = 'c' AND ta.attnum > 0 AND NOT ta.attisdropped + UNION ALL + SELECT rngtypid, rngsubtype FROM pg_range%s + ), column_types(relid, typid) AS ( + SELECT c.oid, ca.atttypid + FROM pg_class c + JOIN pg_namespace n ON c.relnamespace = n.oid + %s + JOIN pg_attribute ca ON ca.attrelid = c.oid + WHERE %s + AND c.relkind = 'r' + AND ca.attnum > 0 AND NOT ca.attisdropped + AND %s + UNION + SELECT ct.relid, d.deptypid + FROM column_types ct + JOIN type_deps d ON d.typid = ct.typid + ), type_defs(typid, def) AS ( + SELECT enumtypid, 'e' || enumtypid::text || ':' || enumsortorder || ':' || enumlabel + FROM pg_enum + UNION ALL + SELECT t.oid, 'c' || t.oid::text || ':' || ta.attnum || ':' || ta.attname || ':' || ta.atttypid + || ':' || ta.atttypmod || ':' || ta.attisdropped + FROM pg_type t + JOIN pg_attribute ta ON ta.attrelid = t.typrelid + WHERE t.typtype = 'c' AND ta.attnum > 0 + ) + SELECT d.relid AS oid, + md5(string_agg(d.def, ',' ORDER BY d.def)) AS columnsmd5, + d.relid IN ( + SELECT ct.relid + FROM column_types ct + JOIN pg_type t ON t.oid = ct.typid + JOIN pg_namespace tn ON tn.oid = t.typnamespace + WHERE tn.nspname = 'pg_catalog' + AND t.typname IN ('regproc', 'regprocedure', 'regoper', 'regoperator', + 'regclass', 'regtype', 'regconfig', 'regdictionary', 'regnamespace', + 'regrole', 'regcollation', 'aclitem') + ) AS refsobjects + FROM ( + SELECT ct.relid, td.def + FROM column_types ct + JOIN type_defs td ON td.typid = ct.typid + UNION ALL + SELECT ca.attrelid, 'a' || ca.attnum::text || ':' || ca.attname || ':' || ca.atttypid + || ':' || ca.atttypmod || ':' || ca.attisdropped + FROM pg_attribute ca + WHERE ca.attnum > 0 + AND ca.attrelid IN (SELECT relid FROM column_types) + ) d + GROUP BY d.relid`, multirangeDeps, amJoin, heapFilter, relationAndSchemaFilterClause()) + + var results []struct { + Oid uint32 + ColumnsMD5 string + RefsObjects bool + } + if connectionPool.Tx[0] != nil { + connectionPool.MustExec("SAVEPOINT gpbackup_heap_columns") + defer connectionPool.MustExec("ROLLBACK TO SAVEPOINT gpbackup_heap_columns") + } + if err := connectionPool.Select(&results, query); err != nil { + return nil, err + } + for _, result := range results { + if !result.RefsObjects { + resultMap[result.Oid] = result.ColumnsMD5 + } + } + return resultMap, nil +} + /* * PAX incremental metadata. * diff --git a/backup/queries_incremental_internal_test.go b/backup/queries_incremental_internal_test.go new file mode 100644 index 00000000..4c6f34ac --- /dev/null +++ b/backup/queries_incremental_internal_test.go @@ -0,0 +1,127 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +*/ + +package backup + +import ( + "errors" + "regexp" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/apache/cloudberry-backup/toc" + "github.com/apache/cloudberry-go-libs/dbconn" + "github.com/apache/cloudberry-go-libs/testhelper" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("backup/queries_incremental heap file hash internals", func() { + var ( + conn *dbconn.DBConn + mock sqlmock.Sqlmock + savedFilter string + columnTypeQuery = regexp.QuoteMeta("WITH RECURSIVE type_deps") + ) + BeforeEach(func() { + testhelper.SetupTestLogger() + conn, mock = testhelper.CreateAndConnectMockDB(1) + testhelper.SetDBVersion(conn, "1.6.0") + conn.Version.Type = dbconn.CBDB + savedFilter = filterRelationClause + filterRelationClause = "n.nspname NOT LIKE 'pg_temp_%'" + }) + AfterEach(func() { + filterRelationClause = savedFilter + Expect(mock.ExpectationsWereMet()).To(Succeed()) + }) + + Describe("keepUnchangedHashes", func() { + It("keeps only hashes that were re-read unchanged", func() { + tables := []heapTable{{Oid: 1, FQN: "public.same"}, {Oid: 2, FQN: "public.changed"}, {Oid: 3, FQN: "public.unreadable"}} + hashes := map[uint32]string{1: "h1", 2: "h2", 3: "h3"} + current := map[uint32]string{1: "h1", 2: "h2b"} + Expect(keepUnchangedHashes(tables, hashes, current)).To(Equal(map[uint32]string{1: "h1"})) + }) + }) + + Describe("getTablesWithActiveWriters", func() { + It("returns the oids holding write locks", func() { + mock.ExpectQuery(regexp.QuoteMeta("FROM pg_catalog.pg_locks")). + WillReturnRows(sqlmock.NewRows([]string{"oid"}).AddRow(16384).AddRow(16390)) + oids, err := getTablesWithActiveWriters(conn) + Expect(err).ToNot(HaveOccurred()) + Expect(oids).To(Equal(map[uint32]bool{16384: true, 16390: true})) + }) + It("returns the query error", func() { + mock.ExpectQuery(regexp.QuoteMeta("FROM pg_catalog.pg_locks")).WillReturnError(errors.New("boom")) + _, err := getTablesWithActiveWriters(conn) + Expect(err).To(HaveOccurred()) + }) + }) + + Describe("getHeapColumnHashes", func() { + It("hashes the table's own columns, follows multirange types and excludes unlogged tables on Cloudberry", func() { + mock.ExpectQuery(columnTypeQuery + "(?s).*rngmultitypid.*relpersistence = 'p'.*FROM pg_attribute ca"). + WillReturnRows(sqlmock.NewRows([]string{"oid", "columnsmd5", "refsobjects"}).AddRow(16384, "abc", false)) + hashes, err := getHeapColumnHashes(conn) + Expect(err).ToNot(HaveOccurred()) + Expect(hashes).To(Equal(map[uint32]string{16384: "abc"})) + }) + It("gives no hash to tables using types that reference catalog objects by oid", func() { + mock.ExpectQuery(columnTypeQuery + "(?s).*'regclass'.*'aclitem'.*AS refsobjects"). + WillReturnRows(sqlmock.NewRows([]string{"oid", "columnsmd5", "refsobjects"}). + AddRow(16384, "abc", false).AddRow(16390, "def", true)) + hashes, err := getHeapColumnHashes(conn) + Expect(err).ToNot(HaveOccurred()) + Expect(hashes).To(Equal(map[uint32]string{16384: "abc"})) + }) + It("rolls back to a savepoint when the query fails inside the backup transaction", func() { + mock.ExpectBegin() + mock.ExpectExec("SET TRANSACTION ISOLATION LEVEL").WillReturnResult(sqlmock.NewResult(0, 0)) + conn.MustBegin(0) + mock.ExpectExec(regexp.QuoteMeta("SAVEPOINT gpbackup_heap_columns")).WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectQuery(columnTypeQuery).WillReturnError(errors.New("boom")) + mock.ExpectExec(regexp.QuoteMeta("ROLLBACK TO SAVEPOINT gpbackup_heap_columns")).WillReturnResult(sqlmock.NewResult(0, 0)) + _, err := getHeapColumnHashes(conn) + Expect(err).To(HaveOccurred()) + }) + }) + + Describe("getHeapIncrementalEntries", func() { + It("combines DDL timestamps and column hashes by oid, dropping tables without a column hash", func() { + mock.ExpectQuery(regexp.QuoteMeta("FROM pg_stat_last_operation")). + WillReturnRows(sqlmock.NewRows([]string{"oid", "tablefqn", "lastddltimestamp"}). + AddRow(1, "public.t", "ts1").AddRow(2, "public.u", "").AddRow(3, "public.refs", "")) + mock.ExpectQuery(columnTypeQuery). + WillReturnRows(sqlmock.NewRows([]string{"oid", "columnsmd5", "refsobjects"}). + AddRow(1, "abc", false).AddRow(2, "def", false).AddRow(3, "ghi", true)) + Expect(getHeapIncrementalEntries(conn)).To(Equal(map[uint32]heapIncrementalEntry{ + 1: {FQN: "public.t", Entry: toc.HeapEntry{LastDDLTimestamp: "ts1", ColumnsMD5: "abc"}}, + 2: {FQN: "public.u", Entry: toc.HeapEntry{LastDDLTimestamp: "", ColumnsMD5: "def"}}, + })) + }) + It("returns nil when the columns cannot be read", func() { + mock.ExpectQuery(regexp.QuoteMeta("FROM pg_stat_last_operation")). + WillReturnRows(sqlmock.NewRows([]string{"oid", "tablefqn", "lastddltimestamp"}).AddRow(1, "public.t", "ts1")) + mock.ExpectQuery(columnTypeQuery).WillReturnError(errors.New("boom")) + Expect(getHeapIncrementalEntries(conn)).To(BeNil()) + }) + }) +}) diff --git a/backup/validate.go b/backup/validate.go index 2902d1aa..a4f84229 100644 --- a/backup/validate.go +++ b/backup/validate.go @@ -156,6 +156,9 @@ func validateFlagCombinations(flags *pflag.FlagSet) { if MustGetFlagBool(options.INCREMENTAL) && !MustGetFlagBool(options.LEAF_PARTITION_DATA) { gplog.Fatal(errors.Errorf("--leaf-partition-data must be specified with --incremental"), "") } + if MustGetFlagBool(options.HEAP_FILE_HASH) && !MustGetFlagBool(options.LEAF_PARTITION_DATA) { + gplog.Fatal(errors.Errorf("--leaf-partition-data must be specified with --heap-file-hash"), "") + } if MustGetFlagBool(options.NO_INHERITS) && !(FlagChanged(options.INCLUDE_RELATION) || FlagChanged(options.INCLUDE_RELATION_FILE)) { gplog.Fatal(errors.Errorf("--no-inherits must be specified with either --include-table or --include-table-file"), "") } diff --git a/backup/wrappers.go b/backup/wrappers.go index 778cc604..30e33096 100644 --- a/backup/wrappers.go +++ b/backup/wrappers.go @@ -771,6 +771,23 @@ func backupIncrementalMetadata() { globalTOC.IncrementalMetadata.AO = aoTableEntries paxTableEntries := GetPAXIncrementalMetadata(connectionPool) globalTOC.IncrementalMetadata.PAX = paxTableEntries + + if heapFileHashes == nil { + return + } + heapTables := getHeapIncrementalEntries(connectionPool) + if heapTables == nil { + return + } + heapEntries := make(map[string]toc.HeapEntry) + for oid, table := range heapTables { + if hash, ok := heapFileHashes[oid]; ok { + table.Entry.FileHashMD5 = hash + heapEntries[table.FQN] = table.Entry + } + } + globalTOC.IncrementalMetadata.Heap = heapEntries + gplog.Verbose("Collected file hashes for %d of %d heap tables", len(heapEntries), len(heapTables)) } func backupStorageServers(metadataFile *utils.FileWithByteCount) { diff --git a/options/flag.go b/options/flag.go index c4ab8de0..dad5ea76 100644 --- a/options/flag.go +++ b/options/flag.go @@ -56,6 +56,7 @@ const ( RESIZE_CLUSTER = "resize-cluster" NO_INHERITS = "no-inherits" REPORT_DIR = "report-dir" + HEAP_FILE_HASH = "heap-file-hash" DEFAULT_HISTORY_SYNC_STANDBY_TIMEOUT = 300 ) @@ -94,6 +95,7 @@ func SetBackupFlagDefaults(flagSet *pflag.FlagSet) { flagSet.Bool(WITH_STATS, false, "Back up query plan statistics") flagSet.Bool(WITHOUT_GLOBALS, false, "Skip backup of global metadata") flagSet.Bool(NO_INHERITS, false, "For a filtered backup, don't back up all tables that inherit included tables") + flagSet.Bool(HEAP_FILE_HASH, false, "Skip unchanged heap tables in incremental backups by hashing their data files. Issues a cluster-wide CHECKPOINT before the backup starts") } func SetRestoreFlagDefaults(flagSet *pflag.FlagSet) { diff --git a/toc/toc.go b/toc/toc.go index 61fcc0b5..c804ea42 100644 --- a/toc/toc.go +++ b/toc/toc.go @@ -53,8 +53,9 @@ type SegmentDataEntry struct { } type IncrementalEntries struct { - AO map[string]AOEntry - PAX map[string]PAXEntry + AO map[string]AOEntry + PAX map[string]PAXEntry + Heap map[string]HeapEntry `yaml:"heap,omitempty"` } type AOEntry struct { @@ -106,6 +107,12 @@ type PAXEntry struct { RelFileNode int64 } +type HeapEntry struct { + FileHashMD5 string `yaml:"fileHashMD5,omitempty"` + LastDDLTimestamp string `yaml:"lastddltimestamp,omitempty"` + ColumnsMD5 string `yaml:"columnsmd5,omitempty"` +} + type UniqueID struct { ClassID uint32 Oid uint32