risingwave_meta/manager/iceberg_compaction/
manual.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
15use anyhow::anyhow;
16use risingwave_connector::sink::catalog::SinkId;
17use risingwave_pb::iceberg_compaction::subscribe_iceberg_compaction_event_request::ReportTask as IcebergReportTask;
18use risingwave_pb::iceberg_compaction::subscribe_iceberg_compaction_event_request::report_task::Status as IcebergReportTaskStatus;
19use risingwave_pb::id::IcebergCompactionTaskId;
20
21use super::*;
22
23impl IcebergCompactionManager {
24    pub(super) fn complete_manual_task_waiter(
25        waiter: ManualCompactionWaiter,
26        report: &IcebergReportTask,
27    ) {
28        let status = IcebergReportTaskStatus::try_from(report.status)
29            .unwrap_or(IcebergReportTaskStatus::Unspecified);
30        let result = match status {
31            IcebergReportTaskStatus::Success => Ok(report.task_id),
32            IcebergReportTaskStatus::BlockedCapacity
33            | IcebergReportTaskStatus::BlockedStatistics
34            | IcebergReportTaskStatus::Busy
35            | IcebergReportTaskStatus::Drained
36            | IcebergReportTaskStatus::Failed
37            | IcebergReportTaskStatus::Unspecified => {
38                let message = report
39                    .error_message
40                    .clone()
41                    .unwrap_or_else(|| "manual iceberg compaction failed".to_owned());
42                Err(anyhow!(
43                    "Manual iceberg compaction task {} for sink {} failed: {}",
44                    report.task_id,
45                    report.sink_id,
46                    message
47                )
48                .into())
49            }
50        };
51
52        let _ = waiter.send(result);
53    }
54
55    pub async fn trigger_manual_compaction(
56        &self,
57        sink_id: SinkId,
58    ) -> MetaResult<IcebergCompactionTaskId> {
59        // Fast-fail before registering a waiter when no compactor can pull the
60        // scheduler task.
61        if self.iceberg_compactor_manager.compactor_num() == 0 {
62            return Err(anyhow!("No iceberg compactor available").into());
63        }
64
65        let waiter = self.start_manual_compaction(sink_id).await?;
66        let cleanup_guard = scopeguard::guard(sink_id, |sink_id| {
67            self.cancel_manual_compaction_waiter(sink_id)
68        });
69
70        tracing::info!(
71            "Manual compaction requested for sink {}, waiting for completion...",
72            sink_id
73        );
74
75        match waiter.await {
76            Ok(result) => {
77                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
78                result
79            }
80            Err(_) => {
81                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
82                Err(anyhow!(
83                    "Manual iceberg compaction waiter dropped unexpectedly for sink {}",
84                    sink_id,
85                )
86                .into())
87            }
88        }
89    }
90}
risingwave_meta/manager/iceberg_compaction/manual.rs

risingwave_meta/manager/iceberg_compaction/
manual.rs
