risingwave_meta/manager/iceberg_compaction/
mod.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod gc;
16mod manual;
17mod schedule;
18mod stream;
19#[cfg(test)]
20pub(crate) mod test_utils;
21
22use std::collections::{BTreeMap, HashMap, HashSet};
23use std::sync::Arc;
24use std::time::Duration;
25
26use anyhow::anyhow;
27use parking_lot::{Mutex, RwLock};
28use risingwave_common::id::WorkerId;
29use risingwave_connector::sink::SinkParam;
30use risingwave_connector::sink::catalog::{SinkCatalog, SinkId};
31use risingwave_connector::sink::iceberg::{ICEBERG_SINK, IcebergConfig};
32use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
33use risingwave_pb::iceberg_compaction::SubscribeIcebergCompactionEventRequest;
34use risingwave_pb::id::IcebergCompactionTaskId;
35use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
36use tonic::Streaming;
37
38use super::MetaSrvEnv;
39use crate::MetaResult;
40use crate::barrier::BarrierScheduler;
41use crate::hummock::IcebergCompactorManagerRef;
42use crate::manager::MetadataManager;
43use crate::rpc::metrics::MetaMetrics;
44
45pub type IcebergCompactionManagerRef = Arc<IcebergCompactionManager>;
46
47pub(crate) type CompactorChangeTx =
48    UnboundedSender<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>;
49
50pub(crate) type CompactorChangeRx =
51    UnboundedReceiver<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>;
52
53type ManualCompactionWaiter = tokio::sync::oneshot::Sender<MetaResult<IcebergCompactionTaskId>>;
54
55use schedule::CompactionTrack;
56pub use schedule::IcebergCompactionScheduleStatus;
57
58pub struct IcebergCompactionManager {
59    pub env: MetaSrvEnv,
60    inner: Arc<RwLock<IcebergCompactionManagerInner>>,
61
62    metadata_manager: MetadataManager,
63    barrier_scheduler: BarrierScheduler,
64    pub iceberg_compactor_manager: IcebergCompactorManagerRef,
65
66    compactor_streams_change_tx: CompactorChangeTx,
67
68    orphan_file_cleanup_locks: Mutex<HashMap<SinkId, Arc<tokio::sync::Mutex<()>>>>,
69
70    pub metrics: Arc<MetaMetrics>,
71}
72
73struct IcebergCompactionManagerInner {
74    sink_schedules: HashMap<SinkId, CompactionTrack>,
75    /// Recovery isolation survives discarding a track. Generations fence async config loads.
76    sink_recovery: HashMap<SinkId, SinkRecoveryState>,
77    /// Never reuse a generation, including after DROP removes a sink's recovery entry.
78    next_recovery_generation: u64,
79    snapshot_expiration_sink_ids: HashSet<SinkId>,
80    manifest_rewrite_sink_ids: HashSet<SinkId>,
81    orphan_file_cleanup_sink_ids: HashSet<SinkId>,
82    manual_compaction_waiters: HashMap<SinkId, ManualCompactionWaiter>,
83}
84
85#[derive(Default)]
86struct SinkRecoveryState {
87    generation: u64,
88    recovering: bool,
89}
90
91impl IcebergCompactionManagerInner {
92    fn next_recovery_generation(&mut self) -> u64 {
93        self.next_recovery_generation += 1;
94        self.next_recovery_generation
95    }
96
97    fn recovery_generation(&self, sink_id: SinkId) -> u64 {
98        self.sink_recovery
99            .get(&sink_id)
100            .map_or(0, |state| state.generation)
101    }
102
103    fn is_recovering(&self, sink_id: SinkId) -> bool {
104        self.sink_recovery
105            .get(&sink_id)
106            .is_some_and(|state| state.recovering)
107    }
108}
109
110impl IcebergCompactionManager {
111    fn report_timeout(&self) -> Duration {
112        Duration::from_secs(self.env.opts.iceberg_compaction_report_timeout_sec)
113    }
114
115    fn config_refresh_interval(&self) -> Duration {
116        Duration::from_secs(self.env.opts.iceberg_compaction_config_refresh_interval_sec)
117    }
118
119    pub fn build(
120        env: MetaSrvEnv,
121        metadata_manager: MetadataManager,
122        iceberg_compactor_manager: IcebergCompactorManagerRef,
123        metrics: Arc<MetaMetrics>,
124        barrier_scheduler: BarrierScheduler,
125    ) -> (Arc<Self>, CompactorChangeRx) {
126        let (compactor_streams_change_tx, compactor_streams_change_rx) =
127            tokio::sync::mpsc::unbounded_channel();
128        (
129            Arc::new(Self {
130                env,
131                inner: Arc::new(RwLock::new(IcebergCompactionManagerInner {
132                    sink_schedules: HashMap::default(),
133                    sink_recovery: HashMap::default(),
134                    next_recovery_generation: 0,
135                    snapshot_expiration_sink_ids: HashSet::default(),
136                    manifest_rewrite_sink_ids: HashSet::default(),
137                    orphan_file_cleanup_sink_ids: HashSet::default(),
138                    manual_compaction_waiters: HashMap::default(),
139                })),
140                metadata_manager,
141                barrier_scheduler,
142                iceberg_compactor_manager,
143                compactor_streams_change_tx,
144                orphan_file_cleanup_locks: Mutex::new(HashMap::default()),
145                metrics,
146            }),
147            compactor_streams_change_rx,
148        )
149    }
150
151    async fn get_sink_param(&self, sink_id: SinkId) -> MetaResult<SinkParam> {
152        let prost_sink_catalog = self
153            .metadata_manager
154            .catalog_controller
155            .get_sink_by_id(sink_id)
156            .await?
157            .ok_or_else(|| anyhow!("Sink not found: {}", sink_id))?;
158        let sink_catalog = SinkCatalog::from(prost_sink_catalog);
159        let param = SinkParam::try_from_sink_catalog(sink_catalog)?;
160        Ok(param)
161    }
162
163    async fn load_iceberg_config(&self, sink_id: SinkId) -> MetaResult<IcebergConfig> {
164        let registration = self
165            .inner
166            .read()
167            .sink_schedules
168            .get(&sink_id)
169            .and_then(|track| track.pk_index_registration().cloned());
170        if let Some(registration) = registration {
171            if !registration.is_ready() {
172                return Err(anyhow!("iceberg pk-index sink {} is not ready", sink_id).into());
173            }
174            return Ok((*registration.config()).clone());
175        }
176        let sink_param = self.get_sink_param(sink_id).await?;
177        let iceberg_config = IcebergConfig::from_btreemap(sink_param.properties)?;
178        Ok(iceberg_config)
179    }
180
181    fn orphan_file_cleanup_lock(&self, sink_id: SinkId) -> Arc<tokio::sync::Mutex<()>> {
182        self.orphan_file_cleanup_locks
183            .lock()
184            .entry(sink_id)
185            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
186            .clone()
187    }
188}
189
190/// User-created iceberg sinks have arbitrary names, so identify them by the
191/// connector property instead of the `__iceberg_sink_` name prefix.
192pub fn is_iceberg_sink(properties: &BTreeMap<String, String>) -> bool {
193    properties
194        .get(UPSTREAM_SOURCE_KEY)
195        .is_some_and(|connector| connector.eq_ignore_ascii_case(ICEBERG_SINK))
196}
risingwave_meta/manager/iceberg_compaction/mod.rs

risingwave_meta/manager/iceberg_compaction/
mod.rs
