{"record":{"id":"ebf19d7a2ba68d15","repo":"risingwavelabs/risingwave","slug":"missing-watermark-serde","errorCode":null,"errorMessage":"Missing watermark serde","messagePattern":"Missing watermark serde","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/common/table/state_table.rs","lineNumber":2172,"sourceCode":"    /// `vnode`, and filters out rows based on watermarks. It calls `iter_with_prefix` and further filters rows\n    /// based on the table watermark retrieved from the state store.\n    ///\n    /// The caller must ensure that `clean_watermark_index` is set before calling this method, otherwise it will return all rows without filtering.\n    pub async fn iter_with_prefix_respecting_watermark(\n        &self,\n        pk_prefix: impl Row,\n        sub_range: &(Bound<impl Row>, Bound<impl Row>),\n        prefetch_options: PrefetchOptions,\n    ) -> StreamExecutorResult<BoxedRowStream<'_>> {\n        let vnode = self.compute_prefix_vnode(&pk_prefix);\n        let Some(clean_watermark_index) = self.clean_watermark_index else {\n            return self\n                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)\n                .await\n                .map(|s| s.boxed());\n        };\n        let Some((watermark_serde, watermark_type)) = &self.watermark_serde else {\n            return Err(StreamExecutorError::from(anyhow!(\n                \"Missing watermark serde\"\n            )));\n        };\n        // Fast path. TableWatermarksIndex::rewrite_range_with_table_watermark has already filtered the rows.\n        if matches!(watermark_type, WatermarkSerdeType::PkPrefix) {\n            return self\n                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)\n                .await\n                .map(|s| s.boxed());\n        }\n\n        let watermark_bytes = self.row_store.state_store.get_table_watermark(vnode);\n        let Some(watermark_bytes) = watermark_bytes else {\n            return self\n                .iter_with_prefix(pk_prefix, sub_range, prefetch_options)\n                .await\n                .map(|s| s.boxed());\n        };","sourceCodeStart":2154,"sourceCodeEnd":2190,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/table/state_table.rs#L2154-L2190","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify the table's `TableCatalog` actually defines a watermark column; add `WITH (watermark = ...)` at creation or recreate the table/MV if it is missing.","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.","Check for version skew: if the table was created before watermark support, resnapshot/recreate the table with the current version.","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."],"exampleFix":"// before: unconditional watermark-aware iteration\nlet stream = table.iter_with_watermark(pk_prefix, sub_range).await?;\n// after: fall back when the table has no watermark serde\nlet stream = if table.has_watermark_serde() {\n    table.iter_with_watermark(pk_prefix, sub_range).await?\n} else {\n    table.iter_with_prefix(pk_prefix, sub_range).await?\n};","handlingStrategy":"fallback","validationCode":"// Before watermark-aware iteration, confirm the table actually has a watermark serde\nif table.watermark_serde.is_none() {\n    tracing::warn!(table_id = %table.table_id(), \"no watermark serde; falling back to prefix iteration\");\n}","typeGuard":"// Rust\nfn can_iter_with_watermark(table: &StateTable) -> bool {\n    table.watermark_serde.is_some()\n}","tryCatchPattern":"// Rust\nmatch table.iter_with_watermark(pk_prefix, sub_range).await {\n    Err(e) if e.to_string().contains(\"Missing watermark serde\") => {\n        // fall back to plain prefix iteration\n        table.iter_with_prefix(pk_prefix, sub_range).await.map(|s| s.boxed())\n    }\n    other => other,\n}","preventionTips":["Only enable watermark-rewrite code paths for tables created WITH (watermark = ...)","Gate executor lookups on `watermark_serde.is_some()` instead of assuming watermarks exist","Recreate pre-watermark-era tables when upgrading so their catalogs include watermark columns","Add an integration test covering watermark-free tables hit by watermark-aware executors"],"tags":["streaming","watermark","state-table","missing-config"],"backgroundTag":"missing-required-config-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}