union operator causes kafka upsert sink to produce inconsistent results

[ly9chee]

Describe the bug

recently, I discovered that some records from a table in starrocks have been deleted but still exist in risingwave. the corresponding mv only do a simple union operation, and there are no delete operations on either side of union operator.

Error message/log

No response

To Reproduce

create table category_level1(
  id bigint primary key,
  name varchar
);

create table category_level2(
  id bigint primary key,
  name varchar
);

insert into category_level1 values (1, '服装');
insert into category_level2 values (2, '女士');

create materialized view category_full as
select *, '1' as typ from category_level1
union
select *, '2' as typ from category_level2;

create sink sink_category_full from category_full
with (
  connector='kafka',
  type='upsert',
  topic='rw_test',
  properties.bootstrap.server='10.100.0.35:9092',
  primary_key='id'
);

-- the following statement will send two out-of-order upsert messages
update category_level1 set name = '酒类';

./bin/kafka-console-consumer.sh --bootstrap-server 10.100.0.35:9092 --topic rw_test --property "print.key=true" --offset latest --partition 0
{"id":1}	{"typ":"1","id":1,"name":"服装"}
{"id":2}	{"typ":"2","id":2,"name":"女士"}
{"id":1}	{"typ":"1","id":1,"name":"酒类"}
{"id":1}	null

Expected behavior

{"id":1}	{"typ":"1","id":1,"name":"服装"}
{"id":2}	{"typ":"2","id":2,"name":"女士"}
{"id":1}	{"typ":"1","id":1,"name":"酒类"}

or

{"id":1}	{"typ":"1","id":1,"name":"服装"}
{"id":2}	{"typ":"2","id":2,"name":"女士"}
{"id":1}	null
{"id":1}	{"typ":"1","id":1,"name":"酒类"}

How did you deploy RisingWave?

docker compose

The version of RisingWave

PostgreSQL 9.5-RisingWave-1.2.0 (0f94fdf)

Additional context

switching from union to union all this problem disappeared, but union all does not have the deduplication capability according to the doc

[BugenZhao]

Same as #12026 (comment) ?