Repository: risingwavelabs/risingwave
File: src/frontend/src/optimizer/plan_node/stream_asof_join.rs (Rust)

Code before:

        let watermark_columns = {
            let l2i = core.l2i_col_mapping();
            let r2i = core.r2i_col_mapping();
            let mut watermark_columns = WatermarkColumns::new();
            for (left_idx, right_idx) in
                eq_join_predicate
                    .eq_indexes()
                    .into_iter()
                    .chain(std::iter::once((
                        inequality_desc.left_idx as usize,
                        inequality_desc.right_idx as usize,
                    )))
            {
                if let Some(l_wtmk_group) = core.left.watermark_columns().get_group(left_idx)
                    && let Some(r_wtmk_group) = core.right.watermark_columns().get_group(right_idx)
                {
                    if let Some(internal) = l2i.try_map(left_idx) {
                        watermark_columns.insert(
                            internal,
                            l_wtmk_group
                                .same_or_else(r_wtmk_group, || ctx.next_watermark_group_id()),
                        );
                    }
                    if let Some(internal) = r2i.try_map(right_idx) {
                        watermark_columns.insert(
                            internal,
                            l_wtmk_group
                                .same_or_else(r_wtmk_group, || ctx.next_watermark_group_id()),
                        );
                    }
                }
            }
            watermark_columns.map_clone(&core.i2o_col_mapping())
        };

        // TODO: derive from input
        let base = PlanBase::new_stream_with_core(
            &core,
            dist,
            stream_kind,
            false, // TODO(rc): derive EOWC property from input
            watermark_columns,
            MonotonicityMap::new(), // TODO: derive monotonicity
        );

        Ok(Self {
            base,
            core,
            eq_join_predicate,
            is_append_only: stream_kind.is_append_only(),

Review comment (suggestion):
The watermark group ID should be computed once and reused for both left and right column insertions. Currently, `same_or_else` is called twice (lines 99-100 and 106-107), which means if `l_wtmk_group != r_wtmk_group`, `ctx.next_watermark_group_id()` will be invoked twice, creating two different watermark group IDs for columns that should be in the same group.

Suggested fix:
```rust
let group_id = l_wtmk_group
    .same_or_else(r_wtmk_group, || ctx.next_watermark_group_id());
if let Some(internal) = l2i.try_map(left_idx) {
    watermark_columns.insert(internal, group_id);
}
if let Some(internal) = r2i.try_map(right_idx) {
    watermark_columns.insert(internal, group_id);
}
```
```suggestion
                    let group_id = l_wtmk_group
                        .same_or_else(r_wtmk_group, || ctx.next_watermark_group_id());
                    if let Some(internal) = l2i.try_map(left_idx) {
                        watermark_columns.insert(internal, group_id);
                    }
                    if let Some(internal) = r2i.try_map(right_idx) {
                        watermark_columns.insert(internal, group_id);
```

Code after:
                eq_join_predicate
                    .eq_indexes()
                    .into_iter()
                    .chain(std::iter::once((
                        inequality_desc.left_idx as usize,
                        inequality_desc.right_idx as usize,
                    )))
            {
                if let Some(l_wtmk_group) = core.left.watermark_columns().get_group(left_idx)
                    && let Some(r_wtmk_group) = core.right.watermark_columns().get_group(right_idx)
                {
                    if let Some(internal) = l2i.try_map(left_idx) {
                        watermark_columns.insert(
                            internal,
                            l_wtmk_group.same_or_else(r_wtmk_group, || {
                                core.ctx().next_watermark_group_id()
                            }),
                        );
                    }
                    if let Some(internal) = r2i.try_map(right_idx) {
                        watermark_columns.insert(
                            internal,
                            l_wtmk_group.same_or_else(r_wtmk_group, || {
                                core.ctx().next_watermark_group_id()
                            }),
                        );
                    }
                }
            }
            watermark_columns.map_clone(&core.i2o_col_mapping())
        };

        // TODO: derive from input
        let base = PlanBase::new_stream_with_core(
            &core,
            dist,
            stream_kind,
            false, // TODO(rc): derive EOWC property from input
            watermark_columns,
            MonotonicityMap::new(), // TODO: derive monotonicity
        );

        Ok(Self {
            base,
            core,
            eq_join_predicate,
            is_append_only: stream_kind.is_append_only(),
            inequality_desc,
        })
    }