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::Drained
33            | IcebergReportTaskStatus::Failed
34            | IcebergReportTaskStatus::Unspecified => {
35                let message = report
36                    .error_message
37                    .clone()
38                    .unwrap_or_else(|| "manual iceberg compaction failed".to_owned());
39                Err(anyhow!(
40                    "Manual iceberg compaction task {} for sink {} failed: {}",
41                    report.task_id,
42                    report.sink_id,
43                    message
44                )
45                .into())
46            }
47        };
48
49        let _ = waiter.send(result);
50    }
51
52    pub async fn trigger_manual_compaction(
53        &self,
54        sink_id: SinkId,
55    ) -> MetaResult<IcebergCompactionTaskId> {
56        // Fast-fail before registering a waiter when no compactor can pull the
57        // scheduler task.
58        if self.iceberg_compactor_manager.compactor_num() == 0 {
59            return Err(anyhow!("No iceberg compactor available").into());
60        }
61
62        let waiter = self.start_manual_compaction(sink_id).await?;
63        let cleanup_guard = scopeguard::guard(sink_id, |sink_id| {
64            self.cancel_manual_compaction_waiter(sink_id)
65        });
66
67        tracing::info!(
68            "Manual compaction requested for sink {}, waiting for completion...",
69            sink_id
70        );
71
72        match waiter.await {
73            Ok(result) => {
74                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
75                result
76            }
77            Err(_) => {
78                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
79                Err(anyhow!(
80                    "Manual iceberg compaction waiter dropped unexpectedly for sink {}",
81                    sink_id,
82                )
83                .into())
84            }
85        }
86    }
87}
risingwave_meta/manager/iceberg_compaction/manual.rs

risingwave_meta/manager/iceberg_compaction/
manual.rs
