Real-Time As-Of Joins in Streaming Databases: How We Solved Out-of-Order Event Matching

In many real-time applications, such as financial trading, risk analytics, and IoT monitoring, timestamps in data streams from different sources may not match exactly. To process these streams effectively, we often need to join them by matching the nearest timestamp. This is where As-Of Joins come into play. RisingWave is one of the very few streaming databases that support streaming as-of joins. Most traditional databases only support batch as-of joins, which work well for historical analysis but are too slow for real-time applications. RisingWave supports both batch and streaming as-of joins, making it a powerful tool for low-latency event processing. This article explains why streaming as-of joins are essential, how they work in RisingWave, and how you can use them effectively. What Is an As-Of Join? An As-Of Join finds the most recent (or closest) event in one stream for an event in another stream based on a timestamp column. Unlike regular SQL joins, which require an exact match, an as-of join allows flexibility in time alignment by matching events to the latest available data before or at the event time. Why Streaming As-Of Joins? Batch vs. Streaming As-Of Joins Most databases support batch as-of joins, which process data in fixed time windows (e.g., hourly or daily). This is fine for historical analysis but useless for real-time applications such as: Stock trading: Traders need the latest bid/ask price at execution, not a delayed batch result. Real-time risk management: Portfolio managers must react to live market conditions instantly. IoT monitoring: Sensor data must be correlated in real-time for alerts and anomaly detection. With streaming as-of joins, RisingWave continuously updates results as new data arrives, ensuring low-latency event matching and real-time analytics. Example: Matching Trades with the Latest Market Price Let’s say we are processing stock trade events in real-time and need to join them with the latest available market price before the trade. Trade Stream (trades) trade_id symbol trade_time trade_price 1 AAPL 2024-03-12 09:30:01 180.50 2 AAPL 2024-03-12 09:30:05 181.00 3 TSLA 2024-03-12 09:30:06 720.25 Market Price Stream (market_prices) symbol update_time price AAPL 2024-03-12 09:30:00 180.25 AAPL 2024-03-12 09:30:03 180.75 TSLA 2024-03-12 09:30:04 719.50 TSLA 2024-03-12 09:30:07 721.00 Expected Output: As-Of Join Result For each trade, we take the most recent market price before or at the trade time. trade_id symbol trade_time trade_price market_price update_time 1 AAPL 2024-03-12 09:30:01 180.50 180.25 2024-03-12 09:30:00 2 AAPL 2024-03-12 09:30:05 181.00 180.75 2024-03-12 09:30:03 3 TSLA 2024-03-12 09:30:06 720.25 719.50 2024-03-12 09:30:04 How to Write the Query in RisingWave In RisingWave, to maintain a continuously updated result, we use CREATE MATERIALIZED VIEW to store and process the stream efficiently. Step 1: Create the Input Streams CREATE TABLE trades ( trade_id BIGINT, symbol TEXT, trade_time TIMESTAMP, trade_price DECIMAL(10, 2) ) WITH (connector='kafka', topic='trades_topic'); CREATE TABLE market_prices ( symbol TEXT, update_time TIMESTAMP, price DECIMAL(10, 2) ) WITH (connector='kafka', topic='market_prices_topic'); Step 2: Create the As-Of Join as a Materialized View CREATE MATERIALIZED VIEW trade_enriched AS SELECT t.trade_id, t.symbol, t.trade_time, t.trade_price, m.price AS market_price, m.update_time FROM trades AS t ASOF JOIN market_prices AS m ON t.symbol = m.symbol AND t.trade_time >= m.update_time; How This Works RisingWave continuously maintains state for market_prices The latest price for each symbol is cached efficiently. When a new trade arrives, it looks up the most recent price before trade_time. The result is immediately available in the materialized view. Now, we can query trade_enriched like a regular table to get continuously updated results. SELECT * FROM trade_enriched WHERE symbol = 'AAPL' ORDER BY trade_time DESC LIMIT 5; Conclusion Streaming As-Of Joins in RisingWave provide real-time, low-latency event matching, making them ideal for finance, trading, and IoT applications. Unlike batch-based joins, RisingWave continuously updates results so that real-time analytics are always fresh and accurate. By using CREATE MATERIALIZED VIEW, RisingWave maintains the join state efficiently, ensuring that queries are always fast and up to date. If you need real-time data enrichment with the latest available information, RisingWave’s As-Of Join is the right tool for the job.

Apr 5, 2025 - 06:43

Real-Time As-Of Joins in Streaming Databases: How We Solved Out-of-Order Event Matching

