fix(1.5.1, over window): fix range cache when some of order by and in…

…put stream key columns are in partition by columns (#14046)

Signed-off-by: Richard Chien <[email protected]>

src/stream/src/executor/over_window/general.rs

@@ -332,6 +332,7 @@ impl<S: StateStore> OverWindowExecutor<S> {
                 &mut cache,
                 this.cache_policy,
                 &this.calls,
+                &this.partition_key_indices,
                 &this.order_key_data_types,
                 &this.order_key_order_types,
                 &this.order_key_indices,

src/stream/src/executor/over_window/over_partition.rs

@@ -243,14 +243,15 @@ impl<'a, S: StateStore> OverPartition<'a, S> {
         cache: &'a mut PartitionCache,
         cache_policy: CachePolicy,
         calls: &'a [WindowFuncCall],
+        partition_key_indices: &'a [usize],
         order_key_data_types: &'a [DataType],
         order_key_order_types: &'a [OrderType],
         order_key_indices: &'a [usize],
         input_pk_indices: &'a [usize],
     ) -> Self {
         // TODO(rc): move the calculation to executor?
         let mut projection = Vec::with_capacity(order_key_indices.len() + input_pk_indices.len());
-        let mut col_dedup = HashSet::new();
+        let mut col_dedup: HashSet<usize> = partition_key_indices.iter().copied().collect();
         for (proj_idx, key_idx) in order_key_indices
             .iter()
             .chain(input_pk_indices.iter())