risingwave_meta/manager/iceberg_compaction/schedule/
report.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::{Context, ensure};
16use risingwave_connector::sink::iceberg::IcebergCommitResult;
17use risingwave_pb::iceberg_compaction::PkIndexCompactionResult;
18
19use super::*;
20use crate::MetaError;
21use crate::barrier::{Command, IcebergPkIndexCompactionOverwrite};
22use crate::manager::iceberg_pk_index_sink::is_iceberg_pk_index_sink;
23
24enum CommandApplicationError {
25    /// Rejected before admission. The command cannot execute later.
26    NotStarted(MetaError),
27    /// Injection or collection may have happened. Recovery must settle it.
28    Unknown(MetaError),
29}
30
31#[derive(Debug)]
32pub(super) struct PkIndexCompactionApply {
33    pub(super) report: IcebergReportTask,
34    pub(super) overwrite: IcebergPkIndexCompactionOverwrite,
35}
36
37fn decode_pk_index_result(
38    result: PkIndexCompactionResult,
39) -> anyhow::Result<IcebergPkIndexCompactionOverwrite> {
40    let output_result = IcebergCommitResult::try_from(
41        result
42            .output_files
43            .as_ref()
44            .context("pk-index compaction result is missing output metadata")?,
45    )?;
46    ensure!(
47        !result.input_file_paths.is_empty(),
48        "pk-index compaction result is missing input file paths"
49    );
50    ensure!(
51        result.read_snapshot_id > 0,
52        "pk-index compaction result has an invalid read snapshot id"
53    );
54    Ok(IcebergPkIndexCompactionOverwrite {
55        output_result,
56        input_file_paths: result.input_file_paths,
57        read_snapshot_id: result.read_snapshot_id,
58    })
59}
60
61impl IcebergCompactionManager {
62    pub fn handle_report_task(self: &Arc<Self>, report: IcebergReportTask) {
63        if let Some(PkIndexCompactionApply {
64            mut report,
65            overwrite,
66        }) = self.prepare_report_task(report)
67        {
68            let manager = self.clone();
69            tokio::spawn(async move {
70                match manager.apply_pk_index_compaction(&report, overwrite).await {
71                    Ok(()) => manager.finish_report_task(report, true),
72                    Err(CommandApplicationError::NotStarted(error)) => {
73                        report.status = IcebergReportTaskStatus::Failed as i32;
74                        report.error_message = Some(format!(
75                            "failed to apply pk-index compaction: {}",
76                            error.as_report()
77                        ));
78                        manager.finish_report_task(report, true);
79                    }
80                    Err(CommandApplicationError::Unknown(error)) => {
81                        manager.mark_apply_unknown(report, error);
82                    }
83                }
84            });
85        }
86    }
87
88    pub(super) fn mark_apply_unknown(&self, report: IcebergReportTask, error: MetaError) {
89        let mut guard = self.inner.write();
90        if let Some(track) = guard.sink_schedules.get_mut(&report.sink_id)
91            && let CompactionTrackState::Applying { task_id, attempt } = &track.state
92            && *task_id == report.task_id
93        {
94            track.state = CompactionTrackState::NeedsRecovery {
95                task_id: *task_id,
96                attempt: attempt.clone(),
97            };
98            tracing::warn!(sink_id = %report.sink_id, task_id = %report.task_id,
99                error = %error.as_report(), "pk-index compaction requires recovery settlement");
100            if let Some(waiter) = guard.manual_compaction_waiters.remove(&report.sink_id) {
101                let _ = waiter.send(Err(anyhow!(error)
102                    .context(
103                        "pk-index compaction application outcome is unknown; recovery required",
104                    )
105                    .into()));
106            }
107        }
108    }
109
110    pub(super) fn prepare_report_task(
111        &self,
112        mut report: IcebergReportTask,
113    ) -> Option<PkIndexCompactionApply> {
114        let sink_id = report.sink_id;
115        let mut guard = self.inner.write();
116        let Some(track) = guard
117            .sink_schedules
118            .get_mut(&sink_id)
119            .filter(|track| track.source_is_ready() && track.is_in_flight_task(report.task_id))
120        else {
121            // In particular, duplicate reports cannot re-apply an Applying task
122            // or complete its manual waiter before the barrier has committed.
123            tracing::warn!(
124                %sink_id,
125                task_id = %report.task_id,
126                "ignoring iceberg compaction report without a matching in-flight task",
127            );
128            return None;
129        };
130        let status = IcebergReportTaskStatus::try_from(report.status)
131            .unwrap_or(IcebergReportTaskStatus::Unspecified);
132        if matches!(
133            status,
134            IcebergReportTaskStatus::Success | IcebergReportTaskStatus::Drained
135        ) && let Some(result) = report.pk_index_result.take()
136        {
137            let CompactionTrackState::InFlight { attempt, .. } = &track.state else {
138                unreachable!()
139            };
140            let overwrite = if !attempt.pk_index_coordinated {
141                Err(anyhow!(
142                    "unexpected pk-index result for an uncoordinated task"
143                ))
144            } else if status == IcebergReportTaskStatus::Drained
145                && attempt.max_file_sequence_number.is_none()
146            {
147                Err(anyhow!("unbounded compaction task cannot report Drained"))
148            } else {
149                decode_pk_index_result(result)
150            };
151            match overwrite {
152                Ok(overwrite) => {
153                    // A rewrite payload cannot prove the round drained, even
154                    // when it was the last group in the planning snapshot.
155                    report.status = IcebergReportTaskStatus::Success as i32;
156                    let result = PkIndexCompactionApply { report, overwrite };
157                    track.state = CompactionTrackState::Applying {
158                        task_id: result.report.task_id,
159                        attempt: attempt.clone(),
160                    };
161                    return Some(result);
162                }
163                Err(error) => {
164                    report.status = IcebergReportTaskStatus::Failed as i32;
165                    report.error_message = Some(format!(
166                        "invalid pk-index compaction result: {}",
167                        error.as_report()
168                    ));
169                }
170            }
171        }
172        // A no-plan task has no rewrite payload, including a manual Full task
173        // reporting Success. There is nothing to apply in that case.
174        drop(guard);
175        self.finish_report_task(report, false);
176        None
177    }
178
179    async fn apply_pk_index_compaction(
180        &self,
181        report: &IcebergReportTask,
182        overwrite: IcebergPkIndexCompactionOverwrite,
183    ) -> Result<(), CommandApplicationError> {
184        let sink = self
185            .prepare_apply_sink(report)
186            .await
187            .map_err(|error| CommandApplicationError::NotStarted(error.into()))?;
188        let completion = {
189            let guard = self.inner.read();
190            if !guard
191                .sink_schedules
192                .get(&report.sink_id)
193                .is_some_and(|track| {
194                    track.source_is_ready()
195                        && matches!(&track.state,
196                    CompactionTrackState::Applying { task_id, .. } if *task_id == report.task_id)
197                })
198            {
199                return Err(CommandApplicationError::NotStarted(
200                    anyhow!("pk-index compaction was isolated before command submission").into(),
201                ));
202            }
203            self.barrier_scheduler
204                .schedule_command(
205                    sink.database_id,
206                    Command::ApplyIcebergPkIndexCompaction {
207                        sink_id: report.sink_id,
208                        task_id: report.task_id,
209                        overwrite,
210                    },
211                )
212                .map_err(CommandApplicationError::NotStarted)?
213        };
214        // Only synchronous rejection proves the command was never admitted.
215        completion.await.map_err(CommandApplicationError::Unknown)
216    }
217
218    async fn prepare_apply_sink(&self, report: &IcebergReportTask) -> anyhow::Result<SinkCatalog> {
219        let sink_id = report.sink_id;
220        let sink = self
221            .metadata_manager
222            .catalog_controller
223            .get_sink_by_id(sink_id)
224            .await?
225            .context("pk-index compaction sink no longer exists")?;
226        let sink = SinkCatalog::from(sink);
227        ensure!(
228            is_iceberg_pk_index_sink(&sink.properties),
229            "compaction sink {} is not a pk-index sink",
230            sink_id
231        );
232        {
233            let mut guard = self.inner.write();
234            let track = guard.sink_schedules.get_mut(&sink_id);
235            match track.map(|track| &mut track.state) {
236                Some(CompactionTrackState::Applying { task_id, .. })
237                    if *task_id == report.task_id => {}
238                _ => anyhow::bail!(
239                    "pk-index compaction task {} was cleared before application",
240                    report.task_id
241                ),
242            }
243        }
244        Ok(sink)
245    }
246}
risingwave_meta/manager/iceberg_compaction/schedule/report.rs

risingwave_meta/manager/iceberg_compaction/schedule/
report.rs
