fix: stream group topn used wrong types for hash key (#7326)

fix #7276

Approved-By: st1page

Co-Authored-By: xxchan <[email protected]>
Co-Authored-By: st1page <[email protected]>

e2e_test/streaming/group_top_n.slt

@@ -198,3 +198,31 @@ drop materialized view mv;
 
 statement ok
 drop table bid;
+
+# https://github.com/risingwavelabs/risingwave/issues/7276
+statement ok
+CREATE TABLE t (
+    x int,
+    ts TIMESTAMP
+);
+
+statement ok
+create materialized view mv as SELECT
+    1 as x
+FROM (
+    SELECT
+    row_number() over (partition by ts order by x) as rank
+    FROM
+    t
+)
+WHERE rank <=1;
+
+statement ok
+INSERT INTO t VALUES
+(1, '2015-07-15 00:00:01');
+
+statement ok
+drop materialized view mv;
+
+statement ok
+drop table t;

src/stream/src/from_proto/group_top_n.rs

@@ -46,7 +46,10 @@ impl ExecutorBuilder for GroupTopNExecutorBuilder {
         let state_table = StateTable::from_table_catalog(table, store, vnodes).await;
         let storage_key = table.get_pk().iter().map(OrderPair::from_prost).collect();
         let [input]: [_; 1] = params.input.try_into().unwrap();
-        let group_key_types = input.schema().data_types()[..group_by.len()].to_vec();
+        let group_key_types = group_by
+            .iter()
+            .map(|i| input.schema()[*i].data_type())
+            .collect();
         let order_by = node.order_by.iter().map(OrderPair::from_prost).collect();
 
         assert_eq!(&params.pk_indices, input.pk_indices());