feat(storage): support non_pk_prefix_watermark state cleaning

[Li0k]

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

What's changed and what's your intention?

related to #18802

This PR supports non_pk_prefix_watermark state cleaning for Hummock.

Since non_pk_prefix_watermark relies on catalogs, this introduces additional overhead. Therefore, this PR does not guarantee read filtering for non_pk_prefix_watermark and only handles expired data.

The changes are as follows:

1. watermarks of type non_pk_prefix_watermark are not added to ReadWatermarkIndex.

2. state table support to write non_pk_prefix_watermark and serialize.

3. compaction catalog agent support to get watermark serde

4. skip watermark iterator supports filtering non_pk_prefix_watermark.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

[hzxa21]

We already have a WaterMarkType define in the version, why don't we just use that to filter out table with non pk prefix watermark?

[hzxa21]

Also, if we filter out non pk prefix watermark here, how can compactor retrieve the non pk prefix watermark? Based on the logic here, it seems that we rely on the fact that non pk prefix watermark is present in the compact task.

[Li0k]

Good catch , we should filter the watermark by WaterMarkType directly.

And, the filtered results are only passed to the picker, while all relevant watermarks are passed to the compactor (pk or non-pk).

[hzxa21]

nits: since SkipWatermarkIterator is only used by compactor, how about moving skip_watermark.rs into src/hummock/compactor?

[Li0k]

Of course, I will propose a separate pr for it

[hzxa21]

IIUC, if cmp_datum returns Euqal | Greater, based on the logic in L360, the watermark will be advanced. I think this is incorrect for non pk prefix watermark because the non pk prefix watermark and the pk doesn't have the same ordering.

[hzxa21]

Actually why don't we do the filtering inside the picker instead like in here if the watermark type is part of TableWatermarks:

risingwave/src/meta/src/hummock/compaction/selector/vnode_watermark_selector.rs

Line 53 in 5ed4920

let table_watermarks =

We can avoid cloning the table watermark, which can be large given that it stores bytes from user data, with no harm.

[hzxa21]

I am afraid this is still incorrect based on the semantic of advance_watermark:

/// Return a flag indicating whether the current key will be filtered by the current watermark.

If we always return false when the table, vnode are the same here, that means none of the keys can be filtered by the watermark. Please clearfully walk through the logics of advance_watermark, should_delete and advance_key_and_watermark. I am still concerned that the implementation of SkipWatermarkState and SkipWatermarkIterator rely on the assumption that the key ordering and watermark ordering is the same and we may still miss some changes.