{"record":{"id":"e6d1cd988156809d","repo":"risingwavelabs/risingwave","slug":"watermark-column-is-expected-to-be-non-null","errorCode":null,"errorMessage":"watermark column is expected to be non-null","messagePattern":"watermark column is expected to be non-null","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/eowc/sort_buffer.rs","lineNumber":53,"sourceCode":"\nuse crate::common::state_cache::{StateCache, StateCacheFiller, TopNStateCache};\nuse crate::common::table::state_table::StateTable;\nuse crate::executor::{StreamExecutorError, StreamExecutorResult};\n\ntype CacheKey = (\n    DefaultOrdered<ScalarImpl>, // sort (watermark) column value\n    MemcmpEncoded,              // memcmp-encoded pk\n);\n\nfn row_to_cache_key<S: StateStore>(\n    sort_column_index: usize,\n    row: impl Row,\n    buffer_table: &StateTable<S>,\n) -> CacheKey {\n    let timestamp_val = row\n        .datum_at(sort_column_index)\n        .to_owned_datum()\n        .expect(\"watermark column is expected to be non-null\");\n    let mut pk = vec![];\n    buffer_table\n        .pk_serde()\n        .serialize((&row).project(buffer_table.pk_indices()), &mut pk);\n    (timestamp_val.into(), pk.into())\n}\n\n// TODO(rc): need to make this configurable?\nconst CACHE_CAPACITY: usize = 2048;\n\n/// [`SortBuffer`] is a common component that consume an unordered stream and produce an ordered\n/// stream by watermark. This component maintains a buffer table passed in, whose schema is same as\n/// [`SortBuffer`]'s input and output. Generally, the component acts as a buffer that output the\n/// data it received with a delay, commonly used to implement emit-on-window-close policy.\npub struct SortBuffer<S: StateStore> {\n    /// The timestamp column to sort on.\n    sort_column_index: usize,\n","sourceCodeStart":35,"sourceCodeEnd":71,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/eowc/sort_buffer.rs#L35-L71","documentation":"`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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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.","Declare the timestamp column NOT NULL (or use a NOT NULL source schema) so nulls are rejected at ingest time.","Fill missing timestamps at the source side (default to processing time or a sentinel) before they reach the sort buffer.","Inspect incoming records to find which producer emits null timestamps and fix it there."],"exampleFix":"// before: nullable event time flows into EOWC MV\nCREATE MATERIALIZED VIEW mv AS SELECT ts, ... FROM src;\n// after\nCREATE MATERIALIZED VIEW mv AS SELECT ts, ... FROM src WHERE ts IS NOT NULL;","handlingStrategy":"validation","validationCode":"-- Ensure the sort/watermark column has no NULLs before it reaches the EOWC executor:\nSELECT COUNT(*) FROM src WHERE ts_col IS NULL; -- must be 0\n-- Or enforce upstream:\nCREATE MATERIALIZED VIEW clean AS SELECT * FROM src WHERE ts_col IS NOT NULL;","typeGuard":"fn watermark_present(row: &impl Row, idx: usize) -> bool {\n    row.datum_at(idx).is_some()\n}","tryCatchPattern":null,"preventionTips":["Declare the event-time column NOT NULL in the source schema.","Filter `WHERE ts IS NOT NULL` in the upstream MV.","Backfill or default missing timestamps at the producer side.","Monitor source data quality for null timestamp fields."],"tags":["streaming","null-value","eowc","watermark","panic"],"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"}