feat(backfill): do not require at least 1 record to be read per epoch, if rate limit enabled

[kwannoel]

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

What's changed and what's your intention?

We have a requirement to read at least 1 record per barrier from snapshot for backfill.

This means we are forced to have the following distribution in our stream:

barrier -> record -> barrier -> record -> barrier

Which means the barrier latency is tied to processing times for the records.
In cases where the processing time is long, e.g. UDF with high latencies, the barrier latency will spike as well.
In the associated test case, we show that when UDF calls have 5s latency, the test should take a long time to complete.
This occurs even when rate limit set, disrupting its usefulness.

The solution is to check if rate limit can currently read a record, before applying the read at least 1 record per epoch step.
If it can't, just skip over. Eventually we will get to read a record.
If it can, then continue to apply this step.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• 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.

[fuyufjh]

I didn't get it.

The motivation of requiring at least 1 record to be read per epoch is to prevent the livelock of scanning from storage #12780

This is so we don't lose tombstone iteration progress for that epoch.

The problem exists regardless of whether rate limit was enabled or not. They are unrelated. Don't understand why associating them together.

[fuyufjh]

Let me guess. Suppose you met a case that even 1 record per epoch is unacceptable. If so, I believe we should find a better implementation for #12780, such as holding a delete-tomb-aware iterator, or something like that.

Fixing a new problem while creating an known problem sounds bad to me.

[kwannoel]

Let me guess. Suppose you met a case that even 1 record per epoch is unacceptable. If so, I believe we should find a better implementation for #12780, such as holding a delete-tomb-aware iterator, or something like that.

Fixing a new problem while creating an known problem sounds bad to me.

This PR does not lead to any regression. Consider the case when we don't read at least 1 record. This is when rate limiter tells us that we have hit the snapshot read rate limit, and should not read a new record.

That's totally fine, because subsequently, when rate limit capacity frees up, we can apply this requirement of 1 record per epoch again. The key thing is that we will always read at least one row, in the rate limit duration, we didn't get rid of this requirement, just generalized it from one row per epoch.

1 record per epoch is unacceptable in cases where the latency of processing a single record is high. If we apply rate limit to this scenario, it still can't change the distribution of the stream.
After this PR however, we can make the distribution like so instead if rate limit is set:

barrier -> record -> barrier -> barrier

Frequency of record now respects rate limit, rather than being forced to be at least 1 per barrier.
At the same time, we eventually read at least 1 record, 
on an epoch when there's free capacity in the rate limiter.

The provided test case will take an extremely long time without the fix in this PR.

For sure we can have this: such as holding a delete-tomb-aware iterator, or something like that.
But this is much more complex than the current solution, which already fixes the issue without causing any regression.

[lmatz]

Does 1.9.0 need to war for this or included in minor version is also ok?

[kwannoel]

Does 1.9.0 need to war for this or included in minor version is also ok?

Minor version is ok

[chenzl25]

LGTM. IIUC, the rate limiter would always be checked before the at least 1 row per barrier read, so only when rate_limit = 0 will stop the at least 1 row per barrier read. I think it is acceptable, because rate_limit = 0 means we want to throttle the streaming DAG entirely.