feat(compaction): introduce iceberg compaction pull strategy #21824
Conversation
There was a problem hiding this comment.
Pull Request Overview
This pull request introduces an iceberg compaction pull strategy to support and trigger iceberg compaction tasks based on commit information.
- Updates protobuf build configuration to include the new IcebergCompactionTask type.
- Refactors the IcebergCompactionManager to record commit counts alongside timing data and to provide a method for retrieving top sink IDs based on commit activity.
- Updates the compaction event handler in the hummock module to asynchronously pull tasks and dispatch new iceberg compaction events.
Reviewed Changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| src/prost/build.rs | Added the new IcebergCompactionTask type for protobuf generation. |
| src/meta/src/manager/iceberg_compaction.rs | Introduced a CommitInfo struct and updated commit recording and retrieval logic with sorting based on commit count and time. |
| src/meta/src/hummock/manager/compaction/compaction_event_loop.rs | Converted the pull task handling to async and integrated the new top-N logic for iceberg compaction events. |
Comments suppressed due to low confidence (2)
src/meta/src/manager/iceberg_compaction.rs:126
- The sorting logic for commit times is unclear; using duration_since with a zero duration may not produce the intended order. Consider comparing the Instants directly or computing durations relative to a common reference for a more natural ordering.
b.1.first_commit_time.duration_since(a.1.first_commit_time).cmp(&std::time::Duration::from_secs(0))
src/meta/src/hummock/manager/compaction/compaction_event_loop.rs:646
- [nitpick] Using .expect here may lead to runtime panics if sink parameters are missing; consider handling the error gracefully to improve robustness.
.await.expect("sink params not found")
| const MIN_SINK_COMMIT_INTERVAL: u64 = 3600; // 1 hour | ||
|
|
||
| sink_ids | ||
| .iter() | ||
| .max_by_key(|&(_, count)| count) | ||
| .map(|(sink_id, _)| *sink_id) | ||
| .filter(|(_, commit_info)| { | ||
| Instant::now().duration_since(commit_info.first_commit_time) | ||
| > std::time::Duration::from_secs(MIN_SINK_COMMIT_INTERVAL) |
There was a problem hiding this comment.
After the initial 1 hour is passed since the sink first commit, are we going to aggressively trigger compaction? I thought we will still trigger compaction based on some interval instead of every time get_top_n_iceberg_commit_sink_ids is called..
There was a problem hiding this comment.
You’re right, so we need to actively clean up this record after sending the msg to avoid repeated execution. I don’t think this is very intuitive.
I also considered recording last_compaction_ts, but this still depends on the external call.
…nto li0k/iceberg_pull_task
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
introduce iceberg compaction pull strategy base on
CommitInfohandle_pull_task_eventfunction forIcebergCompactionEventDispatchertips: Whether compaction is triggered needs to be supported by
sink-param. The pr will be initiated to support compaction settings per sinkChecklist
Documentation
Release note