fix(stream): materialize should not compact input when handling conflict

[st1page]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

fix #13346

          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.

Originally posted by @st1page in #13346 (comment)

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[st1page]

U- 8 1
U+ 8 2
+  8 3

The conflict behavior is ignore. Quesiton is if we need to consider the update as a whole operation. The U- 8 1 is trying to delete a non-existent record, so it is ignored. But if the following U+ 8 2 should be ignored too?

[st1page]

            " i i
            + 1 3
            + 1 4

I think it should be 3 for a ignored behavior?

[codecov]

Codecov Report

Merging #13351 (b237a0c) into main (492076f) will decrease coverage by 0.07%.
Report is 3 commits behind head on main.
The diff coverage is 93.23%.

@@            Coverage Diff             @@
##             main   #13351      +/-   ##
==========================================
- Coverage   67.81%   67.74%   -0.07%     
==========================================
  Files        1524     1524              
  Lines      259436   259469      +33     
==========================================
- Hits       175934   175787     -147     
- Misses      83502    83682     +180     

Flag	Coverage Δ
rust	67.74% <93.23%> (-0.07%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
src/stream/src/executor/mview/materialize.rs	90.04% <93.23%> (+0.12%)	⬆️

... and 21 files with indirect coverage changes

📣 Codecov offers a browser extension for seamless coverage viewing on GitHub. Try it in Chrome or Firefox today!

[BugenZhao]

Would you mind writing some tests?

[wcy-fdu]

LGTM! Thanks for the quick fix~

[BugenZhao]

Can we leverage the snapshot test here?

risingwave/src/stream/tests/integration_tests/snapshot.rs

Lines 85 to 88 in b7c57d0

	/// Similar to [`check_until_pending`], but use a DSL test script as input.
	///
	/// For each input event, it drives the executor until it is pending.
	pub async fn check_with_script<F, Fut>(

[st1page]

Yes, I will take a look later