Struct IcebergCompactionManager 

pub struct IcebergCompactionManager {
    pub env: MetaSrvEnv,
    inner: Arc<RwLock<IcebergCompactionManagerInner>>,
    metadata_manager: MetadataManager,
    barrier_scheduler: BarrierScheduler,
    pub iceberg_compactor_manager: IcebergCompactorManagerRef,
    compactor_streams_change_tx: UnboundedSender<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>,
    orphan_file_cleanup_locks: Mutex<HashMap<SinkId, Arc<Mutex<()>>>>,
    pub metrics: Arc<MetaMetrics>,
}

Fields

env: MetaSrvEnv

inner: Arc<RwLock<IcebergCompactionManagerInner>>

metadata_manager: MetadataManager

barrier_scheduler: BarrierScheduler

iceberg_compactor_manager: IcebergCompactorManagerRef

compactor_streams_change_tx: UnboundedSender<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>

orphan_file_cleanup_locks: Mutex<HashMap<SinkId, Arc<Mutex<()>>>>

metrics: Arc<MetaMetrics>

Implementations

impl IcebergCompactionManager

pub fn gc_loop( manager: Arc<Self>, interval_sec: u64, ) -> (JoinHandle<()>, Sender<()>)

pub fn orphan_file_cleanup_loop( manager: Arc<Self>, interval_sec: u64, ) -> (JoinHandle<()>, Sender<()>)

async fn perform_gc_operations(&self) -> MetaResult<()>

async fn perform_orphan_file_cleanup(&self) -> MetaResult<()>

async fn load_maintenance_config( &self, sink_id: SinkId, ) -> MetaResult<Option<IcebergConfig>>

pub async fn check_and_expire_snapshots( &self, sink_id: SinkId, ) -> MetaResult<()>

pub async fn check_and_remove_orphan_files( &self, sink_id: SinkId, ) -> MetaResult<()>

pub async fn remove_orphan_files(&self, sink_id: SinkId) -> MetaResult<u64>

Remove orphan files immediately, independently of the periodic cleanup switch.

async fn remove_orphan_files_with_config( &self, sink_id: SinkId, iceberg_config: IcebergConfig, ) -> MetaResult<u64>

pub async fn check_and_rewrite_manifests( &self, sink_id: SinkId, ) -> MetaResult<()>

impl IcebergCompactionManager

pub(super) fn complete_manual_task_waiter( waiter: Sender<MetaResult<IcebergCompactionTaskId>>, report: &IcebergReportTask, )

pub async fn trigger_manual_compaction( &self, sink_id: SinkId, ) -> MetaResult<IcebergCompactionTaskId>

pub(super) fn cancel_closed_manual_compaction_waiter(&self, sink_id: SinkId)

impl IcebergCompactionManager

pub fn handle_report_task(self: &Arc<Self>, report: IcebergReportTask)

pub(super) fn mark_apply_unknown( &self, report: IcebergReportTask, error: MetaError, )

pub(super) fn prepare_report_task( &self, report: IcebergReportTask, ) -> Option<PkIndexCompactionApply>

async fn apply_pk_index_compaction( &self, report: &IcebergReportTask, overwrite: IcebergPkIndexCompactionOverwrite, ) -> Result<(), CommandApplicationError>

async fn prepare_apply_sink( &self, report: &IcebergReportTask, ) -> Result<SinkCatalog>

impl IcebergCompactionManager

fn apply_track_finish_action( guard: &mut IcebergCompactionManagerInner, sink_id: SinkId, finish_action: CompactionTrackFinishAction, )

pub(super) fn refresh_schedule_config( &self, track: &mut CompactionTrack, iceberg_config: &IcebergConfig, now: Instant, )

pub fn schedule_on_startup_task( manager: Arc<Self>, ) -> (JoinHandle<()>, Sender<()>)

Schedules existing Iceberg sinks once after meta starts. See iceberg_compaction_schedule_on_startup.

async fn schedule_existing_sinks(&self)

async fn schedule_sink_on_startup(&self, sink_id: SinkId) -> MetaResult<()>

