TimescaleDB Hypertable Creator and Loader

Problem Statement

When operating high-throughput time-series workloads, relational databases like standard PostgreSQL often suffer severe performance degradation as table size increases. As indexes grow larger than available RAM, write throughput plummets, and point-in-time queries require expensive, full-table scans or heavy B-tree index traversals. TimescaleDB solves this by introducing Hypertables—abstractions that automatically partition data into time-based (and optionally space-based) chunks while presenting a unified table interface to SQL clients.

Manually managing hypertable creation, setting appropriate chunk intervals, configuring compression, and bulk-loading datasets across multiple environments is error-prone and hard to automate. This script provides an automated pipeline for creating TimescaleDB hypertables, configuring retention and compression policies, and performing high-throughput batch ingestion from CSV files or auto-generated datasets using optimized PostgreSQL COPY operations.

How It Works

The architecture relies on psycopg2 to execute raw SQL statements against a PostgreSQL instance with the TimescaleDB extension enabled. The workflow follows these steps:

1. Connection & Environment Discovery: Connects to PostgreSQL using environment variables (TSDB_HOST, TSDB_USER, etc.) and verifies that the timescaledb extension is enabled.
2. Schema & Hypertable Setup: Checks if the target table exists. If not, it creates a standardized time-series schema (time, device_id, metric_name, value). It then converts the standard table into a Hypertable using create_hypertable(), specifying a customizable chunk time interval (e.g., 7 days).
3. Policy Configuration: Optionally configures automatic data compression policies (e.g., compress chunks older than 7 days) and retention policies (e.g., drop chunks older than 90 days).
4. Data Ingestion Pipeline: Ingests incoming data using PostgreSQL's native bulk-copy capabilities (COPY FROM STDIN via psycopg2.copy_expert), which bypasses standard SQL parsing overhead for maximum ingestion rates.

[CSV / Input File]
       │
       ▼
[TSDB Ingestion Script]
       │
       ├── 1. Connect & Enable Extension ("timescaledb")
       ├── 2. CREATE TABLE + create_hypertable()
       ├── 3. Set Policies (Compression & Retention)
       └── 4. High-speed Bulk Ingest (COPY command)
       │
       ▼
[TimescaleDB Chunks] (e.g., Chunk 1: Week 1, Chunk 2: Week 2)

Full Script

#!/usr/bin/env python3
"""
TimescaleDB Hypertable Creator and Loader

Automates the creation of TimescaleDB hypertables, manages compression and retention 
policies, and provides high-speed bulk ingestion capabilities for time-series data.
"""

import argparse
import csv
import io
import logging
import os
import random
import sys
from datetime import datetime, timedelta, timezone
from typing import Any, Sequence

import psycopg2
from psycopg2 import sql
from psycopg2.extensions import connection as PgConnection

# Configure Logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("tsdb_loader")

# Environment Variable Configurations with sensible defaults
DB_HOST = os.environ.get("TSDB_HOST", "localhost")
DB_PORT = int(os.environ.get("TSDB_PORT", "5432"))
DB_NAME = os.environ.get("TSDB_DB", "postgres")
DB_USER = os.environ.get("TSDB_USER", "postgres")
DB_PASS = os.environ.get("TSDB_PASSWORD", "postgres")
DB_SSL = os.environ.get("TSDB_SSLMODE", "prefer")


def get_db_connection() -> PgConnection:
    """Establishes and returns a connection to the PostgreSQL/TimescaleDB database."""
    try:
        conn = psycopg2.connect(
            host=DB_HOST,
            port=DB_PORT,
            dbname=DB_NAME,
            user=DB_USER,
            password=DB_PASS,
            sslmode=DB_SSL,
        )
        conn.autocommit = False
        return conn
    except psycopg2.Error as e:
        logger.error(f"Failed to connect to TimescaleDB at {DB_HOST}:{DB_PORT}/{DB_NAME}")
        logger.error(f"Database Error: {e}")
        sys.exit(1)


