ColIndexMapping unable to handle duplicate output column ids

[kwannoel]

Similar to #15075, we could have the same input column mapped to 2 different output indices, if optimizer can't remove it.

In that case, our i2o mapping unable to handle it, because it can only map to 1 or no corresponding element.

/// `ColIndexMapping` is a partial mapping from usize to usize.
///
/// It is used in optimizer for transformation of column index.
#[derive(Clone, PartialEq, Eq, Hash)]
pub struct ColIndexMapping {
    /// The size of the target space, i.e. target index is in the range `(0..target_size)`.
    target_size: usize,
    /// Each subscript is mapped to the corresponding element.
    map: Vec<Option<usize>>,
}

The result is that this PR's e2e test fails: #14846

https://buildkite.com/risingwavelabs/pull-request/builds/44337

thread 'rw-streaming' panicked at src/common/src/util/column_index_mapping.rs:39:13:
assertion failed: target_max < target_size
stack backtrace:
   0: rust_begin_unwind
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/panicking.rs:645:5
   1: core::panicking::panic_fmt
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panicking.rs:72:14
   2: core::panicking::panic
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panicking.rs:144:5
   3: risingwave_common::util::column_index_mapping::ColIndexMapping::new
             at ./src/common/src/util/column_index_mapping.rs:39:13
   4: {async_fn#0}<risingwave_storage::monitor::monitored_store::MonitoredStateStore<alloc::boxed::Box<dyn risingwave_storage::store_impl::boxed_state_store::DynamicDispatchedStateStore, alloc::alloc::Global>>, risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAwareSerde, true, risingwave_stream::common::table::watermark::WatermarkBufferByEpoch<300>, false>
             at ./src/stream/src/common/table/state_table.rs:417:27
   5: {async_fn#0}<risingwave_storage::monitor::monitored_store::MonitoredStateStore<alloc::boxed::Box<dyn risingwave_storage::store_impl::boxed_state_store::DynamicDispatchedStateStore, alloc::alloc::Global>>, risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAwareSerde, risingwave_stream::common::table::watermark::WatermarkBufferByEpoch<300>, false>
             at ./src/stream/src/common/table/state_table.rs:700:96
   6: {async_fn#0}<risingwave_storage::monitor::monitored_store::MonitoredStateStore<alloc::boxed::Box<dyn risingwave_storage::store_impl::boxed_state_store::DynamicDispatchedStateStore, alloc::alloc::Global>>>
             at ./src/stream/src/from_proto/stream_scan.rs:141:21
   7: {async_fn#0}<risingwave_storage::monitor::monitored_store::MonitoredStateStore<alloc::boxed::Box<dyn risingwave_storage::store_impl::boxed_state_store::DynamicDispatchedStateStore, alloc::alloc::Global>>>
             at ./src/stream/src/from_proto/mod.rs:133:5
   8: {async_block#0}<risingwave_storage::monitor::monitored_store::MonitoredStateStore<alloc::boxed::Box<dyn risingwave_storage::store_impl::boxed_state_store::DynamicDispatchedStateStore, alloc::alloc::Global>>>

[kwannoel]

Bug found in #14846

[chenzl25]

Check this method usage for more details.

risingwave/src/common/src/util/column_index_mapping.rs

Lines 234 to 236 in e46a307

	/// Inverse the mapping. If a target corresponds to more than one source, return `None`.
	#[must_use]
	pub fn inverse(&self) -> Option<Self> {

[stdrc]

In my understanding, such column index mapping should contains source indices (or None) from an output/target perspective, so that map[0] == 1 means the output/target column 0 is from the input column 1, map[7] == None means the out column 7 is somehow generated instead of directly duplicate some input column. However, the ColIndexMapping used in optimizer seems act reversely, which I think is not quite intuitive.

I'm not sure😇