From 0e007e9f8c8693224edfc2e4c39155c6781b44c7 Mon Sep 17 00:00:00 2001 From: LiangChen Date: Mon, 9 Nov 2026 18:08:09 +0800 Subject: [PATCH] Add --heap-file-hash to skip unchanged heap tables in incremental backups Every incremental backup re-copies every heap table. With the new opt-in --heap-file-hash flag, a heap table is skipped when its on-disk file state, its last DDL timestamp and a hash of its columns and the types they use all match the previous backup. Without the flag, behavior is unchanged. File hash: - A plpgsql helper, gp_toolkit.gpbackup_file_info(oid), runs on every segment and stats every segment file of the main fork (path, path.1, ... until one is missing), so tables larger than RELSEG_SIZE are fully covered. The path is part of the hash, so a new relfilenode (TRUNCATE, VACUUM FULL, CLUSTER) is detected. The helper is created for each hashing step and dropped afterwards. - Hashes are collected on a separate connection after a CHECKPOINT and before the backup snapshot is taken, so a change committed after the snapshot moves the files on by the next backup. - Tables on which another transaction holds a write lock (RowExclusiveLock or stronger) after the hashes are read get no hash, since rows of in-progress transactions are already in the files. - After the snapshot, a second CHECKPOINT and a re-read drop the hash of any table whose files changed in between. - A file modified in the current second gets no hash, since pg_stat_file reports mtime in whole seconds. - Unlogged tables are never hashed: a regular CHECKPOINT does not write their buffers. Columns and types: restore loads a skipped table's old data into the newest column list, so each heap entry also records, read inside the backup snapshot, the table's last DDL timestamp and a hash of its columns and of the enum labels and composite attributes of the types they use, directly or through domains, arrays, composite, range and multirange types. This covers DROP COLUMN, enum RENAME VALUE, composite ADD/DROP ATTRIBUTE, row types used as columns and DROP DOMAIN ... CASCADE. Tables that use regclass and the other reg* types, or aclitem, whose values are dumped as names of the objects they reference, are never hashed. A heap table is skipped only if the previous restore plan has data for it, both backups carry a hash for it, and the file hash, the DDL timestamp and the column hash all match. Whenever a hash cannot be trusted (CHECKPOINT, lock check or column query failure, a missing segment, Greenplum before 6), no hash is recorded and the table is included, as without the flag. --heap-file-hash requires --leaf-partition-data. New TOC fields are omitempty, so older TOCs and older gprestore are unaffected. --- backup/backup.go | 20 +- backup/global_variables.go | 1 + backup/incremental.go | 40 +- backup/incremental_test.go | 82 +++- backup/queries_incremental.go | 505 ++++++++++++++++++++ backup/queries_incremental_internal_test.go | 127 +++++ backup/validate.go | 3 + backup/wrappers.go | 17 + options/flag.go | 2 + toc/toc.go | 11 +- 10 files changed, 797 insertions(+), 11 deletions(-) create mode 100644 backup/queries_incremental_internal_test.go 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