clickhouse-chorm 0.2.3

A modern, type-safe ORM for ClickHouse with full support for ClickHouse-specific features

CHORM - ClickHouse ORM

A powerful SQLAlchemy-like ORM for ClickHouse, optimized for analytics and high-performance queries.

Features

Core ORM Capabilities

• SQLAlchemy-like Syntax: Familiar API for Python developers
• Native ClickHouse Types: Full support for ClickHouse-specific types (UInt64, Array, Nullable, etc.)
• Type Safety: Explicit type declarations for better IDE support
• Async Support: Full async/await support with AsyncSession
• Query Builder: Fluent API for Select, Insert, Update, Delete
• Compression Codecs: Fine-grained storage optimization via Column(..., codec=ZSTD(1))

Analytics & Advanced Queries

• Window Functions: ROW_NUMBER, RANK, DENSE_RANK, LAG, LEAD, and aggregate window functions
• CTEs (Common Table Expressions): Build complex queries with readable structure
• JOINs: INNER, LEFT, RIGHT, FULL, CROSS with multiple join support
• Subqueries: Scalar and table subqueries with IN/EXISTS support
• Conditional Aggregations: sumIf, countIf, avgIf for multi-metric queries

ClickHouse-Specific Features

• ARRAY JOIN: Efficiently unnest and analyze array columns
• LIMIT BY: Top-N rows per group without window functions
• WITH TOTALS: Add summary rows to GROUP BY results
• PREWHERE: Optimize queries with early filtering
• FINAL, SAMPLE, SETTINGS: Full control over query execution
• OPTIMIZE TABLE: Manual table optimization and deduplication
• INSERT FROM SELECT: Efficient bulk data copying
• Dictionaries: External data source integration
• EXPLAIN/ANALYZE: Query profiling and optimization

Multi-Database Support

• Database Attribute: Specify target database per table with __database__
• Qualified Names: Automatic database.table formatting in all SQL statements
• Database DDL: CREATE DATABASE / DROP DATABASE with full options
• Smart Drop Protection: Automatic size limit bypass with UNDROP warnings
• Safe Migrations: DROP DATABASE never auto-generated, explicit warnings for drops

Installation

pip install clickhouse-chorm

Quick Start

Define Models

from chorm import Table, Column, MergeTree
from chorm.types import UInt64, String

class User(Table):
    __tablename__ = "users"
    __engine__ = MergeTree()
    
    id = Column(UInt64(), primary_key=True)
    name = Column(String())
    email = Column(String())

Synchronous Usage

from chorm import create_engine, Session, select

# Create engine with explicit configuration (recommended)
engine = create_engine(
    host="localhost",
    port=8123,
    username="default",
    password="password",  # No URL encoding needed!
    database="default",
    connect_timeout=5,
    send_receive_timeout=120,
    compress='lz4'
)

# Or using connection URL (backward compatibility)
# engine = create_engine("clickhouse://default:password@localhost:8123/default")

# Create session
session = Session(engine)

# Insert data
user = User(id=1, name="Alice", email="[e-mail]")
session.add(user)
session.commit()

# Query data
stmt = select(User.id, User.name).where(User.id > 0)
result = session.execute(stmt)
users = result.all()

Session Configuration Overrides

You can override connection parameters for a specific session without changing the global engine configuration. This is useful for long-running operations.

# Create a session with a longer timeout
long_session = Session(engine, send_receive_timeout=3600)

long_session.execute("OPTIMIZE TABLE big_table FINAL")
# Uses 3600s timeout, while 'engine' defaults remain unchanged

Asynchronous Usage

from chorm import create_async_engine, AsyncSession

# Create async engine
engine = create_async_engine(
    host="localhost", 
    port=8123, 
    username="default", 
    password="password"
)

# Use async session
async with AsyncSession(engine) as session:
    user = User(id=1, name="Alice", email="[e-mail]")
    session.add(user)
    # Auto-commits on exit

# Query asynchronously
async with AsyncSession(engine) as session:
    stmt = select(User).where(User.id > 0)
    result = await session.execute(stmt)
    users = result.all()

