DEV Community: RisingWave Labs

RisingWave v2.8: Query Your Lakehouse, Backfill Faster, and Tune Jobs Individually

RisingWave Labs — Mon, 20 Apr 2026 11:22:15 +0000

RisingWave v2.8: Query Your Lakehouse, Backfill Faster, and Tune Jobs Individually

If you run streaming pipelines on RisingWave and batch queries on a separate engine, v2.8 changes the equation. This release adds a DataFusion-powered query engine that lets you run batch SQL directly on your data lake (Iceberg, Delta Lake, Hudi) without moving data.

Beyond the lakehouse, we’ve overhauled backfilling to make it significantly faster and introduced per-job resource isolation, giving you granular control over how individual streaming jobs consume CPU and memory.

Here are the highlights of RisingWave v2.8.

Query Your Lakehouse Directly (Powered by DataFusion)

RisingWave has long supported sinking data to data lakes like Apache Iceberg, Delta Lake, and Apache Hudi. However, querying that data usually required an external engine like Trino, StarRocks, or Spark.

In v2.8, we’ve integrated Apache DataFusion as a native batch query engine. You can now use RisingWave to:

Query Sinks Directly: Run SQL queries on the data you’ve already exported to your lakehouse.
Join Streams with Lake Data: Perform complex joins between real-time streaming data and historical data stored in the lake.
Unified Experience: Use the same SQL dialect and connection for both real-time and historical analysis.

This turns RisingWave into a unified engine for the entire data lifecycle—from ingestion and streaming to long-term storage and ad-hoc analysis.

Faster Backfilling with Parallelism

Backfilling—the process of populating a new materialized view with historical data—is a critical but resource-intensive task. In previous versions, backfilling was often a bottleneck for large datasets.

v2.8 introduces Parallel Backfilling. By distributing the backfill task across multiple nodes and utilizing more CPU cores, RisingWave can now process historical data significantly faster. In our internal benchmarks, we’ve seen backfill times drop by up to 50% for large-scale tables, allowing you to get your new materialized views online in record time.

Individual Job Tuning & Resource Isolation

In a multi-tenant or complex streaming environment, one "heavy" job can sometimes impact the performance of others. v2.8 addresses this with Per-Job Resource Isolation.

You can now:

Set Resource Limits: Define specific CPU and memory limits for individual streaming jobs.
Prioritize Critical Jobs: Ensure that your most important pipelines always have the resources they need, regardless of other activity on the cluster.
Granular Monitoring: Track resource consumption at the job level to identify and optimize inefficient queries.

This level of control makes RisingWave more robust for production environments where predictable performance is non-negotiable.

Other Notable Improvements

Enhanced Connector Support: Improved stability and performance for Kafka, Pulsar, and Kinesis connectors.
SQL Enhancements: Support for more window functions and improved query optimization for complex joins.
Dashboard Updates: The RisingWave dashboard now provides more detailed insights into cluster health and job performance.

Get Started with v2.8

RisingWave v2.8 is available now. You can try it out via:

RisingWave Cloud: The easiest way to run RisingWave. Sign up for a free account.
Docker: docker pull risingwavelabs/risingwave:v2.8.0
Binary: Download from our GitHub releases page.

For a full list of changes, check out the v2.8 Release Notes.

RisingWave AI Developer Tools: CLI, Agent Skills, and MCP

RisingWave Labs — Mon, 20 Apr 2026 11:00:32 +0000

We are entering a world where most database code will be written by AI agents, not humans. This is already happening. Developers ask Claude, Copilot, or Cursor to write their SQL, and the agent produces something that looks right. For traditional databases, it usually is right.

For streaming databases, it is not.

AI agents have seen billions of lines of batch SQL in their training data. They know PostgreSQL, MySQL, and SQLite inside out. But streaming SQL is different. The semantics are different. The architecture is different. And the mistakes are subtle -- queries that parse and execute correctly but build pipelines that silently duplicate data, miss events, or never produce results.

We decided to fix this at the source. Instead of writing better documentation and hoping agents would find it, we built three tools that meet AI agents where they already work.

The Cloud CLI That Distributes Knowledge

RisingWave Cloud CLI (rwc) started as a straightforward infrastructure tool -- manage clusters, configure auth, take snapshots. But we realized the CLI is the one tool that every RisingWave Cloud user already installs. So we turned it into a distribution channel.

rwc skill install --target claude-code

One command. Your AI coding assistant now understands RisingWave. No repo to clone, no config file to edit, no documentation to bookmark. The knowledge travels with the tool.

This matters because the hardest problem in developer tooling is not building the tool. It is getting it into the developer's workflow. The CLI was already there.

Teaching Agents What LLMs Cannot Learn From Training Data

RisingWave Agent Skills is the knowledge itself -- an open-source pack of structured reference documents that AI agents load into their context window.

Why does this need to exist? Because streaming SQL is a tiny fraction of what LLMs have been trained on. When an agent encounters RisingWave, it pattern-matches against what it knows: batch SQL. And the results are predictable.

It writes CREATE TABLE when it should write CREATE SOURCE. It uses date_trunc when TUMBLE is the right choice. It sets up CDC by creating one table per source, paying the ingestion cost multiple times instead of sharing a single connection. It forgets to set watermarks, and then wonders why time windows never close.

These are not bugs the agent can debug. They are architectural mistakes baked into the first line of code.

The skill pack encodes 14 rules across schema design, materialized views, streaming SQL patterns, sink configuration, and performance optimization. Each rule exists because we have seen the same mistake made by different agents, in different codebases, repeatedly. They follow the open Agent Skills specification and work with Claude Code, Cursor, GitHub Copilot, Windsurf, Gemini CLI, and 18+ other tools.

Giving Agents Eyes Into Your Database

Static knowledge is necessary but not sufficient. An agent that knows the right patterns still cannot see your actual schema, your existing materialized views, or your cluster state.

RisingWave MCP Server bridges this gap. It connects AI agents directly to a running RisingWave instance through the Model Context Protocol, exposing 100+ tools for querying, schema inspection, DDL operations, streaming job monitoring, and cluster management.

The difference is concrete. Without the MCP server, an agent might suggest creating a materialized view that duplicates one you already have. With it, the agent can inspect your existing views, check your source schemas, and write SQL that fits your actual system -- not a hypothetical one.

It integrates with VS Code Copilot and Claude Desktop, and supports both STDIO and HTTP transports.

Three Layers, One Workflow

These tools are independent -- you can use any one without the others. But they are designed to work together. The CLI installs the skills. The skills tell the agent how to think. The MCP server shows the agent what is actually there.

Why This Matters Now

The window for shaping how AI agents write streaming SQL is right now. Patterns established today will be reinforced in every future codebase these agents touch. If agents learn the wrong patterns, those patterns will propagate. If they learn the right ones, every developer who uses an AI assistant gets the benefit.

We are building these tools because we believe the quality of AI-generated streaming code should not depend on the developer's expertise in streaming systems. The agent should know what to do. That is the standard we are working toward.

All three tools are open source and free to use:

Cloud CLI -- install rwc and get started
Agent Skills -- teach your AI assistant streaming SQL
MCP Server -- connect your agent to a live instance

FAQ

Do I need RisingWave Cloud to use these tools? Only the rwc CLI requires a Cloud account. Agent Skills and the MCP Server work with any RisingWave deployment, including open-source self-hosted.

Which AI coding assistants are supported? Agent Skills work with Claude Code, Cursor, GitHub Copilot, Windsurf, Gemini CLI, and 18+ other tools. The MCP Server works with any MCP-compatible client.

Can I contribute to the agent skills? Yes. Both the Agent Skills and MCP Server repos are open source. We welcome contributions from the community.

Building a Polymarket-Style Prediction Engine with RisingWave

RisingWave Labs — Wed, 17 Dec 2025 17:49:44 +0000

Polymarket and similar platforms have proved something simple and powerful: price behaves like probability. Traders want to bet on elections, sports, and crypto outcomes with the speed and responsiveness of a modern exchange.

While the user experience feels simple, a prediction market is actually a distributed system dealing with two conflicting engineering constraints. First, you have the need for real-time price discovery. You must continuously update odds, order books, and liquidity as trades stream in at high volume. Second, you face the "fan-out" settlement problem. When a real-world event resolves (e.g., "Did Candidate X win?"), a single message from an oracle must instantly settle hundreds of thousands of open positions and update user balances without race conditions.

Most teams attack this with a fragmented stack: Kafka for ingestion, Flink for windowing, Redis for hot state, and Postgres for the ledger. This requires extensive glue code to orchestrate, and often forces the application layer to handle the complex logic of settlement fan-out.

RisingWave offers a different path.

It is a Postgres-compatible streaming database. Instead of writing glue code, you describe your entire backend logic (from live odds to settlement and balances) as Materialized Views in SQL. RisingWave maintains these views incrementally, effectively turning your database into a reactive engine that solves the "fan-out" problem automatically.

