RisingWave Event Streaming Platform

RisingWave is an event streaming platform designed for agentic AI. It replaces the traditional multi-component streaming stack (e.g., Debezium + Kafka + Flink + serving DB) by unifying four key stages into a single system:

1. Ingest: Unifies data from Webhooks (HTTP), Database CDC (PostgreSQL, MySQL, etc.), Event Streams (Kafka, Pulsar, Kinesis), and Historical Data (S3, data warehouses) under a single SQL interface.
2. Process: Performs incremental computation. When upstream data changes, only affected results are recomputed, ensuring end-to-end freshness of under 100 ms.
3. Serve: Maintains results in an internal row store, serving queries via standard SQL with 10-20 ms p99 latency. It is compatible with the PostgreSQL wire protocol (psql, JDBC, etc.).
4. Store: Writes to Apache Iceberg™ tables for long-term retention and analytical access. RisingWave manages the Iceberg REST catalog and handles table maintenance like compaction and snapshot cleanup.

RisingWave is composed of several core Rust crates. Use these crate-level documentations to understand the implementation details of specific components:

Core Components

• Frontend: Handles SQL parsing (risingwave_sqlparser) and frontend logic (risingwave_frontend).
• Meta: Manages metadata (risingwave_meta).
• Compute: Handles expression evaluation (risingwave_expr), batch processing (risingwave_batch), and stream processing (risingwave_stream).
• Storage: Manages data persistence via risingwave_storage, the Hummock SDK (risingwave_hummock_sdk), and object storage (risingwave_object_store).

DML and Connectors

• DML: Data Manipulation Language logic (risingwave_dml).
• Connectors: Logic for data sources and sinks (risingwave_connector).

Common Utilities

• Common: Shared functionalities (risingwave_common, risingwave_common_service).
• RPC: Client implementations for remote procedure calls (risingwave_rpc_client).
• Protobuf: Generated protobuf definitions (risingwave_pb).
• Runtime: Core runtime utilities (risingwave_rt).

Independent Utils

Located in src/utils, these crates simplify development and may be published to crates.io in the future:

• memcomparable
• pgwire

A Stream Key is a set of columns used to uniquely identify records within a RisingWave stream. It allows downstream operators to maintain per-record state by tracking changes (inserts, deletes, and updates) associated with specific key values.

Key Characteristics:

• Identification: It identifies which record is being operated on. For example, in a stream with keys k1, k2, an update is represented by a delete (-) of the old value followed by an insert (+) of the new value for that specific key.
• Consistency Requirements: For a given stream key, operations must be consistent. A stream cannot contain two inserts for the same key without an intervening delete, nor two deletes without an intervening insert.
• Update Integrity: When performing an update, the delete side of the operation must exactly match the previously yielded value for that stream key. Deleting a different value for the same key is invalid because downstream operators rely on the stream key to manage state.
• Composition: A stream key might include columns that are not strictly required for record identification, such as a group key used to specify record distribution.

| op | k1 | k2 | v1 | v2 |
|----|----|----|----|----|
| -  | 1  | 2  | 1  | 1  |
| +  | 1  | 2  | 3  | 4  |

The Storage Primary Key (often referred to as pk_columns in streaming operators) is an internal key used by the storage layer. It is distinct from the SQL PRIMARY KEY defined in a table schema.

Purpose and Behavior:

• Uniqueness and Ordering: Beyond uniquely identifying a record in storage, the Storage Primary Key is used to enforce physical ordering of data.
• Ordering via Prefixes: If a materialized view includes an ORDER BY clause, the Storage Primary Key will include those ordering columns as prefixes. This ensures that when iterating over keys in storage, records are returned in the correct order per partition.
• Relationship to Stream Key: While the Storage Primary Key handles physical ordering and storage, the Stream Key is used to identify which record needs to be updated when an update stream arrives. The system uses the Stream Key to look up the record, retrieves the necessary columns (like those used in the Storage Primary Key), and then updates the materialized state.

Example Comparison:

