risingwavelabs/risingwave · error

watermark column is expected to be non-null

Error message

watermark column is expected to be non-null

What it means

`row_to_cache_key` in the EOWC sort buffer extracts the timestamp/watermark datum from a row to build a cache key and calls `.expect("watermark column is expected to be non-null")`. The executor's invariant is that the sort (watermark) column of rows entering the buffer is never NULL; a NULL there is a schema/data violation, so it panics. Called by insert, delete, update, and refill_cache.

Source

Thrown at src/stream/src/executor/eowc/sort_buffer.rs:53

use crate::common::state_cache::{StateCache, StateCacheFiller, TopNStateCache};
use crate::common::table::state_table::StateTable;
use crate::executor::{StreamExecutorError, StreamExecutorResult};

type CacheKey = (
    DefaultOrdered<ScalarImpl>, // sort (watermark) column value
    MemcmpEncoded,              // memcmp-encoded pk
);

fn row_to_cache_key<S: StateStore>(
    sort_column_index: usize,
    row: impl Row,
    buffer_table: &StateTable<S>,
) -> CacheKey {
    let timestamp_val = row
        .datum_at(sort_column_index)
        .to_owned_datum()
        .expect("watermark column is expected to be non-null");
    let mut pk = vec![];
    buffer_table
        .pk_serde()
        .serialize((&row).project(buffer_table.pk_indices()), &mut pk);
    (timestamp_val.into(), pk.into())
}

// TODO(rc): need to make this configurable?
const CACHE_CAPACITY: usize = 2048;

/// [`SortBuffer`] is a common component that consume an unordered stream and produce an ordered
/// stream by watermark. This component maintains a buffer table passed in, whose schema is same as
/// [`SortBuffer`]'s input and output. Generally, the component acts as a buffer that output the
/// data it received with a delay, commonly used to implement emit-on-window-close policy.
pub struct SortBuffer<S: StateStore> {
    /// The timestamp column to sort on.
    sort_column_index: usize,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Filter out NULL event-time rows before the EOWC executor, e.g. add `WHERE ts_col IS NOT NULL` in the upstream MV or source view.
2. Declare the timestamp column NOT NULL (or use a NOT NULL source schema) so nulls are rejected at ingest time.
3. Fill missing timestamps at the source side (default to processing time or a sentinel) before they reach the sort buffer.
4. Inspect incoming records to find which producer emits null timestamps and fix it there.

Example fix

// before: nullable event time flows into EOWC MV
CREATE MATERIALIZED VIEW mv AS SELECT ts, ... FROM src;
// after
CREATE MATERIALIZED VIEW mv AS SELECT ts, ... FROM src WHERE ts IS NOT NULL;

-- Ensure the sort/watermark column has no NULLs before it reaches the EOWC executor:
SELECT COUNT(*) FROM src WHERE ts_col IS NULL; -- must be 0
-- Or enforce upstream:
CREATE MATERIALIZED VIEW clean AS SELECT * FROM src WHERE ts_col IS NOT NULL;

fn watermark_present(row: &impl Row, idx: usize) -> bool {
    row.datum_at(idx).is_some()
}

When it happens

Trigger: Inserting, updating, deleting, or refilling cache with a row whose column at `sort_column_index` is NULL — typically rows flowing from a source that emitted a NULL event-time/timestamp value into an EMIT ON WINDOW CLOSE pipeline.

Common situations: Kafka/NATS sources delivering events with a missing or null timestamp field; late/malformed records before watermark filtering; a schema where the timestamp column is nullable and no NOT NULL enforcement or filter was applied upstream.

Related errors

• Watermark cannot be NULL
• below watermark check condition eval must return bool array
• unreachable (prev != curr)
• core predicate must exist
• Expected at most 1 clean_watermark_index per table, got {:?}

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