Archival

Implement production-ready data archival strategies for PostgreSQL and MySQL that move historical records to archive tables or cold storage (S3, Azure Blob, GCS) with automated retention policies, compression, compliance tracking, and zero-downtime migration. Reduce primary database size by 50-90% while maintaining query access to archived data.

When to Use This Command

Use /archival when you need to:

• Reduce primary database size by archiving historical records (2+ years old)
• Meet compliance requirements (GDPR, HIPAA) for data retention policies
• Improve query performance on hot tables by removing cold data
• Move infrequently accessed data to cost-effective cold storage (S3)
• Implement tiered storage strategy (hot → warm → cold → deleted)
• Maintain audit trail for archived and deleted records

DON'T use this when:

• You need real-time access to all historical data (denormalize instead)
• Data volume is small (<10GB) and performance is acceptable
• Records are frequently updated (archival is for immutable historical data)
• Compliance requires online access to all data (use read replicas instead)
• You lack backup/restore strategy (test archival process first)

Design Decisions

This command implements automated tiered archival because:

• Separates hot (active) data from cold (historical) data for performance
• Reduces storage costs by moving old data to S3 (1/10th database cost)
• Maintains compliance with automated retention and deletion policies
• Enables point-in-time restore of archived data independent of production
• Provides audit trail for all archival and deletion operations

Alternative considered: Database partitioning only

• Faster queries on partitioned tables (no cross-database joins)
• Simpler backup/restore (single database)
• Still incurs full database storage costs
• Recommended when query performance is primary concern

Alternative considered: Soft deletes (deleted_at flag)

• Simplest implementation (no data movement)
• Maintains referential integrity automatically
• Database size continues growing indefinitely
• Recommended for small datasets with regulatory holds

Prerequisites

Before running this command:

1. Database backup strategy with tested restore procedures
2. Compliance requirements documented (retention periods, deletion policies)
3. Storage destination configured (archive database, S3 bucket, or both)
4. Monitoring for archival job success/failure and storage metrics
5. Testing environment to validate archival process before production

Implementation Process

Step 1: Analyze Table Growth and Identify Archival Candidates

Query table sizes, row counts, and date distribution to identify hot vs cold data.

Step 2: Design Archival Strategy

Choose archive tables, cold storage, or hybrid approach based on access patterns.

Step 3: Implement Archival Automation

Create jobs to move data in batches with transaction safety and error handling.

Step 4: Validate Data Integrity

Compare row counts, checksums, and sample records between source and archive.

Step 5: Monitor and Optimize

Track archival job performance, storage savings, and query latency improvements.

Output Format

The command generates:

• archival/schema.sql - Archive table definitions with identical structure
• archival/archive_job.py - Python script for automated batch archival
• archival/cold_storage_export.py - S3/GCS export with Parquet compression
• archival/retention_policy.sql - SQL procedures for automated deletion
• archival/monitoring_dashboard.json - Grafana dashboard for archival metrics

Code Examples

Example 1: PostgreSQL Automated Archival with Transaction Safety

#!/usr/bin/env python3
"""
Production-ready PostgreSQL archival system with batch processing,
transaction safety, and comprehensive error handling.
"""

import psycopg2
from psycopg2.extras import RealDictCursor
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
…