Architecture

Understand Timeplus Proton’s streaming architecture and how it extends ClickHouse for real-time data processing

Architecture

Timeplus Proton is a unified streaming and historical data processing engine built on top of ClickHouse. It extends ClickHouse’s columnar storage and vectorized query execution with native stream processing capabilities, delivering a single binary solution for real-time analytics.

Overview

Proton combines:

• ClickHouse’s proven OLAP engine for historical data queries and storage
• Native stream processing for real-time data ingestion and continuous queries
• Distributed write-ahead log (WAL) implemented using Kafka for replication and fault tolerance
• Single C++ binary with no JVM, no ZooKeeper dependencies

Core Components

Stream Storage Engine

Proton introduces a new storage engine called StorageStream that extends ClickHouse’s MergeTreeData:

// From src/Storages/Stream/StorageStream.h
class StorageStream : public MergeTreeData
{
    // Distributed data ingestion via Kafka WAL
    // Streaming query support
    // Simplified usability
};

Key capabilities:

• Distributed ingestion: Data is written to a distributed WAL (Kafka) and replicated across shards
• Streaming queries: Native support for continuous queries with WHERE, GROUP BY, and window functions
• MergeTree integration: Uses ClickHouse’s MergeTree family for storage with automatic background merging

Columnar Processing

Like ClickHouse, Proton uses vectorized query execution:

1. Columnar storage

   Data is stored by columns (IColumn interface) with contiguous memory layout for cache efficiency

2. Vectorized execution

   Operations dispatch on arrays (vectors/chunks) rather than individual values for better CPU utilization

3. SIMD optimization

   Single Instruction Multiple Data operations accelerate aggregations and filters

From ClickHouse architecture:

“Operations are dispatched on arrays, rather than on individual values. This helps lower the cost of actual data processing through vectorized query execution.”

Block Streams

Data flows through the system as Blocks - containers of column chunks:

Block = [(IColumn, IDataType, column_name), ...]

Processing pipeline:

1. IBlockInputStream reads blocks from sources (streams, tables, Kafka)
2. Transformations filter, aggregate, join blocks immutably
3. IBlockOutputStream writes results to destinations (views, external streams, storage)

Proton uses a “pull” approach: when you pull a block from a stream, it recursively pulls from nested streams, creating an execution pipeline.

Query Processing Stages

Streaming Query

-- Continuous aggregation over a stream
SELECT device, count(*), avg(temperature)
FROM sensor_data
GROUP BY device;

Execution flow:

1. Parser creates AST from SQL
2. InterpreterSelectQuery builds streaming execution pipeline
3. Stream shards read from WAL or external sources
4. Streaming aggregator maintains state
5. Results emit incrementally as data arrives

Historical Query

-- Query stored data in table mode
SELECT device, count(*), avg(temperature)
FROM table(sensor_data)
GROUP BY device;

Execution flow:

1. Parser creates AST from SQL
2. InterpreterSelectQuery builds batch execution pipeline
3. Read MergeTree parts from disk
4. Apply aggregations on complete data set
5. Return final results

Stream Sharding

Proton distributes streams across multiple shards for parallelism:

// Sharding configuration
UInt32 shards;  // Number of shards
ExpressionActionsPtr sharding_key_expr;  // Sharding expression
std::vector<StreamShardStorePtr> stream_shards;  // Shard stores

Sharding strategies:

• Random: Round-robin distribution across shards
• Expression-based: Hash on specified columns for co-location
• Deterministic: Same key always routes to same shard

Sharding is transparent to queries - Proton automatically routes reads/writes to appropriate shards.

Storage Modes

Proton supports multiple storage backends:

Mode	Description	Use Case
memory	In-memory storage	High-speed streaming, temporary data
default	Disk-based MergeTree	Historical data, persistent streams
append_only	Append-only log	Event sourcing, audit logs
changelog_kv	CDC changelog	Database replication, change tracking

Query Execution Pipeline

Write Path

1. Ingest

   Client sends data via INSERT or external stream connector

2. Shard routing

   Sharding expression evaluates to determine target shard(s)

3. WAL append

   Data appends to distributed WAL (Kafka) with async callbacks

4. Background consumption

   Stream shards consume from WAL and write to MergeTree parts

5. Merge

   Background threads merge parts to optimize storage and queries

Proton does not use MEMTABLE like LSM trees - data writes directly to filesystem as MergeTree parts. Use batch inserts (not single rows) for optimal performance.

Read Path

Streaming Mode

// From StorageStream::read()
void read(
    QueryPlan & query_plan,
    const Names & column_names,
    SelectQueryInfo & query_info,
    ContextPtr context,
    size_t max_block_size,
    size_t num_streams
);

1. Determine shards to read based on query predicates
2. Create parallel stream readers for each shard
3. Apply filters and projections
4. Stream results continuously

Historical Mode

Uses ClickHouse’s standard MergeTree read path:

1. Consult primary index to identify candidate parts
2. Read marks files for column offsets
3. Decompress and read column blocks
4. Apply filters and aggregations

Performance Characteristics

Benchmarks on Apple M2 Max:

• 90 million events/sec ingestion throughput
• 4ms end-to-end latency for streaming queries
• 1 million unique keys high-cardinality aggregation

Optimizations:

• Zero-copy reads from Kafka with direct block conversion
• SIMD-accelerated aggregations and filters
• Sparse primary indexes for range scans
• Column-level compression (LZ4, ZSTD)

Fault Tolerance

Proton inherits ClickHouse’s replication model with streaming enhancements:

Replication mechanism

• Each stream shard maps to a Kafka partition (or native log)
• Data replicates via Kafka replication factor
• On failure, consumers resume from last committed offset
• No distributed consensus required (no ZooKeeper)

Consistency model

• At-least-once delivery by default
• Exactly-once with idempotent keys
• Eventual consistency for replicas

Integration with ClickHouse

Proton maintains full compatibility with ClickHouse:

-- Query ClickHouse external table from Proton
CREATE EXTERNAL TABLE ch_events
SETTINGS type='clickhouse',
         address='clickhouse.example.com:9000',
         database='analytics',
         table='events';

-- Join streaming data with historical ClickHouse data
SELECT s.user_id, s.event, c.user_name
FROM live_events s
JOIN ch_events c ON s.user_id = c.user_id;

Next Steps

Streams

Learn about stream types and operations

External Streams

Connect to Kafka, Pulsar, and other sources

Materialized Views

Build real-time data pipelines

Windows

Time-based aggregations and windowing