PostgreSQL backups: comparing pg_dump speed in different formats and with different compression levels

#!/usr/bin/env python3 """ Script to populate PostgreSQL database with ~10GB of test data for backup performance testing. Inserts 1 million rows into each of 3 tables until reaching 10GB target. """ import psycopg2 import random import string import time import io from datetime import datetime, timedelta import sys # Database connection parameters DB_CONFIG = { "host": "localhost", "port": 7000, "database": "testdb", "user": "postgres", "password": "testpassword", } # Target database size in bytes (10GB) TARGET_SIZE_GB = 10 TARGET_SIZE_BYTES = TARGET_SIZE_GB * 1024 * 1024 * 1024 # Rows per table per round ROWS_PER_TABLE = 100000 def generate_random_string(length): """Generate a random string of specified length.""" return "".join(random.choices(string.ascii_letters + string.digits + " ", k=length)) def generate_random_date(): """Generate a random date within the last 5 years.""" start_date = datetime.now() - timedelta(days=5 * 365) random_days = random.randint(0, 5 * 365) return start_date + timedelta(days=random_days) def create_tables(cursor): """Create the 3 test tables with various data types.""" print("Creating 3 tables...") # Table 1: Large table with mixed data types  cursor.execute(""" CREATE TABLE IF NOT EXISTS large_test_table ( id BIGSERIAL PRIMARY KEY, name VARCHAR(100), description TEXT, email VARCHAR(255), phone VARCHAR(20), address TEXT, city VARCHAR(100), country VARCHAR(100), postal_code VARCHAR(20), birth_date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, salary DECIMAL(10,2), is_active BOOLEAN DEFAULT TRUE, rating FLOAT, notes TEXT, department VARCHAR(100), employee_id VARCHAR(50) ) """) # Table 2: Orders table for transactional data  cursor.execute(""" CREATE TABLE IF NOT EXISTS orders ( id BIGSERIAL PRIMARY KEY, user_id BIGINT, order_number VARCHAR(50) UNIQUE, total_amount DECIMAL(12,2), order_date TIMESTAMP, status VARCHAR(20), shipping_address TEXT, notes TEXT, payment_method VARCHAR(50), shipping_method VARCHAR(50), discount_amount DECIMAL(10,2), product_list TEXT, customer_notes TEXT ) """) # Table 3: Activity logs with lots of text data  cursor.execute(""" CREATE TABLE IF NOT EXISTS activity_logs ( id BIGSERIAL PRIMARY KEY, user_id BIGINT, action VARCHAR(100), details TEXT, ip_address INET, user_agent TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, session_id VARCHAR(100), browser VARCHAR(50), operating_system VARCHAR(50), referrer TEXT, response_time INTEGER, error_message TEXT ) """) print("3 tables created successfully!") def create_indexes(cursor): """Create indexes for all 3 tables.""" print("Creating indexes...") # Indexes for large_test_table  cursor.execute( "CREATE INDEX IF NOT EXISTS idx_large_test_name ON large_test_table(name)" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_large_test_email ON large_test_table(email)" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_large_test_created_at ON large_test_table(created_at)" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_large_test_department ON large_test_table(department)" ) # Indexes for orders  cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_user_id ON orders(user_id)") cursor.execute( "CREATE INDEX IF NOT EXISTS idx_orders_order_date ON orders(order_date)" ) cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status)") # Indexes for activity_logs  cursor.execute( "CREATE INDEX IF NOT EXISTS idx_activity_user_id ON activity_logs(user_id)" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_activity_timestamp ON activity_logs(timestamp)" ) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_activity_action ON activity_logs(action)" ) print("Indexes created successfully!") def drop_indexes(cursor): """Drop indexes for faster bulk loading.""" print("Dropping indexes for bulk loading...") indexes_to_drop = [ "idx_large_test_name", "idx_large_test_email", "idx_large_test_created_at", "idx_large_test_department", "idx_orders_user_id", "idx_orders_order_date", "idx_orders_status", "idx_activity_user_id", "idx_activity_timestamp", "idx_activity_action", ] for index in indexes_to_drop: try: cursor.execute(f"DROP INDEX IF EXISTS {index}") except: pass print("Indexes dropped!") def print_progress(inserted, total, start_time, operation_name): """Print detailed progress information.""" current_time = time.time() elapsed = current_time - start_time rate = inserted / elapsed if elapsed > 0 else 0 percentage = (inserted / total) * 100 if rate > 0: eta_seconds = (total - inserted) / rate eta_minutes = eta_seconds / 60 eta_str = f"ETA: {eta_minutes:.1f}m" else: eta_str = "ETA: calculating..." print( f"{operation_name}: {inserted:,} / {total:,} ({percentage:.1f}%) | " f"Rate: {rate:,.0f} rows/sec | Elapsed: {elapsed / 60:.1f}m | {eta_str}" ) def populate_large_table_batch(cursor, count=ROWS_PER_TABLE): """Populate large_test_table with specified number of rows.""" print(f"Inserting {count:,} rows into large_test_table...") batch_size = 25000 inserted = 0 start_time = time.time() while inserted < count: current_batch_size = min(batch_size, count - inserted) data_buffer = io.StringIO() for i in range(current_batch_size): name = ( generate_random_string(random.randint(20, 100)) .replace("\t", " ") .replace("\n", " ") ) description = ( generate_random_string(random.randint(100, 500)) .replace("\t", " ") .replace("\n", " ") ) email = f"{generate_random_string(10)}@{generate_random_string(10)}.com" phone = f"+1-{random.randint(100, 999)}-{random.randint(100, 999)}-{random.randint(1000, 9999)}" address = ( generate_random_string(random.randint(50, 200)) .replace("\t", " ") .replace("\n", " ") ) city = ( generate_random_string(random.randint(10, 50)) .replace("\t", " ") .replace("\n", " ") ) country = ( generate_random_string(random.randint(5, 30)) .replace("\t", " ") .replace("\n", " ") ) postal_code = f"{random.randint(10000, 99999)}" birth_date = generate_random_date().date() salary = f"{random.randint(30000, 150000) + random.random():.2f}" is_active = "t" if random.choice([True, False]) else "f" rating = f"{random.uniform(1.0, 5.0):.2f}" notes = ( generate_random_string(random.randint(50, 300)) .replace("\t", " ") .replace("\n", " ") ) department = random.choice( ["HR", "Engineering", "Sales", "Marketing", "Finance", "Operations"] ) employee_id = ( f"EMP-{random.randint(1000, 9999)}-{generate_random_string(4)}" ) data_buffer.write( f"{name}\t{description}\t{email}\t{phone}\t{address}\t{city}\t{country}\t{postal_code}\t{birth_date}\t{salary}\t{is_active}\t{rating}\t{notes}\t{department}\t{employee_id}\n" ) data_buffer.seek(0) cursor.copy_from( data_buffer, "large_test_table", columns=( "name", "description", "email", "phone", "address", "city", "country", "postal_code", "birth_date", "salary", "is_active", "rating", "notes", "department", "employee_id", ), sep="\t", ) inserted += current_batch_size if inserted % 100000 == 0: print_progress(inserted, count, start_time, "Large table") elapsed = time.time() - start_time print( f"Completed large_test_table: {inserted:,} rows in {elapsed:.2f}s ({inserted / elapsed:,.0f} rows/sec)" ) def populate_orders_batch(cursor, count=ROWS_PER_TABLE): """Populate orders table with specified number of rows.""" print(f"Inserting {count:,} rows into orders...") batch_size = 25000 inserted = 0 start_time = time.time() while inserted < count: current_batch_size = min(batch_size, count - inserted) data_buffer = io.StringIO() for i in range(current_batch_size): user_id = random.randint(1, 10000000) order_number = f"ORD-{time.time_ns()}-{i:06d}" total_amount = f"{random.uniform(10.0, 5000.0):.2f}" order_date = generate_random_date() status = random.choice( ["pending", "processing", "shipped", "delivered", "cancelled"] ) shipping_address = ( generate_random_string(random.randint(100, 300)) .replace("\t", " ") .replace("\n", " ") ) notes = ( generate_random_string(random.randint(50, 200)) .replace("\t", " ") .replace("\n", " ") if random.random() > 0.5 else "" ) payment_method = random.choice(["credit_card", "paypal", "bank_transfer"]) shipping_method = random.choice(["standard", "express", "overnight"]) discount_amount = ( f"{random.uniform(0, 50) if random.random() > 0.7 else 0:.2f}" ) product_list = ( generate_random_string(random.randint(200, 800)) .replace("\t", " ") .replace("\n", " ") ) customer_notes = ( generate_random_string(random.randint(100, 400)) .replace("\t", " ") .replace("\n", " ") ) data_buffer.write( f"{user_id}\t{order_number}\t{total_amount}\t{order_date}\t{status}\t{shipping_address}\t{notes}\t{payment_method}\t{shipping_method}\t{discount_amount}\t{product_list}\t{customer_notes}\n" ) data_buffer.seek(0) cursor.copy_from( data_buffer, "orders", columns=( "user_id", "order_number", "total_amount", "order_date", "status", "shipping_address", "notes", "payment_method", "shipping_method", "discount_amount", "product_list", "customer_notes", ), sep="\t", ) inserted += current_batch_size if inserted % 100000 == 0: print_progress(inserted, count, start_time, "Orders") elapsed = time.time() - start_time print( f"Completed orders: {inserted:,} rows in {elapsed:.2f}s ({inserted / elapsed:,.0f} rows/sec)" ) def populate_activity_logs_batch(cursor, count=ROWS_PER_TABLE): """Populate activity_logs table with specified number of rows.""" print(f"Inserting {count:,} rows into activity_logs...") batch_size = 25000 inserted = 0 start_time = time.time() actions = [ "login", "logout", "view_product", "add_to_cart", "checkout", "update_profile", "search", "download", ] while inserted < count: current_batch_size = min(batch_size, count - inserted) data_buffer = io.StringIO() for i in range(current_batch_size): user_id = random.randint(1, 10000000) action = random.choice(actions) details = ( generate_random_string(random.randint(100, 500)) .replace("\t", " ") .replace("\n", " ") ) ip_address = f"{random.randint(1, 255)}.{random.randint(1, 255)}.{random.randint(1, 255)}.{random.randint(1, 255)}" user_agent = f"Mozilla/5.0 ({generate_random_string(50)}) {generate_random_string(30)}".replace( "\t", " " ).replace("\n", " ") timestamp = generate_random_date() session_id = generate_random_string(32) browser = random.choice(["Chrome", "Firefox", "Safari", "Edge"]) operating_system = random.choice( ["Windows", "macOS", "Linux", "iOS", "Android"] ) referrer = ( f"https://{generate_random_string(10)}.com" if random.random() > 0.3 else "" ) response_time = random.randint(50, 5000) error_message = ( generate_random_string(random.randint(100, 300)) .replace("\t", " ") .replace("\n", " ") if random.random() > 0.8 else "" ) data_buffer.write( f"{user_id}\t{action}\t{details}\t{ip_address}\t{user_agent}\t{timestamp}\t{session_id}\t{browser}\t{operating_system}\t{referrer}\t{response_time}\t{error_message}\n" ) data_buffer.seek(0) cursor.copy_from( data_buffer, "activity_logs", columns=( "user_id", "action", "details", "ip_address", "user_agent", "timestamp", "session_id", "browser", "operating_system", "referrer", "response_time", "error_message", ), sep="\t", ) inserted += current_batch_size if inserted % 100000 == 0: print_progress(inserted, count, start_time, "Activity logs") elapsed = time.time() - start_time print( f"Completed activity_logs: {inserted:,} rows in {elapsed:.2f}s ({inserted / elapsed:,.0f} rows/sec)" ) def get_database_size_bytes(cursor): """Get the current database size in bytes.""" cursor.execute("SELECT pg_database_size('testdb')") return cursor.fetchone()[0] def get_database_size_mb(cursor): """Get the current database size in MB.""" size_bytes = get_database_size_bytes(cursor) return size_bytes / (1024 * 1024) def format_size_mb(size_bytes): """Format size in bytes to MB string.""" return f"{size_bytes / (1024 * 1024):.1f}MB" def main(): """Main function to populate the database until reaching 10GB.""" print("Starting database population for backup performance testing...") print(f"Target: {TARGET_SIZE_GB * 1024:.0f}MB of data") print( f"Strategy: Insert {ROWS_PER_TABLE:,} rows into each of 3 tables per round until target reached" ) print("-" * 80) try: # Connect to database  print("Connecting to database...") conn = psycopg2.connect(**DB_CONFIG) conn.autocommit = False cursor = conn.cursor() initial_size_mb = get_database_size_mb(cursor) print(f"Initial database size: {initial_size_mb:.1f}MB") overall_start_time = time.time() # Create tables  create_tables(cursor) conn.commit() # Drop indexes for faster loading  drop_indexes(cursor) conn.commit() round_number = 1 current_size_bytes = get_database_size_bytes(cursor) # Keep adding rounds of 1M rows per table until we reach 10GB  while current_size_bytes < TARGET_SIZE_BYTES: print(f"\n{'=' * 20} ROUND {round_number} {'=' * 20}") round_start_time = time.time() # Populate all 3 tables with 1M rows each  populate_large_table_batch(cursor) conn.commit() populate_orders_batch(cursor) conn.commit() populate_activity_logs_batch(cursor) conn.commit() # Check current size  current_size_bytes = get_database_size_bytes(cursor) current_size_mb = format_size_mb(current_size_bytes) round_elapsed = time.time() - round_start_time print(f"\nRound {round_number} completed in {round_elapsed:.2f}s") print(f"Current database size: {current_size_mb}") print( f"Progress: {(current_size_bytes / TARGET_SIZE_BYTES) * 100:.1f}% of target ({TARGET_SIZE_GB * 1024:.0f}MB)" ) if current_size_bytes >= TARGET_SIZE_BYTES: print(f"✅ Target size reached!") break round_number += 1 # Recreate indexes  print(f"\n{'=' * 20} CREATING INDEXES {'=' * 20}") index_start_time = time.time() create_indexes(cursor) conn.commit() index_elapsed = time.time() - index_start_time print(f"Indexes created in {index_elapsed:.2f}s") # Final statistics  overall_elapsed = time.time() - overall_start_time final_size_mb = get_database_size_mb(cursor) print(f"\n{'=' * 60}") print("DATABASE POPULATION COMPLETED!") print(f"Final database size: {final_size_mb:.1f}MB") print(f"Total rounds: {round_number}") print( f"Total time: {overall_elapsed:.2f} seconds ({overall_elapsed / 60:.2f} minutes)" ) # Show table statistics  cursor.execute(""" SELECT relname, n_tup_ins as "Rows", round(pg_total_relation_size('public.'||relname)::numeric / (1024*1024), 1) as "Size_MB" FROM pg_stat_user_tables ORDER BY pg_total_relation_size('public.'||relname) DESC """) print(f"\nTable Statistics:") print(f"{'Table':<20} | {'Rows':<12} | {'Size':<10}") print("-" * 50) for row in cursor.fetchall(): print(f"{row[0]:<20} | {row[1]:>12,} | {row[2]:.1f}MB") cursor.close() conn.close() except psycopg2.Error as e: print(f"Database error: {e}") sys.exit(1) except Exception as e: print(f"Error: {e}") sys.exit(1) if __name__ == "__main__": main() 

 python #!/usr/bin/env python3 """ Comprehensive PostgreSQL backup performance testing script. Tests pg_dump with all possible formats and compression levels. """ import subprocess import time import os import shutil import json import csv from datetime import datetime from pathlib import Path import psycopg2 import argparse def log_with_timestamp(message): """Print a message with timestamp.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") # PostgreSQL binary path PG_BIN_PATH = Path("./postgresql-17/bin") # Database connection parameters DB_CONFIG = { "host": "localhost", "port": 7000, "database": "testdb", "user": "postgres", "password": "testpassword", } # Restore database connection parameters RESTORE_DB_CONFIG = { "host": "localhost", "port": 7001, "database": "testdb", "user": "postgres", "password": "testpassword", } # Test configurations PG_DUMP_FORMATS = [ ("plain", "sql"), ("custom", "dump"), ("directory", "dir"), ("tar", "tar"), ] COMPRESSION_LEVELS = { "gzip": [1, 5, 9], "lz4": [1, 5, 12], "zstd": [1, 5, 15, 22], } # Available compression methods (depends on PostgreSQL version) COMPRESSION_METHODS = ["none", "gzip", "lz4", "zstd"] # Results storage results: list[dict] = [] def ensure_backup_directory(): """Ensure backup directory exists and is clean.""" backup_dir = Path("./backups") if backup_dir.exists(): shutil.rmtree(backup_dir) backup_dir.mkdir(exist_ok=True) return backup_dir def get_database_size(): """Get current database size in bytes.""" try: conn = psycopg2.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute("SELECT pg_database_size('testdb')") size = cursor.fetchone()[0] cursor.close() conn.close() return size except Exception as e: log_with_timestamp(f"Error getting database size: {e}") return 0 def get_restore_database_size(): """Get restore database size in bytes.""" try: conn = psycopg2.connect(**RESTORE_DB_CONFIG) cursor = conn.cursor() cursor.execute("SELECT pg_database_size('testdb')") size = cursor.fetchone()[0] cursor.close() conn.close() return size except Exception as e: log_with_timestamp(f"Error getting restore database size: {e}") return 0 def clean_restore_database(): """Clean the restore database by dropping and recreating it.""" import time try: # Connect to postgres database to drop/create testdb restore_config = RESTORE_DB_CONFIG.copy() restore_config["database"] = "postgres" conn = psycopg2.connect(**restore_config) conn.autocommit = True cursor = conn.cursor() # Terminate any existing connections to testdb with retry logic max_attempts = 5 for attempt in range(max_attempts): try: # Get count of active connections first cursor.execute(""" SELECT COUNT(*) FROM pg_stat_activity WHERE datname = 'testdb' AND pid <> pg_backend_pid() """) active_connections = cursor.fetchone()[0] if active_connections == 0: break log_with_timestamp( f"Found {active_connections} active connections to testdb, terminating..." ) # Terminate connections cursor.execute(""" SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = 'testdb' AND pid <> pg_backend_pid() """) # Fetch all results to ensure the query completes terminated_pids = cursor.fetchall() log_with_timestamp(f"Terminated {len(terminated_pids)} connections") # Wait a bit for connections to actually close time.sleep(2) except Exception as term_error: log_with_timestamp( f"Warning: Error terminating connections (attempt {attempt + 1}): {term_error}" ) if attempt < max_attempts - 1: time.sleep(1) else: # Continue anyway, maybe we can still drop the database break # Try to drop database with retry logic drop_attempts = 3 for attempt in range(drop_attempts): try: cursor.execute("DROP DATABASE IF EXISTS testdb") log_with_timestamp("Database testdb dropped successfully") break except Exception as drop_error: if ( "is being accessed by other users" in str(drop_error) and attempt < drop_attempts - 1 ): log_with_timestamp( f"Database still in use, waiting and retrying... (attempt {attempt + 1})" ) time.sleep(3) else: raise drop_error # Create fresh database cursor.execute("CREATE DATABASE testdb") cursor.close() conn.close() log_with_timestamp("✓ Restore database cleaned and recreated") return True except Exception as e: log_with_timestamp(f"✗ Error cleaning restore database: {e}") return False def wait_for_restore_db(): """Wait for restore database to be ready.""" import time max_attempts = 30 for attempt in range(max_attempts): try: conn = psycopg2.connect(**RESTORE_DB_CONFIG) conn.close() return True except: if attempt < max_attempts - 1: time.sleep(1) else: log_with_timestamp("✗ Restore database not ready after 30 seconds") return False def restore_pg_dump(backup_path, format_name): """Restore a pg_dump backup.""" env = os.environ.copy() env["PGPASSWORD"] = RESTORE_DB_CONFIG["password"] if format_name == "plain": # Use psql for plain format command = [ str(PG_BIN_PATH / "psql.exe"), "-h", RESTORE_DB_CONFIG["host"], "-p", str(RESTORE_DB_CONFIG["port"]), "-U", RESTORE_DB_CONFIG["user"], "-d", RESTORE_DB_CONFIG["database"], "-f", str(backup_path), "-v", "ON_ERROR_STOP=1", ] else: # Use pg_restore for custom, directory, tar formats command = [ str(PG_BIN_PATH / "pg_restore.exe"), "-h", RESTORE_DB_CONFIG["host"], "-p", str(RESTORE_DB_CONFIG["port"]), "-U", RESTORE_DB_CONFIG["user"], "-d", RESTORE_DB_CONFIG["database"], "--verbose", str(backup_path), ] # Add parallel processing for custom and directory formats if format_name in ["custom", "directory"]: command.extend(["-j", "4"]) else: # Only add --single-transaction for tar format (not parallel) command.insert(-1, "--single-transaction") return run_command(command, env=env) def get_file_size(filepath): """Get file or directory size in bytes.""" if os.path.isfile(filepath): return os.path.getsize(filepath) elif os.path.isdir(filepath): total_size = 0 for dirpath, dirnames, filenames in os.walk(filepath): for filename in filenames: file_path = os.path.join(dirpath, filename) total_size += os.path.getsize(file_path) return total_size return 0 def format_size(size_bytes): """Format size in GB with 1 decimal place.""" size_gb = size_bytes / (1024.0 * 1024.0 * 1024.0) return f"{size_gb:.1f} GB" def format_minutes(seconds): """Format time in minutes with 1 decimal place.""" minutes = seconds / 60.0 return f"{minutes:.1f} mins" def run_command(command, timeout=7200, env=None): """Run a command and measure execution time.""" log_with_timestamp(f"Running: {' '.join(command)}") start_time = time.time() try: # Set environment variables for password if env is None: env = os.environ.copy() env["PGPASSWORD"] = DB_CONFIG["password"] result = subprocess.run( command, capture_output=True, text=True, timeout=timeout, env=env ) end_time = time.time() duration = end_time - start_time if result.returncode != 0: log_with_timestamp(f"Command failed with return code {result.returncode}") log_with_timestamp(f"STDERR: {result.stderr}") return None, duration, result.stderr return result, duration, None except subprocess.TimeoutExpired: log_with_timestamp(f"Command timed out after {timeout} seconds") return None, timeout, "Command timed out" except Exception as e: end_time = time.time() duration = end_time - start_time log_with_timestamp(f"Command failed with exception: {e}") return None, duration, str(e) def test_pg_dump(skip_restore=False): """Test pg_dump with all format and compression combinations.""" log_with_timestamp("\n" + "=" * 60) log_with_timestamp("TESTING PG_DUMP") log_with_timestamp("=" * 60) backup_dir = Path("./backups") for format_name, extension in PG_DUMP_FORMATS: log_with_timestamp(f"\nTesting pg_dump format: {format_name}") # Test without compression test_name = f"pg_dump_{format_name}_no_compression" output_path = backup_dir / f"{test_name}.{extension}" command = [ str(PG_BIN_PATH / "pg_dump.exe"), "-h", DB_CONFIG["host"], "-p", str(DB_CONFIG["port"]), "-U", DB_CONFIG["user"], "-d", DB_CONFIG["database"], "-f", str(output_path), "--format", format_name, "--verbose", ] if format_name == "directory": # For directory format, create the directory first output_path.mkdir(exist_ok=True) # Replace the path after -f flag with the directory path for i, item in enumerate(command): if item == "-f" and i + 1 < len(command): command[i + 1] = str(output_path) break # Add parallel processing for directory format command.extend(["-j", "4"]) # Perform backup result, backup_duration, error = run_command(command) if result is not None: backup_size = get_file_size(output_path) db_size = get_database_size() compression_ratio = backup_size / db_size if db_size > 0 else 0 # Clean restore database and perform restore restore_success = False restore_duration = 0 restore_error = None restored_db_size = 0 if not skip_restore: if clean_restore_database() and wait_for_restore_db(): restore_result, restore_duration, restore_error = restore_pg_dump( output_path, format_name ) if restore_result is not None and restore_result.returncode == 0: restore_success = True restored_db_size = get_restore_database_size() else: restore_success = True # Mark as successful if skipped restore_error = "Skipped" results.append( { "tool": "pg_dump", "format": format_name, "compression_method": "none", "compression_level": 0, "backup_duration_seconds": backup_duration, "restore_duration_seconds": restore_duration, "total_duration_seconds": backup_duration + restore_duration, "backup_size_bytes": backup_size, "database_size_bytes": db_size, "restored_db_size_bytes": restored_db_size, "compression_ratio": compression_ratio, "backup_success": True, "restore_success": restore_success, "backup_error": None, "restore_error": restore_error, "timestamp": datetime.now().isoformat(), } ) if skip_restore: log_with_timestamp( f"✓ {test_name}: Backup {format_minutes(backup_duration)}, " f"{format_size(backup_size)}, ratio: {compression_ratio:.3f} (restore skipped)" ) else: log_with_timestamp( f"✓ {test_name}: Backup {format_minutes(backup_duration)}, Restore {format_minutes(restore_duration)}, " f"{format_size(backup_size)}, ratio: {compression_ratio:.3f}" ) else: results.append( { "tool": "pg_dump", "format": format_name, "compression_method": "none", "compression_level": 0, "backup_duration_seconds": backup_duration, "restore_duration_seconds": 0, "total_duration_seconds": backup_duration, "backup_size_bytes": 0, "database_size_bytes": get_database_size(), "restored_db_size_bytes": 0, "compression_ratio": 0, "backup_success": False, "restore_success": False, "backup_error": error, "restore_error": "Backup failed", "timestamp": datetime.now().isoformat(), } ) log_with_timestamp(f"✗ {test_name}: BACKUP FAILED - {error}") # Test with compression (only for formats that support it) if format_name in ["custom", "directory"]: for compression_method in COMPRESSION_METHODS[1:]: # Skip 'none' if compression_method == "gzip" and format_name == "directory": continue # Directory format doesn't support gzip compression directly compression_levels = COMPRESSION_LEVELS.get(compression_method, [1]) for level in compression_levels: test_name = ( f"pg_dump_{format_name}_{compression_method}_level_{level}" ) output_path = backup_dir / f"{test_name}.{extension}" command = [ str(PG_BIN_PATH / "pg_dump.exe"), "-h", DB_CONFIG["host"], "-p", str(DB_CONFIG["port"]), "-U", DB_CONFIG["user"], "-d", DB_CONFIG["database"], "-f", str(output_path), "--format", format_name, "--verbose", ] # Add compression options if format_name == "custom": if compression_method == "gzip": command.extend(["-Z", str(level)]) else: command.extend( [ "--compress", f"{compression_method}:{level}", ] ) elif format_name == "directory": output_path.mkdir(exist_ok=True) # For directory format, replace the path after -f for i, item in enumerate(command): if item == "-f" and i + 1 < len(command): command[i + 1] = str(output_path) break if ( compression_method != "gzip" ): # Directory format supports lz4 and zstd command.extend( [ "--compress", f"{compression_method}:{level}", ] ) # Add parallel processing for directory format command.extend(["-j", "4"]) # Perform backup result, backup_duration, error = run_command(command) if result is not None: backup_size = get_file_size(output_path) db_size = get_database_size() compression_ratio = backup_size / db_size if db_size > 0 else 0 # Clean restore database and perform restore restore_success = False restore_duration = 0 restore_error = None restored_db_size = 0 if not skip_restore: if clean_restore_database() and wait_for_restore_db(): restore_result, restore_duration, restore_error = ( restore_pg_dump(output_path, format_name) ) if ( restore_result is not None and restore_result.returncode == 0 ): restore_success = True restored_db_size = get_restore_database_size() else: restore_success = True # Mark as successful if skipped restore_error = "Skipped" results.append( { "tool": "pg_dump", "format": format_name, "compression_method": compression_method, "compression_level": level, "backup_duration_seconds": backup_duration, "restore_duration_seconds": restore_duration, "total_duration_seconds": backup_duration + restore_duration, "backup_size_bytes": backup_size, "database_size_bytes": db_size, "restored_db_size_bytes": restored_db_size, "compression_ratio": compression_ratio, "backup_success": True, "restore_success": restore_success, "backup_error": None, "restore_error": restore_error, "timestamp": datetime.now().isoformat(), } ) if skip_restore: log_with_timestamp( f"✓ {test_name}: Backup {format_minutes(backup_duration)}, " f"{format_size(backup_size)}, ratio: {compression_ratio:.3f} (restore skipped)" ) else: log_with_timestamp( f"✓ {test_name}: Backup {format_minutes(backup_duration)}, Restore {format_minutes(restore_duration)}, " f"{format_size(backup_size)}, ratio: {compression_ratio:.3f}" ) else: results.append( { "tool": "pg_dump", "format": format_name, "compression_method": compression_method, "compression_level": level, "backup_duration_seconds": backup_duration, "restore_duration_seconds": 0, "total_duration_seconds": backup_duration, "backup_size_bytes": 0, "database_size_bytes": get_database_size(), "restored_db_size_bytes": 0, "compression_ratio": 0, "backup_success": False, "restore_success": False, "backup_error": error, "restore_error": "Backup failed", "timestamp": datetime.now().isoformat(), } ) log_with_timestamp(f"✗ {test_name}: BACKUP FAILED - {error}") def save_tabular_results(): """Save tabular test results to a CSV file.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Save as tabular CSV csv_file = f"backup_performance_tabular_{timestamp}.csv" with open(csv_file, "w", newline="") as f: writer = csv.writer(f) # Write header writer.writerow( [ "format", "backup duration mins", "restore duration mins", "total duration mins", "backup size GB", "db size GB", "restored DB size GB", "db size % from original", ] ) # Write data rows for r in sorted( results, key=lambda x: ( x["format"], x["compression_method"], x["compression_level"], ), ): if not r["backup_success"]: continue format_str = f"{r['format']} {r['compression_method']}" if r["compression_method"] != "none": format_str += f" {r['compression_level']}" backup_mins = round(r["backup_duration_seconds"] / 60.0, 1) restore_mins = round(r["restore_duration_seconds"] / 60.0, 1) total_mins = round(backup_mins + restore_mins, 1) backup_gb = round(r["backup_size_bytes"] / (1024.0 * 1024.0 * 1024.0), 1) db_gb = round(r["database_size_bytes"] / (1024.0 * 1024.0 * 1024.0), 1) restored_db_gb = round( r["restored_db_size_bytes"] / (1024.0 * 1024.0 * 1024.0), 1 ) db_size_percent = round((backup_gb / db_gb) * 100) if db_gb > 0 else 0 writer.writerow( [ format_str, backup_mins, restore_mins, total_mins, backup_gb, db_gb, restored_db_gb, db_size_percent, ] ) print(f"Tabular results saved to {csv_file}") def save_results(): """Save test results to JSON and CSV files.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Save as JSON json_file = f"backup_performance_results_{timestamp}.json" with open(json_file, "w") as f: json.dump(results, f, indent=2) print(f"\nResults saved to {json_file}") # Save as CSV csv_file = f"backup_performance_results_{timestamp}.csv" if results: fieldnames = results[0].keys() with open(csv_file, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(results) print(f"Results saved to {csv_file}") # Save tabular results save_tabular_results() def print_summary(): """Print a summary of test results.""" print("\n" + "=" * 80) print("PERFORMANCE TEST SUMMARY") print("=" * 80) if not results: print("No results to display.") return # Check if any tests had restore attempts restore_attempted = any( r["restore_error"] != "Skipped" for r in results if r["backup_success"] ) if restore_attempted: successful_tests = [ r for r in results if r["backup_success"] and r["restore_success"] ] backup_only_success = [ r for r in results if r["backup_success"] and not r["restore_success"] and r["restore_error"] != "Skipped" ] failed_tests = [r for r in results if not r["backup_success"]] print(f"Total tests: {len(results)}") print(f"Complete success (backup + restore): {len(successful_tests)}") print(f"Backup only success: {len(backup_only_success)}") print(f"Complete failures: {len(failed_tests)}") if successful_tests: print(f"\nBest compression ratios (complete success):") best_compression = sorted( successful_tests, key=lambda x: x["compression_ratio"] )[:5] for test in best_compression: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Ratio: {test['compression_ratio']:.3f}, Backup: {format_minutes(test['backup_duration_seconds'])}, " f"Restore: {format_minutes(test['restore_duration_seconds'])}, Total: {format_minutes(test['total_duration_seconds'])}" ) print(f"\nFastest total time (backup + restore):") fastest_total = sorted( successful_tests, key=lambda x: x["total_duration_seconds"] )[:5] for test in fastest_total: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Total: {format_minutes(test['total_duration_seconds'])}, Ratio: {test['compression_ratio']:.3f}" ) print(f"\nFastest backup times:") fastest_backup = sorted( successful_tests, key=lambda x: x["backup_duration_seconds"] )[:5] for test in fastest_backup: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Backup: {format_minutes(test['backup_duration_seconds'])}, Restore: {format_minutes(test['restore_duration_seconds'])}" ) print(f"\nFastest restore times:") fastest_restore = sorted( successful_tests, key=lambda x: x["restore_duration_seconds"] )[:5] for test in fastest_restore: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Restore: {format_minutes(test['restore_duration_seconds'])}, Backup: {format_minutes(test['backup_duration_seconds'])}" ) if backup_only_success: print(f"\nBackup-only successes (restore failed):") for test in backup_only_success: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Backup: {format_minutes(test['backup_duration_seconds'])}, Restore Error: {test['restore_error']}" ) if failed_tests: print(f"\nComplete failures:") for test in failed_tests: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Backup Error: {test['backup_error']}" ) else: # Restore was skipped for all tests successful_tests = [r for r in results if r["backup_success"]] failed_tests = [r for r in results if not r["backup_success"]] print(f"Total tests: {len(results)}") print(f"Successful backups: {len(successful_tests)}") print(f"Failed backups: {len(failed_tests)}") print("Note: Restore tests were skipped") if successful_tests: print(f"\nBest compression ratios:") best_compression = sorted( successful_tests, key=lambda x: x["compression_ratio"] )[:5] for test in best_compression: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Ratio: {test['compression_ratio']:.3f}, Backup: {format_minutes(test['backup_duration_seconds'])}" ) print(f"\nFastest backup times:") fastest_backup = sorted( successful_tests, key=lambda x: x["backup_duration_seconds"] )[:5] for test in fastest_backup: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Backup: {format_minutes(test['backup_duration_seconds'])}, Ratio: {test['compression_ratio']:.3f}" ) if failed_tests: print(f"\nFailed backups:") for test in failed_tests: print( f" {test['tool']} {test['format']} {test['compression_method']}:{test['compression_level']} - " f"Backup Error: {test['backup_error']}" ) def print_tabular_report(): """Print a tabular report of results similar to the requested format.""" if not results: print("No results to display.") return print("\n" + "=" * 120) print("TABULAR PERFORMANCE REPORT") print("=" * 120) # Print header print( f"{'format':<20}{'backup duration mins':<20}{'restore duration mins':<25}{'total duration mins':<20}" f"{'backup size GB':<15}{'db size GB':<15}{'restored DB size GB':<20}{'db size % from original':<25}" ) print("-" * 120) # Group results by format and compression method+level for r in sorted( results, key=lambda x: (x["format"], x["compression_method"], x["compression_level"]), ): if not r["backup_success"]: continue format_str = f"{r['format']} {r['compression_method']}" if r["compression_method"] != "none": format_str += f" {r['compression_level']}" backup_mins = r["backup_duration_seconds"] / 60.0 restore_mins = r["restore_duration_seconds"] / 60.0 total_mins = backup_mins + restore_mins backup_gb = r["backup_size_bytes"] / (1024.0 * 1024.0 * 1024.0) db_gb = r["database_size_bytes"] / (1024.0 * 1024.0 * 1024.0) restored_db_gb = r["restored_db_size_bytes"] / (1024.0 * 1024.0 * 1024.0) db_size_percent = (backup_gb / db_gb) * 100 if db_gb > 0 else 0 print( f"{format_str:<20}{backup_mins:.1f}{'':<14}{restore_mins:.1f}{'':<19}{total_mins:.1f}{'':<14}" f"{backup_gb:.1f}{'':<9}{db_gb:.1f}{'':<9}{restored_db_gb:.1f}{'':<14}{db_size_percent:.0f}" ) def main(): """Main function to run pg_dump backup performance tests.""" parser = argparse.ArgumentParser( description="PostgreSQL pg_dump backup performance testing" ) parser.add_argument( "--skip-gzip", action="store_true", help="Skip gzip compression tests" ) parser.add_argument( "--skip-lz4", action="store_true", help="Skip lz4 compression tests" ) parser.add_argument( "--skip-zstd", action="store_true", help="Skip zstd compression tests" ) parser.add_argument( "--skip-restore", action="store_true", help="Skip restore tests" ) args = parser.parse_args() # Filter compression methods based on skip flags global COMPRESSION_METHODS filtered_compression_methods = ["none"] # Always include 'none' if not args.skip_gzip: filtered_compression_methods.append("gzip") if not args.skip_lz4: filtered_compression_methods.append("lz4") if not args.skip_zstd: filtered_compression_methods.append("zstd") COMPRESSION_METHODS = filtered_compression_methods log_with_timestamp("PostgreSQL pg_dump Backup Performance Testing") log_with_timestamp("=" * 60) log_with_timestamp( f"Source Database: {DB_CONFIG['database']} on {DB_CONFIG['host']}:{DB_CONFIG['port']}" ) log_with_timestamp( f"Restore Database: {RESTORE_DB_CONFIG['database']} on {RESTORE_DB_CONFIG['host']}:{RESTORE_DB_CONFIG['port']}" ) log_with_timestamp(f"Database size: {format_size(get_database_size())}") log_with_timestamp(f"Compression methods: {', '.join(COMPRESSION_METHODS)}") log_with_timestamp( f"Restore tests: {'Skipped' if args.skip_restore else 'Enabled'}" ) log_with_timestamp(f"Test started: {datetime.now()}") # Ensure backup directory is ready backup_dir = ensure_backup_directory() log_with_timestamp(f"Backup directory: {backup_dir.absolute()}") try: # Run pg_dump tests only test_pg_dump(skip_restore=args.skip_restore) # Save and display results save_results() print_summary() print_tabular_report() except KeyboardInterrupt: log_with_timestamp("\nTest interrupted by user") if results: save_results() print_summary() print_tabular_report() except Exception as e: log_with_timestamp(f"\nTest failed with error: {e}") if results: save_results() print_summary() print_tabular_report() log_with_timestamp(f"\nTest completed: {datetime.now()}") if __name__ == "__main__": main()