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
20use std::collections::{BTreeMap, HashMap, HashSet};
21use std::sync::Arc;
22use std::time::Duration;
23
24use anyhow::anyhow;
25use parking_lot::RwLock;
26use risingwave_common::id::WorkerId;
27use risingwave_connector::sink::SinkParam;
28use risingwave_connector::sink::catalog::{SinkCatalog, SinkId};
29use risingwave_connector::sink::iceberg::{ICEBERG_SINK, IcebergConfig};
30use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
31use risingwave_pb::iceberg_compaction::SubscribeIcebergCompactionEventRequest;
32use risingwave_pb::id::IcebergCompactionTaskId;
33use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
34use tonic::Streaming;
35
36use super::MetaSrvEnv;
37use crate::MetaResult;
38use crate::controller::streaming_job::AbortCreatingJobResult;
39use crate::hummock::IcebergCompactorManagerRef;
40use crate::manager::MetadataManager;
41use crate::rpc::metrics::MetaMetrics;
42
43pub type IcebergCompactionManagerRef = Arc<IcebergCompactionManager>;
44
45pub(crate) type CompactorChangeTx =
46    UnboundedSender<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>;
47
48pub(crate) type CompactorChangeRx =
49    UnboundedReceiver<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>;
50
51type ManualCompactionWaiter = tokio::sync::oneshot::Sender<MetaResult<IcebergCompactionTaskId>>;
52
53use schedule::CompactionTrack;
54pub use schedule::IcebergCompactionScheduleStatus;
55
56pub struct IcebergCompactionManager {
57    pub env: MetaSrvEnv,
58    inner: Arc<RwLock<IcebergCompactionManagerInner>>,
59
60    metadata_manager: MetadataManager,
61    pub iceberg_compactor_manager: IcebergCompactorManagerRef,
62
63    compactor_streams_change_tx: CompactorChangeTx,
64
65    pub metrics: Arc<MetaMetrics>,
66}
67
68struct IcebergCompactionManagerInner {
69    sink_schedules: HashMap<SinkId, CompactionTrack>,
70    snapshot_expiration_sink_ids: HashSet<SinkId>,
71    manifest_rewrite_sink_ids: HashSet<SinkId>,
72    manual_compaction_waiters: HashMap<SinkId, ManualCompactionWaiter>,
73}
74
75impl IcebergCompactionManager {
76    fn report_timeout(&self) -> Duration {
77        Duration::from_secs(self.env.opts.iceberg_compaction_report_timeout_sec)
78    }
79
80    fn config_refresh_interval(&self) -> Duration {
81        Duration::from_secs(self.env.opts.iceberg_compaction_config_refresh_interval_sec)
82    }
83
84    pub fn build(
85        env: MetaSrvEnv,
86        metadata_manager: MetadataManager,
87        iceberg_compactor_manager: IcebergCompactorManagerRef,
88        metrics: Arc<MetaMetrics>,
89    ) -> (Arc<Self>, CompactorChangeRx) {
90        let (compactor_streams_change_tx, compactor_streams_change_rx) =
91            tokio::sync::mpsc::unbounded_channel();
92        (
93            Arc::new(Self {
94                env,
95                inner: Arc::new(RwLock::new(IcebergCompactionManagerInner {
96                    sink_schedules: HashMap::default(),
97                    snapshot_expiration_sink_ids: HashSet::default(),
98                    manifest_rewrite_sink_ids: HashSet::default(),
99                    manual_compaction_waiters: HashMap::default(),
100                })),
101                metadata_manager,
102                iceberg_compactor_manager,
103                compactor_streams_change_tx,
104                metrics,
105            }),
106            compactor_streams_change_rx,
107        )
108    }
109
110    async fn get_sink_param(&self, sink_id: SinkId) -> MetaResult<SinkParam> {
111        let prost_sink_catalog = self
112            .metadata_manager
113            .catalog_controller
114            .get_sink_by_id(sink_id)
115            .await?
116            .ok_or_else(|| anyhow!("Sink not found: {}", sink_id))?;
117        let sink_catalog = SinkCatalog::from(prost_sink_catalog);
118        let param = SinkParam::try_from_sink_catalog(sink_catalog)?;
119        Ok(param)
120    }
121
122    async fn load_iceberg_config(&self, sink_id: SinkId) -> MetaResult<IcebergConfig> {
123        let sink_param = self.get_sink_param(sink_id).await?;
124        let iceberg_config = IcebergConfig::from_btreemap(sink_param.properties)?;
125        Ok(iceberg_config)
126    }
127
128    /// Clear the iceberg maintenance state of the sink aborted by
129    /// `try_abort_creating_streaming_job`, if any.
130    pub fn clear_maintenance_for_aborted_job(&self, abort_result: &AbortCreatingJobResult) {
131        if let Some(sink_id) = abort_result.aborted_sink_id {
132            self.clear_iceberg_maintenance_by_sink_id(sink_id);
133        }
134    }
135}
136
137/// User-created iceberg sinks have arbitrary names, so identify them by the
138/// connector property instead of the `__iceberg_sink_` name prefix.
139pub fn is_iceberg_sink(properties: &BTreeMap<String, String>) -> bool {
140    properties
141        .get(UPSTREAM_SOURCE_KEY)
142        .is_some_and(|connector| connector.eq_ignore_ascii_case(ICEBERG_SINK))
143}
risingwave_meta/manager/iceberg_compaction/mod.rs

risingwave_meta/manager/iceberg_compaction/
mod.rs