Schema Management with Metadata

CHORM provides SQLAlchemy-like metadata for schema management:

from chorm import Table, Column, MetaData, Memory
from chorm.types import UInt64, String

# Create metadata object
metadata = MetaData()

# Define tables using metadata
class User(Table):
    metadata = metadata
    __tablename__ = "users"
    __engine__ = Memory()
    
    id = Column(UInt64(), primary_key=True)
    name = Column(String())

# Create all tables associated with this metadata
metadata.create_all(engine)

# Drop all tables
metadata.drop_all(engine)

Multi-Database Support

Work with tables across multiple databases:

from chorm import Table, Column, create_engine, Session
from chorm.types import UInt64, String
from chorm.table_engines import MergeTree
from chorm.sql import select, insert
from chorm.sql.ddl import create_database, drop_database

# Create engine
engine = create_engine("clickhouse://localhost:8123/default")
session = Session(engine)

# Create a new database
session.execute(create_database("analytics", if_not_exists=True).to_sql())

# Define table in specific database
class Event(Table):
    __tablename__ = "events"
    __database__ = "analytics"  # <-- Target database
    __engine__ = MergeTree()
    __order_by__ = ["id"]
    
    id = Column(UInt64(), primary_key=True)
    name = Column(String())

# Create table (generates: CREATE TABLE analytics.events ...)
session.execute(Event.create_table())

# Insert (generates: INSERT INTO analytics.events ...)
session.execute(insert(Event).values(id=1, name="click").to_sql())

# Select (generates: SELECT analytics.events.id FROM analytics.events ...)
result = session.execute(select(Event.id, Event.name).to_sql())
print(result.all())

# Cleanup
session.execute(drop_database("analytics", if_exists=True).to_sql())

CLI & Migrations

CHORM includes a powerful CLI (similar to Alembic) for managing database migrations.

Initialization

Initialize a new project with the required directory structure:

chorm init

This creates:

• migrations/versions/ - Directory for migration scripts
• migrations/env.py - Configuration script for auto-migrations
• chorm.toml - Configuration file

Configuration (chorm.toml)

Configure your database connection and migration settings:

[chorm]
host = "localhost"
port = 8123
database = "default"
user = "default"
password = ""
secure = false

[migrations]
directory = "migrations"
table_name = "chorm_migrations"
version_style = "uuid" # Options: uuid, int, django

Migration Naming Styles:

• uuid (default): e4d90..._message.py
• int: 1_message.py, 2_message.py (sequential)
• django: 0001_message.py, 0002_message.py (padded sequential)

Creating Migrations

Create a new empty migration file:

chorm make-migration -m "create users table"

Auto-Migrations

Automatically generate migrations by comparing your models with the database.

1. Configure migrations/env.py: To enable auto-discovery, import your MetaData object in env.py:

   from chorm import MetaData
   # Import your application's metadata
   from myapp.models import metadata as target_metadata
   

2. Run Auto-Migrate:

   chorm auto-migrate -m "initial schema"
   

   You can also point to a directory of models if not using env.py:

   chorm auto-migrate --models ./myapp/models -m "update schema"
   

Applying Migrations

Apply pending migrations:

chorm migrate

Managing Migrations

Show migration status:

chorm show-migrations

Rollback the last migration:

chorm downgrade

Rollback multiple steps:

chorm downgrade --steps 3

Connection Pooling

CHORM supports connection pooling for improved performance in high-concurrency applications:

from chorm import create_engine

# Enable connection pooling
engine = create_engine(
    "clickhouse://localhost:8123/default",
    pool_size=10,        # Maximum pooled connections
    max_overflow=20,     # Additional overflow connections
    pool_timeout=30.0,   # Connection acquisition timeout
    pool_recycle=3600    # Recycle connections after 1 hour
)

# Connections are automatically managed
with engine.connection() as conn:
    result = conn.query("SELECT * FROM users")
# Connection automatically returned to pool

Async Connection Pooling:

from chorm import create_async_engine

# Enable async connection pooling
engine = create_async_engine(
    "clickhouse://localhost:8123/default",
    pool_size=10,
    max_overflow=20
)

# Use async context manager
async with engine.connection() as conn:
    result = await conn.query("SELECT * FROM users")
# Connection automatically returned to pool

Benefits:

• 5-10x query throughput improvement for high-frequency queries
• Reduced connection overhead
• Thread-safe (sync) and asyncio-safe (async)
• Automatic connection recycling

See Connection Pooling Guide for detailed examples.

Retry Logic with Exponential Backoff

Automatically retry failed operations with configurable backoff:

from chorm import with_retry, RetryConfig

# Configure retry behavior
retry_config = RetryConfig(
    max_attempts=3,       # Maximum retry attempts
    initial_delay=0.1,    # Initial delay in seconds
    max_delay=10.0,       # Maximum delay cap
    exponential_base=2.0, # Backoff multiplier
    jitter=True           # Add random jitter
)

# Apply to any function
@with_retry(retry_config)
def fetch_critical_data():
    with engine.connection() as conn:
        return conn.query("SELECT * FROM critical_table")

# Automatically retries on transient errors
result = fetch_critical_data()

Async Retry:

from chorm import async_with_retry

@async_with_retry(RetryConfig(max_attempts=5))
async def fetch_data():
    async with engine.connection() as conn:
        return await conn.query("SELECT * FROM users")

result = await fetch_data()

Features:

• Exponential backoff with jitter to avoid thundering herd
• Configurable retryable error types
• Works with connection pooling
• Automatic retry on network errors, timeouts, and memory errors

Health Checks

Monitor ClickHouse connection health:

from chorm import create_engine, HealthCheck

engine = create_engine("clickhouse://localhost:8123/default")
health = HealthCheck(engine)

# Quick ping check
if health.ping():
    print("ClickHouse is reachable")

# Detailed status
status = health.get_status()
print(f"Status: {status['status']}")           # "healthy" or "unhealthy"
print(f"Latency: {status['latency_ms']}ms")    # Response time
print(f"Version: {status['version']}")          # ClickHouse version
print(f"Uptime: {status['uptime_seconds']}s")  # Server uptime

# Server information
info = health.get_server_info()
print(f"Database: {info['current_database']}")
print(f"Memory: {info['used_memory']} / {info['total_memory']}")

Async Health Checks:

from chorm import create_async_engine, AsyncHealthCheck

engine = create_async_engine("clickhouse://localhost:8123/default")
health = AsyncHealthCheck(engine)

# Async health check
if await health.ping():
    status = await health.get_status()
    print(f"Latency: {status['latency_ms']}ms")

Use Cases:

• Kubernetes liveness/readiness probes
• Load balancer health checks
• Monitoring and alerting
• Circuit breaker implementations

Query Execution Metrics

Track and analyze query performance:

from chorm import MetricsCollector, enable_global_metrics

# Create metrics collector
collector = MetricsCollector(
    slow_query_threshold_ms=500.0,  # Queries >500ms are "slow"
    log_all_queries=False            # Only log slow queries
)

# Measure query execution
with collector.measure("SELECT * FROM users WHERE active = 1") as metrics:
    result = session.execute(query)

# Get summary statistics
summary = collector.get_summary()
print(f"Total queries: {summary['total_queries']}")
print(f"Average duration: {summary['avg_duration_ms']:.2f}ms")
print(f"Slow queries: {summary['slow_queries']}")

# Get percentiles
percentiles = collector.get_percentiles([50, 90, 95, 99])
print(f"P95: {percentiles['p95']:.2f}ms")

# Get slow queries
slow_queries = collector.get_slow_queries()
for metric in slow_queries:
    print(f"Slow: {metric.sql[:50]} - {metric.duration_ms:.2f}ms")

Global Metrics:

# Enable global metrics collection
collector = enable_global_metrics(slow_query_threshold_ms=500.0)

