Implementation:ArroyoSystems Arroyo Watermark Generator
| Knowledge Sources | |
|---|---|
| Domains | Streaming, Watermarking, Event_Time |
| Last Updated | 2026-02-08 08:00 GMT |
Overview
WatermarkGenerator is a streaming operator that computes and emits event-time watermarks based on a configurable DataFusion expression, with support for periodic emission intervals and idle source detection.
Description
The WatermarkGenerator operator implements the ArrowOperator trait to generate watermarks from incoming record batches. It evaluates a user-defined expression (a DataFusion PhysicalExpr, typically something like event_time - INTERVAL '5' SECOND) against each batch to compute the watermark value.
The operator's behavior:
- Watermark Computation: For each batch, the expression is evaluated to produce a TimestampNanosecondArray. The minimum value from this array becomes the candidate watermark. The operator tracks max_watermark (the highest watermark seen so far) in its state cache.
- Emission Control: Watermarks are emitted only when the time elapsed since the last emission (tracked by last_watermark_emitted_at) exceeds the configured interval, preventing excessive watermark traffic. The batch is always forwarded downstream regardless of watermark emission.
- Idle Detection: If idle_time is configured, the operator monitors wall-clock time since the last event via a 1-second tick. When the idle threshold is exceeded, it broadcasts Watermark::Idle to signal that this partition has no data and downstream operators should not wait for it.
- End of Data: On close with SignalMessage::EndOfData, the operator emits a near-maximum watermark (year ~2554) to flush all downstream windows.
State is checkpointed to a global keyed state table "s" storing WatermarkGeneratorState (a bincode-serializable struct with last_watermark_emitted_at and max_watermark).
Usage
Automatically inserted by the Arroyo planner at the source of a streaming query when the user specifies a watermark strategy using an expression (e.g., WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND). Constructed via WatermarkGeneratorConstructor from an ExpressionWatermarkConfig.
Code Reference
Source Location
- Repository: ArroyoSystems_Arroyo
- File: crates/arroyo-worker/src/arrow/watermark_generator.rs
Signature
#[derive(Encode, Decode, Copy, Clone, Debug, PartialEq)]
pub struct WatermarkGeneratorState {
last_watermark_emitted_at: SystemTime,
max_watermark: SystemTime,
}
pub struct WatermarkGenerator {
interval: Duration,
state_cache: WatermarkGeneratorState,
idle_time: Option<Duration>,
last_event: SystemTime,
idle: bool,
expression: Arc<dyn PhysicalExpr>,
}
impl WatermarkGenerator {
pub fn expression(
interval: Duration,
idle_time: Option<Duration>,
expression: Arc<dyn PhysicalExpr>,
) -> WatermarkGenerator;
}
pub struct WatermarkGeneratorConstructor;
impl OperatorConstructor for WatermarkGeneratorConstructor {
type ConfigT = ExpressionWatermarkConfig;
fn with_config(
&self,
config: Self::ConfigT,
registry: Arc<Registry>,
) -> anyhow::Result<ConstructedOperator>;
}
Import
use arroyo_worker::arrow::watermark_generator::{
WatermarkGenerator, WatermarkGeneratorConstructor, WatermarkGeneratorState,
};
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| record_batch | RecordBatch | Yes | Input batch passed through and used to compute the watermark expression |
| checkpoint | CheckpointBarrier | Yes | Triggers persistence of WatermarkGeneratorState to global keyed state "s" |
Outputs
| Name | Type | Description |
|---|---|---|
| record_batch | RecordBatch | The input batch forwarded unchanged to downstream operators |
| watermark | Watermark::EventTime | Emitted when the emission interval has elapsed, containing the minimum expression value |
| idle_watermark | Watermark::Idle | Emitted when no events have been received for longer than idle_time |
Usage Examples
// Created from SQL watermark definition:
// CREATE TABLE input (
// event_time TIMESTAMP,
// WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
// )
let constructor = WatermarkGeneratorConstructor;
let operator = constructor.with_config(watermark_config, registry)?;
// Or constructed directly:
let generator = WatermarkGenerator::expression(
Duration::from_secs(1), // emission interval
Some(Duration::from_secs(30)), // idle timeout
expression, // DataFusion PhysicalExpr
);