Struct IcebergCompactionManager

pub struct IcebergCompactionManager {
    pub env: MetaSrvEnv,
    inner: Arc<RwLock<IcebergCompactionManagerInner>>,
    metadata_manager: MetadataManager,
    pub iceberg_compactor_manager: IcebergCompactorManagerRef,
    compactor_streams_change_tx: UnboundedSender<(u32, Streaming<SubscribeIcebergCompactionEventRequest>)>,
    pub metrics: Arc<MetaMetrics>,
}

Fields

env: MetaSrvEnv

inner: Arc<RwLock<IcebergCompactionManagerInner>>

metadata_manager: MetadataManager

iceberg_compactor_manager: IcebergCompactorManagerRef

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

metrics: Arc<MetaMetrics>

Implementations

impl IcebergCompactionManager

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

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

pub fn update_iceberg_commit_info(&self, msg: IcebergSinkCompactionUpdate)

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

Get the top N iceberg commit sink ids Sorted by commit count and next compaction time

pub fn clear_iceberg_commits_by_sink_id(&self, sink_id: SinkId)

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

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

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

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

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

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

GC loop for expired snapshots management This is a separate loop that periodically checks all tracked Iceberg tables and performs garbage collection operations like expiring old snapshots

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

Perform GC operations on all tracked Iceberg tables

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

Check snapshot count for a specific table and trigger expiration if needed