fn resolve_clean_watermark_indices_in_jk( desc: Option<&HashJoinWatermarkHandleDesc>, join_key_len: usize, ) -> Vec<(usize, bool)>