fix: stream group topn used wrong types for hash key

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/common/src/hash/dispatcher.rs

@@ -64,7 +64,9 @@ pub trait HashKeyDispatcher: Sized {
     fn data_types(&self) -> &[DataType];
 
     fn dispatch(self) -> Self::Output {
-        match calc_hash_key_kind(self.data_types()) {
+        let k = calc_hash_key_kind(self.data_types());
+        tracing::debug!("dispatch {:?} to {:?}", self.data_types(), k);
+        match k {
             HashKeyKind::Key8 => self.dispatch_impl::<hash::Key8>(),
             HashKeyKind::Key16 => self.dispatch_impl::<hash::Key16>(),
             HashKeyKind::Key32 => self.dispatch_impl::<hash::Key32>(),

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());