Records the head of the commit branch as one sink commit, so the sink is compacted after the compaction interval. The compaction commits nothing if the table is already clean.

pub async fn update_iceberg_commit_info(&self, msg: IcebergSinkCompactionUpdate)

pub fn on_sink_config_changed(&self, sink_id: SinkId, config: IcebergConfig)

Apply maintenance-only ALTER options to a registered PK-index sink, even when ingestion is idle. Readiness publication remains the only initialization path.

pub fn on_recovery_start(&self, sink_ids: Option<&[SinkId]>)

Discard PK-index scheduling before replacing coordinators. Persisted sink state, not the old attempt, determines which rewrites recovery commits.

pub(crate) fn on_sink_ready( &self, sink_id: SinkId, registration: Arc<IcebergPkIndexSinkRegistration>, observed_snapshot: Option<IcebergCommittedSnapshot>, ) -> bool

Publish only after CREATE post-collect or successful database recovery unblocks its queue. The coordinator registry serializes publication with retirement. Keeping its registration also prevents updates/dispatch from using a retired instance before schedule cleanup.

async fn prepare_sink_update( &self, sink_id: SinkId, kind: SinkUpdateKind, now: Instant, ) -> PreparedSinkUpdate

fn apply_sink_update( &self, guard: &mut IcebergCompactionManagerInner, prepared_update: PreparedSinkUpdate, ) -> bool

pub(super) fn create_compaction_track( &self, iceberg_config: &IcebergConfig, now: Instant, ) -> CompactionTrack

fn resolve_schedule_config( &self, iceberg_config: &IcebergConfig, now: Instant, ) -> CompactionScheduleConfig

pub(super) async fn start_manual_compaction( &self, sink_id: SinkId, ) -> MetaResult<Receiver<MetaResult<IcebergCompactionTaskId>>>

fn finish_timed_out_compaction_tasks( guard: &mut IcebergCompactionManagerInner, now: Instant, ) -> Vec<(SinkId, Sender<MetaResult<IcebergCompactionTaskId>>)>

pub(crate) fn get_top_n_iceberg_commit_sink_ids( &self, n: usize, ) -> Vec<IcebergCompactionHandle>

pub fn clear_iceberg_maintenance_by_sink_id(&self, sink_id: SinkId)

fn remove_sink_schedule( guard: &mut IcebergCompactionManagerInner, sink_id: SinkId, ) -> Option<ScheduledCompactionTask>

fn cancel_scheduled_task_if_any( &self, sink_id: SinkId, task: Option<ScheduledCompactionTask>, )

pub fn list_compaction_statuses(&self) -> Vec<IcebergCompactionScheduleStatus>

fn finish_report_task(&self, report: IcebergReportTask, applying: bool)

impl IcebergCompactionManager

pub fn compaction_stat_loop( manager: Arc<Self>, rx: UnboundedReceiver<IcebergSinkCompactionUpdate>, ) -> (JoinHandle<()>, Sender<()>)

pub fn add_compactor_stream( &self, context_id: WorkerId, req_stream: Streaming<SubscribeIcebergCompactionEventRequest>, )

pub fn iceberg_compaction_event_loop( iceberg_compaction_manager: Arc<Self>, compactor_streams_change_rx: UnboundedReceiver<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>, ) -> Vec<(JoinHandle<()>, Sender<()>)>

impl IcebergCompactionManager

fn report_timeout(&self) -> Duration

fn config_refresh_interval(&self) -> Duration

pub fn build( env: MetaSrvEnv, metadata_manager: MetadataManager, iceberg_compactor_manager: IcebergCompactorManagerRef, metrics: Arc<MetaMetrics>, barrier_scheduler: BarrierScheduler, ) -> (Arc<Self>, UnboundedReceiver<(WorkerId, Streaming<SubscribeIcebergCompactionEventRequest>)>)

async fn get_sink_param(&self, sink_id: SinkId) -> MetaResult<SinkParam>

async fn load_iceberg_config( &self, sink_id: SinkId, ) -> MetaResult<IcebergConfig>

fn orphan_file_cleanup_lock(&self, sink_id: SinkId) -> Arc<Mutex<()>>