# All queries are now tracked automatically
with collector.measure("SELECT 1"):
    result = session.execute(query)

# Get summary
summary = collector.get_summary()

Connection Pool Statistics

Monitor pool utilization and performance:

# Get pool statistics
stats = engine.pool.get_statistics()

print(f"Pool size: {stats['pool_size']}")
print(f"Connections in use: {stats['connections_in_use']}")
print(f"Overflow: {stats['overflow']}")
print(f"Utilization: {stats['utilization_percent']}%")

Complete Monitoring Example:

from chorm import (
    create_engine, MetricsCollector, HealthCheck
)

# Setup with pooling
engine = create_engine(
    "clickhouse://localhost:8123/default",
    pool_size=10,
    max_overflow=20
)

# Setup monitoring
collector = MetricsCollector(slow_query_threshold_ms=500.0)
health = HealthCheck(engine)

# Check health
if health.ping():
    status = health.get_status()
    print(f"✓ Healthy - Latency: {status['latency_ms']}ms")
else:
    print("✗ Unhealthy - Cannot reach ClickHouse")

# Execute queries with metrics
with collector.measure("SELECT * FROM users"):
    with engine.connection() as conn:
        result = conn.query("SELECT * FROM users LIMIT 100")

# Monitor pool
pool_stats = engine.pool.get_statistics()
print(f"Pool utilization: {pool_stats['utilization_percent']}%")

# Check for slow queries
if collector.get_slow_queries():
    print("⚠ Slow queries detected!")

See examples/monitoring_demo.py for complete examples.

ClickHouse-Optimized Batch Insert

Efficiently insert large volumes of data with ClickHouse-optimized batching:

from chorm import create_engine, bulk_insert

engine = create_engine("clickhouse://localhost:8123/default")
client = engine._client

# Prepare data (1M rows)
data = [
    [i, f"User{i}", f"user{i}@example.com"]
    for i in range(1_000_000)
]

# Insert with optimal batch size (100k rows)
stats = bulk_insert(
    client, "users", data,
    columns=["id", "name", "email"],
    batch_size=100_000  # ClickHouse-optimized
)

print(f"Inserted {stats['total_rows']:,} rows in {stats['batches_sent']} batches")

Advanced Batch Insert:

from chorm import ClickHouseBatchInsert

# Create batch inserter
batch = ClickHouseBatchInsert(
    client,
    "users",
    columns=["id", "name", "email"],
    batch_size=100_000
)

# Add rows (auto-flushes at 100k rows)
for i in range(1_000_000):
    batch.add_row([i, f"User{i}", f"user{i}@example.com"])

# Flush remaining rows
stats = batch.finish()

DataFrame Support:

import pandas as pd
from chorm import bulk_insert

df = pd.DataFrame({
    'id': range(1_000_000),
    'name': [f'User{i}' for i in range(1_000_000)]
})

# Automatically detected as DataFrame
stats = bulk_insert(client, "users", df, batch_size=100_000)

Performance Features:

• Uses native client.insert() (5-10x faster than SQL VALUES)
• Default batch size: 100,000 rows (ClickHouse-optimized)
• Automatic batching and flushing
• DataFrame support built-in
• Optional OPTIMIZE TABLE after insertion

See examples/batch_insert_demo.py for complete examples.

Pandas Integration

CHORM integrates seamlessly with Pandas for both inserting and querying data.

Query to DataFrame:

from chorm import select

# Synchronous Session
df = session.query_df(select(User).where(User.id > 0))

# Asynchronous Session
df = await session.query_df(select(User).where(User.id > 0))

# Using raw SQL
df = session.query_df("SELECT * FROM users WHERE id > 0")

Insert from DataFrame:

See DataFrame Support in Batch Insert section.

Advanced Features

Performance Operations

from chorm import optimize_table, insert, select

# Optimize table (manual merge)
stmt = optimize_table(User, final=True)
session.execute(stmt.to_sql())

# Deduplicate data
stmt = optimize_table(User, deduplicate=True, final=True)
session.execute(stmt.to_sql())