In this post, we will walk through how to build a production-grade prediction market architecture, moving from raw streams to instant settlement.

1. The Anatomy of a Prediction Market

To understand the architecture, it helps to first understand the two data streams that drive a prediction market.

Trading Stream: All the Bets

The first is the Trading Stream. Every time a user places a bet, whether matched via an AMM or a CLOB (Central Limit Order Book), the system appends a record like:

{
  "user_id": 12345,
  "market_id": 42,
  "side": "YES",
  "size": 100,
  "price": 0.57,
  "timestamp": "2025-06-01T12:34:56Z"
}

This stream is high volume and continuous. It is what drives the live price you see on the frontend.

Oracle Stream: The Final Answer

The second is the Oracle Stream. It is low volume but high impact. At some point, each market receives a resolution event, for example:

{
  "market_id": 42,
  "outcome": "YES",
  "resolved_at": "2025-06-02T03:00:00Z"
}

This single message decides who won and who lost in that market.

Where the Hard Part Lives: When Streams Meet

The core engineering challenge is what happens when these two streams meet.

An oracle event arrives for market_id = 42. The system must:

Find every trade for that market
Compute Profit and Loss (PnL) for each user based on their entry price
Update user balances safely and atomically

In a traditional architecture, this often triggers a large batch job or a heavy, locking SQL update once the market resolves.

With a streaming database, you can instead treat this as a continuous join between the Trading Stream and the Oracle Stream. When the oracle event appears, the fan-out settlement is performed automatically as part of the normal stream processing.

2. RisingWave: Streaming SQL with Tiered Storage

RisingWave looks and feels like Postgres, but it processes data as it arrives. You connect it to sources like Kafka or database CDC, define Materialized Views (MVs) to express your logic, and then query those views directly.

Crucially for this use case, RisingWave offloads its internal state to object storage (S3). This "Tiered Storage" architecture means you can maintain the state of millions of open positions or historical trades without being limited by RAM or managing complex RocksDB instances.

For our market, we will define four core views to handle prices, settlement, risk, and the ledger.

3. Real-Time Market State

The most immediate requirement is that when a user buys "YES", the price on the dashboard should tick from 0.50 to 0.51 instantly.

We start by ingesting the raw trades from Kafka.

CREATE SOURCE trades (
  trade_id   BIGINT,
  user_id    BIGINT,
  market_id  BIGINT,
  side       VARCHAR, -- 'YES' or 'NO'
  size       NUMERIC,
  price      NUMERIC,
  ts         TIMESTAMPTZ
) WITH (
  connector = 'kafka',
  topic = 'trades',
  properties.bootstrap.server = '...',
  format = 'json'
);

To drive the UI, we create a materialized view that calculates the latest price and 24-hour volume. We use last_value ordered by the timestamp to ensure the price is deterministic.

CREATE MATERIALIZED VIEW market_prices AS
SELECT
  market_id,
  last_value(price ORDER BY ts) AS last_price,
  AVG(price)                    AS avg_price,
  SUM(size)                     AS total_volume_24h
FROM trades
WHERE ts > now() - interval '24 hours'
GROUP BY market_id;

Because this view is materialized, your API can serve the frontend by simply querying SELECT * FROM market_prices WHERE market_id = 42. The result is always fresh, eliminating the need to sync data to a separate Redis cache.

4. Instant Settlement: The Streaming Join

This is where the architecture diverges from the traditional approach. Instead of writing a batch script that runs "after the game" to calculate winners, we define settlement as a continuous join between the trades stream and the oracle_feed.

First, we define the oracle source:

CREATE SOURCE oracle_feed (
  market_id   BIGINT,
  outcome     VARCHAR,
  resolved_at TIMESTAMPTZ
) WITH (
  connector = 'kafka',
  topic = 'oracle_feed',
  format = 'json'
);

Next, we create the settlement engine. RisingWave maintains the state of all open trades. When an outcome message arrives, it immediately matches against that index and emits the calculated PnL for every affected trade.

CREATE MATERIALIZED VIEW settled_trades AS
SELECT
  t.trade_id,
  t.user_id,
  t.market_id,
  -- If you bought YES and outcome is YES, you get $1.00 per share.
  -- Otherwise, you lose your principal (price * size).
  CASE
    WHEN o.outcome = 'YES' AND t.side = 'YES' THEN (1 - t.price) * t.size
    WHEN o.outcome = 'NO'  AND t.side = 'NO'  THEN (1 - t.price) * t.size
    ELSE -1 * (t.price * t.size)
  END AS pnl,
  o.resolved_at
FROM trades t
JOIN oracle_feed o
  ON t.market_id = o.market_id;

This view is the engine of the platform. It handles the 1-to-N fan-out automatically. There is no loop logic to write and no state consistency to manage manually. As soon as the oracle message hits the Kafka topic, the PnL is calculated and ready for the ledger.

5. User Balances: The Ledger Pattern

A common anti-pattern in SQL is to attempt joining trades, transfers, and settlements all at once to calculate a user's balance. In a streaming context, this is inefficient.

A more robust approach is the Ledger Pattern. We treat every financial event—whether it's a deposit, a withdrawal, or a trading profit—as a line item in a unified ledger. We can then use a UNION ALL to combine these streams into a single flow of balance changes.

CREATE MATERIALIZED VIEW balance_ledger AS
-- 1. Cash movements from the transfers stream
SELECT user_id, amount AS change, ts FROM transfers
UNION ALL
-- 2. Realized PnL from the settled trades view we created above
SELECT user_id, pnl AS change, resolved_at as ts FROM settled_trades

With the streams unified, calculating the live balance becomes a simple aggregation:

CREATE MATERIALIZED VIEW user_balances AS
SELECT
  user_id,
  SUM(change) AS current_balance
FROM balance_ledger
GROUP BY user_id;

Finally, to show users their active risk, we can aggregate the trades that have not yet settled to show their current exposure:

CREATE MATERIALIZED VIEW user_positions AS
SELECT
  user_id,
  market_id,
  SUM(CASE WHEN side = 'YES' THEN size ELSE -size END) AS net_exposure
FROM trades
GROUP BY user_id, market_id;

6. End-to-End Architecture

By pushing the complexity down into the database, the resulting architecture is remarkably clean.

In a typical stack, the box labeled MV: settled_trades would be a complex microservice or a heavy Spark/Flink job responsible for the fan‑out settlement logic. Here, that complexity is absorbed by the database’s join operator.

Similarly, the box labeled MV: user_balances replaces what is usually a fragile orchestration of updates between Redis and a permanent data store.

Because RisingWave offloads the internal state of these joins to S3, you can scale to millions of open positions without managing expensive infrastructure. The result is a system where correctness is derived from the stream itself, providing built‑in auditability through the immutable MV: balance_ledger pattern.

7. Conclusion

To build a prediction market like Polymarket, you need streaming performance for the UI and relational correctness for the money. RisingWave bridges this gap by allowing you to ingest raw streams, define complex business logic using standard SQL, and serve the results directly to your application.

If you are designing a high-frequency trading application or a prediction market, consider pushing the "hard parts“ (state management and stream joining) down into the database. It allows you to focus on the market mechanics rather than the plumbing.

Get Started with RisingWave

Try RisingWave Today:
- Download the open-sourced version of RisingWave to deploy on your own infrastructure.
- Get started quickly with RisingWave Cloud for a fully managed experience.

EXPLAIN ANALYZE in RisingWave: Profiling Real‑Time Query Performance

RisingWave Labs — Tue, 02 Dec 2025 12:58:00 +0000

You’ve meticulously crafted a real-time streaming pipeline in RisingWave. You’ve created a materialized view to power a critical dashboard, but something is wrong. The data on the dashboard is lagging, alerts are delayed, and you suspect a performance bottleneck somewhere in your query.

You reach for a familiar tool: EXPLAIN. You check the query plan, and it looks perfectly optimal. The join order is correct, filters are being pushed down, and the distribution strategy makes sense. Yet, the problem persists.

This is a common frustration in stream processing. A static execution plan only tells you what the system intends to do, not how well it's actually doing it under a live workload. Is one of your joins struggling to keep up? Is a specific operator getting overwhelmed and buffering too much data? EXPLAIN can't answer these questions.

To bridge this gap between plan and reality, we introduced EXPLAIN ANALYZE, a powerful diagnostic tool that enriches the execution plan with live, real-time performance statistics.

What Is `EXPLAIN ANALYZE`?

While EXPLAIN shows the logical, physical, and distributed plan for a query, EXPLAIN ANALYZE goes a step further. It executes the query (or, for a materialized view, attaches to the already-running job) and collects live metrics from each operator in the dataflow graph. With this information, you can stop guessing and start diagnosing.

Metrics Collection and Reporting

When you run:

**EXPLAIN ANALYZE MATERIALIZED VIEW my_view;**

RisingWave performs the following steps:

The serving node sends an RPC to the meta service to fetch fragment assignments and streaming worker locations.
It sends two RPC calls to each streaming node:
- The first collects a baseline metric snapshot.
- The second (after a brief delay) captures a follow-up snapshot.
The frontend computes deltas (e.g., records/sec) from these two samples.
It maps the collected metrics to a reconstructed streaming execution graph, using fragment metadata.

The result is a per-operator breakdown of live performance, including:

Records per second (RPS): Throughput per operator.
Average output pending ratio: How full the output buffer is; indicates potential backpressure or downstream delays.
Actor IDs: Actors contributing to each operator.

Example

***-- Create a table and a materialized view*CREATE TABLE t(v1 int);
CREATE MATERIALIZED VIEW m1 AS SELECT * FROM t x JOIN t y USING(v1) JOIN t z USING(v1) JOIN t c USING(v1);
*-- Analyze the materialized view*EXPLAIN ANALYZE MATERIALIZED VIEW m1;**

Example output:

**identity                                        |  actor_ids  | output_rps | avg_output_pending_ratio
*------------------------------------------------+-------------+------------+--------------------------*
 StreamMaterialize                              | 12,11,10,9  | 41589.1    | 0.00
 └─ MergeExecutor                               | 12,11,10,9  | 41589.1    | 0.81
    └─ Dispatcher                               |             | 0          | NaN
       └─ StreamHashJoin                        | 16,15,14,13 | 41560.1    | 0.85
          ├─ StreamHashJoin                     | 16,15,14,13 | 152.6      | 0.93
          │  ├─ MergeExecutor                   | 16,15,14,13 | 25.6       | 1.63
          │  │  └─ Dispatcher                   |             | 0          | NaN
          │  │     └─ StreamFilter              | 20,19,18,17 | 0          | 0
          │  │        └─ StreamTableScan        | 20,19,18,17 | 0          | 0
          │  │           ├─ BatchPlanNode       | 20,19,18,17 | 0          | 0
          │  │           └─ Upstream            | 20,19,18,17 | 0          | 0
          │  └─ MergeExecutor                   | 16,15,14,13 | 0          | 0
          │     └─ Dispatcher                   |             | 0          | NaN
          │        └─ StreamFilter              | 28,27,26,25 | 0          | 0
          │           └─ StreamTableScan        | 28,27,26,25 | 0          | 0
          │              ├─ BatchPlanNode       | 28,27,26,25 | 0          | 0
          │              └─ Upstream            | 28,27,26,25 | 0          | 0
          └─ MergeExecutor                      | 16,15,14,13 | 51.2       | 0.71
             └─ Dispatcher                      |             | 0          | NaN
                └─ StreamHashJoin               | 24,23,22,21 | 51.2       | 0.71
                   ├─ MergeExecutor             | 24,23,22,21 | 0          | 0
                   │  └─ Dispatcher             |             | 0          | NaN
                   │     └─ StreamFilter        | 36,35,34,33 | 0          | 0
                   │        └─ StreamTableScan  | 36,35,34,33 | 0          | 0
                   │           ├─ BatchPlanNode | 36,35,34,33 | 0          | 0
                   │           └─ Upstream      | 36,35,34,33 | 0          | 0
                   └─ MergeExecutor             | 24,23,22,21 | 0          | 0
                      └─ Dispatcher             |             | 0          | NaN
                         └─ StreamFilter        | 32,31,30,29 | 0          | 0
                            └─ StreamTableScan  | 32,31,30,29 | 0          | 0
                               ├─ BatchPlanNode | 32,31,30,29 | 0          | 0
                               └─ Upstream      | 32,31,30,29 | 0          | 0
(32 rows)**

Practical Use Cases

With EXPLAIN ANALYZE, stream processing engineers can:

Pinpoint bottlenecks: Use low output_rps or high avg_output_pending_ratio to identify slow or blocked operators.
Tune query plans: Compare runtime performance across strategies (e.g., local vs. global aggregation, filter pushdown, partitioning schemes).
Diagnose backpressure: A consistently high pending ratio may indicate downstream congestion or under-provisioned sinks.
Test at scale: Run synthetic load tests on new materialized views and verify they can scale before rollout.

Try It Today

If you're running RisingWave 2.4 or later versions, you can start using EXPLAIN ANALYZE immediately on materialized views, sinks, and indexes. It’s a low-overhead, high-impact tool to keep your streaming jobs efficient and transparent.

Want to explore further?

Check out our EXPLAIN ANALYZE documentation for detailed syntax and parameters.
Join the conversation in our Slack community to ask questions and connect with others.

What Are Process-Time Temporal Joins?

RisingWave Labs — Tue, 02 Dec 2025 03:00:00 +0000

Imagine you run an online store where product prices change frequently. A customer places an order, and five minutes later, you update the price of an item they bought. When you review the order, how do you determine the correct price the customer paid? A standard database join would likely show you the newest price, which is incorrect for that past order.

This is a common challenge in real-time data systems. You need to know what the data looked like at the specific moment when the data is processed. In RisingWave, you can solve this easily with process-time temporal joins.

This blog post will explain what process-time temporal joins are and demonstrate how to use them to achieve point-in-time correctness in your applications.

What Are Process-Time Temporal Joins?

A process-time temporal join is a special type of join that connects a data stream to a table based on the time the system processes each event. Instead of joining with the latest version of a record, it joins with the version of the record that was active at the moment the event from the stream is processed.

This is perfect for our e-commerce scenario. We can join an orders stream with a product_prices table to find the price that was active at the exact moment each order was processed.

A Step-by-Step Demo: Tracking Prices for Orders

Let’s build a real-time materialized view that correctly calculates the total cost of an order using the product price at the time of purchase.

Step 1: Set up Your Tables

First, we need two data structures. One will be a table to store product prices, which can be updated. The other will be an append-only stream of new orders.

The product_prices table requires a PRIMARY KEY. This ensures that new rows with the same product_id are treated as updates to the existing product's price.

-- 1. Tracks product prices with automatic processing time
CREATE TABLE product_prices (
    product_id INT PRIMARY KEY, -- NOTICE: a primary key is needed
    price FLOAT
);

The orders table will represent an append-only stream, meaning we only insert new orders and never update or delete existing ones.

-- 2. Contains historical orders
CREATE TABLE orders (
        order_id INT,
        product_id INT,
        quantity INT
) APPEND ONLY; -- NOTICE: orders is a source instead of table

Step 2: Insert Sample Data

Let’s add some products and their initial prices. We’ll then immediately update them to simulate a price change.

-- Initial prices (we assume that these prices are updated at 08:00)
INSERT INTO product_prices VALUES
(101, 110.0),
(102, 200.0);

If you query product_prices now, you'll only see the latest values (110.0 and 200.0).

Now, let’s add two customer orders. These will be processed by the system after the prices have been updated.

-- We assume that these orders are processed at 08:30
INSERT INTO orders VALUES
(1, 101, 2),
(2, 102, 1);

Step 3: Create a Materialized View with a Temporal Join

This is where the magic happens. We’ll create a materialized view that joins orders to product_prices. The key is the FOR SYSTEM_TIME AS OF PROCTIME() syntax. This tells RisingWave: "For each order, join it with the version of the product price that was valid at the moment the order is processed."

CREATE MATERIALIZED VIEW order_with_price AS
SELECT
    o.order_id,
    o.product_id,
    p.price AS price_at_purchase
FROM orders o
-- Temporal join: Get price version active when the order was processed
JOIN product_prices FOR SYSTEM_TIME AS OF PROCTIME() p
ON o.product_id = p.product_id;

Step 4: Query the Results and See It in Action

Let’s check our materialized view.

SELECT * FROM order_with_price;

The output shows the orders matched with the prices that were active when they were placed. Notice that the last two columns of data are mocked up data used merely for demonstrating the temporal logic and will not appear in the actual results.

order_id | product_id |price_at_purchase| price_update_time | order_process_time
                                          -- (mock up data)      (mock up data)
----------+------------+-----------------+-------------------+------------------
        1 |        101 |           110   |--    08:00:00     |      08:30:00
        2 |        102 |           200   |--    08:00:00     |      08:30:00

Now, let’s update the prices again.

-- We assume that these prices are updated at 09:00
INSERT INTO product_prices VALUES
(101, 150.0),
(102, 250.0);

If we query the view again, the existing results remain unchanged because the temporal join “locks in” the price from when the order was first processed.

SELECT * FROM order_with_price;
-- The output is the same as before!

But what about a new order? Let’s add one that arrives after the latest price change.

-- We assume that these orders are processed at 09:30
INSERT INTO orders VALUES
(3, 102, 3);

Query the view one last time. The new order (ID 3) correctly uses the latest price (150.0), while the old orders retain their original prices.

SELECT * FROM order_with_price;

-- Final Result
 order_id | product_id | current_price | price_update_time | order_process_time
                                        -- (mock up data)      (mock up data)