def ensure_extension(conn: PgConnection) -> None:
    """Ensures the TimescaleDB extension is enabled in the target database."""
    with conn.cursor() as cursor:
        cursor.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")
    conn.commit()
    logger.info("Verified TimescaleDB extension is active.")


def create_base_table_if_not_exists(conn: PgConnection, table_name: str, time_col: str) -> None:
    """Creates a default metric table if it does not already exist."""
    query = sql.SQL("""
        CREATE TABLE IF NOT EXISTS {table} (
            {time_field} TIMESTAMPTZ NOT NULL,
            device_id VARCHAR(64) NOT NULL,
            metric_name VARCHAR(64) NOT NULL,
            value DOUBLE PRECISION NOT NULL
        );
    """).format(
        table=sql.Identifier(table_name),
        time_field=sql.Identifier(time_col),
    )
    
    with conn.cursor() as cursor:
        cursor.execute(query)
    conn.commit()
    logger.info(f"Ensured base table '{table_name}' exists.")


def convert_to_hypertable(
    conn: PgConnection,
    table_name: str,
    time_col: str,
    chunk_interval: str,
) -> None:
    """Converts a standard PostgreSQL table into a TimescaleDB Hypertable."""
    query = """
        SELECT create_hypertable(
            %s::regclass,
            %s,
            chunk_time_interval => %s::interval,
            if_not_exists => TRUE
        );
    """
    with conn.cursor() as cursor:
        cursor.execute(query, (table_name, time_col, chunk_interval))
    conn.commit()
    logger.info(f"Hypertable '{table_name}' initialized with chunk interval '{chunk_interval}'.")


def configure_hypertable_policies(
    conn: PgConnection,
    table_name: str,
    enable_compression: bool,
    retention_days: int | None,
) -> None:
    """Sets up compression and data retention policies on the hypertable."""
    with conn.cursor() as cursor:
        if enable_compression:
            # Enable compression on the table
            alter_query = sql.SQL("""
                ALTER TABLE {table} SET (
                    timescaledb.compress,
                    timescaledb.compress_segmentby = 'device_id, metric_name'
                );
            """).format(table=sql.Identifier(table_name))
            
            try:
                cursor.execute(alter_query)
                # Add compression policy (compress chunks older than 7 days)
                cursor.execute(
                    "SELECT add_compression_policy(%s, INTERVAL '7 days', if_not_exists => TRUE);",
                    (table_name,),
                )
                logger.info(f"Enabled compression policy on '{table_name}' (older than 7 days).")
            except psycopg2.Error as e:
                logger.warning(f"Could not configure compression: {e}")
                conn.rollback()

        if retention_days and retention_days > 0:
            retention_interval = f"{retention_days} days"
            cursor.execute(
                "SELECT add_retention_policy(%s, INTERVAL %s, if_not_exists => TRUE);",
                (table_name, retention_interval),
            )
            logger.info(f"Enabled retention policy on '{table_name}' (drop after {retention_days} days).")
            
    conn.commit()


def bulk_load_csv(conn: PgConnection, table_name: str, file_path: str) -> int:
    """Loads CSV data into the hypertable using COPY FROM for maximum throughput."""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Input file not found: {file_path}")

    logger.info(f"Starting bulk load from '{file_path}' into '{table_name}'...")
    
    copy_query = sql.SQL("COPY {table} FROM STDIN WITH (FORMAT csv, HEADER true)").format(
        table=sql.Identifier(table_name)
    )

    count = 0
    with open(file_path, "r", encoding="utf-8") as f:
        # Count lines for logging feedback
        count = sum(1 for _ in f) - 1
        f.seek(0)
        
        with conn.cursor() as cursor:
            cursor.copy_expert(sql=copy_query.as_string(conn), file=f)
            
    conn.commit()
    logger.info(f"Successfully loaded {count:,} records into '{table_name}'.")
    return count


