PostgreSQL Scaling Strategies Best Practices¶

Core Principles¶

Vertical Scaling: Optimize single-instance performance
Horizontal Scaling: Implement read replicas and sharding
Load Balancing: Distribute workload across instances
Caching: Implement multi-layer caching strategies
Monitoring: Monitor scaling effectiveness and bottlenecks

Vertical Scaling Strategies¶

Hardware Optimization¶

-- Create system resource monitoring table
CREATE TABLE system_resources (
    id SERIAL PRIMARY KEY,
    metric_name VARCHAR(100) NOT NULL,
    metric_value NUMERIC NOT NULL,
    metric_unit VARCHAR(20) NOT NULL,
    recorded_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to monitor system resources
CREATE OR REPLACE FUNCTION monitor_system_resources()
RETURNS VOID AS $$
BEGIN
    -- Monitor CPU usage
    INSERT INTO system_resources (metric_name, metric_value, metric_unit)
    VALUES ('cpu_usage', 0, 'percent'); -- Placeholder for actual CPU monitoring

    -- Monitor memory usage
    INSERT INTO system_resources (metric_name, metric_value, metric_unit)
    VALUES ('memory_usage', pg_database_size(current_database()) / 1024 / 1024, 'MB');

    -- Monitor disk usage
    INSERT INTO system_resources (metric_name, metric_value, metric_unit)
    VALUES ('disk_usage', pg_database_size(current_database()) / 1024 / 1024 / 1024, 'GB');

    -- Monitor active connections
    INSERT INTO system_resources (metric_name, metric_value, metric_unit)
    VALUES ('active_connections', (SELECT COUNT(*) FROM pg_stat_activity), 'count');
END;
$$ LANGUAGE plpgsql;

Configuration Tuning¶

-- Create configuration optimization table
CREATE TABLE configuration_optimization (
    id SERIAL PRIMARY KEY,
    parameter_name VARCHAR(100) NOT NULL,
    current_value TEXT NOT NULL,
    recommended_value TEXT NOT NULL,
    optimization_type VARCHAR(50) NOT NULL,
    impact_level VARCHAR(20) NOT NULL,
    description TEXT,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to analyze configuration
CREATE OR REPLACE FUNCTION analyze_configuration()
RETURNS TABLE (
    parameter_name VARCHAR(100),
    current_value TEXT,
    recommended_value TEXT,
    optimization_type VARCHAR(50),
    impact_level VARCHAR(20)
) AS $$
BEGIN
    -- Analyze shared_buffers
    RETURN QUERY
    SELECT 
        'shared_buffers'::VARCHAR(100),
        current_setting('shared_buffers')::TEXT,
        '25% of total RAM'::TEXT,
        'memory'::VARCHAR(50),
        'high'::VARCHAR(20);

    -- Analyze work_mem
    RETURN QUERY
    SELECT 
        'work_mem'::VARCHAR(100),
        current_setting('work_mem')::TEXT,
        '4MB'::TEXT,
        'memory'::VARCHAR(50),
        'medium'::VARCHAR(20);

    -- Analyze maintenance_work_mem
    RETURN QUERY
    SELECT 
        'maintenance_work_mem'::VARCHAR(100),
        current_setting('maintenance_work_mem')::TEXT,
        '256MB'::TEXT,
        'memory'::VARCHAR(50),
        'medium'::VARCHAR(20);

    -- Analyze effective_cache_size
    RETURN QUERY
    SELECT 
        'effective_cache_size'::VARCHAR(100),
        current_setting('effective_cache_size')::TEXT,
        '75% of total RAM'::TEXT,
        'memory'::VARCHAR(50),
        'high'::VARCHAR(20);
END;
$$ LANGUAGE plpgsql;

Horizontal Scaling Strategies¶

Read Replica Management¶

-- Create read replica configuration table
CREATE TABLE read_replicas (
    id SERIAL PRIMARY KEY,
    replica_name VARCHAR(100) UNIQUE NOT NULL,
    host VARCHAR(100) NOT NULL,
    port INTEGER NOT NULL,
    database_name VARCHAR(100) NOT NULL,
    username VARCHAR(100) NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    lag_threshold INTEGER DEFAULT 1000, -- milliseconds
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to add read replica
CREATE OR REPLACE FUNCTION add_read_replica(
    p_replica_name VARCHAR(100),
    p_host VARCHAR(100),
    p_port INTEGER,
    p_database_name VARCHAR(100),
    p_username VARCHAR(100),
    p_lag_threshold INTEGER DEFAULT 1000
)
RETURNS VOID AS $$
BEGIN
    INSERT INTO read_replicas (
        replica_name, host, port, database_name, username, lag_threshold
    ) VALUES (
        p_replica_name, p_host, p_port, p_database_name, p_username, p_lag_threshold
    );
END;
$$ LANGUAGE plpgsql;

-- Create function to get read replica
CREATE OR REPLACE FUNCTION get_read_replica(p_workload_type VARCHAR(50))
RETURNS TABLE (
    replica_name VARCHAR(100),
    host VARCHAR(100),
    port INTEGER,
    database_name VARCHAR(100),
    username VARCHAR(100)
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        rr.replica_name,
        rr.host,
        rr.port,
        rr.database_name,
        rr.username
    FROM read_replicas rr
    WHERE rr.is_active = TRUE
    ORDER BY RANDOM()
    LIMIT 1;
END;
$$ LANGUAGE plpgsql;

Load Balancing¶

-- Create load balancer configuration table
CREATE TABLE load_balancer_config (
    id SERIAL PRIMARY KEY,
    service_name VARCHAR(100) NOT NULL,
    target_host VARCHAR(100) NOT NULL,
    target_port INTEGER NOT NULL,
    weight INTEGER DEFAULT 1,
    is_healthy BOOLEAN DEFAULT TRUE,
    last_health_check TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to check replica health
CREATE OR REPLACE FUNCTION check_replica_health(p_replica_name VARCHAR(100))
RETURNS BOOLEAN AS $$
DECLARE
    replica_host VARCHAR(100);
    replica_port INTEGER;
    replica_db VARCHAR(100);
    replica_user VARCHAR(100);
    lag_milliseconds INTEGER;
BEGIN
    -- Get replica configuration
    SELECT host, port, database_name, username
    INTO replica_host, replica_port, replica_db, replica_user
    FROM read_replicas
    WHERE replica_name = p_replica_name AND is_active = TRUE;

    IF replica_host IS NULL THEN
        RETURN FALSE;
    END IF;

    -- Check replication lag (simplified)
    -- In practice, this would connect to the replica and check lag
    lag_milliseconds := 0; -- Placeholder

    -- Update health status
    UPDATE read_replicas
    SET last_health_check = CURRENT_TIMESTAMP
    WHERE replica_name = p_replica_name;

    RETURN lag_milliseconds < 1000; -- 1 second threshold
END;
$$ LANGUAGE plpgsql;

Sharding Strategies¶

Shard Management¶

-- Create shard configuration table
CREATE TABLE shard_config (
    id SERIAL PRIMARY KEY,
    shard_name VARCHAR(100) UNIQUE NOT NULL,
    shard_key VARCHAR(100) NOT NULL,
    shard_range_start BIGINT NOT NULL,
    shard_range_end BIGINT NOT NULL,
    host VARCHAR(100) NOT NULL,
    port INTEGER NOT NULL,
    database_name VARCHAR(100) NOT NULL,
    username VARCHAR(100) NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to add shard
CREATE OR REPLACE FUNCTION add_shard(
    p_shard_name VARCHAR(100),
    p_shard_key VARCHAR(100),
    p_range_start BIGINT,
    p_range_end BIGINT,
    p_host VARCHAR(100),
    p_port INTEGER,
    p_database_name VARCHAR(100),
    p_username VARCHAR(100)
)
RETURNS VOID AS $$
BEGIN
    INSERT INTO shard_config (
        shard_name, shard_key, shard_range_start, shard_range_end,
        host, port, database_name, username
    ) VALUES (
        p_shard_name, p_shard_key, p_range_start, p_range_end,
        p_host, p_port, p_database_name, p_username
    );
END;
$$ LANGUAGE plpgsql;

-- Create function to find shard for key
CREATE OR REPLACE FUNCTION find_shard_for_key(p_shard_key VARCHAR(100), p_key_value BIGINT)
RETURNS TABLE (
    shard_name VARCHAR(100),
    host VARCHAR(100),
    port INTEGER,
    database_name VARCHAR(100),
    username VARCHAR(100)
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        sc.shard_name,
        sc.host,
        sc.port,
        sc.database_name,
        sc.username
    FROM shard_config sc
    WHERE sc.shard_key = p_shard_key
    AND sc.is_active = TRUE
    AND p_key_value >= sc.shard_range_start
    AND p_key_value < sc.shard_range_end;
END;
$$ LANGUAGE plpgsql;

Shard Routing¶

-- Create function to route queries to shards
CREATE OR REPLACE FUNCTION route_to_shard(
    p_table_name VARCHAR(100),
    p_shard_key VARCHAR(100),
    p_key_value BIGINT
)
RETURNS TEXT AS $$
DECLARE
    shard_info RECORD;
    connection_string TEXT;
BEGIN
    -- Find appropriate shard
    SELECT * INTO shard_info
    FROM find_shard_for_key(p_shard_key, p_key_value);

    IF shard_info IS NULL THEN
        RAISE EXCEPTION 'No shard found for key % with value %', p_shard_key, p_key_value;
    END IF;

    -- Build connection string
    connection_string := format('host=%s port=%s dbname=%s user=%s',
                               shard_info.host, shard_info.port, 
                               shard_info.database_name, shard_info.username);

    RETURN connection_string;
END;
$$ LANGUAGE plpgsql;

Caching Strategies¶

Multi-Layer Caching¶

-- Create cache configuration table
CREATE TABLE cache_config (
    id SERIAL PRIMARY KEY,
    cache_name VARCHAR(100) UNIQUE NOT NULL,
    cache_type VARCHAR(50) NOT NULL,
    ttl_seconds INTEGER NOT NULL,
    max_size_mb INTEGER NOT NULL,
    is_enabled BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to configure cache
CREATE OR REPLACE FUNCTION configure_cache(
    p_cache_name VARCHAR(100),
    p_cache_type VARCHAR(50),
    p_ttl_seconds INTEGER,
    p_max_size_mb INTEGER,
    p_is_enabled BOOLEAN DEFAULT TRUE
)
RETURNS VOID AS $$
BEGIN
    INSERT INTO cache_config (
        cache_name, cache_type, ttl_seconds, max_size_mb, is_enabled
    ) VALUES (
        p_cache_name, p_cache_type, p_ttl_seconds, p_max_size_mb, p_is_enabled
    ) ON CONFLICT (cache_name) 
    DO UPDATE SET 
        cache_type = EXCLUDED.cache_type,
        ttl_seconds = EXCLUDED.ttl_seconds,
        max_size_mb = EXCLUDED.max_size_mb,
        is_enabled = EXCLUDED.is_enabled;
END;
$$ LANGUAGE plpgsql;

-- Create function to get cache configuration
CREATE OR REPLACE FUNCTION get_cache_config(p_cache_name VARCHAR(100))
RETURNS TABLE (
    cache_name VARCHAR(100),
    cache_type VARCHAR(50),
    ttl_seconds INTEGER,
    max_size_mb INTEGER,
    is_enabled BOOLEAN
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        cc.cache_name,
        cc.cache_type,
        cc.ttl_seconds,
        cc.max_size_mb,
        cc.is_enabled
    FROM cache_config cc
    WHERE cc.cache_name = p_cache_name;
END;
$$ LANGUAGE plpgsql;

Query Result Caching¶

-- Create query cache table
CREATE TABLE query_cache (
    id SERIAL PRIMARY KEY,
    query_hash VARCHAR(64) UNIQUE NOT NULL,
    query_sql TEXT NOT NULL,
    result_data JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMPTZ NOT NULL,
    access_count INTEGER DEFAULT 0,
    last_accessed TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create function to cache query result
CREATE OR REPLACE FUNCTION cache_query_result(
    p_query_sql TEXT,
    p_result_data JSONB,
    p_ttl_seconds INTEGER DEFAULT 3600
)
RETURNS VARCHAR(64) AS $$
DECLARE
    query_hash VARCHAR(64);
    expires_at TIMESTAMPTZ;
BEGIN
    -- Generate query hash
    query_hash := encode(digest(p_query_sql, 'sha256'), 'hex');

    -- Calculate expiration time
    expires_at := CURRENT_TIMESTAMP + (p_ttl_seconds || ' seconds')::interval;

    -- Insert or update cache entry
    INSERT INTO query_cache (query_hash, query_sql, result_data, expires_at)
    VALUES (query_hash, p_query_sql, p_result_data, expires_at)
    ON CONFLICT (query_hash) 
    DO UPDATE SET 
        result_data = EXCLUDED.result_data,
        expires_at = EXCLUDED.expires_at,
        access_count = query_cache.access_count + 1,
        last_accessed = CURRENT_TIMESTAMP;

    RETURN query_hash;
END;
$$ LANGUAGE plpgsql;

-- Create function to get cached result
CREATE OR REPLACE FUNCTION get_cached_result(p_query_sql TEXT)
RETURNS JSONB AS $$
DECLARE
    query_hash VARCHAR(64);
    result_data JSONB;
BEGIN
    -- Generate query hash
    query_hash := encode(digest(p_query_sql, 'sha256'), 'hex');

    -- Get cached result
    SELECT qc.result_data INTO result_data
    FROM query_cache qc
    WHERE qc.query_hash = query_hash
    AND qc.expires_at > CURRENT_TIMESTAMP;

    -- Update access statistics
    IF result_data IS NOT NULL THEN
        UPDATE query_cache
        SET access_count = access_count + 1,
            last_accessed = CURRENT_TIMESTAMP
        WHERE query_hash = query_hash;
    END IF;

    RETURN result_data;
END;
$$ LANGUAGE plpgsql;

Scaling Automation¶

Auto-Scaling Implementation¶

# scaling/postgres_scaler.py
import psycopg2
import json
import time
from datetime import datetime, timedelta
import logging

class PostgreSQLScaler:
    def __init__(self, connection_params):
        self.conn_params = connection_params
        self.setup_logging()

    def setup_logging(self):
        """Setup logging."""
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def monitor_scaling_metrics(self):
        """Monitor scaling metrics."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Monitor active connections
                cur.execute("SELECT COUNT(*) FROM pg_stat_activity")
                active_connections = cur.fetchone()[0]

                # Monitor database size
                cur.execute("SELECT pg_database_size(current_database())")
                db_size = cur.fetchone()[0]

                # Monitor query performance
                cur.execute("""
                    SELECT 
                        AVG(EXTRACT(EPOCH FROM (now() - query_start))) as avg_query_time,
                        COUNT(*) as active_queries
                    FROM pg_stat_activity 
                    WHERE state = 'active' AND query_start IS NOT NULL
                """)
                query_metrics = cur.fetchone()

                return {
                    'active_connections': active_connections,
                    'database_size': db_size,
                    'avg_query_time': query_metrics[0] if query_metrics[0] else 0,
                    'active_queries': query_metrics[1] if query_metrics[1] else 0
                }

        except Exception as e:
            self.logger.error(f"Error monitoring scaling metrics: {e}")
            return {}
        finally:
            conn.close()

    def check_scaling_thresholds(self, metrics):
        """Check if scaling thresholds are met."""
        thresholds = {
            'max_connections': 100,
            'max_query_time': 5.0,  # seconds
            'max_db_size': 1073741824  # 1GB
        }

        scaling_needed = []

        if metrics.get('active_connections', 0) > thresholds['max_connections']:
            scaling_needed.append('connection_limit')

        if metrics.get('avg_query_time', 0) > thresholds['max_query_time']:
            scaling_needed.append('query_performance')

        if metrics.get('database_size', 0) > thresholds['max_db_size']:
            scaling_needed.append('storage_capacity')

        return scaling_needed

    def scale_horizontally(self, scaling_reasons):
        """Scale horizontally by adding read replicas."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Add read replica
                replica_name = f"replica_{int(time.time())}"
                cur.execute("""
                    SELECT add_read_replica(%s, %s, %s, %s, %s, %s)
                """, (replica_name, 'replica-host', 5432, 'production', 'replica_user', 1000))

                conn.commit()
                self.logger.info(f"Added read replica: {replica_name}")

        except Exception as e:
            self.logger.error(f"Error scaling horizontally: {e}")
            conn.rollback()
            raise
        finally:
            conn.close()

    def scale_vertically(self, scaling_reasons):
        """Scale vertically by optimizing configuration."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Optimize configuration based on scaling reasons
                if 'connection_limit' in scaling_reasons:
                    cur.execute("ALTER SYSTEM SET max_connections = 200")
                    self.logger.info("Increased max_connections to 200")

                if 'query_performance' in scaling_reasons:
                    cur.execute("ALTER SYSTEM SET shared_buffers = '256MB'")
                    cur.execute("ALTER SYSTEM SET work_mem = '8MB'")
                    self.logger.info("Optimized memory settings for query performance")

                if 'storage_capacity' in scaling_reasons:
                    cur.execute("ALTER SYSTEM SET checkpoint_segments = 32")
                    self.logger.info("Optimized checkpoint settings for storage")

                conn.commit()

        except Exception as e:
            self.logger.error(f"Error scaling vertically: {e}")
            conn.rollback()
            raise
        finally:
            conn.close()

    def auto_scale(self):
        """Automatically scale based on metrics."""
        metrics = self.monitor_scaling_metrics()
        scaling_reasons = self.check_scaling_thresholds(metrics)

        if scaling_reasons:
            self.logger.info(f"Scaling needed for: {scaling_reasons}")

            # Scale horizontally for connection limits
            if 'connection_limit' in scaling_reasons:
                self.scale_horizontally(scaling_reasons)

            # Scale vertically for performance
            if 'query_performance' in scaling_reasons or 'storage_capacity' in scaling_reasons:
                self.scale_vertically(scaling_reasons)
        else:
            self.logger.info("No scaling needed")

# Usage
if __name__ == "__main__":
    scaler = PostgreSQLScaler({
        'host': 'localhost',
        'database': 'production',
        'user': 'scaler_user',
        'password': 'scaler_password'
    })

    # Run auto-scaling
    scaler.auto_scale()

Scaling Monitoring¶

Performance Monitoring¶

# monitoring/scaling_monitor.py
import psycopg2
import json
from datetime import datetime
import logging

class ScalingMonitor:
    def __init__(self, connection_params):
        self.conn_params = connection_params
        self.setup_logging()

    def setup_logging(self):
        """Setup logging."""
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def get_scaling_metrics(self):
        """Get scaling metrics."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Get system metrics
                cur.execute("""
                    SELECT 
                        COUNT(*) as active_connections,
                        pg_database_size(current_database()) as database_size,
                        AVG(EXTRACT(EPOCH FROM (now() - query_start))) as avg_query_time
                    FROM pg_stat_activity 
                    WHERE state = 'active'
                """)

                system_metrics = cur.fetchone()

                # Get replica status
                cur.execute("""
                    SELECT 
                        replica_name,
                        is_active,
                        last_health_check
                    FROM read_replicas
                    ORDER BY replica_name
                """)

                replica_status = cur.fetchall()

                # Get shard status
                cur.execute("""
                    SELECT 
                        shard_name,
                        is_active,
                        shard_range_start,
                        shard_range_end
                    FROM shard_config
                    ORDER BY shard_name
                """)

                shard_status = cur.fetchall()

                return {
                    'system_metrics': system_metrics,
                    'replica_status': replica_status,
                    'shard_status': shard_status
                }

        except Exception as e:
            self.logger.error(f"Error getting scaling metrics: {e}")
            return {}
        finally:
            conn.close()

    def generate_scaling_report(self):
        """Generate scaling report."""
        metrics = self.get_scaling_metrics()

        report = {
            'report_timestamp': datetime.now().isoformat(),
            'scaling_metrics': metrics
        }

        return report

# Usage
if __name__ == "__main__":
    monitor = ScalingMonitor({
        'host': 'localhost',
        'database': 'production',
        'user': 'monitor_user',
        'password': 'monitor_password'
    })

    report = monitor.generate_scaling_report()
    print(json.dumps(report, indent=2))

TL;DR Runbook¶

Quick Start¶

-- 1. Configure read replicas
SELECT add_read_replica('replica_1', 'replica-host', 5432, 'production', 'replica_user');

-- 2. Configure sharding
SELECT add_shard('shard_1', 'user_id', 0, 1000000, 'shard-host', 5432, 'production', 'shard_user');

-- 3. Configure caching
SELECT configure_cache('query_cache', 'memory', 3600, 100, TRUE);

-- 4. Monitor scaling metrics
SELECT * FROM monitor_system_resources();

Essential Patterns¶

# Complete PostgreSQL scaling setup
def setup_postgresql_scaling():
    # 1. Vertical scaling strategies
    # 2. Horizontal scaling strategies
    # 3. Sharding strategies
    # 4. Caching strategies
    # 5. Scaling automation
    # 6. Performance monitoring
    # 7. Load balancing
    # 8. Auto-scaling

    print("PostgreSQL scaling setup complete!")