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

risingwave_meta/manager/iceberg_compaction/
manual.rs