If you define a table with id as the primary key, but create a materialized view with ORDER BY i, id, the internal plan will reflect:

• stream_key: [id] (used to track the record)
• pk_columns: [i, id] (used to order the data in storage)

create table t1(id bigint primary key, i bigint);
create materialized view mv1 as select id, i from t1 order by i, id;

RisingWave uses a specialized KV state store called Hummock to store data for all streaming executors. Hummock is a cloud-native, LSM-Tree-based storage engine co-designed with the RisingWave streaming engine and optimized specifically for streaming workloads.

Key characteristics include:

• S3-Compatible Backend: Stores all data (SST files) on S3-compatible services.
• Node-Local Isolation: Every streaming executor reads and writes only its own portion of data. Data is generally not shared across nodes, meaning a get or scan on one node only guarantees immediate visibility of writes from that same node.
• Epoch-Based Persistence: Data is committed in serial using a barrier-based checkpoint algorithm, persisting state epoch by epoch.
• Distributed Architecture: Consists of a Hummock manager (on the meta node), Hummock clients (on worker nodes like compute, frontend, and compactor nodes), and shared storage for SST files.

Backfill is the mechanism RisingWave uses to merge historical data with real-time data streams. This is critical when creating new Materialized Views (MVs) on top of existing ones or tables.

The Problem it Solves

When a new Materialized View is created, the system must process all existing historical data before it can accurately apply real-time updates. Without backfill, the system would have to either:

1. Buffer all real-time updates: This risks Out-of-Memory (OOM) errors if historical processing takes a long time.
2. Block the upstream stream: This would pause the entire stream graph, causing latency and blocking other operations.

How Backfill Works

Backfill allows the system to process historical data in chunks (epochs) while simultaneously processing real-time updates.

1. Historical Processing: The system reads a subset of historical data (e.g., rows 1 and 2) from the source table.
2. Real-time Merging: Instead of buffering all updates, the system only applies real-time updates that correspond to the historical data already processed. For example, if row 1 was part of the historical batch, any real-time DELETE or UPDATE for row 1 is applied immediately.
3. Progress Tracking: The system tracks the last processed row (e.g., pk_offset) to ensure it can resume from the correct position in the next epoch without re-processing data.

-- Example: Creating a Materialized View that triggers backfill
CREATE TABLE t (
  id INT PRIMARY KEY,
  name VARCHAR
);

-- This command initiates the backfill process to fetch historical data from 't'
CREATE MATERIALIZED VIEW mv AS SELECT * FROM t;

Streaming freshness indicates when data produced 'now' will be visible in downstream materialized views and sinks. The canonical way to measure this is by tracking the lifecycle of a barrier through the system.

RisingWave breaks barrier latency into three distinct stages. To reason about end-to-end freshness, you should monitor these stages separately as Prometheus histograms cannot be summed into a single distribution:

1. Pre-dispatch (Queueing): meta_barrier_send_duration_seconds measures the time a barrier spends in the meta scheduler queue before being dispatched. High values indicate the meta checkpoint manager is overloaded or blocked.
2. Processing (Collection): meta_barrier_duration_seconds measures the time from dispatch until all compute nodes have ACKed the barrier. This is the primary metric for barrier latency.
3. Post-collection (Commit): meta_barrier_wait_commit_duration_seconds measures the wait for Hummock to commit after the barrier is collected. High values indicate Hummock write bottlenecks.

Interpretation Thresholds (relative to configured barrier interval):

• < 1x interval: Healthy.
• 1–10x interval: Streaming graph is building up back-pressure.
• ≥ 10x interval (and rising): The graph is likely stuck; expect 'Too Many Barriers' alerts.

# Barrier collection latency p50 / p99, per database
histogram_quantile(0.5,  sum(rate(meta_barrier_duration_seconds_bucket[$__rate_interval])) by (le, database_id))
histogram_quantile(0.99, sum(rate(meta_barrier_duration_seconds_bucket[$__rate_interval])) by (le, database_id))