risingwavelabs/risingwave · error · StreamExecutorError

Watermark cannot be NULL

Error message

Watermark cannot be NULL

What it means

This error is raised in RisingWave's watermark update path (StateTableInner) when the single-column watermark row extracted from the incoming watermark chunk is NULL. The update_watermark contract requires a non-NULL value because NULL cannot serve as a monotonic watermark bound; the code explicitly checks `watermark_value.is_none()` after cloning row[0].

Source

Thrown at src/stream/src/common/table/state_table.rs:2201

        let watermark_bytes = self.row_store.state_store.get_table_watermark(vnode);
        let Some(watermark_bytes) = watermark_bytes else {
            return self
                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)
                .await
                .map(|s| s.boxed());
        };
        let watermark_row = watermark_serde.deserialize(&watermark_bytes)?;
        if watermark_row.len() != 1 {
            return Err(StreamExecutorError::from(format!(
                "Watermark row should have exactly 1 column, got {}",
                watermark_row.len()
            )));
        }
        let watermark_value = watermark_row[0].clone();
        // StateTableInner::update_watermark should ensure that the watermark is not NULL
        if watermark_value.is_none() {
            return Err(StreamExecutorError::from(anyhow!(
                "Watermark cannot be NULL"
            )));
        }
        let order_type = watermark_serde.get_order_types().get(0).ok_or_else(|| {
            StreamExecutorError::from(anyhow!(
                "Watermark serde should have at least one order type"
            ))
        })?;

        let direction = if order_type.is_ascending() {
            WatermarkDirection::Ascending
        } else {
            WatermarkDirection::Descending
        };
        let clean_watermark_index_in_pk = self
            .pk_indices
            .iter()
            .position(|&i| i == clean_watermark_index);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check upstream watermark column source for expressions/inputs that can emit NULL and coalesce or filter them
2. Verify the watermark column is declared NOT NULL in the table schema
3. Audit the executor feeding update_watermark to ensure it never forwards a NULL watermark datum
4. If reproducible, file a bug with the plan/graph: NULL watermarks should be dropped upstream, not reach the state table

Example fix

// before: executor forwards datum directly
let watermark_row = vec![watermark_datum];
// after: guard at the producer
if watermark_datum.is_none() { return Ok(()); }
let watermark_row = vec![watermark_datum];

fn ensure_watermark_not_null(row: &[Datum]) -> Result<(), StreamExecutorError> {
    match row.first() {
        Some(Some(_)) => Ok(()),
        _ => Err(StreamExecutorError::from(anyhow!("watermark datum is NULL"))),
    }
}

fn has_watermark(row: &[Datum]) -> bool { matches!(row.first(), Some(Some(_))) }

When it happens

Trigger: A stream executor (e.g. WatermarkFilter or a window executor) calls the state table's watermark update path with a watermark row whose first (and only) column deserialized to None — i.e. the upstream produced a NULL watermark datum.

Common situations: A user-defined watermark expression or column emits NULL; an upstream operator propagates a NULL watermark after schema or data-type changes; buggy aggregator output on a watermark column.

Related errors

• Watermark serde should have at least one order type
• below watermark check condition eval must return bool array
• unreachable (prev != curr)
• watermark column is expected to be non-null
• Watermark should not be produced by a table function

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/9dd13a84b96472e6. Report an issue: GitHub.