In many real-time applications, such as financial trading, risk analytics, and IoT monitoring, timestamps in data streams from different sources may not match exactly. To process these streams effectively, we often need to join them by matching the nearest timestamp.

This is where As-Of Joins come into play. RisingWave is one of the very few streaming databases that support streaming as-of joins. Most traditional databases only support batch as-of joins, which work well for historical analysis but are too slow for real-time applications.

RisingWave supports both batch and streaming as-of joins, making it a powerful tool for low-latency event processing. This article explains why streaming as-of joins are essential, how they work in RisingWave, and how you can use them effectively.

What Is an As-Of Join?

An As-Of Join finds the most recent (or closest) event in one stream for an event in another stream based on a timestamp column.

Unlike regular SQL joins, which require an exact match, an as-of join allows flexibility in time alignment by matching events to the latest available data before or at the event time.

Why Streaming As-Of Joins?

Batch vs. Streaming As-Of Joins

Most databases support batch as-of joins, which process data in fixed time windows (e.g., hourly or daily). This is fine for historical analysis but useless for real-time applications such as:

Stock trading: Traders need the latest bid/ask price at execution, not a delayed batch result.
Real-time risk management: Portfolio managers must react to live market conditions instantly.
IoT monitoring: Sensor data must be correlated in real-time for alerts and anomaly detection.

With streaming as-of joins, RisingWave continuously updates results as new data arrives, ensuring low-latency event matching and real-time analytics.

Example: Matching Trades with the Latest Market Price

Let’s say we are processing stock trade events in real-time and need to join them with the latest available market price before the trade.

Trade Stream (`trades`)

trade_id	symbol	trade_time	trade_price
1	AAPL	2024-03-12 09:30:01	180.50
2	AAPL	2024-03-12 09:30:05	181.00
3	TSLA	2024-03-12 09:30:06	720.25

Market Price Stream (`market_prices`)

symbol	update_time	price
AAPL	2024-03-12 09:30:00	180.25
AAPL	2024-03-12 09:30:03	180.75
TSLA	2024-03-12 09:30:04	719.50
TSLA	2024-03-12 09:30:07	721.00

Expected Output: As-Of Join Result

For each trade, we take the most recent market price before or at the trade time.

trade_id	symbol	trade_time	trade_price	market_price	update_time
1	AAPL	2024-03-12 09:30:01	180.50	180.25	2024-03-12 09:30:00
2	AAPL	2024-03-12 09:30:05	181.00	180.75	2024-03-12 09:30:03
3	TSLA	2024-03-12 09:30:06	720.25	719.50	2024-03-12 09:30:04

How to Write the Query in RisingWave

In RisingWave, to maintain a continuously updated result, we use CREATE MATERIALIZED VIEW to store and process the stream efficiently.

Step 1: Create the Input Streams

CREATE TABLE trades (
    trade_id BIGINT,
    symbol TEXT,
    trade_time TIMESTAMP,
    trade_price DECIMAL(10, 2)
) WITH (connector='kafka', topic='trades_topic');

CREATE TABLE market_prices (
    symbol TEXT,
    update_time TIMESTAMP,
    price DECIMAL(10, 2)
) WITH (connector='kafka', topic='market_prices_topic');

Step 2: Create the As-Of Join as a Materialized View

CREATE MATERIALIZED VIEW trade_enriched AS
SELECT t.trade_id, t.symbol, t.trade_time, t.trade_price,
       m.price AS market_price, m.update_time
FROM trades AS t
ASOF JOIN market_prices AS m
ON t.symbol = m.symbol
AND t.trade_time >= m.update_time;

How This Works

RisingWave continuously maintains state for market_prices
- The latest price for each symbol is cached efficiently.
When a new trade arrives, it looks up the most recent price before trade_time.
The result is immediately available in the materialized view.

Now, we can query trade_enriched like a regular table to get continuously updated results.

SELECT * FROM trade_enriched WHERE symbol = 'AAPL' ORDER BY trade_time DESC LIMIT 5;

Conclusion

Streaming As-Of Joins in RisingWave provide real-time, low-latency event matching, making them ideal for finance, trading, and IoT applications. Unlike batch-based joins, RisingWave continuously updates results so that real-time analytics are always fresh and accurate.

By using CREATE MATERIALIZED VIEW, RisingWave maintains the join state efficiently, ensuring that queries are always fast and up to date.

If you need real-time data enrichment with the latest available information, RisingWave’s As-Of Join is the right tool for the job.