Events Table Architecture#

Langfuse is migrating its ClickHouse storage layer from a separate observations + traces model to a unified events-based model. In this new architecture, every span, observation, and trace root is a row in a single events table, with trace-level fields denormalized onto each row. The migration uses a dual-write pattern so the old and new pipelines run in parallel, allowing a gradual cutover.

ClickHouse Tables#

Three tables form the core of the new architecture (DDL in dev-tables.sh):

observations_batch_staging uses 3-minute partitions keyed on s3_first_seen_timestamp and has a 12-hour TTL (ttl_only_drop_parts=1) — complete partitions are dropped automatically, not individual rows 1.

events_full is the write target. Its primary key is (project_id, toStartOfMinute(start_time), xxHash32(trace_id)), enabling efficient range scans and trace-ID lookups 2. All trace-level fields (trace_name, user_id, session_id, tags, etc.) are denormalized directly onto each span row.

events_core is populated by a materialized view (events_core_mv) that truncates input/output/metadata_values to 200 characters 3. Most read queries use events_core for speed; detail views fetch from events_full.

Dual-Write During Ingestion#

When processing incoming events, IngestionService writes to both the legacy observations table and the new observations_batch_staging table. The dual-write is gated behind the writeToStagingTables / LANGFUSE_EXPERIMENT_INSERT_INTO_EVENTS_TABLE feature flags 4.

Two record types are staged:

• Observation records — written verbatim to staging with a partition-aware timestamp 4.
• Trace records — converted to observation-shaped records via convertTraceToStagingObservation so the trace root becomes a pseudo-span with span_id = trace_id 5.

Partition-Aware Timestamp#

The getPartitionAwareTimestamp() method controls which partition a staged record lands in. If the original createdAtTimestamp is within the last 2 minutes, it is used as-is (so the record lands in its natural 3-minute partition). Older events are stamped with Date.now(), forwarding them to the current partition. This implements a "partition lock" that prevents churning old partitions after they have been processed 6.

EventPropagation Job: Staging → events_full#

The handleEventPropagationJob worker runs every minute via BullMQ and propagates completed partitions from observations_batch_staging into events_full.

Processing Flow#

Redis cursor → next partition → JOIN with traces → INSERT INTO events_full → update cursor

1. Cursor read — The job reads langfuse:event-propagation:last-processed-partition from Redis to find the last completed partition 7.

2. Partition selection — It queries system.parts for the oldest active partition in observations_batch_staging that is (a) older than a configurable delay (default 10 minutes via LANGFUSE_EXPERIMENT_EVENT_PROPAGATION_PARTITION_DELAY_MINUTES) and (b) after the cursor 8.

3. JOIN + INSERT — The partition's rows are joined against traces (within a ±1-day time window, capped at 7 days max) and inserted into events_full. Trace-level fields (name, user_id, session_id, tags, release, metadata) are merged onto each observation row at this step 9.

4. Parent span resolution — parent_span_id is resolved using COALESCE: if parent_observation_id is null the trace's root pseudo-span (concat('t-', trace_id)) is used as fallback, keeping the tree connected 10.

5. Cursor update — After a successful insert, the processed partition timestamp is written back to Redis 11.

Concurrency and Ordering#

The EventPropagationQueue enforces global concurrency = 1, ensuring partitions are always processed sequentially in timestamp order 12. The queue runs on a * * * * * cron (every minute) 13. The worker is only started when both QUEUE_CONSUMER_EVENT_PROPAGATION_QUEUE_IS_ENABLED=true and LANGFUSE_EXPERIMENT_INSERT_INTO_EVENTS_TABLE=true.

Query Layer#

The EventsQueryBuilder and EventsAggregationQueryBuilder provide fluent query building for the new tables. Key behaviors:

• Table selection: defaults to events_core; switches to events_full automatically when full I/O or metadata expansion is requested 14.
• ORDER BY optimization: automatically prepends project_id and toStartOfMinute(start_time) to match the table's primary key 15.
• xxHash32 optimization: adds xxHash32(trace_id) = xxHash32(...) for trace-ID equality filters to enable primary key pruning 16.
• Split queries: buildEventsFullTableSplitQuery() filters on events_core, then fetches full I/O only for matched rows from events_full — avoiding large I/O reads on the full table.

For trace-level aggregations (traces list UI), EventsAggregationQueryBuilder groups by (trace_id, project_id) and uses argMaxIf on event_ts to pick the latest value for mutable fields 17.

Key Source Files#