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.
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.
Talk to Our Experts: Have a complex use case or want to see a personalized demo? Contact us to discuss how RisingWave can address your specific challenges.
Join Our Community: Connect with fellow developers, ask questions, and share your experiences in our vibrant Slack community.
If you’d like to see a personalized demo or discuss how this could work for your use case, please contact our sales team.