# INSERT FROM SELECT (bulk data copying)
source_query = select(SourceTable.id, SourceTable.name).where(SourceTable.active == 1)
stmt = insert(TargetTable).from_select(source_query, columns=["id", "name"])
session.execute(stmt.to_sql())

Advanced Aggregates

from chorm.sql.expression import top_k, group_bitmap, any_last

# Top-K most frequent values
query = select(
    User.country,
    func.count().label("count")
).select_from(User).group_by(User.country).order_by(func.count().desc()).limit(10)

# Bitmap aggregation for unique counts
query = select(
    Event.date,
    group_bitmap(Event.user_id).label("unique_users")
).select_from(Event).group_by(Event.date)

# Sampling aggregates
query = select(
    User.country,
    any_last(User.last_login).label("latest_login")
).select_from(User).group_by(User.country)

Dictionary Support

from chorm import create_dictionary
from chorm.sql.expression import dict_get, dict_has

# Create dictionary
stmt = create_dictionary(
    "user_dict",
    "ClickHouse(HOST 'localhost' PORT 9000 USER 'default' TABLE 'users' DB 'default')",
    "HASHED",
    [("id", "UInt64"), ("name", "String"), ("country", "String")],
    lifetime=300  # Cache lifetime in seconds
)
session.execute(stmt.to_sql())

# Use dictionary in queries
query = select(
    Order.id,
    dict_get("user_dict", "name", Order.user_id).label("user_name"),
    dict_get("user_dict", "country", Order.user_id).label("user_country")
).select_from(Order)

Materialized Views

CHORM provides first-class support for Materialized Views.

Declarative Definition:

from chorm import Table, Column, MaterializedView, MergeTree, select
from chorm.types import UInt64, String

# 1. Define Source Table
class User(Table):
    __tablename__ = "users"
    __engine__ = MergeTree()
    id = Column(UInt64(), primary_key=True)
    name = Column(String())

# 2. Define Materialized View
class UserMV(Table):
    __tablename__ = "users_mv"
    # Reference the source table class directly
    __engine__ = MaterializedView(to_table=User)
    
    # Auto-generate 'SELECT * FROM users'
    __from_table__ = User
    
    # OR define custom logic
    # __select__ = select(User.id, User.name).where(User.id > 100)
    
    # Define columns if needed (introspected automatically if simple view)

Introspection:

Run chorm introspect to generate clean model code for existing Materialized Views, including class references and dependencies.

Query Observability

# Analyze query execution
query = select(User).where(User.country == 'US')

# Get query plan
explain_stmt = query.explain(explain_type="PLAN")
result = session.execute(explain_stmt.to_sql())

# Profile query (pipeline analysis)
explain_stmt = query.analyze()  # Shortcut for EXPLAIN PIPELINE
result = session.execute(explain_stmt.to_sql())

Documentation

• Migration Guide - Migrate from raw SQL to CHORM
• Performance Guide - Query optimization, bulk operations, indexing strategies
• Best Practices - Schema design, query patterns, error handling
• Analytics Guide - Window functions, CTEs, advanced analytics

Development

Setup

# Clone repository
git clone https://github.com/yourusername/chorm.git
cd chorm

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -e ".[dev]"

Running Tests

Start ClickHouse with Docker Compose:

docker-compose up -d

Wait for ClickHouse to be ready (check with docker-compose ps), then run tests:

pytest

Stop ClickHouse:

docker-compose down

Running Integration Tests

Integration tests require a running ClickHouse instance:

# Start ClickHouse
docker-compose up -d

# Wait for healthcheck
docker-compose ps

# Run all tests including integration
pytest

# Run only integration tests
pytest tests/integration/

# Stop ClickHouse
docker-compose down -v  # -v removes volumes

Query Examples

SELECT with Filters

from chorm import select, func

# Simple select
stmt = select(User.id, User.name).where(User.id > 10)

# With multiple conditions
stmt = select(User).where(
    (User.id > 10) & (User.name.like("A%"))
)

