Imagine you are computing the total number of orders placed in each one-minute window. Your system is consuming a Kafka topic and every second new events arrive. At 10:02:00, you decide to close the 10:01:00 window and emit the result. But three seconds later, an event with timestamp 10:01:47 shows up. It was delayed in transit. What do you do?
If you wait forever, you never emit results. If you close the window immediately and ignore late arrivals, your counts are wrong. Every production streaming system faces this problem, and watermarks are the standard solution.
This article explains what watermarks are, why late data makes stream processing hard, and how RisingWave's WATERMARK syntax lets you declare event-time progress directly in SQL. All SQL in this guide is tested against RisingWave 2.8.0.
The Problem: Processing Time vs. Event Time
Before understanding watermarks, you need to understand the distinction between two clocks that every stream processing system tracks.
Event time is when an event actually happened, according to the data itself. This is the timestamp recorded by the producer: the sensor reading at 10:01:47, the click event at 10:01:52, the payment processed at 10:01:59.
Processing time is when the stream processor receives and processes the event. Due to network delays, producer batching, broker replication lag, and consumer scheduling, processing time is always later than event time and the gap is unpredictable.
| Clock | Controlled by | Predictable? | Use for |
| Processing time | The system clock | Yes (wall clock) | System health metrics, latency monitoring |
| Event time | The data producer | No | Business logic, correctness-sensitive aggregations |
For business analytics, you almost always want event time. "Orders placed between 10:00 and 10:01" means the event-time window, not "orders received by the processor between 10:00 and 10:01." These are different sets of events and can differ significantly under load.
Why Late Data Happens
Late data is not an edge case. It is a structural property of distributed systems. Common causes include:
- Mobile clients that batch events and flush when connectivity is restored
- Multi-region ingestion where data travels across continents before reaching a central broker
- Kafka consumer lag during backpressure or rebalancing
- IoT devices with intermittent network connections
- Log aggregators that buffer writes for efficiency
In each of these cases, you have real, valid events that arrive seconds, minutes, or even hours after their event timestamp. Your windowed aggregations need a principled way to handle them.
What Is a Watermark?
A watermark is a lower bound on event time. It is the system's best estimate of the statement: "I believe I have seen all events with a timestamp earlier than time T."
When the watermark advances to time T, the system can safely close all windows that end at or before T. Any events with a timestamp before T that arrive after the watermark has passed are considered late.
Events arriving (by processing time):
10:00:05 → 10:01:10 → 10:01:47 → 10:00:58 → 10:02:03 → ...
Watermark (lagging by 30 seconds):
At processing time 10:00:35: watermark = 10:00:05
At processing time 10:01:40: watermark = 10:01:10
At processing time 10:02:17: watermark = 10:01:47
At processing time 10:02:33: watermark = 10:02:03 ← window [10:01, 10:02] closes here
10:00:58 was late but already included
The watermark formula is simple: watermark = max(observed event_time) - lag_tolerance.
The lag tolerance is your knob. A larger lag tolerates later arrivals at the cost of higher latency in emitting results. A smaller lag gives faster results but discards more late data.
The Tradeoff Triangle
Every watermark configuration balances three forces:
- Completeness: how many late events are included in window results
- Latency: how long you wait before emitting a window result
- Correctness: whether emitted results accurately reflect the true aggregate
You cannot maximize all three simultaneously. A five-second watermark lag emits results within five seconds of a window closing but will discard events that arrive more than five seconds late. A five-minute watermark captures nearly all late arrivals but delays results by five minutes.
The right value depends on your data and requirements. For an IoT pipeline with reliable network connectivity, a 10-second lag may be sufficient. For a mobile analytics pipeline with users on spotty connections, a 5-minute lag may be necessary.
How RisingWave Implements Watermarks
RisingWave is a streaming SQL database that lets you declare watermarks directly in your table or source definition using standard SQL. The watermark expression is part of the schema, not scattered across application code.
The syntax is:
WATERMARK FOR <column_name> AS <column_name> - INTERVAL '<value>' <unit>
This declaration tells RisingWave:
- Which column carries event time (
column_name) - How much to lag behind the maximum observed value (
INTERVAL)
RisingWave advances the watermark automatically as new data arrives. You do not write any code to track or update it. The watermark propagates through the query plan and tells downstream window operators when they can safely close and emit results.
Tables with Watermarks Must Be Append-Only
In RisingWave, declaring a watermark on a table requires the table to be append-only. This makes sense conceptually: watermarks track monotonic progress through an ordered event stream. If rows could be updated or deleted, the ordering guarantees that watermarks depend on would break.
Add the APPEND ONLY clause to enable watermarks on a table.
Practical Example: Order Revenue by Minute
Here is a complete working example. You will create an orders table with a watermark, insert events, and build a materialized view that counts orders and revenue per one-minute tumbling window.
Step 1: Create the Table with a Watermark
CREATE TABLE wm_orders (
order_id VARCHAR,
customer_id VARCHAR,
product VARCHAR,
amount NUMERIC,
event_time TIMESTAMP,
WATERMARK FOR event_time AS event_time - INTERVAL '5' MINUTE
) APPEND ONLY;
The WATERMARK FOR event_time AS event_time - INTERVAL '5' MINUTE clause sets a five-minute tolerance. The system waits five minutes past the maximum observed event timestamp before treating that time boundary as final. Events arriving up to five minutes late will still be counted in their correct windows.
Step 2: Insert Some Events
INSERT INTO wm_orders VALUES
('o1', 'c1', 'Widget A', 29.99, '2026-04-01 10:00:00'),
('o2', 'c2', 'Widget B', 49.99, '2026-04-01 10:01:30'),
('o3', 'c3', 'Widget C', 19.99, '2026-04-01 10:02:00'),
('o4', 'c1', 'Widget A', 29.99, '2026-04-01 10:03:45'),
('o5', 'c4', 'Widget D', 99.99, '2026-04-01 10:04:55');
Step 3: Define the Tumbling Window Materialized View
CREATE MATERIALIZED VIEW wm_mv_orders_per_minute AS
SELECT
window_start,
window_end,
COUNT(*) AS order_count,
SUM(amount) AS total_revenue
FROM TUMBLE(wm_orders, event_time, INTERVAL '1' MINUTE)
GROUP BY window_start, window_end;
TUMBLE(wm_orders, event_time, INTERVAL '1' MINUTE) creates non-overlapping one-minute buckets aligned to wall-clock minutes. Each row lands in exactly one window based on its event_time value.
Step 4: Query the Results
SELECT * FROM wm_mv_orders_per_minute ORDER BY window_start;
Output:
window_start | window_end | order_count | total_revenue
---------------------+---------------------+-------------+---------------
2026-04-01 10:00:00 | 2026-04-01 10:01:00 | 1 | 29.99
2026-04-01 10:01:00 | 2026-04-01 10:02:00 | 1 | 49.99
2026-04-01 10:02:00 | 2026-04-01 10:03:00 | 1 | 19.99
2026-04-01 10:03:00 | 2026-04-01 10:04:00 | 1 | 29.99
2026-04-01 10:04:00 | 2026-04-01 10:05:00 | 1 | 99.99
Each minute-window shows its order count and revenue. If a late event arrives within the five-minute watermark tolerance, RisingWave will update the affected window's row in the materialized view.
Sliding Windows with HOP
Tumbling windows are non-overlapping: each event belongs to exactly one window. Sometimes you want a sliding window that updates more frequently. The HOP function creates overlapping windows with a configurable slide interval.
CREATE MATERIALIZED VIEW wm_mv_orders_rolling AS
SELECT
window_start,
window_end,
COUNT(*) AS order_count,
SUM(amount) AS total_revenue
FROM HOP(wm_orders, event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)
GROUP BY window_start, window_end;
HOP(table, time_col, hop_size, window_size) creates five-minute windows that slide by one minute. At any point, there are five active windows covering overlapping time ranges. This is useful for "rolling N-minute total" metrics that update every minute.
Querying wm_mv_orders_rolling returns:
window_start | window_end | order_count | total_revenue
---------------------+---------------------+-------------+---------------
2026-04-01 09:56:00 | 2026-04-01 10:01:00 | 1 | 29.99
2026-04-01 09:57:00 | 2026-04-01 10:02:00 | 2 | 79.98
2026-04-01 09:58:00 | 2026-04-01 10:03:00 | 3 | 99.97
2026-04-01 09:59:00 | 2026-04-01 10:04:00 | 4 | 129.96
2026-04-01 10:00:00 | 2026-04-01 10:05:00 | 5 | 229.95
2026-04-01 10:01:00 | 2026-04-01 10:06:00 | 4 | 199.96
2026-04-01 10:02:00 | 2026-04-01 10:07:00 | 3 | 149.97
2026-04-01 10:03:00 | 2026-04-01 10:08:00 | 2 | 129.98
2026-04-01 10:04:00 | 2026-04-01 10:09:00 | 1 | 99.99
Each row represents a five-minute window, and consecutive windows overlap by four minutes. The watermark on wm_orders governs when each of these windows finalizes.
Example: Short-Interval Watermarks for Web Analytics
For high-frequency event streams like web clickstream data, shorter watermark intervals make sense. Network latency between a browser and your ingestion endpoint is typically under a second in the same region.
CREATE TABLE wm_page_events (
user_id VARCHAR,
page VARCHAR,
action VARCHAR,
event_time TIMESTAMP,
WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) APPEND ONLY;
A ten-second watermark is aggressive: any event arriving more than ten seconds late is dropped from its window. But for web analytics in a low-latency environment, this gives you near-real-time results with minimal data loss.
CREATE MATERIALIZED VIEW wm_mv_page_events_per_30s AS
SELECT
window_start,
window_end,
page,
action,
COUNT(*) AS event_count
FROM TUMBLE(wm_page_events, event_time, INTERVAL '30' SECOND)
GROUP BY window_start, window_end, page, action;
After inserting events:
INSERT INTO wm_page_events VALUES
('u1', '/home', 'view', '2026-04-01 12:00:00'),
('u2', '/product', 'view', '2026-04-01 12:00:05'),
('u3', '/cart', 'click', '2026-04-01 12:00:08'),
('u1', '/product', 'click', '2026-04-01 12:00:15'),
('u2', '/cart', 'view', '2026-04-01 12:00:22'),
('u4', '/home', 'view', '2026-04-01 12:00:30');
Querying wm_mv_page_events_per_30s:
window_start | window_end | page | action | event_count
---------------------+---------------------+----------+--------+-------------
2026-04-01 12:00:00 | 2026-04-01 12:00:30 | /cart | click | 1
2026-04-01 12:00:00 | 2026-04-01 12:00:30 | /cart | view | 1
2026-04-01 12:00:00 | 2026-04-01 12:00:30 | /home | view | 1
2026-04-01 12:00:00 | 2026-04-01 12:00:30 | /product | click | 1
2026-04-01 12:00:00 | 2026-04-01 12:00:30 | /product | view | 1
2026-04-01 12:00:30 | 2026-04-01 12:01:00 | /home | view | 1
The last event (at 12:00:30) falls into the next 30-second window because 12:00:30 is the boundary of the first window and the TUMBLE function uses half-open intervals [start, end).
Using Watermarks with Kafka Sources
In production, your event data comes from Kafka rather than direct inserts. The watermark declaration works identically on sources:
CREATE SOURCE wm_orders_source (
order_id VARCHAR,
customer_id VARCHAR,
product VARCHAR,
amount NUMERIC,
event_time TIMESTAMP,
WATERMARK FOR event_time AS event_time - INTERVAL '5' MINUTE
) WITH (
connector = 'kafka',
topic = 'orders',
properties.bootstrap.server = 'broker:9092',
scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;
The watermark expression is the same regardless of whether the data comes from a Kafka topic, a PostgreSQL CDC stream, or direct inserts. RisingWave handles the event-time tracking automatically.
Once the source is defined, you can create the same TUMBLE or HOP materialized views over it. The watermark on the source propagates to all downstream materialized views that reference it.
Watermark Lag: Choosing the Right Value
There is no universally correct watermark lag. You need to profile your data to pick a value that suits your workload.
A useful first step is to measure the distribution of delays in your pipeline. The delay for a given event is processing_time - event_time. If p99 of your delays is under 30 seconds, a 60-second watermark captures nearly all data with a reasonable buffer.
| Environment | Typical delay distribution | Suggested watermark lag |
| Same-region web analytics | p99 < 1s | 5-10 seconds |
| Cross-region mobile apps | p99 < 60s | 2-5 minutes |
| IoT sensors on cellular | p99 < 5 minutes | 10-15 minutes |
| Log aggregation pipelines | p99 < 30 minutes | 1 hour |
You can also monitor late data rates by counting events that arrive below the current watermark. If you see a spike in late arrivals, your watermark lag may be too short for current network conditions.
How Watermarks Relate to Exactly-Once Processing
Watermarks handle the correctness of event-time windows. A separate concern is delivery guarantees: does each event contribute to aggregations exactly once, even across failures?
RisingWave uses barrier-based checkpointing to ensure exactly-once semantics. Barriers flow through the processing graph alongside data. When all operators in the graph process a barrier, RisingWave takes a checkpoint. On recovery from failure, processing resumes from the last checkpoint. Watermarks are also part of the checkpoint state, so event-time progress is correctly restored after a restart.
The combination of watermarks and exactly-once checkpointing means your time-windowed aggregations are both correct (they account for late data appropriately) and complete (no events are double-counted or dropped due to failures). For a deeper look at how exactly-once works in stream processing, see Exactly-Once Processing in Stream Processing: How It Really Works.
Watermarks vs. Processing-Time Windows
If you do not declare a watermark, you can still write time-based queries in RisingWave using processing time. For example, you can filter rows using NOW() for sliding temporal filters:
SELECT *
FROM wm_orders
WHERE event_time > NOW() - INTERVAL '5' MINUTE;
This returns events that arrived within the last five minutes by processing time. It is simple and has zero lag, but it does not handle late arrivals. An event that arrives six minutes after it happened will not appear in this query even if it belongs to a window that should have included it.
Processing-time queries are appropriate for operational use cases like "show me alerts from the last five minutes" where you care about recency, not correctness. For anything that requires accurate aggregations over event time (revenue totals, user session counts, SLA compliance metrics), event-time windows with watermarks are the right tool.
Architecture: Where Watermarks Fit
In a typical RisingWave streaming pipeline, watermarks live at the ingestion layer and propagate downstream automatically.
graph LR
A[Kafka Topic] -->|Events with event_time| B[RisingWave Source / Table]
B -->|WATERMARK declaration| C[Watermark Operator]
C -->|Advances watermark| D[TUMBLE / HOP Window]
D -->|Closes completed windows| E[Materialized View]
E -->|Query results| F[Dashboard / API]
style A fill:#231F20,stroke:#231F20,color:#fff
style B fill:#1A6AFF,stroke:#1A6AFF,color:#fff
style C fill:#1A6AFF,stroke:#1A6AFF,color:#fff
style D fill:#1A6AFF,stroke:#1A6AFF,color:#fff
style E fill:#1A6AFF,stroke:#1A6AFF,color:#fff
style F fill:#4CAF50,stroke:#4CAF50,color:#fff
You declare the watermark once in the source or table definition. Every materialized view built on top of that source automatically inherits its watermark semantics. You do not re-declare the watermark in each view.
This is a meaningful difference from systems like Apache Flink, where watermark generation is configured in application code for each data stream. In RisingWave, it is a schema-level concern expressed in SQL, which makes it reviewable, versionable, and accessible to anyone who understands SQL. For a direct comparison of the two approaches, see Apache Flink vs. RisingWave: Which Is Right for Your Use Case?.
Further Reading
- RisingWave Watermarks documentation: Complete reference for the
WATERMARKclause, including generated column usage and TTL behavior. - RisingWave Time Windows documentation: Full reference for
TUMBLE,HOP, andSESSIONwindow functions with all parameters. - Incremental Materialized Views Explained: How RisingWave keeps materialized views continuously up to date without full recomputation.
- Building Real-Time Dashboards from Kafka Data with SQL: End-to-end guide connecting Kafka, RisingWave, and Grafana.
FAQ
Can I declare a watermark on a regular table (not append-only)?
No. RisingWave requires that tables with watermarks are declared APPEND ONLY. This is because watermarks track monotonic progress through an ordered event stream, and updates or deletes would violate the ordering assumptions the watermark depends on. If your use case requires both watermarks and updates (for example, CDC streams where orders can be modified), use a separate lookup table for mutable data and an append-only table for the event log.
What happens to events that arrive after the watermark has passed their window?
They are dropped from event-time windows. The window has already closed and emitted its result. If you need to capture late arrivals even after the watermark has passed, you would need to implement a correction pattern: write both an initial result and a corrected result when late data arrives. RisingWave does not automatically reopen closed windows.
Does a shorter watermark lag always give lower query latency?
Yes, but the tradeoff is more late data being dropped. A one-second watermark lag gives results within one second of a window closing, but any event arriving more than one second late is excluded. Whether that is acceptable depends on your SLAs. If your p99 event delay is 500ms, a one-second lag captures the vast majority of data. If your p99 is 30 seconds, a one-second lag discards a large fraction of events.
Can I use different watermark lags for different parts of the pipeline?
Each table or source has a single watermark declaration. Downstream materialized views inherit that watermark. If you need different latency tolerances for different consumers of the same raw data, one approach is to ingest the raw stream without a watermark and define separate watermarked views with different window sizes to approximate different lag tolerances. Another approach is to create two sources pointing at the same Kafka topic with different watermark configurations.
