bug: stream over window emit inconsistent changes

[st1page]

reproduce:

create table t(a int, b int, c int);
create materialized view mv as SELECT * FROM (
    SELECT
        *,
        row_number() OVER (PARTITION BY a ORDER BY b) AS rank
        FROM t
) WHERE rank <= 1;
insert into t values (1,1,1);
insert into t values (1,0,1);
insert into t values (1,0,1);

complete psql

dev=> create table t(a int, b int, c int);
create materialized view mv as SELECT * FROM (
    SELECT
        *,
        row_number() OVER (PARTITION BY a ORDER BY b) AS rank
        FROM t
) WHERE rank <= 1;
CREATE_TABLE
CREATE_MATERIALIZED_VIEW
dev=> insert into t values (1,1,1);
INSERT 0 1
dev=> insert into t values (1,0,1);
INSERT 0 1
dev=> insert into t values (1,0,1);
ERROR:  QueryError: internal error: internal error: error trying to connect: tcp connect error: Connection refused (os error 111)

explain:

dev=> explain create materialized view mv as SELECT * FROM (
    SELECT                                    
        *,
        row_number() OVER (PARTITION BY a ORDER BY b) AS rank
        FROM t                                               
) WHERE rank <= 1;
                                                                    QUERY PLAN                                                                    
--------------------------------------------------------------------------------------------------------------------------------------------------
 StreamMaterialize { columns: [a, b, c, t._row_id(hidden), rank], stream_key: [a], pk_columns: [a], pk_conflict: NoCheck }
 └─StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY t.a ORDER BY t.b ASC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] }
   └─StreamGroupTopN { order: [t.b ASC], limit: 1, offset: 0, group_key: [t.a] }
     └─StreamExchange { dist: HashShard(t.a) }
       └─StreamTableScan { table: t, columns: [a, b, c, _row_id] }
(5 rows)

log:

thread 'risingwave-streaming-actor' panicked at /home/sp/risingwave/src/stream/src/common/table/state_table.rs:697:17:
mem-table operation inconsistent! table_id: 1002, vnode: 121, key: OwnedRow([Some(Int32(1))]), prev: Insert(Ok([Some(Int32(1)), Some(Int32(0)), Some(Int32(1)), Some(Serial(Serial(324502861839073280))), Some(Int64(1))])), new: Delete(Ok([Some(Int32(1)), Some(Int32(1)), Some(Int32(1)), Some(Serial(Serial(324502809553403904))), Some(Int64(1))]))
stack backtrace:
2023-09-13T18:58:31.774286151+08:00  INFO risingwave_storage::hummock::compactor: running_task_count=0 pull_task_ack=false pending_pull_task_count=2
   0: rust_begin_unwind
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/std/src/panicking.rs:619:5
   1: core::panicking::panic_fmt
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panicking.rs:72:14
   2: risingwave_stream::common::table::state_table::StateTableInner<S,SD,_,W,_>::handle_mem_table_error
             at ./src/stream/src/common/table/state_table.rs:697:17
   3: risingwave_stream::common::table::state_table::StateTableInner<S,SD,_,W,_>::delete_inner::{{closure}}
             at ./src/stream/src/common/table/state_table.rs:728:33
   4: core::result::Result<T,E>::unwrap_or_else
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/result.rs:1429:23
   5: risingwave_stream::common::table::state_table::StateTableInner<S,SD,_,W,_>::delete_inner
             at ./src/stream/src/common/table/state_table.rs:726:9
   6: risingwave_stream::common::table::state_table::StateTableInner<S,SD,_,W,_>::write_chunk
             at ./src/stream/src/common/table/state_table.rs:865:29
   7: risingwave_stream::executor::mview::materialize::MaterializeExecutor<S,SD>::execute_inner::{{closure}}
             at ./src/stream/src/executor/mview/materialize.rs:175:29