risingwavelabs/risingwave · error · StreamExecutorError

Missing watermark serde

Error message

Missing watermark serde

What it means

This error comes from the state table's watermark-aware iteration path. When the table has table watermarks configured, iterating requires a `watermark_serde` (and its `WatermarkSerdeType`) to encode/decode watermark prefixes; if the state table was built without one (`self.watermark_serde` is `None`) but a watermark-aware lookup (`iter_with_watermark`) is invoked, the call fails with `Missing watermark serde`. It indicates the table configuration and the executor's access pattern are mismatched.

Source

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

    /// `vnode`, and filters out rows based on watermarks. It calls `iter_with_prefix` and further filters rows
    /// based on the table watermark retrieved from the state store.
    ///
    /// The caller must ensure that `clean_watermark_index` is set before calling this method, otherwise it will return all rows without filtering.
    pub async fn iter_with_prefix_respecting_watermark(
        &self,
        pk_prefix: impl Row,
        sub_range: &(Bound<impl Row>, Bound<impl Row>),
        prefetch_options: PrefetchOptions,
    ) -> StreamExecutorResult<BoxedRowStream<'_>> {
        let vnode = self.compute_prefix_vnode(&pk_prefix);
        let Some(clean_watermark_index) = self.clean_watermark_index else {
            return self
                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)
                .await
                .map(|s| s.boxed());
        };
        let Some((watermark_serde, watermark_type)) = &self.watermark_serde else {
            return Err(StreamExecutorError::from(anyhow!(
                "Missing watermark serde"
            )));
        };
        // Fast path. TableWatermarksIndex::rewrite_range_with_table_watermark has already filtered the rows.
        if matches!(watermark_type, WatermarkSerdeType::PkPrefix) {
            return self
                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)
                .await
                .map(|s| s.boxed());
        }

        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());
        };

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the table's `TableCatalog` actually defines a watermark column; add `WITH (watermark = ...)` at creation or recreate the table/MV if it is missing.
2. Guard the calling executor path so watermark-aware iteration is only used when `watermark_serde` is `Some` — fall back to `iter_with_prefix` for non-watermark tables.
3. Check for version skew: if the table was created before watermark support, resnapshot/recreate the table with the current version.
4. If the table should have a watermark serde but doesn't, dump the catalog to check `watermark_column_index` / clean watermark indices and fix the planner/catalog bug.

Example fix

// before: unconditional watermark-aware iteration
let stream = table.iter_with_watermark(pk_prefix, sub_range).await?;
// after: fall back when the table has no watermark serde
let stream = if table.has_watermark_serde() {
    table.iter_with_watermark(pk_prefix, sub_range).await?
} else {
    table.iter_with_prefix(pk_prefix, sub_range).await?
};

// Before watermark-aware iteration, confirm the table actually has a watermark serde
if table.watermark_serde.is_none() {
    tracing::warn!(table_id = %table.table_id(), "no watermark serde; falling back to prefix iteration");
}

// Rust
fn can_iter_with_watermark(table: &StateTable) -> bool {
    table.watermark_serde.is_some()
}

// Rust
match table.iter_with_watermark(pk_prefix, sub_range).await {
    Err(e) if e.to_string().contains("Missing watermark serde") => {
        // fall back to plain prefix iteration
        table.iter_with_prefix(pk_prefix, sub_range).await.map(|s| s.boxed())
    }
    other => other,
}

When it happens

Trigger: Calling watermark-aware iteration/range read on a `StateTable` whose `TableCatalog` has no watermark columns (so no `watermark_serde` was constructed at build time), typically from `rewrite_range_with_table_watermark` or an executor that assumes watermarks exist.

Common situations: Executor code paths that unconditionally perform watermark-prefixed lookups on tables created without `watermark` clauses; version skew where a table created before watermark support is accessed by newer executor logic; planner producing watermark rewrites on tables lacking watermark columns.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Expected at most 1 clean_watermark_index per table, got {:?}
• multiple clean watermark columns are not supported yet
• Watermark cannot be NULL
• Watermark serde should have at least one order type
• clean watermark column index {} is not included in table val

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