Materialize executor unexpectedly compact the UPDATE and DELETE operations from CDC

[StrikeW]

Describe the bug

Materialize executor unexpectedly compact the UPDATE and DELETE operations of same pk from CDC

Error message/log

The chunk trace of the upstream UPDATE and DELETE operations.

2023-11-09T19:15:45.28921+08:00 DEBUG actor{otel.name="Actor 6" actor_id=6 prev_epoch=5393176590614528 curr_epoch=5393176656216064}:executor{otel.name="DmlExecutor 600000001 (actor 6)" actor_id=6}:executor{otel>
+---+-----+------+-----------------------+
|   | id  | name | description           |
+---+-----+------+-----------------------+
| + | 101 | RW   | Small 2-wheel scooter |
| - | 101 | RW   | Small 2-wheel scooter |
+---+-----+------+-----------------------+
 cardinality=2 capacity=2
2023-11-09T19:15:45.289386+08:00 DEBUG actor{otel.name="Actor 6" actor_id=6 prev_epoch=5393176590614528 curr_epoch=5393176656216064}:executor{otel.name="DmlExecutor 600000001 (actor 6)" actor_id=6}: events::str>
+---+-----+------+-----------------------+
|   | id  | name | description           |
+---+-----+------+-----------------------+
| + | 101 | RW   | Small 2-wheel scooter |
| - | 101 | RW   | Small 2-wheel scooter |
+---+-----+------+-----------------------+
 cardinality=2 capacity=2
2023-11-09T19:15:45.289835+08:00 DEBUG actor{otel.name="Actor 2" actor_id=2 prev_epoch=5393176590614528 curr_epoch=5393176656216064}:executor{otel.name="Materialize 200000003 (actor 2)" actor_id=2}:executor{ote>
+---+--------------------+----------------------+-----------------------------+
|   | mytest.products.id | mytest.products.name | mytest.products.description |
+---+--------------------+----------------------+-----------------------------+
| + | 101                | RW                   | Small 2-wheel scooter       |
| - | 101                | RW                   | Small 2-wheel scooter       |
+---+--------------------+----------------------+-----------------------------+

To Reproduce

1. create a table in mysql and insert one row

mysql> select * from products;
+-----+---------+-----------------------+
| id  | name    | description           |
+-----+---------+-----------------------+
| 101 | scooter | Small 2-wheel scooter |
+-----+---------+-----------------------+

3. Ingest the table into RW

create table products ( id INT,
 name STRING,
 description STRING,
 PRIMARY KEY (id)
) with (
 connector = 'mysql-cdc',
 hostname = 'localhost',
 port = '8306',
 username = 'dbz',
 password = '123456',
 database.name = 'mytest',
 table.name = 'products',
 server.id = '5085'
);

4. Shutdown the cluster
5. Execute following commands in mysql

UPDATE products SET name='RW' where id=101;
DELETE from products where id=101;

6. Restart the cluster
7. Select the table in RW

The expected result is the row should be deleted from RW, but it still exist.

Expected behavior

The expected result is the row should be deleted from RW, but it still exist.

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[wcy-fdu]

cc @st1page

[BugenZhao]

This is really a severe fault. Chaos unit test might be a good-to-have if code review cannot be enforced.

[st1page]

It is because we use a MaterializeBuffer to compact the input chunk and because the upsert stream does not obey the associative property, the compact operation with the retractable stream rules is illegal. e.g.
INSERT 1, DELETE 1 can be compacted into No-op. but UPSERT 1, DELETE 1 can not because we can not know if the record that should be deleted is inserted by the upsert operation.
I think we should remove the MaterializeBuffer structure but just leave a HashSet to help to get all the keys in the streamChunk and fetch them into the cache. And then we need to do all real operations in the state cache.

[fuyufjh]

closed by #13351