# With ordering and limit
stmt = select(User).order_by(User.name).limit(10)

# With aggregation
stmt = select(func.count(User.id)).where(User.id > 0)

GROUP BY and Aggregation

from chorm import select, func

# Simple GROUP BY
stmt = select(User.city, func.count(User.id)).group_by(User.city)

# GROUP BY with HAVING
stmt = (
    select(User.city, func.count(User.id))
    .group_by(User.city)
    .having(func.count(User.id) > 10)
)

# Multiple aggregations
stmt = (
    select(
        User.city,
        func.count(User.id).label("count"),
        func.avg(User.age).label("avg_age")
    )
    .group_by(User.city)
)

INSERT

from chorm import insert

# Single insert
stmt = insert(User).values(id=1, name="Alice", email="[e-mail]")
session.execute(stmt)

# Bulk insert via session
users = [
    User(id=1, name="Alice", email="[e-mail]"),
    User(id=2, name="Bob", email="[e-mail]"),
]
for user in users:
    session.add(user)
session.commit()

UPDATE (ClickHouse ALTER TABLE)

from chorm import update

stmt = update(User).where(User.id == 1).values(name="Alice Updated")
session.execute(stmt)

DELETE (ClickHouse ALTER TABLE)

from chorm import delete

stmt = delete(User).where(User.id == 1)
session.execute(stmt)

DDL Operations

CHORM provides comprehensive DDL (Data Definition Language) operations for managing database schema:

DROP TABLE

from chorm import drop_table

# Drop table with IF EXISTS
stmt = drop_table(User)
session.execute(stmt.to_sql())

# Drop without IF EXISTS
stmt = drop_table(User, if_exists=False)
session.execute(stmt.to_sql())

TRUNCATE TABLE

from chorm import truncate_table

# Remove all data from table
stmt = truncate_table(User)
session.execute(stmt.to_sql())

ALTER TABLE - Column Operations

from chorm import add_column, drop_column, modify_column, rename_column

# Add a column
stmt = add_column(User, "age UInt8", after="name")
session.execute(stmt.to_sql())

# Add column with default value
stmt = add_column(User, "status String DEFAULT 'active'")
session.execute(stmt.to_sql())

# Drop a column
stmt = drop_column(User, "old_field")
session.execute(stmt.to_sql())

# Modify column type
stmt = modify_column(User, "age UInt16")
session.execute(stmt.to_sql())

# Rename column
stmt = rename_column(User, "old_name", "new_name")
session.execute(stmt.to_sql())

ALTER TABLE - Index Operations

from chorm import add_index, drop_index
from chorm.sql.expression import Identifier

# Add minmax index
stmt = add_index(User, "idx_email", Identifier("email"))
session.execute(stmt.to_sql())

# Add bloom filter index
stmt = add_index(
    User, 
    "idx_name", 
    Identifier("name"),
    index_type="bloom_filter",
    granularity=4
)
session.execute(stmt.to_sql())

# Drop index
stmt = drop_index(User, "idx_email")
session.execute(stmt.to_sql())

Using DDL in Migrations

DDL operations integrate seamlessly with the migration system:

from chorm.migration import Migration
from chorm.session import Session
from chorm.sql.expression import Identifier

class AddUserAgeColumn(Migration):
    id = "20231203_001"
    name = "Add age column to users"
    
    def upgrade(self, session: Session) -> None:
        # Add column using helper method
        self.add_column(session, 'users', 'age UInt8', after='name')
        
        # Add index
        self.add_index(
            session, 
            'users', 
            'idx_age', 
            Identifier('age'),
            index_type='minmax'
        )
    
    def downgrade(self, session: Session) -> None:
        # Drop index first
        self.drop_index(session, 'users', 'idx_age')
        
        # Then drop column
        self.drop_column(session, 'users', 'age')

ClickHouse-Specific Features

PREWHERE Clause

stmt = select(User).prewhere(User.id > 1000).where(User.name == "Alice")

FINAL Modifier

stmt = select(User).final()

SAMPLE Clause