----------+------------+---------------+-------------------+-------------------
        1 |        101 |           110 |--    08:00:00     |      08:30:00
        2 |        102 |           200 |--    08:00:00     |      08:30:00
        3 |        102 |           250 |--    09:00:00     |      09:30:00

Optimizing Your Joins: Append-Only vs. Non-Append-Only

RisingWave is smart about how it handles temporal joins. The type of join it uses depends on the properties of your data streams.

Append-Only Temporal Join: This is what we used in our demo. It’s highly efficient because the left side of the join (orders) is append-only. RisingWave knows that old orders will never be updated or deleted, so it doesn't need to maintain a large state to track changes. This is ideal for streaming data like events, logs, or orders.
Non-Append-Only Temporal Join: If the table on the left side can receive updates or deletes (i.e., it’s not strictly append-only), RisingWave can still perform a temporal join. However, it requires more resources because it has to maintain an internal state to manage potential retractions or changes to past results.

For the best performance, use an append-only source or table on the left side of your temporal join whenever possible.

Conclusion

Process-time temporal joins are a powerful feature in RisingWave for building applications that require point-in-time accuracy. By using the FOR SYSTEM_TIME AS OF PROCTIME() syntax, you can ensure that your joins reflect the correct state of your data at the exact moment it was processed. This eliminates a whole class of data consistency problems in real-time systems.

Convert millisecond price gaps into real profits

RisingWave Labs — Tue, 25 Nov 2025 07:08:15 +0000

Building a Real-Time Crypto Arbitrage Monitoring System - DEV Community

Arbitrage is a simple strategy. You buy an asset on one exchange where the price is low and sell it...

dev.to

Building a Real-Time Crypto Arbitrage Monitoring System

RisingWave Labs — Tue, 25 Nov 2025 07:06:01 +0000

Arbitrage is a simple strategy. You buy an asset on one exchange where the price is low and sell it on another where the price is high.

In crypto markets, these price differences, or spreads, appear and vanish in milliseconds. If your data pipeline takes five seconds to process a batch of prices, the opportunity is already gone. This post demonstrates how to use RisingWave—an open-source real-time event streaming platform—to detect arbitrage opportunities with sub-second latency using standard SQL.

The Engineering Bottleneck

Arbitrage requires monitoring fragmented liquidity across Binance, Coinbase, OKX, and DEXs simultaneously. This creates three specific engineering hurdles.

Velocity: During volatility, you might ingest over 10,000 price ticks per second.
Synchronization: You cannot compare a Binance price from 10:00:01 with a Coinbase price from 10:00:05. The comparison must be time-aligned.
Latency: The profit potential decays effectively to zero within hundreds of milliseconds.

Traditional ETL pipelines are too slow. We need a system that processes data as it arrives, not in batches.

Why RisingWave Fits

RisingWave is a streaming database designed exactly for this velocity. Unlike a traditional Postgres database where a materialized view is a static snapshot, a RisingWave materialized view updates continuously.

It connects natively to event streams like Kafka or WebSockets.
It calculates results incrementally, processing only new data points.
It handles time-windowing logic natively.
It uses SQL, so you don’t need to write complex Java or Flink jobs to define trading strategies.

How to build the monitor

1. Ingesting Market Data

We assume market data is flowing into a message queue like Kafka. We define two sources: one for Binance and one for Coinbase.

We treat these Kafka topics as tables. RisingWave consumes the JSON events directly.

