feat(subscription): support blocking cursor

[xxhZs]

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

What's changed and what's your intention?

#18208

In this pr, we support blocking subscription cursor and all cursor timeout.
the stmt of our fetch cursor becomes
fetch next/num from cursor_name with (timeout = 'xx');

about subscription cursor's timeout :

1. When there are a lot of values in the subscription and the fetch is set to a large number of rows, the fetch statement may take longer than timeout time to execute. This time we will returns all the values currently fetched.
2. When subscription has no new data at the moment and fetch is used, there are two scenarios

• If you have already fetched more than one row, the fetch statement does not block, but returns all the values currently fetched.
• If there is currently no value, we block until there is a new value, or until the blocking time exceeds timeout

about query cursor's timeout :
only support 1

When with (timeout = 'xx') is not set,
The query timeout is u64::max
The block timeout is 0

timeout is a string in interval format.

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

Support
fetch next/num from cursor_name with (timeout = 'xx');
About subscription cursor's timeout :

1. When there are a lot of values in the subscription and the fetch is set to a large number of rows, the fetch statement may take longer than timeout time to execute. This time we will returns all the values currently fetched.
2. When subscription has no new data at the moment and fetch is used, there are two scenarios

• If we have already fetched more than one row, the fetch statement does not block, but returns all the values currently fetched.
• If there is currently no value, we block until there is a new value, or until the blocking time exceeds timeout

About query cursor's timeout :
only support 1

When with (timeout = 'xx') is not set,
In the case of 1, this corresponds to timeout = u64::max
In the case of 2, this corresponds to timeout = 0

timeout is a string in interval format.

[hzxa21]

Currently one CursorNotifier per session and within each CursorNotifier, we have one channel per table. Assume we have N sessions and M tables in total, we can end up creating N*M channels. Not to mention that we have a task in CursorNotifier to maintain the per table channel. These seem inefficient and complicated. IMO, a better way to implement the notification is:

• Create one and only one watch channel to listen on table changed log updates.
• When cursor needs to wait for changelog epoch changes for a table, it only needs to subscribe to this channel.

Implementation-wise, maybe we can store HummockSnapshotManager in each cursor instance and implement a wait_for_epoch similar to the method here.

[wenym1]

Actually, I think the current cursor implementation in frontend can be greatly simplified. The most complicated part in the current cursor is the state machine. The async feature of rust is a very powerful tool to generate state machine. So for a cursor, it can be implemented as a stream

#[try_stream(...)]
async fn cursor_stream(...) {
    if need_snapshot {
         for row in create_snapshot_iter(table_id) { yield row; }
    }
    let mut current_epoch = ...;
    loop {
        for row in change_log(current_epoch, table_id) { yield row; }
        current_epoch = snapshot_manager.wait_next_epoch(current_epoch, table_id);
    }
}

When a cursor is created, we can create such stream and hold it in the session. On a fetch, we just call cursor_stream.next().await, and the timeout can be applied with simply timeout(timeout_duration, cursor_stream.next()).await.

We can be aware of table drop in snapshot_manager, when we see that the table_id does not exist in the state_table_info of FrontendHummockVersion anymore, and then changes on catalog in this PR can be totally avoided.

[hzxa21]

Does None here always mean no data? I saw that we will return None when field mismatches here.

[xxhZs]

Yes, none means that there is no data for this call to next_row

[wenym1]

Actually, I think the current cursor implementation in frontend can be greatly simplified. The most complicated part in the current cursor is the state machine. The async feature of rust is a very powerful tool to generate state machine. So for a cursor, it can be implemented as a stream

#[try_stream(...)]
async fn cursor_stream(...) {
    if need_snapshot {
         for row in create_snapshot_iter(table_id) { yield row; }
    }
    let mut current_epoch = ...;
    loop {
        for row in change_log(current_epoch, table_id) { yield row; }
        current_epoch = snapshot_manager.wait_next_epoch(current_epoch, table_id);
    }
}

When a cursor is created, we can create such stream and hold it in the session. On a fetch, we just call cursor_stream.next().await, and the timeout can be applied with simply timeout(timeout_duration, cursor_stream.next()).await.

We can be aware of table drop in snapshot_manager, when we see that the table_id does not exist in the state_table_info of FrontendHummockVersion anymore, and then changes on catalog in this PR can be totally avoided.

[hzxa21]

LGTM