stmt = select(User).sample(0.1)  # 10% sample

SETTINGS

stmt = select(User).settings(max_threads=4, max_memory_usage=10000000000)

Analytics Examples

Window Functions

from chorm import window, func

# Ranking products by sales within each category
w = window(
    partition_by=[Product.category],
    order_by=[func.sum(Order.amount).desc()]
)

stmt = (
    select(
        Product.category,
        Product.name,
        func.sum(Order.amount).label('total_sales'),
        func.row_number().over(w).label('rank')
    )
    .select_from(Order)
    .join(Product, on=Order.product_id == Product.id)
    .group_by(Product.category, Product.name)
)

# Running totals
w_running = window(
    partition_by=[Order.user_id],
    order_by=[Order.created_at],
    rows_between=('UNBOUNDED PRECEDING', 'CURRENT ROW')
)

stmt = select(
    Order.id,
    Order.amount,
    func.sum(Order.amount).over(w_running).label('running_total')
).select_from(Order)

CTEs (Common Table Expressions)

# Multi-step analytics with CTEs
active_users = (
    select(User.id, User.name)
    .where(User.last_login > func.now() - Literal("INTERVAL 30 DAY"))
    .cte('active_users')
)

high_value_orders = (
    select(Order.user_id, func.sum(Order.amount).label('total'))
    .group_by(Order.user_id)
    .having(func.sum(Order.amount) > 1000)
    .cte('high_value')
)

stmt = (
    select(
        Identifier('active_users.name'),
        Identifier('high_value.total')
    )
    .select_from(Identifier('active_users'))
    .join(Identifier('high_value'), 
          on=Identifier('active_users.id') == Identifier('high_value.user_id'))
    .with_cte(active_users, high_value)
)

ARRAY JOIN

# Analyze tags from array column
stmt = (
    select(
        Product.category,
        Identifier('tag'),
        func.count().label('tag_count')
    )
    .select_from(Product)
    .array_join(Product.tags, alias='tag')
    .group_by(Product.category, Identifier('tag'))
)

Conditional Aggregations

from chorm import func

# Multiple metrics in one query
stmt = select(
    User.city,
    func.sumIf(Order.amount, Order.status == 'completed').label('completed_sales'),
    func.sumIf(Order.amount, Order.status == 'pending').label('pending_sales'),
    func.countIf(Order.status == 'cancelled').label('cancelled_count'),
    func.avgIf(Order.amount, Order.status == 'completed').label('avg_order')
).select_from(Order).join(User, on=Order.user_id == User.id).group_by(User.city)

LIMIT BY

# Top 3 orders per user
stmt = (
    select(Order.user_id, Order.id, Order.amount)
    .select_from(Order)
    .order_by(Order.amount.desc())
    .limit_by(3, Order.user_id)
)

Table Engines

CHORM supports all major ClickHouse table engines. Choose the right engine based on your use case:

MergeTree Family

MergeTree - Default engine for most use cases:

from chorm import MergeTree

class User(Table):
    __tablename__ = "users"
    __engine__ = MergeTree()
    # ORDER BY is required for MergeTree

ReplacingMergeTree - For data with updates (deduplication):

from chorm import ReplacingMergeTree

class User(Table):
    __tablename__ = "users"
    __engine__ = ReplacingMergeTree(version_column="updated_at")
    # Keeps only the latest version based on version_column

SummingMergeTree - For pre-aggregated metrics:

from chorm import SummingMergeTree

class Metrics(Table):
    __tablename__ = "metrics"
    __engine__ = SummingMergeTree(columns=["value", "count"])
    # Automatically sums numeric columns during merges

AggregatingMergeTree - For pre-aggregated data with aggregate functions:

from chorm import AggregatingMergeTree

class AggregatedStats(Table):
    __tablename__ = "aggregated_stats"
    __engine__ = AggregatingMergeTree()
    # Stores pre-computed aggregates

CollapsingMergeTree - For data with sign-based updates:

from chorm import CollapsingMergeTree

