feat(sink): introduce exactly once iceberg sink

[wcy-fdu]

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

What's changed and what's your intention?

This pr introduce exactly once consistency semantics for iceberg sink.

Related RFC: Exactly Once File Sink

implementation detail

The implementation of this feature can be mainly divided into four parts:

1. introduced a table that exists in the meta store.
2. normal write process(when no recovery occurs):

• Pre-commit the relevant metadata to the meta store before the Iceberg commit occurs. (Since the actual commit of Iceberg is completed by the coordinator after collecting all parallel tasks in the meta store, this is directly implemented within the coordinator.commit() interface.)
• Immediately delete the written metadata after the Iceberg commit returns successfully. (Since those data is successfully written into iceberg, deleting metadata is safe).

3. Recovery logic
   All logic is completed in the coordinator.init() interface when recovery occurs. This is because all memory states are cleared during recovery, and then the sink writer is built first, and then the sink writer builds the coordinator.

• check whether the pre-commit metadata recorded in the meta store < the current epoch contains any uncommitted data, and do re_commit.
  • If yes, this means that the previous commit failed, and iceberg's repeated submission is idempotent, so re-commit.
  • If no, this means that the previous commit success, just skip.
• Modify the log store rewind so that it can start rewinding from a certain epoch.
• After re-commit or skip is completed, and pass the max_epoch called last_recommit_epoch in the code, is passed to CN through the existing StartCoordinationResponse for log store rewind. The rewind here is to avoid repeated consumption of the log store, because this batch of data must have been in the downstream iceberg at this time.

test logic

The testing part is still being improved and will be continuously updated in PR.
After some offline discussions, we decided to test it this way: simulate error injection through madsim, and then sink the same batch of source data into iceberg twice, showing two iceberg tables. One of them has errors injected, and the other does not. After comparing the two tables, they are exactly the same.

Notes:

1. Since iceberg is currently hosted in RisingWave, the GC cycle is 2 weeks. This PR currently assumes that the original s3 file that failed to be committed to iceberg will not be GC before the next successful commit. In later pr, the re-commit step after recovery will be changed to rewind to the start epoch.
2. All metadata deletions happen immediately after iceberg commit successfully. correct me if I am wrong.

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

[wcy-fdu]

The chaos mesh test is still being improved. Please review the functional part first when you have time. @chenzl25 @wenym1

[hzxa21]

Left some early comments

[hzxa21]

Will passing DatabaseConnection::Disconnected cause the iceberg sink bench to fail?

[wcy-fdu]

The iceberg sink will only write to the system table in one case, that is, is_exactly_once = true is set in the create sink statement. Otherwise, this DatabaseConnection will not be used.