Architecture Teardown: PostgreSQL 17 Partitioning – How It Handles 1TB per Table for Python 3.14 Apps

import psycopg
from psycopg.types.datetime import Date
from datetime import date, timedelta
import logging
from typing import List, Dict

# Configure logging for error tracking
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# PostgreSQL 17 connection parameters
DB_PARAMS = {
    "host": "localhost",
    "port": 5432,
    "user": "app_user",
    "password": "secure_password",
    "dbname": "telemetry_db"
}

def create_partitioned_telemetry_table() -> None:
    """
    Creates a 1TB-scale time-series telemetry table partitioned by month in PostgreSQL 17.
    Includes error handling for connection failures, duplicate table errors, and permission issues.
    """
    conn = None
    try:
        # Connect to PostgreSQL 17 instance using psycopg3 3.2.1 (Python 3.14 compatible)
        conn = psycopg.connect(**DB_PARAMS, autocommit=False)
        cur = conn.cursor()

        # Create parent partitioned table (PostgreSQL 17 supports native partitioning for all types)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS device_telemetry (
                device_id UUID NOT NULL,
                metric_name VARCHAR(64) NOT NULL,
                metric_value NUMERIC(18, 6) NOT NULL,
                recorded_at TIMESTAMPTZ NOT NULL,
                PRIMARY KEY (device_id, recorded_at)
            ) PARTITION BY RANGE (recorded_at);
        """)
        logger.info("Created parent partitioned table device_telemetry")

        # Create monthly partitions for 24 months (covers ~1TB of data at 12GB/month)
        start_date = date(2024, 1, 1)
        for month_offset in range(24):
            partition_start = start_date + timedelta(days=32 * month_offset)
            partition_end = (partition_start + timedelta(days=32)).replace(day=1)
            partition_name = f"device_telemetry_{partition_start.strftime('%Y%m')}"

            # PostgreSQL 17 supports automatic partition creation, but we explicitly define for 1TB scale
            cur.execute(f"""
                CREATE TABLE IF NOT EXISTS {partition_name}
                PARTITION OF device_telemetry
                FOR VALUES FROM ('{partition_start}') TO ('{partition_end}');
            """)
            logger.info(f"Created partition {partition_name}")

        # Create BRIN index on recorded_at (optimized for 1TB partitioning in PostgreSQL 17)
        cur.execute("""
            CREATE INDEX IF NOT EXISTS idx_telemetry_recorded_at_brin
            ON device_telemetry USING BRIN (recorded_at)
            WITH (pages_per_range = 128);
        """)

        conn.commit()
        logger.info("Successfully created partitioned table structure for 1TB telemetry data")

    except psycopg.OperationalError as e:
        logger.error(f"Connection failed: {e}")
        raise
    except psycopg.errors.DuplicateTableError as e:
        logger.warning(f"Table already exists: {e}")
        conn.rollback() if conn else None
    except psycopg.errors.InsufficientPrivilegeError as e:
        logger.error(f"Permission denied: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        conn.rollback() if conn else None
        raise
    finally:
        if conn:
            conn.close()
            logger.info("Database connection closed")

if __name__ == "__main__":
    create_partitioned_telemetry_table()

import psycopg
from psycopg.types.datetime import DateTime
from datetime import datetime, timedelta
import uuid
import random
import logging
from typing import List, Generator
import time

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DB_PARAMS = {
    "host": "localhost",
    "port": 5432,
    "user": "app_user",
    "password": "secure_password",
    "dbname": "telemetry_db"
}

def generate_telemetry_batch(batch_size: int = 10000) -> Generator[Dict, None, None]:
    """
    Generates synthetic telemetry data batches for 1TB scale testing.
    Simulates 10,000 devices reporting every 10 seconds for 6 months.
    """
    device_ids = [uuid.uuid4() for _ in range(10000)]
    metric_names = ["temperature", "humidity", "pressure", "voltage", "current"]
    end_time = datetime.now()
    start_time = end_time - timedelta(days=180)  # 6 months of data

    current_time = start_time
    while current_time <= end_time:
        batch = []
        for device_id in device_ids[:batch_size]:
            batch.append({
                "device_id": device_id,
                "metric_name": random.choice(metric_names),
                "metric_value": round(random.uniform(0.0, 100.0), 6),
                "recorded_at": current_time
            })
        yield batch
        current_time += timedelta(seconds=10)
        if current_time > end_time:
            break

def ingest_telemetry_to_partitioned_table() -> None:
    """
    Ingests 1TB of telemetry data into PostgreSQL 17 partitioned table using batch inserts.
    Includes retry logic, batch size tuning, and partition pruning validation.
    """
    conn = None
    total_rows = 0
    start_time = time.time()

    try:
        conn = psycopg.connect(**DB_PARAMS, autocommit=False)
        cur = conn.cursor()

        # Validate partition pruning works before ingestion (PostgreSQL 17 feature)
        cur.execute("""
            EXPLAIN (FORMAT JSON) SELECT * FROM device_telemetry
            WHERE recorded_at BETWEEN '2024-06-01' AND '2024-06-02';
        """)
        explain_plan = cur.fetchone()[0][0]
        if "Partition Pruning" not in str(explain_plan):
            logger.warning("Partition pruning not detected in explain plan")
        else:
            logger.info("Partition pruning validated for date range query")

        # Ingest batches
        for batch_num, batch in enumerate(generate_telemetry_batch()):
            try:
                # Use execute_values for bulk insert (psycopg3 optimized for partitioning)
                psycopg.execute_values(
                    cur,
                    """INSERT INTO device_telemetry (device_id, metric_name, metric_value, recorded_at)
                    VALUES %s""",
                    [(
                        row["device_id"],
                        row["metric_name"],
                        row["metric_value"],
                        row["recorded_at"]
                    ) for row in batch],
                    page_size=1000
                )
                total_rows += len(batch)
                conn.commit()
                if batch_num % 100 == 0:
                    logger.info(f"Ingested {total_rows} rows, batch {batch_num}")
            except psycopg.errors.UniqueViolationError as e:
                logger.warning(f"Duplicate row in batch {batch_num}: {e}")
                conn.rollback()
            except psycopg.OperationalError as e:
                logger.error(f"Connection lost during batch {batch_num}: {e}")
                # Retry logic for transient errors
                for retry in range(3):
                    try:
                        conn = psycopg.connect(**DB_PARAMS, autocommit=False)
                        cur = conn.cursor()
                        logger.info(f"Reconnected after retry {retry}")
                        break
                    except Exception as retry_e:
                        logger.error(f"Retry {retry} failed: {retry_e}")
                        time.sleep(2 ** retry)
                if not conn:
                    raise

        logger.info(f"Completed ingestion: {total_rows} rows in {time.time() - start_time:.2f}s")

    except Exception as e:
        logger.error(f"Ingestion failed: {e}")
        raise
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    ingest_telemetry_to_partitioned_table()

import psycopg
from datetime import datetime, timedelta
import logging
import time
from typing import List, Dict
from prometheus_client import start_http_server, Gauge

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheus metrics for query performance
QUERY_LATENCY = Gauge('postgres_partitioned_query_latency_ms', 'Query latency in ms', ['query_type'])
PARTITIONS_ACCESSED = Gauge('postgres_partitions_accessed', 'Number of partitions accessed', ['query_type'])

DB_PARAMS = {
    "host": "localhost",
    "port": 5432,
    "user": "app_user",
    "password": "secure_password",
    "dbname": "telemetry_db"
}

def query_partitioned_telemetry(query_type: str, start: datetime, end: datetime) -> List[Dict]:
    """
    Executes partition-aware queries on PostgreSQL 17 partitioned table.
    Collects metrics on latency and partition count for benchmarking.
    """
    conn = None
    start_time = time.time()

    try:
        conn = psycopg.connect(**DB_PARAMS, autocommit=True)
        cur = conn.cursor(row_factory=psycopg.rows.dict_row)

        # Query 1: Aggregate metrics (leverages PostgreSQL 17 partition-wise aggregation)
        if query_type == "aggregate":
            cur.execute("""
                EXPLAIN (FORMAT JSON) SELECT
                    metric_name,
                    AVG(metric_value) as avg_value,
                    COUNT(*) as sample_count
                FROM device_telemetry
                WHERE recorded_at BETWEEN %s AND %s
                GROUP BY metric_name;
            """, (start, end))
            explain_plan = cur.fetchone()[0][0]
            # Extract number of partitions accessed from explain plan
            partitions = len([node for node in explain_plan["Plan"]["Plans"] if "Partition" in node.get("Node Type", "")])
            PARTITIONS_ACCESSED.labels(query_type=query_type).set(partitions)

            cur.execute("""
                SELECT
                    metric_name,
                    AVG(metric_value) as avg_value,
                    COUNT(*) as sample_count
                FROM device_telemetry
                WHERE recorded_at BETWEEN %s AND %s
                GROUP BY metric_name;
            """, (start, end))
            result = cur.fetchall()

        # Query 2: Point lookup (tests partition pruning)
        elif query_type == "point_lookup":
            cur.execute("""
                EXPLAIN (FORMAT JSON) SELECT *
                FROM device_telemetry
                WHERE device_id = %s AND recorded_at = %s;
            """, (uuid.UUID("a1b2c3d4-e5f6-7890-abcd-1234567890ef"), start))
            explain_plan = cur.fetchone()[0][0]
            partitions = 1 if "Partition Pruning" in str(explain_plan) else 0
            PARTITIONS_ACCESSED.labels(query_type=query_type).set(partitions)

            cur.execute("""
                SELECT *
                FROM device_telemetry
                WHERE device_id = %s AND recorded_at = %s;
            """, (uuid.UUID("a1b2c3d4-e5f6-7890-abcd-1234567890ef"), start))
            result = cur.fetchall()

        else:
            raise ValueError(f"Unsupported query type: {query_type}")

        latency_ms = (time.time() - start_time) * 1000
        QUERY_LATENCY.labels(query_type=query_type).set(latency_ms)
        logger.info(f"Query {query_type} completed in {latency_ms:.2f}ms, accessed {partitions} partitions")
        return result

    except psycopg.errors.InvalidDatetimeFormatError as e:
        logger.error(f"Invalid date format: {e}")
        raise
    except psycopg.errors.UndefinedTableError as e:
        logger.error(f"Table not found: {e}")
        raise
    except Exception as e:
        logger.error(f"Query failed: {e}")
        raise
    finally:
        if conn:
            conn.close()

def run_benchmark_queries() -> None:
    """
    Runs benchmark queries against 1TB partitioned table and logs results.
    """
    start_http_server(8000)  # Expose Prometheus metrics
    logger.info("Started Prometheus metrics server on port 8000")

    # Benchmark 1: 1 month of data (aggregate)
    start = datetime(2024, 6, 1)
    end = datetime(2024, 7, 1)
    agg_result = query_partitioned_telemetry("aggregate", start, end)
    logger.info(f"Aggregate query returned {len(agg_result)} metric groups")

    # Benchmark 2: Point lookup for specific device
    point_result = query_partitioned_telemetry("point_lookup", start, start + timedelta(hours=1))
    logger.info(f"Point lookup returned {len(point_result)} rows")

if __name__ == "__main__":
    run_benchmark_queries()

-- PostgreSQL 17 BRIN index for partitioned time-series table
CREATE INDEX idx_telemetry_recorded_at_brin
ON device_telemetry USING BRIN (recorded_at)
WITH (pages_per_range = 128, autosummarize = on);

import psycopg
from datetime import datetime

# Good: Explicit partition key in WHERE clause (enables pruning)
cur.execute("""
    SELECT * FROM device_telemetry
    WHERE recorded_at BETWEEN %s AND %s
""", (datetime(2024,6,1), datetime(2024,6,2)))

# Bad: Function on partition key (disables pruning)
cur.execute("""
    SELECT * FROM device_telemetry
    WHERE DATE(recorded_at) = %s
""", ('2024-06-01',))

def check_partition_health() -> None:
    """Checks partition count and age using pg_partman functions via psycopg3."""
    conn = psycopg.connect(**DB_PARAMS)
    cur = conn.cursor()
    cur.execute("""
        SELECT partition_table_name, partition_range_start, partition_range_end
        FROM partman.partition_list
        WHERE parent_table = 'device_telemetry'
        ORDER BY partition_range_start;
    """)
    for row in cur.fetchall():
        logger.info(f"Partition {row[0]}: {row[1]} to {row[2]}")
    conn.close()