Struct IcebergSinkCommitter 

pub struct IcebergSinkCommitter {
    catalog: Arc<dyn Catalog>,
    table: Table,
    pub last_commit_epoch: u64,
    pub(crate) sink_id: SinkId,
    pub(crate) config: IcebergConfig,
    pub(crate) param: SinkParam,
    commit_retry_num: u32,
    pub(crate) iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
}

Fields

catalog: Arc<dyn Catalog>

table: Table

last_commit_epoch: u64

sink_id: SinkId

config: IcebergConfig

param: SinkParam

commit_retry_num: u32

iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>

Implementations

impl IcebergSinkCommitter

async fn reload_table( catalog: &dyn Catalog, table_ident: &TableIdent, schema_id: i32, partition_spec_id: i32, ) -> Result<Table>

impl IcebergSinkCommitter

Methods Required to Achieve Exactly Once Semantics

fn pre_commit_inner( &mut self, _epoch: u64, metadata: Vec<SinkMetadata>, ) -> Result<Option<(Vec<IcebergCommitResult>, i64)>>

async fn commit_data_impl( &mut self, epoch: u64, write_results: Vec<IcebergCommitResult>, snapshot_id: i64, ) -> Result<()>

async fn is_snapshot_id_in_iceberg( &self, iceberg_config: &IcebergConfig, snapshot_id: i64, ) -> Result<bool>

During pre-commit metadata, we record the snapshot_id corresponding to each batch of files. Therefore, the logic for checking whether all files in this batch are present in Iceberg has been changed to verifying if their corresponding snapshot_id exists in Iceberg.

fn check_schema_change_applied( &self, schema_change: &PbSinkSchemaChange, ) -> Result<bool>

Check if the specified columns already exist in the iceberg table’s current schema. This is used to determine if schema change has already been applied.

async fn commit_schema_change_impl( &mut self, schema_change: PbSinkSchemaChange, ) -> Result<()>

Commit schema changes (e.g., add columns) to the iceberg table. This function uses Transaction API to atomically update the table schema with optimistic locking to prevent concurrent conflicts.

fn count_snapshots_since_rewrite(&self) -> usize

Check if the number of snapshots since the last rewrite/overwrite operation exceeds the limit Returns the number of snapshots since the last rewrite/overwrite

async fn wait_for_snapshot_limit(&mut self) -> Result<()>

Wait until snapshot count since last rewrite is below the limit

Trait Implementations

impl SinglePhaseCommitCoordinator for IcebergSinkCommitter

fn init<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Initialize the sink committer coordinator.

fn commit_data<'life0, 'async_trait>( &'life0 mut self, epoch: u64, metadata: Vec<SinkMetadata>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Commit data directly using single-phase strategy.

fn commit_schema_change<'life0, 'async_trait>( &'life0 mut self, epoch: u64, schema_change: PbSinkSchemaChange, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Idempotent implementation is required, because commit_schema_change in the same epoch could be called multiple times.

impl TwoPhaseCommitCoordinator for IcebergSinkCommitter

fn init<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Initialize the sink committer coordinator.

fn pre_commit<'life0, 'async_trait>( &'life0 mut self, epoch: u64, metadata: Vec<SinkMetadata>, _schema_change: Option<PbSinkSchemaChange>, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Return serialized commit metadata to be passed to commit.

fn commit_data<'life0, 'async_trait>( &'life0 mut self, epoch: u64, commit_metadata: Vec<u8>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Idempotent implementation is required, because commit_data in the same epoch could be called multiple times.

fn commit_schema_change<'life0, 'async_trait>( &'life0 mut self, epoch: u64, schema_change: PbSinkSchemaChange, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Idempotent implementation is required, because commit_schema_change in the same epoch could be called multiple times.

fn abort<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, _commit_metadata: Vec<u8>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Idempotent implementation is required, because abort in the same epoch could be called multiple times.