{"record":{"id":"9dd13a84b96472e6","repo":"risingwavelabs/risingwave","slug":"watermark-cannot-be-null","errorCode":null,"errorMessage":"Watermark cannot be NULL","messagePattern":"Watermark cannot be NULL","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/common/table/state_table.rs","lineNumber":2201,"sourceCode":"\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        };\n        let watermark_row = watermark_serde.deserialize(&watermark_bytes)?;\n        if watermark_row.len() != 1 {\n            return Err(StreamExecutorError::from(format!(\n                \"Watermark row should have exactly 1 column, got {}\",\n                watermark_row.len()\n            )));\n        }\n        let watermark_value = watermark_row[0].clone();\n        // StateTableInner::update_watermark should ensure that the watermark is not NULL\n        if watermark_value.is_none() {\n            return Err(StreamExecutorError::from(anyhow!(\n                \"Watermark cannot be NULL\"\n            )));\n        }\n        let order_type = watermark_serde.get_order_types().get(0).ok_or_else(|| {\n            StreamExecutorError::from(anyhow!(\n                \"Watermark serde should have at least one order type\"\n            ))\n        })?;\n\n        let direction = if order_type.is_ascending() {\n            WatermarkDirection::Ascending\n        } else {\n            WatermarkDirection::Descending\n        };\n        let clean_watermark_index_in_pk = self\n            .pk_indices\n            .iter()\n            .position(|&i| i == clean_watermark_index);","sourceCodeStart":2183,"sourceCodeEnd":2219,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/table/state_table.rs#L2183-L2219","documentation":"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].","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check upstream watermark column source for expressions/inputs that can emit NULL and coalesce or filter them","Verify the watermark column is declared NOT NULL in the table schema","Audit the executor feeding update_watermark to ensure it never forwards a NULL watermark datum","If reproducible, file a bug with the plan/graph: NULL watermarks should be dropped upstream, not reach the state table"],"exampleFix":"// before: executor forwards datum directly\nlet watermark_row = vec![watermark_datum];\n// after: guard at the producer\nif watermark_datum.is_none() { return Ok(()); }\nlet watermark_row = vec![watermark_datum];","handlingStrategy":"validation","validationCode":"fn ensure_watermark_not_null(row: &[Datum]) -> Result<(), StreamExecutorError> {\n    match row.first() {\n        Some(Some(_)) => Ok(()),\n        _ => Err(StreamExecutorError::from(anyhow!(\"watermark datum is NULL\"))),\n    }\n}","typeGuard":"fn has_watermark(row: &[Datum]) -> bool { matches!(row.first(), Some(Some(_))) }","tryCatchPattern":null,"preventionTips":["Declare the watermark column NOT NULL","Filter NULL watermark datums in the producing executor","Add unit tests feeding NULL datums to the watermark path"],"tags":["rust","streaming","watermark","null-value"],"backgroundTag":"null-argument","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}