stream_key not equal to downstream primary key even though they should be the same

[lmatz]

Describe the bug

dev=> set streaming_parallelism = 1;
SET_VARIABLE
dev=> CREATE TABLE IF NOT EXISTS agg (user_id VARCHAR PRIMARY KEY) ON CONFLICT DO UPDATE IF NOT NULL;   
CREATE_TABLE
dev=> create table t(user_id varchar);                                                                                
CREATE_TABLE
dev=> create sink s into agg as select user_id from t group by user_id with ( type = 'append-only',  force_append_only = 'true',);
CREATE_SINK

dev=> explain create sink s into agg_user_insights as select user_id from t group by user_id with ( type = 'append-only',  force_append_only = 'true',);
                             QUERY PLAN                              
---------------------------------------------------------------------
 StreamProject { exprs: [t.user_id] }
 └─StreamSink { type: append-only, columns: [user_id] }
   └─StreamProject { exprs: [t.user_id] }
     └─StreamHashAgg { group_key: [t.user_id], aggs: [count] }
       └─StreamExchange { dist: HashShard(t.user_id) }
         └─StreamTableScan { table: t, columns: [user_id, _row_id] }
(6 rows)

I guess "StreamProject" is the issue here?

which leads to unnecessary sink buffering.

Error message/log

No response

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response