fix(stream): clear upstream buffer if no snapshot progress

[kwannoel]

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

What's changed and what's your intention?

As per title. Thanks to @Little-Wallace for discovering and reporting this bug. Found in #11398.

The observation was that multiple rows with the same PK were inserted into upstream mview.
Consider the following sequence of events:

1. Upstream update for row with pk=123. -------> Upstream buffer contains pk=123
2. Snapshot read no progress in this epoch.
3. Barrier (and upstream buffer not cleaned)
4. new snapshot with current_pos=None. Pk will be at least 123.
5. snapshot read with 123, flush pk=123 to downstream.
6. No more snapshot.
7. Flush remaining rows in upstream buffer. ---------> Upstream buffer contains 123, so we flush it twice.

Then we will have inconsistent table.

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 contains user-facing changes.

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• Installation and deployment
• Connector (sources & sinks)
• SQL commands, functions, and operators
• RisingWave cluster configuration changes
• Other (please specify in the release note below)

Release note

[chenzl25]

LGTM

[codecov]

Codecov Report

Merging #11625 (0e9ca49) into main (f4388cc) will decrease coverage by 0.01%.
Report is 5 commits behind head on main.
The diff coverage is n/a.

@@            Coverage Diff             @@
##             main   #11625      +/-   ##
==========================================
- Coverage   70.30%   70.30%   -0.01%     
==========================================
  Files        1359     1359              
  Lines      227141   227141              
==========================================
- Hits       159694   159690       -4     
- Misses      67447    67451       +4     

Flag	Coverage Δ
rust	70.30% <ø> (-0.01%)	⬇️

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

Files Changed	Coverage Δ
...tream/src/executor/backfill/no_shuffle_backfill.rs	0.00% <ø> (ø)

... and 5 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more