{"record":{"id":"9df6f68cf8d047d3","repo":"risingwavelabs/risingwave","slug":"expected-at-most-1-clean-watermark-index-per-table","errorCode":null,"errorMessage":"Expected at most 1 clean_watermark_index per table, got {:?}","messagePattern":"Expected at most 1 clean_watermark_index per table, got (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/optimizer/plan_node/stream_hash_join.rs","lineNumber":486,"sourceCode":"            if is_left && *clean_left {\n                let col_idx = pair.left_idx;\n                if !clean_watermark_indices.contains(&col_idx) {\n                    inequal_clean_watermark_indices.push(col_idx);\n                }\n            } else if !is_left && *clean_right {\n                let col_idx = pair.right_idx;\n                if !clean_watermark_indices.contains(&col_idx) {\n                    inequal_clean_watermark_indices.push(col_idx);\n                }\n            }\n        }\n\n        clean_watermark_indices.extend(inequal_clean_watermark_indices.clone());\n\n        // Verify: only 1 column per table is allowed to do state cleaning.\n        // This invariant is enforced by `derive_watermark_for_hash_join`.\n        if clean_watermark_indices.len() > 1 {\n            bail!(\n                \"Expected at most 1 clean_watermark_index per table, got {:?}\",\n                clean_watermark_indices\n            )\n        }\n\n        Ok((\n            clean_watermark_indices,\n            eq_join_key_clean_watermark_indices,\n            inequal_clean_watermark_indices,\n        ))\n    }\n\n    /// Infer which join keys can be used for state cleaning based on equal conditions.\n    fn infer_eq_join_key_clean_watermark_indices(&self, join_key_indices: &[usize]) -> Vec<usize> {\n        let mut clean_indices = vec![];\n        for (idx_in_jk, do_state_cleaning) in &self.watermark_indices_in_jk {\n            if *do_state_cleaning {\n                let col_idx = join_key_indices[*idx_in_jk];","sourceCodeStart":468,"sourceCodeEnd":504,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/optimizer/plan_node/stream_hash_join.rs#L468-L504","documentation":"In stream HashJoin watermark inference, each table may have at most one column used for state cleaning (clean_watermark_index). If the collected `clean_watermark_indices` vector has more than one entry, `infer_clean_watermark_indices` returns this error via `bail!`. The invariant is supposed to be guaranteed upstream by `derive_watermark_for_hash_join`, so this error means the join's inequality/equality conditions yielded multiple cleanable columns per table.","triggerScenarios":"Creating a `CREATE MATERIALIZED VIEW` with a stream hash join whose watermark derivation produces more than one clean_watermark_index for one side of the join, i.e. multiple inequality conditions each mapping to cleanable columns on the same table, when `derive_watermark_for_hash_join` did not collapse them.","commonSituations":"Queries with multiple non-equi join conditions on watermark columns; upgrading RisingWave after changes to watermark/now-offset derivation rules; writing SQL that combines several inequality predicates (e.g. `t1.ts > t2.ts AND t1.id > t2.id`) on streaming joins with watermarks.","solutions":["Rewrite the query so each joined table has at most one inequality/watermark-cleaning condition (drop or merge extra non-equi predicates).","Check `derive_watermark_for_hash_join` output for the failing join to see which conditions produced multiple indices.","If the invariant should hold, fix the derivation logic so it enforces at most one cleaning column per table before inference runs.","Simplify the join (e.g. pre-filter inputs or split the join) to avoid multiple cleanable columns."],"exampleFix":"-- before: multiple cleaning columns per table\nCREATE MV AS SELECT * FROM t1 JOIN t2 ON t1.ts > t2.ts AND t1.id > t2.id;\n-- after: keep only one inequality suitable for watermark cleaning\nCREATE MV AS SELECT * FROM t1 JOIN t2 ON t1.ts > t2.ts AND t1.id = t2.id;","handlingStrategy":"validation","validationCode":"-- SQL: ensure at most one non-equi (watermark-cleaning) condition per joined table\n-- Inspect the join predicates before creating the MV:\n-- SELECT pg_get_expr(ev.ev, ev.oid) FROM ...  -- or review the ON clause manually\n-- Count inequality predicates per table; rewrite if > 1 per side.","typeGuard":null,"tryCatchPattern":"// Rust caller: treat bail! result as an error, not panic\nmatch node.infer_clean_watermark_indices() {\n    Ok(indices) => proceed(indices),\n    Err(e) => return Err(context!(e, \"hash join watermark inference\")),\n}","preventionTips":["Keep at most one inequality predicate per table in streaming join conditions.","Convert additional non-equi conditions into equality or pre-filters.","After upgrading, re-derive watermarks with EXPLAIN to check cleaning columns.","Add planner tests for joins with multiple inequality conditions."],"tags":["streaming","watermark","hash-join","sql","optimizer"],"backgroundTag":"invalid-argument-value","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"}