CREATE SOURCE binance_prices (
    symbol VARCHAR,
    price DECIMAL,
    ts TIMESTAMP
) WITH (
    connector = 'kafka',
    topic = 'binance.ticker',
    properties.bootstrap.server = 'localhost:9092'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE coinbase_prices (
    symbol VARCHAR,
    price DECIMAL,
    ts TIMESTAMP
) WITH (
    connector = 'kafka',
    topic = 'coinbase.ticker',
    properties.bootstrap.server = 'localhost:9092'
) FORMAT PLAIN ENCODE JSON;

2. The Core Logic: Detecting the Spread

We need to find instances where the price difference for the same symbol crosses a profitability threshold (e.g., 0.5%).

Crucially, we must ensure we compare prices that exist at roughly the same moment. A price mismatch is not a valid opportunity if the data points are seconds apart.

We use an interval join condition: ABS(EXTRACT(EPOCH FROM (b.ts - c.ts))) < 1. This restricts matches to price ticks that occurred within one second of each other.

CREATE MATERIALIZED VIEW arbitrage_opportunities AS
SELECT
    b.symbol,
    b.price AS binance_price,
    c.price AS coinbase_price,
    (c.price - b.price) / b.price * 100 AS spread_percent,
    b.ts
FROM binance_prices b
JOIN coinbase_prices c
  ON b.symbol = c.symbol
AND ABS(EXTRACT(EPOCH FROM (b.ts - c.ts))) < 1  *-- align within 1s window*WHERE ABS((c.price - b.price) / b.price * 100) > 0.5;  *-- threshold 0.5%*

As new ticks arrive, RisingWave updates this view instantly. If a spread appears, it shows up here immediately.

3. Market-Wide Aggregation

You may also want to see the global "Best Bid and Offer" (BBO) across all exchanges to visualize market fragmentation.

We combine streams using UNION ALL and apply a hopping window via HOP to calculate spread statistics over short intervals.

CREATE MATERIALIZED VIEW best_bid_offer AS
WITH t AS (SELECT symbol, price, ts FROM binance_prices
           UNION ALL
           SELECT symbol, price, ts FROM coinbase_prices)
SELECT
    symbol,
    MIN(price) AS best_bid,
    MAX(price) AS best_offer,
    MAX(price) - MIN(price) AS spread
FROM HOP(t, ts, INTERVAL '1 second', INTERVAL '5 seconds') 
GROUP BY symbol, window_start;

4. Actionable Output (Subscriptions and Sinks)

Detecting the spread is half the battle. You need to trigger your trading bot immediately.

Traditionally, you would sink this data into Redis or Kafka and have your bot listen there. RisingWave simplifies this with Subscriptions.

Subscriptions allow your application to listen directly to changes in a materialized view. Your trading bot connects to RisingWave and receives a stream of updates—new arbitrage opportunities—as they are generated. This removes the need for an intermediate message queue, reducing both latency and infrastructure complexity.

CREATE SUBSCRIPTION arb_subscription
FROM arbitrage_opportunities
WITH (
    retention = '1h'
);

However, if you need to archive data for backtesting or integrate with other existing systems, you can still use Sinks. For example, you might sink to S3 for long-term storage or to Redis if multiple external services need to access the state.

CREATE SINK arb_signals
FROM arbitrage_opportunities
WITH (
    connector = 'redis',
    primary_key = 'arb_signals',
    redis.url= 'redis://localhost:6379/'
) FORMAT PLAIN ENCODE JSON (
    force_append_only='true'
);

Expanding the Strategy

Once the pipeline is running, you can use SQL to prototype more complex strategies.

Triangular Arbitrage: Join three pairs (BTC/USDT, ETH/BTC, ETH/USDT) within the same exchange.
Funding Rate Arbitrage: Monitor divergence between spot prices and perpetual swap funding rates.
Latency Monitoring: Compare exchange API timestamps against your ingestion timestamps to detect network lag.

Summary

Speed is the primary competitive advantage in crypto trading. By moving processing logic into RisingWave, you achieve sub-second detection of market anomalies without complex application code. The database handles the windowing and joining state, allowing you to define strategies in pure SQL.

From designing Kafka Consumers to building a unified Source–Sink streaming engine that powers real-time data pipelines.

RisingWave Labs — Wed, 29 Oct 2025 07:12:04 +0000

A Hands-On Guide to Building the Speed Layer of the Lambda Architecture - DEV Community

In the previous article, we introduced the design principles and implementation of the Lambda...

dev.to

A Hands-On Guide to Building the Speed Layer of the Lambda Architecture

RisingWave Labs — Wed, 29 Oct 2025 07:07:44 +0000

In the previous article, we introduced the design principles and implementation of the Lambda Architecture.

We talked about its three layers:

Batch Layer – slow but precise; processes full historical data to ensure eventual consistency
Speed Layer – handles incremental data in real time
Serving Layer – merges results from both layers and provides a unified query interface

Starting today, we’ll dive into the implementation of the Speed Layer.

We’ll begin by writing a Kafka Consumer and gradually explore the core mechanisms of stream processing.

In today’s article, we’ll write a Kafka Consumer — a key component of the Speed Layer — responsible for instantly receiving order data and writing it into the Serving DB (while the Batch Layer focuses on processing full historical data each day).

⚠️ Important

All the code in this article is for educational and conceptual demonstration only (pseudo code).

It’s designed to help you understand and solve real-world problems, but it’s not production-ready.

Focus on grasping the overall architecture and design logic — you can skim over low-level code details.

Step 1: Speed Layer Kafka Consumer: Ingesting Orders into the Serving DB

Let’s start with the simplest version:

from kafka import KafkaConsumer
import json

# Subscribe to the orders topic
consumer = KafkaConsumer('orders')

print("[Speed Layer] Waiting for fresh orders...")

for message in consumer:
    order = json.loads(message.value.decode('utf-8'))

    insert_db(order)
    conn.commit()
    print(f"[Speed Layer] Inserted order {order['id']}")

This logic is straightforward:

Kafka provides a real-time stream of order data
The Consumer consumes and processes the stream
The Serving DB stores the processed results
Core task: write incoming data into storage as fast as possible

Step 2: Designing the Serving DB

Here we have two key tables:

orders_batch_summary – pre-aggregated historical data computed daily by the Batch Layer
orders_realtime – detailed real-time orders sent by the Speed Layer

When querying from the dashboard, the system merges both tables while filtering out invalid orders with status = 'removed':

SELECT status, SUM(count) AS total
FROM (
    SELECT status, count(*)
    FROM orders_batch_summary
    WHERE status != 'removed'

    UNION ALL

    SELECT status, COUNT(*) AS count
    FROM orders_realtime
    WHERE status != 'removed'
    GROUP BY status
) t
GROUP BY status;

However, as the company grows, more Consumers are written independently by different developers, each with their own logic. Over time, the codebase becomes tangled and unmaintainable.

We need to refactor — to establish a unified Stream Processing architecture so everyone can build upon a consistent framework.

Designing a Source Abstraction Layer

On a team project, different Kafka Consumer implementations can make integration harder.

The solution is to define a unified Source Interface.

Source Architecture Design

    ┌─────────────┐
    │ BaseSource  │  ◄── Abstract Interface
    │             │
    │ + run()     │
    └─────────────┘
           △
           │ implements
    ┌─────────────┐
    │KafkaSource  │  ◄── Concrete Implementation
    │             │
    │ + run()     │
    └─────────────┘

Step-by-Step Explanation of the Source Core

Step 1: Define the BaseSource Abstract Interface

from abc import ABC, abstractmethod

class BaseSource(ABC):
    def __init__(self, name: str):
        self.name = name

    @abstractmethod
    def run(self):
        pass

Key Points:

Each Source must have a unique name
run() is abstract — forcing subclasses to implement it

Step 2: Initialize SimpleKafkaSource

class SimpleKafkaSource(BaseSource):
    def __init__(self, name: str, topic: str, broker_address: str = "localhost:9092"):
        super().__init__(name)
        self.topic = topic
        self.broker_address = broker_address
        self.consumer = None
        self.message_handler = self._default_handler

Key Points:

Inherits from BaseSource to conform to a unified interface
message_handler is replaceable, offering flexibility in message processing logic

Step 3: Set Up the Kafka Consumer

def _setup_consumer(self):
    self.consumer = KafkaConsumer(
        self.topic,
        bootstrap_servers=self.broker_address,
        group_id=f"simple-source-{self.name}",
        auto_offset_reset='latest',
        value_deserializer=lambda m: json.loads(m.decode('utf-8')) if m else None
    )

Key Points:

group_id is automatically generated to avoid conflicts
auto_offset_reset='latest' ensures consumption starts from the latest messages
Automatic JSON deserialization

Step 4: Core Runtime Logic

def run(self):
    self._setup_consumer()  # initialize Consumer

    for message in self.consumer:  # continuously listen for messages
        self.message_handler({
            'key': message.key,
            'value': message.value,
            'topic': message.topic,
            'offset': message.offset
        })

Execution Flow:

Initialize Kafka Consumer
Continuously read messages from the Topic
Wrap messages in a standard format
Call message_handler for processing

Key Points:

The Source only handles data ingestion; message processing logic is injected externally through message_handler, enabling high flexibility.

import logging
import json
from abc import ABC, abstractmethod
from typing import Optional, Callable, Any
from kafka import KafkaConsumer

logger = logging.getLogger(__name__)

class BaseSource(ABC):
    """Base abstract class for all Sources"""

    def __init__(self, name: str):
        self.name = name
        self._running = False

    @abstractmethod
    def run(self):
        """Main execution method"""
        pass

    def stop(self):
        """Stop the Source"""
        self._running = False
        logger.info(f"Source {self.name} stopped")

class SimpleKafkaSource(BaseSource):
    """Simple Kafka Source implementation"""

    def __init__(
        self,
        name: str,
        topic: str,
        broker_address: str = "localhost:9092",
        consumer_group: Optional[str] = None,
        message_handler: Optional[Callable[[Any], None]] = None
    ):
        super().__init__(name)
        self.topic = topic
        self.broker_address = broker_address
        self.consumer_group = consumer_group or f"simple-source-{name}"
        self.message_handler = message_handler or self._default_handler
        self.consumer: Optional[KafkaConsumer] = None

    def _default_handler(self, message):
        print(f"[{self.name}] Received message: {message}")

    def _setup_consumer(self):
        try:
            self.consumer = KafkaConsumer(
                self.topic,
                bootstrap_servers=self.broker_address,
                group_id=self.consumer_group,
                auto_offset_reset='latest',
                value_deserializer=lambda m: json.loads(m.decode('utf-8')) if m else None,
                key_deserializer=lambda m: m.decode('utf-8') if m else None
            )
            logger.info(f"Consumer setup for topic: {self.topic}, group: {self.consumer_group}")
        except Exception as e:
            logger.error(f"Failed to setup consumer: {e}")
            raise

    def run(self):
        logger.info(f"Starting Source {self.name} for topic {self.topic}")
        self._setup_consumer()
        self._running = True

        try:
            while self._running:
                message_batch = self.consumer.poll(timeout_ms=1000)
                for topic_partition, messages in message_batch.items():
                    for message in messages:
                        if not self._running:
                            break
                        try:
                            self.message_handler({
                                'key': message.key,
                                'value': message.value,
                                'topic': message.topic,
                                'partition': message.partition,
                                'offset': message.offset,
                                'timestamp': message.timestamp
                            })
                        except Exception as e:
                            logger.error(f"Error processing message: {e}")
        except KeyboardInterrupt:
            logger.info("Received interrupt signal")
        except Exception as e:
            logger.error(f"Error in run loop: {e}")
        finally:
            if self.consumer:
                self.consumer.close()
            logger.info(f"Source {self.name} finished")

    def stop(self):
        super().stop()
        if self.consumer:
            self.consumer.close()

Designing the Sink Abstraction Layer

In the Speed Layer architecture, Sources handle data input, while Sinks handle data output.

To avoid inconsistent implementations, we define a unified Sink interface.

Sink Architecture Design

    ┌─────────────┐
    │  BaseSink   │  ◄── Abstract Interface
    │             │
    │ + write()   │
    └─────────────┘
           △
           │ implements
    ┌──────────────────┐
    │SimplePostgreSQL  │  ◄── Concrete Implementation
    │Sink              │
    │ + write()        │
    └──────────────────┘

Step-by-Step Explanation of the Sink Core

Step 1: Define the Base Sink Interface

from abc import ABC, abstractmethod

class BaseSink(ABC):
    def __init__(self, name: str):
        self.name = name

    @abstractmethod
    def write(self, message):
        pass

    def setup(self):
        pass  # default no-op

Key Points:

Each Sink must have a unique name
write() handles actual data writes
setup() is optional and can be overridden

Step 2: Initialize the Simple PostgreSQL Sink

class SimplePostgreSQLSink(BaseSink):
    def __init__(self, name: str, host: str, dbname: str, table_name: str):
        super().__init__(name)
        self.host = host
        self.dbname = dbname
        self.table_name = table_name
        self.connection = None

Key Points:

Inherits from BaseSink for interface consistency
Stores database connection information
Lazy connection initialization (connection = None)

Step 3: Core Logic of write()

def write(self, message):
    # Automatically detect columns and insert into DB
    data = message.get('value', {})
    # ... dynamically generate SQL and insert

Key Points:

Automatically detects field structures in message['value'], dynamically generates INSERT SQL, and writes to the database.

This allows the Sink to adapt to various data schemas automatically.

Complete Sink Implementation

import logging
from abc import ABC, abstractmethod
from typing import Any, Dict

try:
    import psycopg2
    from psycopg2.extras import Json
    from psycopg2 import sql
except ImportError:
    psycopg2 = None
    print("Warning: psycopg2 not installed. Run: pip install psycopg2-binary")

logger = logging.getLogger(__name__)

class BaseSink(ABC):
    """Base abstract class for all Sinks"""

    def __init__(self, name: str):
        self.name = name

    @abstractmethod
    def write(self, message: Dict[str, Any]):
        """Write a message"""
        pass

    def setup(self):
        """Setup connection"""
        pass

    def close(self):
        """Close connection"""
        pass

class SimplePostgreSQLSink(BaseSink):
    """PostgreSQL Sink with automatic schema detection"""

    def __init__(
        self,
        name: str,
        host: str,
        port: int,
        dbname: str,
        user: str,
        password: str,
        table_name: str
    ):
        super().__init__(name)
        self.host = host
        self.port = port
        self.dbname = dbname
        self.user = user
        self.password = password
        self.table_name = table_name
        self.connection = None

    def setup(self):
        if psycopg2 is None:
            raise ImportError("psycopg2 is required")

        self.connection = psycopg2.connect(
            host=self.host,
            port=self.port,
            dbname=self.dbname,
            user=self.user,
            password=self.password
        )
        logger.info(f"Connected to PostgreSQL: {self.host}:{self.port}/{self.dbname}")

    def write(self, message: Dict[str, Any]):
        """Automatically detect fields and write to PostgreSQL"""
        data = message.get('value', {})
        # ... dynamic field detection and SQL execution logic

    def close(self):
        if self.connection:
            self.connection.close()
            logger.info("PostgreSQL connection closed")

Simple Streaming Engine: Unified Management Layer

Between the Source (data input) and the Sink (data output), we need a unified management layer to handle orchestration, monitoring, and lifecycle management.

This component is the SimpleStreamingEngine.

Simple Streaming Engine Architecture Design

    ┌─────────────────────┐
    │SimpleStreamingEngine│  ◄── Central Manager
    │                     │
    │    +add_source()    │
    │    +add_sink()      │
    │    + run()          │
    └─────────────────────┘
           │
           │ manages
           ▼
    ┌──────────────┐    ┌──────────────┐
    │    Source    │───▶│     Sink     │
    │              │    │              │
    │ KafkaSource  │    │PostgreSQLSink│
    └──────────────┘    └──────────────┘

Step-by-Step Breakdown of the Simple Streaming Engine Core Code

Step 1: Initializing the Simple Streaming Engine

class SimpleStreamingEngine:
    def __init__(self, name: str = "simple-streaming-app"):
        self.name = name
        self._sources = []  # List of Sources
        self._sinks = []    # List of Sinks

Key Points:

The SimpleStreamingEngine manages two lists: Sources and Sinks.
It provides unified registration interfaces for both.

Step 2: Registering Sources and Sinks

def add_source(self, source: BaseSource):
    self._sources.append(source)

def add_sink(self, sink: BaseSink):
    self._sinks.append(sink)

Key Points:

Simple list-based management supporting multiple sources and sinks.
Follows a unified interface — any class implementing BaseSource or BaseSink can be registered.

Step 3: Core Execution Logic

def run(self):
    # Initialize all sinks
    for sink in self._sinks:
        sink.setup()

    # Set message handler for each source
    for source in self._sources:
        source.message_handler = self._create_message_handler()
        source.run()  # Start consuming data

Execution Flow:

Initialize connections for all sinks.
Assign message handlers to sources.
Start all sources to begin consuming data.

Step 4: Message Handler Core Logic

def _create_message_handler(self):
    def handler(message):
        # Forward message to all sinks
        for sink in self._sinks:
            sink.write(message)
    return handler

Detailed Data Flow Explanation:

When SimpleStreamingEngine starts:

# Inside SimpleStreamingEngine.run()
for source in self._sources:
    source.message_handler = self._create_message_handler()  # Inject handler
    source.run()  # Start source

When Source receives data:

# Inside SimpleKafkaSource.run()
for message in self.consumer:  # Fetch messages from Kafka
    formatted_message = {
        'key': message.key,
        'value': message.value
    }
    self.message_handler(formatted_message)  # Call injected handler

When handler forwards messages:

# Handler returned by _create_message_handler()
def handler(message):  # message = formatted data from source
    for sink in self._sinks:
        sink.write(message)

Overall Data Flow:

Kafka → Source.run() → message_handler() → Sink.write()

Key Point:

The SimpleStreamingEngine uses function injection so that sources are unaware of sinks, achieving complete decoupling between components.

Why Do We Need the Simple Streaming Engine?

Decoupled Design: Sources and sinks are fully independent and interchangeable.
Scalability: Supports multiple sinks simultaneously (e.g., PostgreSQL + Elasticsearch).
Unified Management: Offers a consistent API for registration and execution.

Complete Simple Streaming Engine Code

import logging
from typing import List
from .source import BaseSource
from .sink import BaseSink

logger = logging.getLogger(__name__)

class SimpleStreamingEngine:
    """
    A minimal streaming processing engine
    """

    def __init__(self, name: str = "simple-streaming-engine"):
        self.name = name
        self._sources: List[BaseSource] = []  # List of sources
        self._sinks: List[BaseSink] = []      # List of sinks

    def add_source(self, source: BaseSource):
        """
        Register a source with the streaming engine
        """
        self._sources.append(source)

    def add_sink(self, sink: BaseSink):
        """
        Register a sink with the streaming engine
        """
        self._sinks.append(sink)

    def run(self):
        """
        Start the streaming engine and process data streams
        """
        # Initialize all sinks
        for sink in self._sinks:
            sink.setup()

        # Assign message handler to each source
        for source in self._sources:
            source.message_handler = self._create_message_handler()
            source.run()  # Start consuming data

    def _create_message_handler(self):
        """
        Create a message handler that dispatches data to all sinks
        """
        def handler(message):
            for sink in self._sinks:
                sink.write(message)
        return handler

Putting It All Together: Automatic Data Flow in Action

At this point, we have:

Source: Ingest data from Kafka
Sink: Deliver data into PostgreSQL
SimpleStreamingEngine: Connects, manages, and orchestrates both

Once assembled, we can kick off the automatic end-to-end data flow.

# 1. Create the SimpleStreamingEngine
engine = SimpleStreamingEngine(...)

# 2. Create a Kafka Source
orders_source = SimpleKafkaSource(...)

# 3. Create a PostgreSQL Sink
pg_sink = SimplePostgreSQLSink(...)

# 4. Assemble and start
engine.add_source(orders_source)
engine.add_sink(pg_sink)
engine.run()  # Start processing: Kafka → PostgreSQL

Summary

In this section, we explored the core implementation of the Speed Layer:

The Batch Layer provides reliable historical data processing.
The Speed Layer delivers real-time responsiveness and streaming data handling.
Without the Speed Layer, true real-time capability in the Lambda Architecture is impossible.

Through the Source–Sink–SimpleStreamingEngine architecture, we built:

A unified data processing interface
A scalable stream processing framework
A fully functional Speed Layer implementation

Day 5 Preview: Tackling Performance Bottlenecks

At first, the system runs smoothly — the consumer processes order data without issue.

But under heavy traffic:

The console starts showing latency warnings.
The consumer’s processing power hits its limit.
Orders begin to queue up for processing.

In the next part, we’ll explore Speed Layer performance challenges and how to optimize for high-throughput workloads.

Fixing the Half-Transaction Problem in Real-Time CDC

RisingWave Labs — Tue, 14 Oct 2025 06:43:59 +0000

Modern real-time data systems rely heavily on Change Data Capture (CDC) to power downstream stream processing. CDC continuously captures INSERT, UPDATE, and DELETE operations from databases, allowing downstream systems to subscribe to these event streams to:

Build real-time reports and monitoring dashboards
Drive risk control and alerting logic
Synchronize data to data warehouses, search engines, or caches
Support applications like real-time analytics, recommendations, and trade matching

However, a critical issue is often overlooked: transactional boundaries.

Database writes are transactional. But when these changes are broken down into individual CDC events, downstream systems might split a single transaction, leading to the processing and exposure of “half-finished” data. Such transient inconsistencies can cause fluctuating reports, false alarms, and even the ingestion of incomplete data into data warehouses.

This article details the common problems with transactional boundaries when processing CDC streams. We will then explore how to recognize and respect these boundaries in practice to guarantee atomicity and consistency, ensuring that materialized views and join queries always reflect the true, complete state of the database.

Two Examples

Consider a typical payment scenario where a user completes a transfer, and the system executes a transaction to:

Debit an account
Credit another account
Insert a record into an audit table

If we send each UPDATE and INSERT event downstream as soon as it's received, a materialized view might see the debit before the credit, or the audit log before the balances are updated. For real-time risk control and monitoring systems, this could mean:

False alarms: A risk control rule sees an anomalous balance after the debit but before the credit, mistakenly flagging it as a fund loss.
Incorrect analysis: A real-time report shows a decrease in total assets, leading to wrong decisions by operations or traders.
Data corruption: Writing data to a warehouse mid-transaction might only capture half of the transaction, breaking consistency and foreign key constraints.

Let’s look at another join scenario. Suppose we have a real-time job that joins an orders table with a payments table to calculate Gross Merchandise Volume (GMV):

SELECT o.order_id, o.user_id, p.amount
FROM orders o
JOIN payments p ON o.order_id = p.order_id;

If a transaction first inserts an order and then a payment record, but the CDC stream is split by a Barrier — with the order landing in one Epoch and the payment in the next — the join’s intermediate state could return an empty result for a cycle because the payment hasn’t arrived yet. A downstream dashboard would show a temporary drop in GMV, triggering unnecessary alerts or manual investigations.

This problem is particularly critical in domains like finance, payments, e-commerce, and logistics:

Transaction Monitoring: Trade matching, payment processing, and risk validation must be based on complete transactions, not partial data.
Real-time Reporting: Metrics like GMV, inventory levels, and warehouse movements require transactional consistency; otherwise, they will fluctuate, misleading operations.
Risk Control & Alerting: Partial transactions can trigger false alarms, leading to frozen funds or intercepted orders, which negatively impacts the user experience.

Therefore, preserving transactional atomicity and order within CDC streams is fundamental to ensuring the reliability of real-time computation and analysis.

Barriers and Epochs: The “Commit Points” of Stream Processing

To understand the problem of transactional boundaries, we must first understand how stream processing systems organize data and ensure consistency. Unlike batch processing, stream processing deals with an infinite stream of events, making it impossible to simply “commit” everything at once as in a database transaction. To achieve consistency and fault tolerance, these systems introduce the concept of a Barrier.

A Barrier is a special control message injected into the data stream. It carries no business data but propagates through the entire stream graph. When an operator receives a barrier, it first processes all data in its buffer, then snapshots its state, and finally passes the barrier downstream. Once all operators in the topology have received the barrier, it signifies that all data up to that point has been fully processed and can be safely made visible. The current state can also be persisted as a consistent checkpoint.

The data between two consecutive barriers is called an Epoch. You can think of it as a “micro-batch” in stream processing:

Within an Epoch, an operator’s state is continuously updated, but the output is buffered until the epoch is complete.
The boundary of an Epoch is the barrier, which acts like a commit point in a database.

This design allows stream processing systems to achieve exactly-once semantics (ensuring each event is processed only once) and recover from the latest checkpoint after a failure, avoiding data loss or duplicate processing.

However, Barriers are typically injected based on a fixed schedule (e.g., every second) or event count (e.g., after every N records). They are completely unaware of when a transaction in the upstream database begins or ends. If a transaction’s updates span a barrier, they are split into two Epochs, causing temporary inconsistencies in downstream states and materialized views. This is the “half-a-transaction” problem.

The Technical Challenge: Conflicts Between Barriers and Transactions

We now understand the roles of Barriers and Epochs — they guarantee exactly-once semantics and recoverability, but they are oblivious to upstream transactional boundaries. Herein lies the problem: Barriers are injected periodically, while transaction commits are unpredictable. If a transaction spans two Barriers, it gets split into two Epochs.

Imagine a typical bank transfer: first a debit, then a credit, then an audit log entry. If the debit event falls into one Epoch, while the credit and log events fall into the next, downstream materialized views and joins will show incorrect results for a period: Account A’s balance decreases, Account B’s balance remains unchanged, and the total assets mysteriously shrink. This not only triggers false risk alerts but can also corrupt the downstream data warehouse by causing ETL jobs to write incomplete data.

Cross-table transactions make things even worse. CDC often splits changes from different tables into separate, parallel streams. Without unified coordination, a join operator might see an update from Table A before seeing the corresponding update from Table B, causing the join to return empty or partially matched results in some windows. For GMV calculations, order-payment reconciliation, or inventory-shipment consistency analysis, such transient inconsistencies are unacceptable.

Therefore, the real challenge is this: how can we aggregate all events of a transaction into the same Epoch to ensure atomic commits, without breaking the fault-tolerance semantics of stream processing? This also requires handling cross-table transactions by merging changes from multiple tables into a single logical stream; otherwise, downstream joins cannot guarantee consistency.

RisingWave’s Solution: Perceiving and Respecting Transactional Boundaries

To ensure that CDC streams correctly reflect the transaction semantics of the upstream database, RisingWave’s design considers several key aspects: ensuring input stream order, preventing Barriers from splitting transactions, and handling cross-table transactions, schema evolution, and fault tolerance. Let’s dive into how RisingWave tackles these challenges.

1. Acquiring a Strictly Ordered Transaction Stream

The first step is to ensure the input stream is correctly ordered. While the change log of an upstream database is inherently ordered, middleware like a Debezium Connector with Kafka can split changes from different tables into different topics. With transaction metadata in a separate topic, the downstream system must reassemble the transaction, which adds complexity and introduces risks of out-of-order events and delays.

RisingWave embeds the Debezium Embedded Engine, running the parsing logic directly within the connector node. This provides a single, strictly linear stream of events whose order is identical to the commit order of the upstream database, complete with transaction markers like BEGIN, COMMIT, and ROLLBACK. As a result, the downstream system doesn't need any extra "assembly" logic; all transaction events are naturally complete and contiguous.

Technically, this step eliminates the complexities of multi-threaded consumption and multi-topic merging, allowing for precise perception of transactional boundaries. This is the foundation for everything that follows.

2. Pausing Barriers to Keep Transactions in the Same Epoch

With an ordered transaction stream, the next problem is preventing barriers from splitting transactions. RisingWave’s approach is straightforward: pause the propagation of barriers during a transaction.

Specifically, when the Source Executor receives a BEGIN message, it enters a "transaction mode." All incoming events are buffered in memory instead of being immediately dispatched downstream. Any arriving Barrier is propagated downstream, but without any associated events. Only upon receiving a COMMIT message are all buffered events atomically flushed. Downstream operators then immediately consume these events, update their state, and generate a new Epoch snapshot.

The key advantages of this design are its simplicity and correctness:

Atomicity: All events within a transaction are either completely invisible or become visible all at once.
Consistency: Materialized views and joins always reflect the true state of the database, without any intermediate “half-transaction” states.
Recoverability: Since Barriers only advance after a transaction is complete, checkpoints always align with transaction boundaries. Upon recovery, transactions won’t be lost or re-applied.

Of course, this approach has a trade-off: a particularly large transaction can delay Barrier propagation, causing an Epoch to last longer than usual, which in turn delays checkpoints and downstream visibility. However, we believe this is a reasonable trade-off. A delay of a few hundred milliseconds, or even a few seconds, is more acceptable than exposing partial transactions.

3. Multi-Table CDC Source for Cross-Table Transaction Guarantees

Ensuring single-table transactionality is not enough. In real-world business scenarios, multi-table transactions are common: updating an order’s status, inserting a payment record, and modifying inventory all within a single transaction. If these tables are ingested using different sources, the events within the transaction will be scattered across different streams, and downstream joins cannot guarantee consistency.

To address this, RisingWave introduces the multi-table CDC source. Users can specify multiple tables within a single source, and RisingWave will use one connection to subscribe to their change logs. This way, all events within a transaction appear in the same physical stream with their order fully preserved.

For example:

CREATE SOURCE my_cdc WITH (
  connector = 'postgres-cdc',
  database.name = 'mydb',
  table.name = 'public.orders, public.payments, public.inventory'
);

CREATE TABLE orders (...) FROM SOURCE my_cdc TABLE 'public.orders';
CREATE TABLE payments (...) FROM SOURCE my_cdc TABLE 'public.payments';
CREATE TABLE inventory (...) FROM SOURCE my_cdc TABLE 'public.inventory';

Now, if a transaction updates all three tables simultaneously, these changes will be buffered and flushed together, landing in the same Epoch. Downstream joins and aggregations will see an atomically consistent result.

4. Streaming Graph Implementation Details

From an implementation perspective, this means that when RisingWave builds a streaming graph, it places the source node and all its dependent materialized views into the same “transaction domain.” A Barrier is intercepted upon entering this domain and only propagates downstream after the transaction is complete. This ensures the integrity of the transaction and keeps checkpoints consistently aligned.

RisingWave’s Meta Service manages the lifecycle of these “transaction domains.” During a DDL transaction, if a user creates multiple tables that reference the same CDC source, the system recognizes this and runs them under the same physical source job, sharing a single connection and barrier stream.

5. Schema Evolution and Transactional Boundaries

Transaction awareness also simplifies schema evolution. When an upstream table’s schema changes (e.g., adding a new column), a user can execute ALTER TABLE in RisingWave. The system will replan the execution graph at the next transaction boundary, ensuring the schema change and data processing remain consistent. This avoids the awkward situation where half a transaction uses the old schema and the other half uses the new one.

6. Trade-offs: Large Transactions and Latency

While pausing Barriers guarantees transactional atomicity, it has an inevitable side effect: if a transaction is very large, the time needed to receive and buffer all its events might exceed a normal Barrier interval. This forces events that could have been spread across multiple epochs to be processed atomically in a single, longer epoch, thus increasing the system’s end-to-end latency.

We evaluated several options during the design phase:

Propagate Barriers early: Send the Barrier downstream first and then send the data after the transaction commits. This maintains a fixed checkpoint interval but exposes an incomplete transaction downstream, destroying consistency.
Optimistic execution with transaction rollback: Send events downstream for preliminary computation, confirm visibility upon transaction commit, or roll back otherwise. This requires almost all stateful operators to support multi-version storage and rollback logic, which is extremely complex.
Strict buffering with delayed Barriers: Simple and direct, ensuring a one-to-one alignment between transactions and Epochs at the cost of increased latency.

RisingWave chose the third option because it guarantees consistency without introducing additional state management complexity. We found that in real-world scenarios, large and cross-table transactions are limited, and most transactions complete within a normal Barrier interval, so the added latency is acceptable. For the rare, extremely large transaction, trading latency for atomicity is a reasonable engineering trade-off.

7. Recovery and Fault Tolerance

Another benefit of this design is simpler recovery logic. When the system fails, RisingWave recovers from the snapshot corresponding to the most recent Barrier, which is always located after a transaction boundary. This ensures that committed transactions are not re-applied and that no partial transaction data is lost upon recovery. In contrast, if a Barrier were to land in the middle of a transaction, recovery would require special logic to skip events that have already been sent, making the process complex and error-prone.

Conclusion: Making CDC Reliable

Correctly processing CDC data isn’t just about pushing individual change events downstream; it’s about ensuring that the combination of these changes is consistent with the true state of the database. Without respecting transactional boundaries, downstream systems may see intermediate states that never actually existed, leading to fluctuating reports, false risk alerts, data inconsistencies, and even business incidents.

RisingWave’s design philosophy is simple:

Use the Debezium Embedded Engine to obtain a strictly ordered transaction stream.
Pause Barriers during transactions to align transactions with Epochs.
Support multi-table CDC sources to handle cross-table transactions within a single logical stream.

This way, materialized views and join queries always reflect the true state of the database; what you see at any moment is a complete, atomic result. This enhances data reliability and makes real-time analytics, risk control, trade matching, and monitoring scenarios dependable.

For engineers, this means you can confidently use RisingWave to drive real-time reports, risk control, and trading decisions without worrying about “half a transaction” wandering through your system and breaking business logic. In other words, we bring the atomicity of databases to the streaming world.

Why Engineers Try HTTP for Streaming — And Where It Breaks

RisingWave Labs — Mon, 13 Oct 2025 10:09:47 +0000

In the previous article, we discussed the evolution of stream processing engines. Today, let’s talk about an interesting phenomenon: why do engineers often think of HTTP when faced with real-time processing requirements?

HTTP: The engineer’s Swiss army knife

The product manager may excitedly say: "We need a real-time order statistics system!"

An engineer’s first instinct would be: "Just POST new orders to the stats service, update the counts, respond with success, done."

Why this instinct? Because HTTP feels like a Swiss Army knife for engineers:

Familiarity: Both frontend and backend engineers use it daily.
Rich tooling: Postman, curl, Swagger—whatever you need, it’s there.
Minimal learning curve: Way easier than learning Kafka.
Easy debugging: A single curl command is enough to test.

This naturally leads to a synchronous model.

A closer look at HTTP approach

1. Architecture

Backend (Producer): POST new orders to the stats service.
Stats service (Server): Receive request → update in-memory counters → respond success.
Dashboard (Consumer): Sends a GET request to /stats every few seconds to fetch the latest data.

Flowchart:
Shop System --sync HTTP--> Stats Service <--sync HTTP-- Dashboard

Looks perfect, right? Any engineer could implement this in 5 minutes with no new tech to learn.

But…

2. Problems emerge when traffic grows

The HTTP synchronous model works fine under low traffic, but when the system scales:

(1) Tight coupling

The shop system has to wait for the stats service to process the request before completing the order. If the stats service slows down, the entire order API slows down.

(2) Traffic spike challenge

During high traffic, say 100,000 orders in a short period, the stats service would need to handle 100,000 concurrent HTTP requests—an obvious bottleneck.

(3) Fault tolerance issues

Network glitches or service restarts can result in lost stats, requiring additional retry or compensation mechanisms.

3. Why switch to an asynchronous model?

The core problem with the synchronous model is that transmission and processing are locked to the same timeline.

If one side slows down, the other suffers—like two people running tied together with a chain.

A better approach is to introduce a middle layer (message queue/event streaming platform):

Backend: Just push new orders into the event stream, no need to wait for stats processing.
Stats service: Pull data from the event stream and process at its own pace.

Three major benefits of asynchronous architecture:

Stress-resistant: During spikes, queue orders first and process them gradually.
Decoupled: Shop system and stats service don’t block each other.
Fault-tolerant: Event streams can persist data, so services can recover after crashes.

Conclusion

The HTTP synchronous model is a great starting point and works well for small-scale systems. But as traffic grows, synchronous processing introduces coupling and scaling challenges.

That’s when asynchronous, event-driven architectures come into play. By decoupling producers and consumers via a middle layer, the system becomes much more resilient.

In the next article, we’ll dive into the classic Lambda Architecture, exploring how it elegantly addresses these problems while ensuring both speed and accuracy.

Most people know what streaming frameworks can do — but few dig into why they work. Shih-min (Stan) Hsu, Team Lead of Data Engineering at SHOPLINE is embarking on a 30-day challenge to Dive Deep into Streaming Pipelines.

RisingWave Labs — Fri, 10 Oct 2025 04:32:13 +0000

Why Do Data Pipelines Need Streaming — Isn’t Batch Processing Enough? - DEV Community

About this series This time, I’m taking on a 30-day challenge to focus on streaming...

dev.to

DEV Community: RisingWave Labs

RisingWave v2.8: Query Your Lakehouse, Backfill Faster, and Tune Jobs Individually

RisingWave v2.8: Query Your Lakehouse, Backfill Faster, and Tune Jobs Individually

Query Your Lakehouse Directly (Powered by DataFusion)

Faster Backfilling with Parallelism

Individual Job Tuning & Resource Isolation

Other Notable Improvements

Get Started with v2.8

RisingWave AI Developer Tools: CLI, Agent Skills, and MCP

The Cloud CLI That Distributes Knowledge

Teaching Agents What LLMs Cannot Learn From Training Data

Giving Agents Eyes Into Your Database

Three Layers, One Workflow

Why This Matters Now

FAQ

Building a Polymarket-Style Prediction Engine with RisingWave

1. The Anatomy of a Prediction Market

Trading Stream: All the Bets

Oracle Stream: The Final Answer

Where the Hard Part Lives: When Streams Meet

2. RisingWave: Streaming SQL with Tiered Storage

3. Real-Time Market State

4. Instant Settlement: The Streaming Join

5. User Balances: The Ledger Pattern

6. End-to-End Architecture

7. Conclusion

Get Started with RisingWave

EXPLAIN ANALYZE in RisingWave: Profiling Real‑Time Query Performance

What Is EXPLAIN ANALYZE?

Metrics Collection and Reporting

Practical Use Cases

Try It Today

What Are Process-Time Temporal Joins?

What Are Process-Time Temporal Joins?

A Step-by-Step Demo: Tracking Prices for Orders

Step 1: Set up Your Tables

Step 2: Insert Sample Data

Step 3: Create a Materialized View with a Temporal Join

Step 4: Query the Results and See It in Action

Optimizing Your Joins: Append-Only vs. Non-Append-Only

Conclusion

Convert millisecond price gaps into real profits

Building a Real-Time Crypto Arbitrage Monitoring System - DEV Community

Building a Real-Time Crypto Arbitrage Monitoring System

The Engineering Bottleneck

Why RisingWave Fits

How to build the monitor

1. Ingesting Market Data

2. The Core Logic: Detecting the Spread

3. Market-Wide Aggregation

4. Actionable Output (Subscriptions and Sinks)

Expanding the Strategy

Summary

From designing Kafka Consumers to building a unified Source–Sink streaming engine that powers real-time data pipelines.

A Hands-On Guide to Building the Speed Layer of the Lambda Architecture - DEV Community

A Hands-On Guide to Building the Speed Layer of the Lambda Architecture

Step 1: Speed Layer Kafka Consumer: Ingesting Orders into the Serving DB

Step 2: Designing the Serving DB

Designing a Source Abstraction Layer

Source Architecture Design

Step-by-Step Explanation of the Source Core

Designing the Sink Abstraction Layer

Sink Architecture Design

Step-by-Step Explanation of the Sink Core

Complete Sink Implementation

Simple Streaming Engine: Unified Management Layer

Simple Streaming Engine Architecture Design

Step-by-Step Breakdown of the Simple Streaming Engine Core Code

Step 1: Initializing the Simple Streaming Engine

Step 2: Registering Sources and Sinks

Step 3: Core Execution Logic

Step 4: Message Handler Core Logic

Why Do We Need the Simple Streaming Engine?

Complete Simple Streaming Engine Code

Putting It All Together: Automatic Data Flow in Action

Summary

Day 5 Preview: Tackling Performance Bottlenecks

Fixing the Half-Transaction Problem in Real-Time CDC

Two Examples

Barriers and Epochs: The “Commit Points” of Stream Processing

What Is `EXPLAIN ANALYZE`?