class Events(Table):
    __tablename__ = "events"
    __engine__ = CollapsingMergeTree(sign_column="sign")
    # sign=1 for insert, sign=-1 for delete

VersionedCollapsingMergeTree - CollapsingMergeTree with versioning:

from chorm import VersionedCollapsingMergeTree

class Events(Table):
    __tablename__ = "events"
    __engine__ = VersionedCollapsingMergeTree(sign_column="sign", version_column="version")

GraphiteMergeTree - For Graphite metrics:

from chorm import GraphiteMergeTree

class GraphiteData(Table):
    __tablename__ = "graphite"
    __engine__ = GraphiteMergeTree(config_element="graphite_rollup")

Replicated Engines

All MergeTree engines have replicated versions for high availability:

from chorm import (
    ReplicatedMergeTree,
    ReplicatedReplacingMergeTree,
    ReplicatedSummingMergeTree,
    ReplicatedAggregatingMergeTree,
    ReplicatedCollapsingMergeTree,
    ReplicatedVersionedCollapsingMergeTree,
    ReplicatedGraphiteMergeTree
)

class ReplicatedUser(Table):
    __tablename__ = "users"
    __engine__ = ReplicatedMergeTree(
        zookeeper_path="/clickhouse/tables/users",
        replica_name="replica1"
    )

Log Engines

For small tables and temporary data:

from chorm import Log, TinyLog, StripeLog

# Log - General purpose log engine
class TempData(Table):
    __tablename__ = "temp"
    __engine__ = Log()

# TinyLog - Minimal overhead
class SmallTable(Table):
    __tablename__ = "small"
    __engine__ = TinyLog()

# StripeLog - Better for writes
class WriteHeavy(Table):
    __tablename__ = "writes"
    __engine__ = StripeLog()

Special Engines

Memory - In-memory storage:

from chorm import Memory

class Cache(Table):
    __tablename__ = "cache"
    __engine__ = Memory()

Distributed - Distributed tables across cluster:

from chorm import Distributed

class DistributedUser(Table):
    __tablename__ = "users_distributed"
    __engine__ = Distributed(
        cluster="my_cluster",
        database="default",
        table="users"
    )

Kafka - Kafka integration:

from chorm import Kafka

class KafkaEvents(Table):
    __tablename__ = "kafka_events"
    __engine__ = Kafka()

External Data Sources:

from chorm import MySQL, PostgreSQL, ODBC, JDBC

# MySQL
class MySQLTable(Table):
    __engine__ = MySQL(
        host="mysql.example.com",
        database="db",
        table="table",
        user="user",
        password="pass"
    )

# PostgreSQL
class PostgresTable(Table):
    __engine__ = PostgreSQL(
        host="postgres.example.com",
        database="db",
        table="table",
        user="user",
        password="pass"
    )

Other Engines:

• File - File-based storage
• Null - Discards writes, returns empty reads
• Set - Set data structure
• Join - Join table for JOIN operations
• View - Materialized view

See Best Practices Guide for guidance on choosing the right engine.

ClickHouse-Specific Features

Type System

CHORM supports all ClickHouse types:

from chorm.types import (
    # Integers
    UInt8, UInt16, UInt32, UInt64,
    Int8, Int16, Int32, Int64,
    
    # Floats
    Float32, Float64,
    
    # Strings
    String, FixedString,
    
    # Dates
    Date, DateTime,
    
    # Special
    UUID, Decimal, JSON,
    
    # Composite
    Array, Nullable, Map, Tuple, LowCardinality
)

class Event(Table):
    __tablename__ = "events"
    __engine__ = MergeTree()
    
    id = Column(UInt64(), primary_key=True)
    tags = Column(Array(String()))
    metadata = Column(Map(String(), String()))
    optional_field = Column(Nullable(String()))

Documentation

• Analytics Guide - Patterns for cohort analysis, funnels, time-series, and more
• Performance Guide - Optimization tips and best practices
• Examples - Working code examples for all features

License

MIT

Contributing

Contributions are welcome! Please read our contributing guidelines and submit pull requests.