def generate_mock_csv(file_path: str, num_rows: int = 10000) -> None:
    """Generates sample time-series data for demonstration or testing purposes."""
    logger.info(f"Generating {num_rows:,} mock metrics records at '{file_path}'...")
    
    devices = [f"sensor-node-{i:03d}" for i in range(1, 11)]
    metrics = ["cpu_usage", "memory_usage", "temperature", "humidity"]
    
    start_time = datetime.now(timezone.utc) - timedelta(days=30)
    
    with open(file_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["ts", "device_id", "metric_name", "value"])
        
        for i in range(num_rows):
            current_time = start_time + timedelta(seconds=i * 15)
            device = random.choice(devices)
            metric = random.choice(metrics)
            value = round(random.uniform(10.0, 99.9), 2)
            
            writer.writerow([current_time.isoformat(), device, metric, value])
            
    logger.info("Mock dataset generated successfully.")


def parse_args() -> argparse.Namespace:
    """Parses command-line arguments."""
    parser = argparse.ArgumentParser(
        description="TimescaleDB Hypertable Creator and Loader Utility"
    )
    parser.add_argument(
        "--table",
        type=str,
        default="metrics",
        help="Name of the target database table/hypertable (default: metrics)",
    )
    parser.add_argument(
        "--time-col",
        type=str,
        default="ts",
        help="Column name representing time dimension (default: ts)",
    )
    parser.add_argument(
        "--chunk-interval",
        type=str,
        default="7 days",
        help="TimescaleDB chunk time interval (default: '7 days')",
    )
    parser.add_argument(
        "--input",
        type=str,
        default=None,
        help="Path to CSV file to ingest. If omitted, mock data will be generated.",
    )
    parser.add_argument(
        "--enable-compression",
        action="store_true",
        help="Enable automatic hypertable compression policy",
    )
    parser.add_argument(
        "--retention-days",
        type=int,
        default=None,
        help="Optional data retention policy in days (e.g., 90)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    
    conn = get_db_connection()
    try:
        ensure_extension(conn)
        create_base_table_if_not_exists(conn, args.table, args.time_col)
        convert_to_hypertable(conn, args.table, args.time_col, args.chunk_interval)
        configure_hypertable_policies(
            conn, args.table, args.enable_compression, args.retention_days
        )
        
        input_file = args.input
        if not input_file:
            input_file = "/tmp/sample_metrics.csv"
            generate_mock_csv(input_file, num_rows=25000)
            
        bulk_load_csv(conn, args.table, input_file)
        
    except psycopg2.Error as e:
        logger.error(f"Database operation failed: {e}")
        conn.rollback()
        sys.exit(1)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        sys.exit(1)
    finally:
        conn.close()
        logger.info("Database connection closed.")


if __name__ == "__main__":
    main()

Configuration

The script is controlled via environment variables for database credentials and CLI flags for table mechanics:

Environment Variables

CLI Arguments

Usage Examples

Example 1: Basic Ingestion with Auto-Generated Mock Data

Creates a default hypertable metrics with time column ts and ingests auto-generated sample data:

python3 tsdb_loader.py --table metrics --time-col ts

Expected Output:

2023-10-24 14:00:01,102 [INFO] Verified TimescaleDB extension is active.
2023-10-24 14:00:01,115 [INFO] Ensured base table 'metrics' exists.
2023-10-24 14:00:01,142 [INFO] Hypertable 'metrics' initialized with chunk interval '7 days'.
2023-10-24 14:00:01,143 [INFO] Generating 25,000 mock metrics records at '/tmp/sample_metrics.csv'...
2023-10-24 14:00:01,310 [INFO] Mock dataset generated successfully.
2023-10-24 14:00:01,311 [INFO] Starting bulk load from '/tmp/sample_metrics.csv' into 'metrics'...
2023-10-24 14:00:01,580 [INFO] Successfully loaded 25,000 records into 'metrics'.
2023-10-24 14:00:01,582 [INFO] Database connection closed.

Example 2: Loading Custom Dataset with Compression & Retention Policies

In an environment with custom environment credentials, load custom data into a hypertable with 1-day chunks, compression enabled, and a 90-day retention window:

export TSDB_HOST="timescale.internal.net"
export TSDB_USER="telemetry