Discussion: state cleaning based on watermark

[hzxa21]

The watermark mechanism ensures that no further upstream messages will be sent with column payload <= watermark in the watermark column. That means if the watermark column is part of the state table PK, we can safely clean the states whose column payload <= downstream watermark in the watermark column. There are two cases to consider here:

1. The watermark column is the PK prefix (| watermark_col | other_pk_col1 | other_pk_col2 | ...).
2. The watermark column is not the PK prefix (| other_pk_col1 | watermark_col | ...).

Currently we support 1 via hummock table vnode watermark but 2 is not supported so we rely on streaming to ensure that the watermark column is the PK prefix to make state cleaning effective.

However, in the following cases (cc @stdrc @st1page @yuhao-su to comfirm), streaming is not possible to put watermark column as the PK prefix but technically state cleaning should be performed:

• Interval Join. Example
• Over Window. Example
• ASOF JOIN
• MIN/MAX agg

This issue is created to capture and discuss ideas to perform state cleaning for case 2 (| other_pk_col1 | watermark_col | ...).

[hzxa21]

Ideas:

1. Implement a mechanism to periodically run delete * from __internal_xxx where watermark_col < x on the state table.

   • Pros:
     • Non-intrusive to executor and storage
   • Cons:
     • Scan the whole table periodically may waste resources and have some overheads
     • Can generate many point delete tombstones in storage, which may affect storage read performance and increase compaction load.

2. Executor maintains an index for the watermark column. Before it emits downstream watermark, the executor performs a scan on the
   index and issue point deletes to its state table for state cleaning. The implementation is similar to maintain an internal dynamic filter.

   • Pros:
     • Non-intrusive to storage
     • Easy to implement
   • Cons:
     • Happen in the critical path of streaming, which may affect freshness
     • Can generate many point delete tombstones in storage, which may affect storage read performance and increase compaction load.

3. Implement a mechanism in compaction to be aware of watermark and clean the entries below watermark asynchronously during compaction.

   • Pros:
     • Non-intrusive to executor
     • Minimize overhead on read because no point delete tombstones will be generated
     • Less waste on resources because it happens along with the regular compaction
   • Cons:
     • Since the state cleaning is asynchronous, executor needs to make sure data below watermark must not be touched or executor needs to filter data on its own.
     • Compaction needs to be more aware of the table schema and needs to do key deserialization
     • May complicate compaction strategy if the task generation needs to consider state clean watermark, in addition to the LSM tree balanceness.

Personally prefer 3 > 1 > 2

[st1page]

related discussion for interval join https://github.com/risingwavelabs/rfcs/pull/32/files#r1059266190

[st1page]

related discussion https://github.com/risingwavelabs/rfcs/pull/32/files#r1059266190

And I think the 2nd method is exactly what Flink has been done

[st1page]

Since the state cleaning is asynchronous, executor needs to make sure data below watermark must not be touched or executor needs to filter data on its own.

That is feasible

[yuhao-su]

asof join #18503

[fuyufjh]

Let's use #18802 afterwards