Struct IcebergSinkCommitter

pub struct IcebergSinkCommitter {
    catalog: Arc<dyn Catalog>,
    table: Table,
    pub last_commit_epoch: u64,
    pub(crate) is_exactly_once: bool,
    pub(crate) sink_id: u32,
    pub(crate) config: IcebergConfig,
    pub(crate) param: SinkParam,
    pub(crate) db: DatabaseConnection,
    commit_notifier: Option<UnboundedSender<()>>,
    commit_retry_num: u32,
    _compact_task_guard: Option<Sender<()>>,
    pub(crate) committed_epoch_subscriber: Option<SinkCommittedEpochSubscriber>,
}

Fields

catalog: Arc<dyn Catalog>

table: Table

last_commit_epoch: u64

is_exactly_once: bool

sink_id: u32

config: IcebergConfig

param: SinkParam

db: DatabaseConnection

commit_notifier: Option<UnboundedSender<()>>

commit_retry_num: u32

_compact_task_guard: Option<Sender<()>>

committed_epoch_subscriber: Option<SinkCommittedEpochSubscriber>

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

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

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

async fn persist_pre_commit_metadata( &self, db: DatabaseConnection, start_epoch: u64, end_epoch: u64, pre_commit_metadata: Vec<u8>, snapshot_id: i64, ) -> Result<()>

async fn mark_row_is_committed_by_sink_id_and_end_epoch( &self, db: &DatabaseConnection, sink_id: u32, end_epoch: u64, ) -> Result<()>

async fn delete_row_by_sink_id_and_end_epoch( &self, db: &DatabaseConnection, sink_id: u32, end_epoch: u64, ) -> Result<()>

async fn iceberg_sink_has_pre_commit_metadata( &self, db: &DatabaseConnection, sink_id: u32, ) -> Result<bool>

async fn get_pre_commit_info_by_sink_id( &self, db: &DatabaseConnection, sink_id: u32, ) -> Result<Vec<(u64, Vec<u8>, i64, bool)>>

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.

Trait Implementations

impl SinkCommitCoordinator for IcebergSinkCommitter

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

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

Initialize the sink committer coordinator, return the log store rewind start offset.

fn commit<'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,

After collecting the metadata from each sink writer, a coordinator will call commit with the set of metadata. The metadata is serialized into bytes, because the metadata is expected to be passed between different gRPC node, so in this general trait, the metadata is serialized bytes.